mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-10-05 20:28:23 +00:00
refactor: align client-app tests with suite style, tighten comments
Flatten the test classes into plain test functions matching the rest of tests/unit, add LoggingMiddleware integration tests covering request.state.client_app and the unknown fallback, and trim docstrings and comments to house density.
This commit is contained in:
9qeklajc
@@ -185,11 +185,9 @@ class RequestIdFilter(logging.Filter):
|
||||
class ClientAppFilter(logging.Filter):
|
||||
"""Filter to add the requesting client app to all log records.
|
||||
|
||||
The client app (the app or agent that made the request) is resolved by the
|
||||
logging middleware from the OpenRouter-convention identity headers
|
||||
(``X-Title``/``HTTP-Referer``, falling back to ``User-Agent``) and stored
|
||||
in a context variable, so every log line emitted while handling a request
|
||||
carries it — including error messages raised deep in the wallet/mint code.
|
||||
The middleware resolves it from the OpenRouter-convention identity headers
|
||||
(X-Title, then Referer, then User-Agent) into a context variable, so even
|
||||
errors raised deep in the wallet/mint code carry it.
|
||||
"""
|
||||
|
||||
def filter(self, record: logging.LogRecord) -> bool:
|
||||
|
||||
+10
-14
@@ -14,30 +14,26 @@ logger = get_logger(__name__)
|
||||
# Context variable to store request ID across async context
|
||||
request_id_context: ContextVar[str | None] = ContextVar("request_id")
|
||||
|
||||
# Context variable holding the client app that made the current request, so
|
||||
# every log line emitted while handling it (including deep wallet/mint errors)
|
||||
# can say who triggered it. "unknown" when the client sent no identity headers.
|
||||
# Context variable holding the client app behind the current request, so log
|
||||
# lines emitted while handling it (wallet/mint errors included) can say who
|
||||
# triggered it.
|
||||
client_app_context: ContextVar[str | None] = ContextVar("client_app")
|
||||
|
||||
UNKNOWN_CLIENT_APP = "unknown"
|
||||
|
||||
# Client identity headers, in priority order. Follows the OpenRouter
|
||||
# convention: apps identify themselves with ``X-Title`` (human-readable app
|
||||
# name) and/or ``HTTP-Referer`` (app URL); ``User-Agent`` is the fallback for
|
||||
# SDKs and scripts that set neither.
|
||||
# OpenRouter-convention identity headers, in priority order: X-Title carries
|
||||
# the app name, HTTP-Referer its URL; User-Agent covers SDKs and scripts that
|
||||
# set neither.
|
||||
_CLIENT_APP_HEADERS: tuple[str, ...] = ("x-title", "referer", "user-agent")
|
||||
|
||||
# Header values are attacker-controlled free text; cap the length so a single
|
||||
# request can't bloat every log line, and strip control characters so a crafted
|
||||
# header can't inject fake log records.
|
||||
# Headers are attacker-controlled free text: cap the length so one request
|
||||
# can't bloat every log line, strip control characters so a crafted value
|
||||
# can't forge log records.
|
||||
_CLIENT_APP_MAX_LENGTH = 120
|
||||
|
||||
|
||||
def client_app_from_headers(headers: Headers) -> str:
|
||||
"""Resolve the client app identity from request headers.
|
||||
|
||||
Priority: ``X-Title`` > ``HTTP-Referer`` > ``User-Agent`` > "unknown".
|
||||
"""
|
||||
"""Resolve the requesting app from identity headers, or "unknown"."""
|
||||
for header in _CLIENT_APP_HEADERS:
|
||||
raw = headers.get(header)
|
||||
if raw is None:
|
||||
|
||||
@@ -1,21 +1,20 @@
|
||||
"""Unit tests for client-app identification in request logging."""
|
||||
"""Tests for client-app identification in request logging."""
|
||||
|
||||
import logging
|
||||
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.testclient import TestClient
|
||||
from starlette.datastructures import Headers
|
||||
|
||||
from routstr.core.logging import ClientAppFilter
|
||||
from routstr.core.middleware import (
|
||||
UNKNOWN_CLIENT_APP,
|
||||
LoggingMiddleware,
|
||||
client_app_context,
|
||||
client_app_from_headers,
|
||||
)
|
||||
|
||||
|
||||
def _headers(**kwargs: str) -> Headers:
|
||||
return Headers({k.replace("_", "-"): v for k, v in kwargs.items()})
|
||||
|
||||
|
||||
def _record() -> logging.LogRecord:
|
||||
return logging.LogRecord(
|
||||
name="routstr.test",
|
||||
@@ -28,68 +27,123 @@ def _record() -> logging.LogRecord:
|
||||
)
|
||||
|
||||
|
||||
class TestClientAppFromHeaders:
|
||||
def test_x_title_wins_over_all(self) -> None:
|
||||
headers = _headers(
|
||||
x_title="Goose",
|
||||
referer="https://myapp.example.com",
|
||||
user_agent="python-httpx/0.27",
|
||||
)
|
||||
assert client_app_from_headers(headers) == "Goose"
|
||||
|
||||
def test_referer_used_when_no_x_title(self) -> None:
|
||||
headers = _headers(
|
||||
referer="https://myapp.example.com", user_agent="python-httpx/0.27"
|
||||
)
|
||||
assert client_app_from_headers(headers) == "https://myapp.example.com"
|
||||
|
||||
def test_user_agent_is_last_fallback(self) -> None:
|
||||
headers = _headers(user_agent="python-httpx/0.27")
|
||||
assert client_app_from_headers(headers) == "python-httpx/0.27"
|
||||
|
||||
def test_unknown_when_no_identity_headers(self) -> None:
|
||||
assert client_app_from_headers(Headers({})) == UNKNOWN_CLIENT_APP
|
||||
assert UNKNOWN_CLIENT_APP == "unknown"
|
||||
|
||||
def test_blank_header_falls_through_to_next(self) -> None:
|
||||
headers = _headers(x_title=" ", user_agent="curl/8.4.0")
|
||||
assert client_app_from_headers(headers) == "curl/8.4.0"
|
||||
|
||||
def test_all_blank_resolves_to_unknown(self) -> None:
|
||||
headers = _headers(x_title=" ", user_agent="\t")
|
||||
assert client_app_from_headers(headers) == UNKNOWN_CLIENT_APP
|
||||
|
||||
def test_value_is_truncated(self) -> None:
|
||||
headers = _headers(x_title="a" * 500)
|
||||
assert client_app_from_headers(headers) == "a" * 120
|
||||
|
||||
def test_control_characters_are_stripped(self) -> None:
|
||||
# A crafted header must not be able to inject fake log records.
|
||||
headers = _headers(user_agent="evil-app\x1b[0m fake INFO line")
|
||||
assert client_app_from_headers(headers) == "evil-app[0m fake INFO line"
|
||||
# ---------------------------------------------------------------------------
|
||||
# client_app_from_headers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestClientAppFilter:
|
||||
def test_uses_context_variable(self) -> None:
|
||||
token = client_app_context.set("Goose")
|
||||
try:
|
||||
record = _record()
|
||||
assert ClientAppFilter().filter(record) is True
|
||||
assert record.client_app == "Goose" # type: ignore[attr-defined]
|
||||
finally:
|
||||
client_app_context.reset(token)
|
||||
def test_x_title_takes_priority() -> None:
|
||||
"""X-Title wins over Referer and User-Agent."""
|
||||
headers = Headers(
|
||||
{
|
||||
"x-title": "Goose",
|
||||
"referer": "https://myapp.example.com",
|
||||
"user-agent": "python-httpx/0.27",
|
||||
}
|
||||
)
|
||||
assert client_app_from_headers(headers) == "Goose"
|
||||
|
||||
def test_defaults_to_unknown_outside_request_context(self) -> None:
|
||||
|
||||
def test_referer_used_when_no_x_title() -> None:
|
||||
headers = Headers(
|
||||
{"referer": "https://myapp.example.com", "user-agent": "python-httpx/0.27"}
|
||||
)
|
||||
assert client_app_from_headers(headers) == "https://myapp.example.com"
|
||||
|
||||
|
||||
def test_user_agent_is_last_fallback() -> None:
|
||||
assert (
|
||||
client_app_from_headers(Headers({"user-agent": "curl/8.4.0"})) == "curl/8.4.0"
|
||||
)
|
||||
|
||||
|
||||
def test_unknown_when_no_identity_headers() -> None:
|
||||
assert client_app_from_headers(Headers({})) == UNKNOWN_CLIENT_APP
|
||||
|
||||
|
||||
def test_blank_header_falls_through_to_next() -> None:
|
||||
"""A whitespace-only X-Title must not shadow a usable User-Agent."""
|
||||
headers = Headers({"x-title": " ", "user-agent": "curl/8.4.0"})
|
||||
assert client_app_from_headers(headers) == "curl/8.4.0"
|
||||
|
||||
|
||||
def test_all_blank_resolves_to_unknown() -> None:
|
||||
headers = Headers({"x-title": " ", "user-agent": "\t"})
|
||||
assert client_app_from_headers(headers) == UNKNOWN_CLIENT_APP
|
||||
|
||||
|
||||
def test_value_is_truncated_to_120_chars() -> None:
|
||||
assert client_app_from_headers(Headers({"x-title": "a" * 500})) == "a" * 120
|
||||
|
||||
|
||||
def test_control_characters_are_stripped() -> None:
|
||||
"""A crafted header must not be able to forge log records."""
|
||||
headers = Headers({"user-agent": "evil-app\x1b[0m fake INFO line"})
|
||||
assert client_app_from_headers(headers) == "evil-app[0m fake INFO line"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ClientAppFilter
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_filter_reads_context_variable() -> None:
|
||||
token = client_app_context.set("Goose")
|
||||
try:
|
||||
record = _record()
|
||||
assert ClientAppFilter().filter(record) is True
|
||||
assert record.client_app == UNKNOWN_CLIENT_APP # type: ignore[attr-defined]
|
||||
assert record.client_app == "Goose" # type: ignore[attr-defined]
|
||||
finally:
|
||||
client_app_context.reset(token)
|
||||
|
||||
def test_explicit_extra_is_not_overwritten(self) -> None:
|
||||
token = client_app_context.set("context-app")
|
||||
try:
|
||||
record = _record()
|
||||
record.client_app = "explicit-app" # type: ignore[attr-defined]
|
||||
assert ClientAppFilter().filter(record) is True
|
||||
assert record.client_app == "explicit-app" # type: ignore[attr-defined]
|
||||
finally:
|
||||
client_app_context.reset(token)
|
||||
|
||||
def test_filter_defaults_to_unknown_outside_request_context() -> None:
|
||||
record = _record()
|
||||
assert ClientAppFilter().filter(record) is True
|
||||
assert record.client_app == UNKNOWN_CLIENT_APP # type: ignore[attr-defined]
|
||||
|
||||
|
||||
def test_filter_keeps_explicit_extra() -> None:
|
||||
"""extra={"client_app": ...} on a log call wins over the context value."""
|
||||
token = client_app_context.set("context-app")
|
||||
try:
|
||||
record = _record()
|
||||
record.client_app = "explicit-app" # type: ignore[attr-defined]
|
||||
assert ClientAppFilter().filter(record) is True
|
||||
assert record.client_app == "explicit-app" # type: ignore[attr-defined]
|
||||
finally:
|
||||
client_app_context.reset(token)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# LoggingMiddleware integration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_middleware_exposes_client_app_on_request_state() -> None:
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/whoami")
|
||||
async def whoami(request: Request) -> dict:
|
||||
return {"client_app": request.state.client_app}
|
||||
|
||||
app.add_middleware(LoggingMiddleware)
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/whoami", headers={"X-Title": "Goose"})
|
||||
assert response.json() == {"client_app": "Goose"}
|
||||
|
||||
|
||||
def test_middleware_reports_unknown_without_identity_headers() -> None:
|
||||
app = FastAPI()
|
||||
|
||||
@app.get("/whoami")
|
||||
async def whoami(request: Request) -> dict:
|
||||
return {"client_app": request.state.client_app}
|
||||
|
||||
app.add_middleware(LoggingMiddleware)
|
||||
# TestClient sets its own User-Agent; blank it out to simulate a bare client.
|
||||
client = TestClient(app, headers={"user-agent": ""})
|
||||
|
||||
response = client.get("/whoami")
|
||||
assert response.json() == {"client_app": UNKNOWN_CLIENT_APP}
|
||||
|
||||
Reference in New Issue
Block a user