fix(cli): mask query tokens in uvicorn access log
This commit is contained in:
@@ -24,6 +24,7 @@ import asyncio
|
||||
import contextlib
|
||||
import functools
|
||||
import logging
|
||||
import re
|
||||
import signal
|
||||
from contextlib import AsyncExitStack
|
||||
from typing import TYPE_CHECKING, Any
|
||||
@@ -286,8 +287,30 @@ def _plain_postgres_url(url: str) -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
_TOKEN_IN_QUERY = re.compile(r"(token=)[^&\s\"]+")
|
||||
|
||||
|
||||
class ScrubQueryTokens(logging.Filter):
|
||||
"""Mask ``?token=…`` in uvicorn access lines.
|
||||
|
||||
Webhook senders that cannot set headers put the secret in the URL, and
|
||||
``docker logs`` is not a vault.
|
||||
"""
|
||||
|
||||
def filter(self, record: logging.LogRecord) -> bool:
|
||||
if isinstance(record.args, tuple):
|
||||
record.args = tuple(
|
||||
_TOKEN_IN_QUERY.sub(r"\1<…>", a) if isinstance(a, str) else a
|
||||
for a in record.args
|
||||
)
|
||||
elif isinstance(record.msg, str):
|
||||
record.msg = _TOKEN_IN_QUERY.sub(r"\1<…>", record.msg)
|
||||
return True
|
||||
|
||||
|
||||
async def _serve_root(gateway: Gateway, *, extra: dict[str, ASGIApp]) -> None:
|
||||
app = build_root_app(gateway.frontends, extra=extra)
|
||||
logging.getLogger("uvicorn.access").addFilter(ScrubQueryTokens())
|
||||
config = uvicorn.Config(app, host=gateway.host, port=gateway.port, log_level="info")
|
||||
_log.info(
|
||||
"gateway on http://%s:%d - %s",
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""``require_token`` accepts ``?token=`` only when no auth header is present."""
|
||||
|
||||
import logging
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
@@ -57,3 +58,20 @@ async def test_bootstrap_entry_can_carry_a_scope() -> None:
|
||||
assert admin is not None and admin.scope == "*"
|
||||
assert hook is not None and hook.scope == "api" and hook.name == "komodo"
|
||||
assert not hook.allows("admin") and hook.allows("api")
|
||||
|
||||
|
||||
def test_access_log_filter_masks_query_tokens() -> None:
|
||||
from beaver_gateway.cli import ScrubQueryTokens
|
||||
|
||||
record = logging.LogRecord(
|
||||
"uvicorn.access",
|
||||
logging.INFO,
|
||||
__file__,
|
||||
1,
|
||||
'%s - "%s %s HTTP/%s" %d',
|
||||
("1.2.3.4:1", "POST", "/hooks/komodo?token=s3cret&x=1", "1.1", 202),
|
||||
None,
|
||||
)
|
||||
assert ScrubQueryTokens().filter(record)
|
||||
assert "s3cret" not in record.getMessage()
|
||||
assert "/hooks/komodo?token=<…>&x=1" in record.getMessage()
|
||||
|
||||
Reference in New Issue
Block a user