diff --git a/saas_web.py b/saas_web.py index 2f177f8..f10b95c 100644 --- a/saas_web.py +++ b/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.""" diff --git a/tests/test_saas_web.py b/tests/test_saas_web.py index 06852fa..1fd22be 100644 --- a/tests/test_saas_web.py +++ b/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()