diff --git a/saas_web.py b/saas_web.py index 63265e9..071b141 100644 --- a/saas_web.py +++ b/saas_web.py @@ -114,7 +114,7 @@ async def require_api_key(request: Request, call_next): 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 + hmac.compare_digest(provided_key.encode("utf-8"), key.encode("utf-8")) for key in configured_keys ): return JSONResponse( status_code=401, diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py index 3b57e03..c70f9b5 100644 --- a/tests/test_saas_web.py +++ b/tests/test_saas_web.py @@ -688,6 +688,54 @@ def _post_shrink(self, headers=None): headers=headers or {}, ) + def test_non_ascii_header_does_not_crash(self): + import asyncio + import json + from starlette.requests import Request + from saas_web import require_api_key + + async def dummy_call_next(request: Request): + from starlette.responses import JSONResponse + return JSONResponse({"status": "ok"}) + + with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}): + # Test non-ASCII header fails closed with 401 (not 500) + scope = { + "type": "http", + "method": "POST", + "path": "/shrink", + "headers": [(b"x-api-key", "secret-key😀".encode("utf-8"))], + } + req = Request(scope) + res = asyncio.run(require_api_key(req, dummy_call_next)) + self.assertEqual(res.status_code, 401) + body = json.loads(res.body.decode("utf-8")) + self.assertEqual(body, {"error": "Invalid or missing API key"}) + + # Test matching configured ASCII key success + scope_success = { + "type": "http", + "method": "POST", + "path": "/shrink", + "headers": [(b"x-api-key", b"secret-key")], + } + req_success = Request(scope_success) + res_success = asyncio.run(require_api_key(req_success, dummy_call_next)) + self.assertEqual(res_success.status_code, 200) + body_success = json.loads(res_success.body.decode("utf-8")) + self.assertEqual(body_success, {"status": "ok"}) + + # Test missing/incorrect ASCII 401 + scope_missing = { + "type": "http", + "method": "POST", + "path": "/shrink", + "headers": [(b"x-api-key", b"wrong-key")], + } + req_missing = Request(scope_missing) + res_missing = asyncio.run(require_api_key(req_missing, dummy_call_next)) + self.assertEqual(res_missing.status_code, 401) + def test_no_env_var_leaves_endpoints_open(self): with patch.dict(os.environ): os.environ.pop("CODEC_CARVER_API_KEYS", None)