fix(proxy): read WebSocket auth headers exactly as the HTTP gate does

The middleware built its own lowercased dict from the raw ASGI header pairs,
which keeps the LAST value for a repeated header. Starlette's Headers — what
the HTTP gate reads — returns the FIRST. So the two transports could disagree
about which Authorization counted, which is precisely the drift the shared
read_proxy_token was introduced to prevent.

Not a credential bypass on its own (an attacker still has to present the real
token), but it is the header-desync shape that goes wrong when a fronting proxy
and the origin disagree about which duplicate wins.

Use Headers(scope=scope) so parity holds by construction rather than by
re-implementing its semantics, and pin it with a test.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

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
         ):

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()