Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 23 additions & 1 deletion python/packages/devui/agent_framework_devui/_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -445,7 +467,7 @@ 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):
if not self._token_matches(token, expected_token):
return JSONResponse(
status_code=401,
content={
Expand Down
36 changes: 36 additions & 0 deletions python/packages/devui/tests/devui/test_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -841,6 +841,42 @@ 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"),
],
)
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.
response = client.get(
"/v1/entities",
headers={"Authorization": b"Bearer " + presented_token.encode("utf-8")},

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

lisry-1 This header mapping mixes a string key with a bytes value. HTTPX accepts string/string or bytes/bytes mappings, so the request works at runtime but the current test fails Pyright and Pyrefly and blocks the merge gatekeeper. Please use a bytes key—{b"Authorization": b"Bearer " + presented_token.encode("utf-8")}—to preserve the raw token bytes without violating the declared header type.

)

assert response.status_code == expected_status


def test_meta_endpoint_requires_auth(monkeypatch):
"""/meta exposes capability flags (deployment, instrumentation, version) — gate it behind auth.

Expand Down
Loading