fix(auth): compare API keys at the raw ASGI boundary

CHANGELOG.md

@@ -10,5 +10,6 @@
 - 순수 영숫자 토큰은 정규식 호출을 건너뛰되 다국어·문장부호 토큰화 결과는 기존 의미와 동일하게 유지합니다. 근거, 한계, APA 7 참고문헌은 [`docs/doctoring/token-fast-path-equivalence.md`](docs/doctoring/token-fast-path-equivalence.md)에 기록했습니다.
 
 ### Fixed
+- API-key 인증은 raw ASGI `X-API-Key` bytes를 configured UTF-8 key와 비교하고 중복 헤더를 거절하여 non-ASCII 입력의 예외와 Unicode credential 손상을 함께 방지합니다.
 - 단일·일괄 대상 크기 입력을 비웠을 때 이전 custom validity와 `aria-invalid` 상태를 즉시 초기화해 현재 필수 입력 상태를 정확히 전달합니다.
-- 업로드 파일명의 경로 구분자를 정규화하여 POSIX에서도 Windows 형식의 클라이언트 경로가 일관된 basename으로 기록되도록 수정했습니다.
+- 업로드 파일명의 경로 구분자를 정규화하여 POSIX에서도 Windows 형식의 클라이언트 경로가 일관된 basename으로 기록되도록 수정했습니다.

saas_web.py

@@ -83,6 +83,7 @@ async def limited_receive():
     except RequestTooLarge:
         return JSONResponse(status_code=413, content={"error": "Payload Too Large"})
 
+
 def get_configured_api_keys():
     """Return the API keys configured via the CODEC_CARVER_API_KEYS env var.
 
@@ -98,23 +99,38 @@ def get_configured_api_keys():
     return [key.strip() for key in raw.split(",") if key.strip()]
 
 
+def _raw_api_key_header(request: Request) -> bytes | None:
+    """Return the sole raw X-API-Key value, or None for missing/duplicate input."""
+
+    values = [
+        value
+        for name, value in request.scope.get("headers", ())
+        if name.lower() == b"x-api-key"
+    ]
+    if len(values) != 1:
+        return None
+    return values[0]
+
+
 @app.middleware("http")
 async def require_api_key(request: Request, call_next):
     """Enforce opt-in API-key authentication on all endpoints except GET /.
 
     When one or more keys are configured via CODEC_CARVER_API_KEYS, every
-    request other than GET / (the upload UI page) must carry an X-API-Key
-    header matching a configured key; comparison uses hmac.compare_digest to
-    stay constant-time. Requests failing the check receive a 401 JSON error
-    without echoing any key material. When no keys are configured, all
-    requests pass through unchanged.
+    request other than GET / (the upload UI page) must carry exactly one raw
+    X-API-Key header matching a configured UTF-8 key. Comparison uses
+    hmac.compare_digest on bytes so Unicode credentials survive the ASGI header
+    boundary without a Latin-1 round-trip. Missing or duplicated credentials
+    fail closed with a 401 and no key material is echoed. When no keys are
+    configured, all requests pass through unchanged.
     """
 
     configured_keys = get_configured_api_keys()
     if configured_keys and not (request.method == "GET" and request.url.path == "/"):
-        provided_key = request.headers.get("x-api-key", "")
-        if not any(
-            hmac.compare_digest(provided_key, key) for key in configured_keys
+        provided_bytes = _raw_api_key_header(request)
+        if provided_bytes is None or not any(
+            hmac.compare_digest(provided_bytes, key.encode("utf-8"))
+            for key in configured_keys
         ):
             return JSONResponse(
                 status_code=401,

tests/test_api_key_header_multiplicity.py

@@ -0,0 +1,58 @@
+import asyncio
+import os
+import unittest
+from unittest.mock import patch
+
+try:
+    import saas_web
+    from starlette.requests import Request
+    from starlette.responses import Response
+
+    _HAS_WEB_STACK = True
+except ImportError:  # pragma: no cover - optional integration dependency boundary
+    _HAS_WEB_STACK = False
+
+
+@unittest.skipUnless(_HAS_WEB_STACK, "web integration dependencies are not installed")
+class TestApiKeyHeaderMultiplicity(unittest.TestCase):
+    """Lock the credential boundary to exactly one raw X-API-Key header."""
+
+    @staticmethod
+    def _request(raw_headers: list[tuple[bytes, bytes]]) -> "Request":
+        return Request(
+            {
+                "type": "http",
+                "http_version": "1.1",
+                "method": "POST",
+                "scheme": "https",
+                "path": "/shrink",
+                "raw_path": b"/shrink",
+                "query_string": b"",
+                "headers": raw_headers,
+                "client": ("127.0.0.1", 12345),
+                "server": ("codec-carver.test", 443),
+            }
+        )
+
+    def _assert_duplicate_is_rejected(self, raw_values: list[bytes]) -> None:
+        async def call_next(_request: "Request") -> "Response":
+            self.fail("duplicated credentials must not reach the protected handler")
+
+        request = self._request([(b"x-api-key", value) for value in raw_values])
+        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "안녕"}):
+            response = asyncio.run(saas_web.require_api_key(request, call_next))
+
+        self.assertEqual(response.status_code, 401)
+
+    def test_duplicate_api_key_headers_fail_closed_when_values_differ(self) -> None:
+        self._assert_duplicate_is_rejected(
+            ["안녕".encode("utf-8"), "다름".encode("utf-8")]
+        )
+
+    def test_duplicate_api_key_headers_fail_closed_when_values_match(self) -> None:
+        value = "안녕".encode("utf-8")
+        self._assert_duplicate_is_rejected([value, value])
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

tests/test_api_key_unicode_contract.py

@@ -0,0 +1,76 @@
+import asyncio
+import os
+import unittest
+from unittest.mock import patch
+
+try:
+    import saas_web
+    from starlette.requests import Request
+    from starlette.responses import Response
+
+    _HAS_WEB_STACK = True
+except ImportError:  # pragma: no cover - optional integration dependency boundary
+    _HAS_WEB_STACK = False
+
+
+@unittest.skipUnless(_HAS_WEB_STACK, "web integration dependencies are not installed")
+class TestUnicodeApiKeyContract(unittest.TestCase):
+    """Exercise API-key authentication from the raw ASGI header boundary."""
+
+    @staticmethod
+    def _request(raw_api_key: bytes) -> "Request":
+        return Request(
+            {
+                "type": "http",
+                "http_version": "1.1",
+                "method": "POST",
+                "scheme": "https",
+                "path": "/shrink",
+                "raw_path": b"/shrink",
+                "query_string": b"",
+                "headers": [(b"x-api-key", raw_api_key)],
+                "client": ("127.0.0.1", 12345),
+                "server": ("codec-carver.test", 443),
+            }
+        )
+
+    def test_configured_unicode_key_matches_its_raw_utf8_header(self) -> None:
+        reached_handler = False
+
+        async def call_next(_request: "Request") -> "Response":
+            nonlocal reached_handler
+            reached_handler = True
+            return Response(status_code=204)
+
+        configured_key = "안녕"
+        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": configured_key}):
+            response = asyncio.run(
+                saas_web.require_api_key(
+                    self._request(configured_key.encode("utf-8")),
+                    call_next,
+                )
+            )
+
+        self.assertTrue(
+            reached_handler,
+            "a valid configured Unicode key must not be rejected after ASGI header decoding",
+        )
+        self.assertEqual(response.status_code, 204)
+
+    def test_different_raw_utf8_key_is_rejected(self) -> None:
+        async def call_next(_request: "Request") -> "Response":
+            self.fail("invalid credentials must not reach the protected handler")
+
+        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "안녕"}):
+            response = asyncio.run(
+                saas_web.require_api_key(
+                    self._request("다름".encode("utf-8")),
+                    call_next,
+                )
+            )
+
+        self.assertEqual(response.status_code, 401)
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

tests/test_saas_web.py

@@ -715,6 +715,26 @@ def test_wrong_key_rejected(self):
         self.assertEqual(response.json(), {"error": "Invalid or missing API key"})
         self.assertNotIn("secret-key", response.text)
 
+    def test_non_ascii_key_does_not_crash(self):
+        import asyncio
+        from unittest.mock import patch
+        import os
+        from starlette.requests import Request
+        from saas_web import require_api_key
+
+        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
+            scope = {
+                "type": "http",
+                "method": "POST",
+                "path": "/shrink",
+                "headers": [(b"x-api-key", b"wrong-\xff-key")],
+            }
+            req = Request(scope)
+            async def call_next(request): return None
+
+            res = asyncio.run(require_api_key(req, call_next))
+            self.assertEqual(res.status_code, 401)
+
     def test_correct_key_reaches_handler(self):
         with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
             response = self._post_shrink(headers={"X-API-Key": "secret-key"})