🛡️ Sentinel: [HIGH] API 인증 시 hmac.compare_digest TypeError로 인한 DoS 취약점 수정

saas_web.py

@@ -6,6 +6,7 @@
 import os
 import shutil
 import tempfile
+import threading
 import uuid
 import zipfile
 from datetime import datetime, timezone
@@ -83,38 +84,77 @@ 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.
+class APIKeyRegistry:
+    """Keep runtime API keys in process memory behind a thread-safe boundary.
 
-    The variable holds a comma-separated list of keys. Whitespace around each
-    key is stripped and empty entries are ignored. Keys are read from the
-    environment at request time (not import time) so tests can patch the
-    environment easily and key rotation needs no server restart. Returns an
-    empty list when the variable is unset or contains no usable keys, which
-    leaves the service open (today's default behaviour).
+    The environment variable is bootstrap transport only. Creating a registry
+    consumes and removes that variable so request handling and later environment
+    changes cannot silently rotate credentials.
     """
 
-    raw = os.environ.get("CODEC_CARVER_API_KEYS", "")
-    return [key.strip() for key in raw.split(",") if key.strip()]
+    def __init__(self) -> None:
+        """Create a registry and consume any one-time environment bootstrap."""
+
+        self._keys: tuple[str, ...] = ()
+        self._lock = threading.Lock()
+        self._bootstrap_from_environment()
+
+    def _bootstrap_from_environment(self) -> None:
+        """Load comma-separated keys once, then remove the transport variable."""
+
+        raw_keys = os.environ.pop("CODEC_CARVER_API_KEYS", None)
+        if raw_keys is not None:
+            self.set_keys(raw_keys.split(","))
+
+    def set_keys(self, keys: list[str]) -> None:
+        """Atomically replace keys after trimming whitespace and empty entries.
+
+        Args:
+            keys: Candidate plaintext keys supplied by trusted bootstrap or
+                credential-management code.
+        """
+
+        normalized_keys = tuple(
+            key.strip() for key in keys if key.strip()
+        )
+        with self._lock:
+            self._keys = normalized_keys
+
+    def get_keys(self) -> list[str]:
+        """Return a snapshot of registered keys without exposing mutable state."""
+
+        with self._lock:
+            return list(self._keys)
+
+
+API_KEY_REGISTRY = APIKeyRegistry()
+
+
+def get_configured_api_keys() -> list[str]:
+    """Return the current in-memory API-key snapshot used by middleware."""
+
+    return API_KEY_REGISTRY.get_keys()
 
 
 @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.
+    When one or more keys are present in the in-memory credential registry,
+    every request other than GET / (the upload UI page) must carry an X-API-Key
+    header matching a configured key. Byte-oriented hmac.compare_digest keeps
+    comparison constant-time and prevents non-ASCII header values from raising
+    TypeError. Failed requests receive a 401 response without echoing key
+    material; an empty registry preserves the open default.
     """
 
     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", "")
+        provided_key_bytes = provided_key.encode("utf-8")
         if not any(
-            hmac.compare_digest(provided_key, key) for key in configured_keys
+            hmac.compare_digest(provided_key_bytes, key.encode("utf-8"))
+            for key in configured_keys
         ):
             return JSONResponse(
                 status_code=401,

tests/test_saas_web.py

@@ -676,7 +676,18 @@ def test_get_ui_includes_batch_upload_form(self):
     _HAS_FASTAPI, "fastapi not installed (optional integration dependency)"
 )
 class TestApiKeyAuth(unittest.TestCase):
-    """Tests for the opt-in CODEC_CARVER_API_KEYS authentication middleware."""
+    """Verify registry-backed API-key authentication and bootstrap boundaries."""
+
+    def setUp(self):
+        """Isolate the process-wide credential registry for each test."""
+
+        self.original_keys = saas_web.API_KEY_REGISTRY.get_keys()
+        saas_web.API_KEY_REGISTRY.set_keys([])
+
+    def tearDown(self):
+        """Restore credentials without reading the environment."""
+
+        saas_web.API_KEY_REGISTRY.set_keys(self.original_keys)
 
     def _post_shrink(self, headers=None):
         """POST a minimal /shrink request and return the response."""
@@ -688,105 +699,112 @@ def _post_shrink(self, headers=None):
             headers=headers or {},
         )
 
-    def test_no_env_var_leaves_endpoints_open(self):
-        with patch.dict(os.environ):
-            os.environ.pop("CODEC_CARVER_API_KEYS", None)
-            response = self._post_shrink()
-
+    def test_empty_registry_leaves_endpoints_open(self):
+        response = self._post_shrink()
         self.assertEqual(response.status_code, 200)
         self.assertEqual(
             response.json(),
             {"error": "Invalid target_bytes value. Must be greater than 0."},
         )
 
-    def test_missing_header_rejected_when_keys_configured(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
-            response = self._post_shrink()
-
-        self.assertEqual(response.status_code, 401)
-        self.assertEqual(response.json(), {"error": "Invalid or missing API key"})
-        self.assertNotIn("secret-key", response.text)
-
-    def test_wrong_key_rejected(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
-            response = self._post_shrink(headers={"X-API-Key": "wrong-key"})
-
-        self.assertEqual(response.status_code, 401)
-        self.assertEqual(response.json(), {"error": "Invalid or missing API key"})
-        self.assertNotIn("secret-key", response.text)
-
-    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"})
+    def test_missing_and_wrong_keys_are_rejected_without_secret_echo(self):
+        saas_web.API_KEY_REGISTRY.set_keys(["secret-key"])
+        for headers in ({}, {"X-API-Key": "wrong-key"}):
+            response = self._post_shrink(headers=headers)
+            self.assertEqual(response.status_code, 401)
+            self.assertEqual(
+                response.json(), {"error": "Invalid or missing API key"}
+            )
+            self.assertNotIn("secret-key", response.text)
 
+    def test_registered_key_reaches_handler(self):
+        saas_web.API_KEY_REGISTRY.set_keys(["secret-key"])
+        response = self._post_shrink(headers={"X-API-Key": "secret-key"})
         self.assertEqual(response.status_code, 200)
-        self.assertEqual(
-            response.json(),
-            {"error": "Invalid target_bytes value. Must be greater than 0."},
-        )
-
-    def test_get_ui_always_open_without_key(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
-            response = client.get("/")
 
+    def test_get_ui_remains_open_without_key(self):
+        saas_web.API_KEY_REGISTRY.set_keys(["secret-key"])
+        response = client.get("/")
         self.assertEqual(response.status_code, 200)
         self.assertIn(b"Codec Carver SaaS", response.content)
 
-    def test_job_api_requires_key_when_configured(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "secret-key"}):
-            response = client.get("/jobs/missing")
-            allowed = client.get("/jobs/missing", headers={"X-API-Key": "secret-key"})
-
-        self.assertEqual(response.status_code, 401)
-        self.assertEqual(response.json(), {"error": "Invalid or missing API key"})
+    def test_job_api_requires_registered_key(self):
+        saas_web.API_KEY_REGISTRY.set_keys(["secret-key"])
+        rejected = client.get("/jobs/missing")
+        allowed = client.get(
+            "/jobs/missing", headers={"X-API-Key": "secret-key"}
+        )
+        self.assertEqual(rejected.status_code, 401)
         self.assertEqual(allowed.status_code, 404)
 
-    def test_multiple_comma_separated_keys_all_valid(self):
+    def test_registry_normalizes_multiple_whitespace_and_empty_keys(self):
+        saas_web.API_KEY_REGISTRY.set_keys(
+            [" key-one ", "", "  ", "key-two"]
+        )
+        self.assertEqual(
+            saas_web.get_configured_api_keys(), ["key-one", "key-two"]
+        )
+        for key in ("key-one", "key-two"):
+            response = self._post_shrink(headers={"X-API-Key": key})
+            self.assertEqual(response.status_code, 200, key)
+        self.assertEqual(
+            self._post_shrink(headers={"X-API-Key": " key-one "}).status_code,
+            401,
+        )
+
+    def test_bootstrap_consumes_transport_variable_once(self):
         with patch.dict(
-            os.environ, {"CODEC_CARVER_API_KEYS": "key-one,key-two,key-three"}
+            os.environ, {"CODEC_CARVER_API_KEYS": " a ,, b ,"}
         ):
-            for key in ("key-one", "key-two", "key-three"):
-                response = self._post_shrink(headers={"X-API-Key": key})
-                self.assertEqual(response.status_code, 200, key)
-            rejected = self._post_shrink(headers={"X-API-Key": "key-four"})
+            registry = saas_web.APIKeyRegistry()
+            self.assertEqual(registry.get_keys(), ["a", "b"])
+            self.assertNotIn("CODEC_CARVER_API_KEYS", os.environ)
 
-        self.assertEqual(rejected.status_code, 401)
+    def test_bootstrap_without_transport_starts_empty(self):
+        with patch.dict(os.environ, {}, clear=True):
+            registry = saas_web.APIKeyRegistry()
+        self.assertEqual(registry.get_keys(), [])
 
-    def test_whitespace_around_keys_is_stripped(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "  key-one , key-two  "}):
-            response = self._post_shrink(headers={"X-API-Key": "key-one"})
-            self.assertEqual(response.status_code, 200)
-            response = self._post_shrink(headers={"X-API-Key": "key-two"})
-            self.assertEqual(response.status_code, 200)
-            rejected = self._post_shrink(headers={"X-API-Key": " key-one "})
+    def test_environment_changes_cannot_rotate_runtime_credentials(self):
+        with patch.dict(
+            os.environ, {"CODEC_CARVER_API_KEYS": "bootstrap-key"}
+        ):
+            registry = saas_web.APIKeyRegistry()
+        with patch.dict(
+            os.environ, {"CODEC_CARVER_API_KEYS": "untrusted-new-key"}
+        ):
+            self.assertEqual(registry.get_keys(), ["bootstrap-key"])
+        registry.set_keys(["updated-key"])
+        self.assertEqual(registry.get_keys(), ["updated-key"])
 
-        self.assertEqual(rejected.status_code, 401)
+    def test_non_ascii_header_is_rejected_without_type_error(self):
+        import asyncio
 
-    def test_empty_entries_are_ignored(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "key-one,,  ,"}):
-            response = self._post_shrink(headers={"X-API-Key": "key-one"})
-            self.assertEqual(response.status_code, 200)
-            rejected = self._post_shrink(headers={"X-API-Key": ""})
+        from starlette.requests import Request
 
-        self.assertEqual(rejected.status_code, 401)
+        saas_web.API_KEY_REGISTRY.set_keys(["secret-key"])
+        request = Request(
+            {
+                "type": "http",
+                "method": "GET",
+                "path": "/jobs/123",
+                "headers": [
+                    (b"x-api-key", "malicious_key_한글".encode("utf-8"))
+                ],
+            }
+        )
 
-    def test_only_empty_entries_leave_endpoints_open(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": " , ,"}):
-            response = self._post_shrink()
+        async def call_next(_request):
+            return "unexpected success"
 
-        self.assertEqual(response.status_code, 200)
+        response = asyncio.run(
+            saas_web.require_api_key(request, call_next)
+        )
+        self.assertEqual(response.status_code, 401)
         self.assertEqual(
-            response.json(),
-            {"error": "Invalid target_bytes value. Must be greater than 0."},
+            response.body, b'{"error":"Invalid or missing API key"}'
         )
 
-    def test_get_configured_api_keys_parsing(self):
-        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": " a ,, b ,"}):
-            self.assertEqual(saas_web.get_configured_api_keys(), ["a", "b"])
-        with patch.dict(os.environ):
-            os.environ.pop("CODEC_CARVER_API_KEYS", None)
-            self.assertEqual(saas_web.get_configured_api_keys(), [])
-
 
 @unittest.skipUnless(
     _HAS_FASTAPI, "fastapi not installed (optional integration dependency)"