feat: opt-in API-key auth for saas_web via CODEC_CARVER_API_KEYS

saas_web.py

@@ -1,5 +1,7 @@
 """FastAPI upload UI for shrinking one media file through Codec Carver."""
 
+import hmac
+import os
 import tempfile
 import logging
 import shutil
@@ -54,6 +56,46 @@ 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.
+
+    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).
+    """
+
+    raw = os.environ.get("CODEC_CARVER_API_KEYS", "")
+    return [key.strip() for key in raw.split(",") if key.strip()]
+
+
+@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.
+    """
+
+    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
+        ):
+            return JSONResponse(
+                status_code=401,
+                content={"error": "Invalid or missing API key"},
+            )
+    return await call_next(request)
+
+
 @app.middleware("http")
 async def add_security_headers(request: Request, call_next):
     """Attach conservative browser security headers to every response."""

tests/test_saas_web.py

@@ -1,5 +1,6 @@
 import asyncio
 import io
+import os
 import unittest
 from unittest.mock import patch, MagicMock
 from pathlib import Path
@@ -307,5 +308,107 @@ def test_get_ui_includes_preset_buttons(self):
         self.assertIn('onclick="setTargetBytes(1073741824)"', html)
         self.assertIn('function setTargetBytes(bytes)', html)
 
+class TestApiKeyAuth(unittest.TestCase):
+    """Tests for the opt-in CODEC_CARVER_API_KEYS authentication middleware."""
+
+    def _post_shrink(self, headers=None):
+        """POST a minimal /shrink request and return the response."""
+
+        return client.post(
+            "/shrink",
+            files={"file": ("input.wav", io.BytesIO(b"dummy wav data"), "audio/wav")},
+            data={"target_bytes": 0},
+            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()
+
+        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"})
+
+        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("/")
+
+        self.assertEqual(response.status_code, 200)
+        self.assertIn(b"Codec Carver SaaS", response.content)
+
+    def test_multiple_comma_separated_keys_all_valid(self):
+        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": "key-one,key-two,key-three"}):
+            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"})
+
+        self.assertEqual(rejected.status_code, 401)
+
+    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 "})
+
+        self.assertEqual(rejected.status_code, 401)
+
+    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": ""})
+
+        self.assertEqual(rejected.status_code, 401)
+
+    def test_only_empty_entries_leave_endpoints_open(self):
+        with patch.dict(os.environ, {"CODEC_CARVER_API_KEYS": " , ,"}):
+            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_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(), [])
+
+
 if __name__ == '__main__':
     unittest.main()