diff --git a/python/packages/devui/agent_framework_devui/_server.py b/python/packages/devui/agent_framework_devui/_server.py index bc94dee9a21..0c32bda57e4 100644 --- a/python/packages/devui/agent_framework_devui/_server.py +++ b/python/packages/devui/agent_framework_devui/_server.py @@ -137,6 +137,28 @@ def _loopback_allowed_hosts(self) -> frozenset[str] | None: return None return self._LOOPBACK_HOSTS + @staticmethod + def _token_matches(presented: str, expected: str) -> bool: + """Compare two bearer tokens in constant time without raising on non-ASCII input. + + `secrets.compare_digest` accepts only ASCII `str` values and raises `TypeError` when + either side holds a non-ASCII character. Uncaught, that surfaces as HTTP 500 rather + than HTTP 401, and it makes a non-ASCII `DEVUI_AUTH_TOKEN` unusable even when the + client presents it correctly. Comparing bytes has no such restriction. + + Headers travel as bytes and Starlette hands them back decoded as latin-1, so + re-encoding with latin-1 recovers what the client actually put on the wire. The + expected token is text (an argument, or `DEVUI_AUTH_TOKEN` from the environment), so + UTF-8 is its wire form. ASCII tokens behave exactly as before, since both encodings + agree on ASCII. + """ + try: + presented_bytes = presented.encode("latin-1") + except UnicodeEncodeError: + # Not something a header could have carried; treat it as a wrong token. + return False + return secrets.compare_digest(presented_bytes, expected.encode("utf-8")) + @classmethod def _resolve_auth_token(cls, host: str, auth_enabled: bool, auth_token: str | None) -> str | None: """Resolve the active Bearer token. Returns None when auth is disabled.""" @@ -444,8 +466,13 @@ async def auth_middleware(request: Request, call_next: Callable[[Request], Await }, ) - token = auth_header.replace("Bearer ", "", 1).strip() - if not secrets.compare_digest(token, expected_token): + # Trim only HTTP optional-whitespace (space/tab). A bare `.strip()` + # would also drop non-ASCII whitespace such as NBSP (U+00A0), which + # is a valid byte in a UTF-8 token (e.g. the second byte of "à" is + # 0xA0), so a correct non-ASCII token would be truncated before the + # byte comparison and rejected with 401. + token = auth_header.replace("Bearer ", "", 1).strip(" \t") + if not self._token_matches(token, expected_token): return JSONResponse( status_code=401, content={ diff --git a/python/packages/devui/tests/devui/test_server.py b/python/packages/devui/tests/devui/test_server.py index 56055fd137a..24bcecb5e5e 100644 --- a/python/packages/devui/tests/devui/test_server.py +++ b/python/packages/devui/tests/devui/test_server.py @@ -841,6 +841,49 @@ def test_devserver_accepts_request_with_valid_bearer_token(monkeypatch): assert response.status_code == 200 +@pytest.mark.parametrize( + ("configured_token", "presented_token", "expected_status"), + [ + pytest.param("s3cret", "s3cret", 200, id="ascii-matches"), + pytest.param("s3cret", "s\u00e9cret", 401, id="non-ascii-presented-is-rejected"), + pytest.param("s\u00e9cret", "s\u00e9cret", 200, id="non-ascii-configured-matches"), + pytest.param("s\u00e9cret", "s3cret", 401, id="non-ascii-configured-wrong-token"), + # U+00E0 ("à") encodes to UTF-8 bytes 0xC3 0xA0, whose second byte is NBSP + # (U+00A0). A bare `.strip()` would drop that trailing 0xA0 and reject the + # correct token; trimming only HTTP whitespace must keep it intact. + pytest.param("secret\u00e0", "secret\u00e0", 200, id="trailing-a-grave-matches"), + pytest.param("secret\u00e0", "secret\u00e9", 401, id="trailing-a-grave-wrong-token"), + ], +) +def test_bearer_token_comparison_survives_non_ascii_tokens( + monkeypatch, + configured_token: str, + presented_token: str, + expected_status: int, +) -> None: + """A non-ASCII bearer token must be compared, not answered with HTTP 500. + + `secrets.compare_digest` raises `TypeError` for any `str` holding a non-ASCII + character, so a merely wrong token produced 500 instead of 401, and a non-ASCII + `DEVUI_AUTH_TOKEN` never matched even when the client presented it correctly. + """ + monkeypatch.delenv("DEVUI_AUTH_TOKEN", raising=False) + + server = DevServer(auth_token=configured_token) + app = server.get_app() + + with TestClient(app, base_url="http://127.0.0.1", raise_server_exceptions=False) as client: + # Sent as raw bytes: httpx refuses to encode a non-ASCII header value itself. + # A bytes key keeps the header mapping bytes/bytes so the raw token bytes are + # preserved on the wire and the mapping stays type-consistent for Pyright/Pyrefly. + response = client.get( + "/v1/entities", + headers={b"Authorization": b"Bearer " + presented_token.encode("utf-8")}, + ) + + assert response.status_code == expected_status + + def test_meta_endpoint_requires_auth(monkeypatch): """/meta exposes capability flags (deployment, instrumentation, version) — gate it behind auth.