diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py index ba6de0636..e4c94d70c 100644 --- a/headroom/proxy/server.py +++ b/headroom/proxy/server.py @@ -2662,11 +2662,14 @@ class WebSocketAuthMiddleware: await self.app(scope, receive, send) return - headers = { - name.decode("latin-1").lower(): value.decode("latin-1") - for name, value in scope.get("headers", []) - } - provided = read_proxy_token(headers) + # Starlette's own Headers rather than a hand-built dict: on a repeated + # header it returns the FIRST occurrence, which is what the HTTP gate + # sees. Building a dict here instead took the LAST one, so the two + # transports disagreed about which `Authorization` counted — exactly the + # drift the shared reader below exists to prevent. + from starlette.datastructures import Headers + + provided = read_proxy_token(Headers(scope=scope)) if provided is not None and hmac.compare_digest( provided.encode("utf-8", "replace"), self.token_bytes ): diff --git a/tests/test_proxy_hardening.py b/tests/test_proxy_hardening.py index 1b07c0b76..85ddec4d8 100644 --- a/tests/test_proxy_hardening.py +++ b/tests/test_proxy_hardening.py @@ -194,6 +194,30 @@ class TestWebSocketAuthMiddleware: assert downstream.called is True assert not _closed_with_policy_violation(sent) + async def test_repeated_header_resolves_like_the_http_gate(self): + """A duplicated Authorization must mean the same thing on both transports. + + Starlette's Headers (what the HTTP gate reads) returns the FIRST + occurrence. A hand-built dict returns the last, which would let the two + paths disagree about which credential counted. + """ + downstream = _SpyApp() + mw = WebSocketAuthMiddleware(downstream, proxy_token="s3cr3t-token") + + sent = await _drive( + mw, + _ws_scope( + headers=[ + ("authorization", "Bearer s3cr3t-token"), + ("authorization", "Bearer wrong"), + ] + ), + ) + + # First header wins → authenticated, same as the HTTP gate. + assert downstream.called is True + assert not _closed_with_policy_violation(sent) + async def test_no_token_configured_is_a_passthrough(self): """Default deployment must gain no new challenge.""" downstream = _SpyApp()