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

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)"