fix(cli): mask query tokens in uvicorn access log

This commit is contained in:
hh
2026-08-30 18:30:26 +02:00
parent c438d4cb92
commit 878f7d6473
2 changed files with 41 additions and 0 deletions
+23
View File
@@ -24,6 +24,7 @@ import asyncio
import contextlib import contextlib
import functools import functools
import logging import logging
import re
import signal import signal
from contextlib import AsyncExitStack from contextlib import AsyncExitStack
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, Any
@@ -286,8 +287,30 @@ def _plain_postgres_url(url: str) -> str | None:
return 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: async def _serve_root(gateway: Gateway, *, extra: dict[str, ASGIApp]) -> None:
app = build_root_app(gateway.frontends, extra=extra) 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") config = uvicorn.Config(app, host=gateway.host, port=gateway.port, log_level="info")
_log.info( _log.info(
"gateway on http://%s:%d - %s", "gateway on http://%s:%d - %s",
+18
View File
@@ -1,5 +1,6 @@
"""``require_token`` accepts ``?token=`` only when no auth header is present.""" """``require_token`` accepts ``?token=`` only when no auth header is present."""
import logging
from types import SimpleNamespace from types import SimpleNamespace
import pytest 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 admin is not None and admin.scope == "*"
assert hook is not None and hook.scope == "api" and hook.name == "komodo" assert hook is not None and hook.scope == "api" and hook.name == "komodo"
assert not hook.allows("admin") and hook.allows("api") 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()