diff --git a/CHANGELOG.md b/CHANGELOG.md index 9313538b..31802387 100644 --- a/CHANGELOG.md +++ b/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으로 기록되도록 수정했습니다. \ No newline at end of file diff --git a/saas_web.py b/saas_web.py index 63265e94..213e9fe9 100644 --- a/saas_web.py +++ b/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, diff --git a/tests/test_api_key_header_multiplicity.py b/tests/test_api_key_header_multiplicity.py new file mode 100644 index 00000000..f43d0352 --- /dev/null +++ b/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() diff --git a/tests/test_api_key_unicode_contract.py b/tests/test_api_key_unicode_contract.py new file mode 100644 index 00000000..58e58d79 --- /dev/null +++ b/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() diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py index 3b57e033..ad0fb131 100644 --- a/tests/test_saas_web.py +++ b/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"})