fix(api): harden auth header parsing for proxy edge cases

apps/api/test_utils.py

@@ -0,0 +1,35 @@
+from types import SimpleNamespace
+
+from django.test import SimpleTestCase
+
+from apps.api.utils import _get_api_key_from_headers
+
+
+class APIHeaderParsingTestCase(SimpleTestCase):
+    def _request(self, headers):
+        return SimpleNamespace(headers=headers)
+
+    def test_get_api_key_from_headers_prefers_x_api_key(self):
+        request = self._request({"X-API-Key": "abc123"})
+
+        self.assertEqual(_get_api_key_from_headers(request), "abc123")
+
+    def test_get_api_key_from_headers_accepts_bearer_scheme(self):
+        request = self._request({"Authorization": "Bearer abc123"})
+
+        self.assertEqual(_get_api_key_from_headers(request), "abc123")
+
+    def test_get_api_key_from_headers_handles_non_string_authorization_safely(self):
+        request = self._request({"Authorization": 12345})
+
+        self.assertIsNone(_get_api_key_from_headers(request))
+
+    def test_get_api_key_from_headers_handles_non_string_x_api_key_safely(self):
+        request = self._request({"X-API-Key": object(), "Authorization": "Token fallback-key"})
+
+        self.assertEqual(_get_api_key_from_headers(request), "fallback-key")
+
+    def test_get_api_key_from_headers_handles_missing_headers_attribute(self):
+        request = SimpleNamespace()
+
+        self.assertIsNone(_get_api_key_from_headers(request))

apps/api/utils.py

@@ -5,24 +5,56 @@
 logger = get_agent_commons_logger(__name__)
 
 
+def _coerce_header_value(value: object) -> str | None:
+    if value is None:
+        return None
+
+    if isinstance(value, bytes):
+        try:
+            value = value.decode("utf-8")
+        except UnicodeDecodeError:
+            return None
+
+    if not isinstance(value, str):
+        return None
+
+    value = value.strip()
+    if not value:
+        return None
+
+    return value
+
+
+def _header_get(headers, name: str) -> object:
+    value = headers.get(name)
+    if value is None:
+        value = headers.get(name.lower())
+    return value
+
+
 def _get_api_key_from_headers(request: HttpRequest) -> str | None:
-    key = request.headers.get("X-API-Key")
+    headers = getattr(request, "headers", None)
+    if not headers:
+        return None
+
+    key = _coerce_header_value(_header_get(headers, "X-API-Key"))
     if key:
         return key
 
-    auth_header = request.headers.get("Authorization")
+    auth_header = _coerce_header_value(_header_get(headers, "Authorization"))
     if not auth_header:
         return None
 
     parts = auth_header.split(None, 1)
     if len(parts) != 2:
         return None
 
-    scheme, value = parts[0].lower(), parts[1].strip()
+    scheme = parts[0].lower()
+    value = _coerce_header_value(parts[1])
     if not value:
         return None
 
     if scheme in {"api-key", "apikey", "bearer", "token"}:
         return value
 
-    return None
+    return None