diff --git a/CLAUDE.md b/CLAUDE.md index b198e7d5..f4062d8c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -177,10 +177,11 @@ DISABLE_HMR="false" # set "true" for dev environments that need HMR # Backend (env var, no .env file) SONIC_ANALYZER_PORT=8100 -GEMINI_API_KEY="your_key_here" # read by server.py at runtime, not in browser bundle +GEMINI_API_KEY="your_key_here" # read by server.py at runtime, not in browser bundle +SONIC_ANALYZER_ADMIN_KEY="optional" # if set, DELETE /api/analysis-runs/{run_id} accepts an X-Admin-Key header that bypasses ownership for operator-level purge. Unset by default; admin path is closed. ``` -Phase 2 is gated by `VITE_ENABLE_PHASE2_GEMINI`. `GEMINI_API_KEY` is backend-only. +Phase 2 is gated by `VITE_ENABLE_PHASE2_GEMINI`. `GEMINI_API_KEY` is backend-only. `SONIC_ANALYZER_ADMIN_KEY` is backend-only and never exposed to clients. ## Key Guardrails diff --git a/apps/backend/ARCHITECTURE.md b/apps/backend/ARCHITECTURE.md index a77c0985..7c86de5a 100644 --- a/apps/backend/ARCHITECTURE.md +++ b/apps/backend/ARCHITECTURE.md @@ -70,7 +70,7 @@ Custom routes: - `POST /api/analysis-runs/estimate` - `POST /api/analysis-runs` — multipart upload OR URL ingestion. Provide *exactly one* of `track` (multipart `UploadFile`) or `url` (form field with a public `http`/`https` URL). URL mode is SSRF-guarded against private/loopback/link-local addresses and enforces the same 100 MiB cap via streaming. See [`url_ingest.py`](url_ingest.py). - `GET /api/analysis-runs/{run_id}` -- `DELETE /api/analysis-runs/{run_id}` +- `DELETE /api/analysis-runs/{run_id}` — owner can delete their own run; operators with `SONIC_ANALYZER_ADMIN_KEY` set can supply `X-Admin-Key` to delete any run. Admin path is closed when the env var is unset. - `GET /api/analysis-runs/{run_id}/artifacts` and `…/artifacts/{artifact_id}` - `GET /api/analysis-runs/{run_id}/export/csv/{field_path}` — CSV export of a Phase 1 time-series field. See [`docs/adr/0001-phase1-json-schema-v1.md`](../../docs/adr/0001-phase1-json-schema-v1.md) and the registry in [`csv_export.py`](csv_export.py). - `POST /api/analysis-runs/{run_id}/spectral-enhancements/{kind}` diff --git a/apps/backend/auth_context.py b/apps/backend/auth_context.py index bde2f4e0..05f606b6 100644 --- a/apps/backend/auth_context.py +++ b/apps/backend/auth_context.py @@ -1,11 +1,20 @@ from __future__ import annotations +import hmac +import os from dataclasses import dataclass from runtime_profile import resolve_runtime_profile, should_require_authenticated_user LOCAL_DEV_USER_ID = "local-dev" +# Operator-level privilege gate. When this env var is set and non-empty, +# requests carrying a matching ``X-Admin-Key`` header may perform privileged +# operations (e.g. delete runs they do not own). When unset, no admin path +# exists and routes fall back to per-user-ownership enforcement. Default for +# local development is unset. +ADMIN_KEY_ENV_VAR = "SONIC_ANALYZER_ADMIN_KEY" + class AuthenticationRequiredError(PermissionError): pass @@ -42,3 +51,42 @@ def resolve_api_user_context( email=email or None, runtime_profile=runtime_profile, ) + + +def _read_configured_admin_key() -> str: + """Return the configured admin key, or empty string if unset.""" + return (os.getenv(ADMIN_KEY_ENV_VAR) or "").strip() + + +def admin_key_is_configured() -> bool: + """True iff :data:`ADMIN_KEY_ENV_VAR` is set to a non-empty value. + + Use this to decide *whether* to expose admin-gated routes, not to + *authorize* a specific request. For authorization, call + :func:`admin_key_matches`. + """ + return bool(_read_configured_admin_key()) + + +def admin_key_matches(provided: str | None) -> bool: + """Constant-time check: does ``provided`` match the configured admin key? + + Returns ``False`` when: + - the admin key is not configured (the env var is unset/empty), or + - the caller did not supply a header value, or + - the supplied value does not match the configured key. + + Uses :func:`hmac.compare_digest` to resist timing-based probing of + the configured key. Strips whitespace from both sides — both the + header value and the env var — to avoid trailing-newline surprises + when keys are sourced from secret managers. + """ + if provided is None: + return False + configured = _read_configured_admin_key() + if not configured: + return False + stripped = provided.strip() + if not stripped: + return False + return hmac.compare_digest(configured, stripped) diff --git a/apps/backend/server.py b/apps/backend/server.py index 53bc39aa..f3aa4870 100644 --- a/apps/backend/server.py +++ b/apps/backend/server.py @@ -29,6 +29,7 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import FileResponse, JSONResponse, Response +import auth_context from auth_context import AuthenticationRequiredError, UserContext, resolve_api_user_context from analysis_runtime import ( AnalysisRuntime, @@ -2157,13 +2158,38 @@ async def delete_analysis_run( run_id: str, x_asa_user_id: str | None = Header(None), x_asa_user_email: str | None = Header(None), + x_admin_key: str | None = Header(None), ) -> JSONResponse: - user_context = _resolve_route_user_context(x_asa_user_id, x_asa_user_email) - if isinstance(user_context, JSONResponse): - return user_context + """Delete a run, its artifacts, and interrupt any active stages. + + Two paths grant access: + + 1. **Owner.** The user resolved from ``X-ASA-User-Id`` (or local-mode + default) is the run's owner — current behavior, unchanged. + 2. **Operator.** ``X-Admin-Key`` matches the configured + :data:`auth_context.ADMIN_KEY_ENV_VAR` env var. When this matches, + the ownership check is skipped — operators can purge any run + regardless of who owns it. The env var is unset by default; if + unset, no admin path exists and ownership remains the only gate. + + The admin path still returns ``RUN_NOT_FOUND`` for nonexistent runs, + so a privileged caller cannot use this endpoint to enumerate + run IDs (the response is identical whether the run doesn't exist + or the caller lacks ownership without the key). + """ + is_admin = auth_context.admin_key_matches(x_admin_key) + if is_admin: + # Skip user-context resolution; the admin key is the auth. + owner_user_id: str | None = None + else: + user_context = _resolve_route_user_context(x_asa_user_id, x_asa_user_email) + if isinstance(user_context, JSONResponse): + return user_context + owner_user_id = user_context.user_id + runtime = get_analysis_runtime() try: - runtime.get_run(run_id, owner_user_id=user_context.user_id) + runtime.get_run(run_id, owner_user_id=owner_user_id) terminated_stages = _interrupt_active_child_processes(run_id) runtime.delete_run(run_id) except (KeyError, PermissionError): @@ -2173,6 +2199,7 @@ async def delete_analysis_run( content={ "runId": run_id, "deleted": True, + "deletedBy": "admin" if is_admin else "owner", "interrupt": { "stagesTerminated": terminated_stages, }, diff --git a/apps/backend/tests/test_auth_context.py b/apps/backend/tests/test_auth_context.py new file mode 100644 index 00000000..0ec28cee --- /dev/null +++ b/apps/backend/tests/test_auth_context.py @@ -0,0 +1,134 @@ +"""Tests for ``auth_context.admin_key_matches`` and ``admin_key_is_configured``. + +These guard a security-critical surface: the admin key gates privileged +operations like cross-user run deletion. Bugs here trade ownership for +access — or vice versa. +""" + +import os +import sys +import unittest +from pathlib import Path +from unittest import mock + + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +if str(_BACKEND_ROOT) not in sys.path: + sys.path.insert(0, str(_BACKEND_ROOT)) + + +import auth_context # noqa: E402 — load after sys.path is set + + +class AdminKeyIsConfiguredTests(unittest.TestCase): + """``admin_key_is_configured()`` reflects only the env-var state.""" + + def test_unset_env_var(self): + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop(auth_context.ADMIN_KEY_ENV_VAR, None) + self.assertFalse(auth_context.admin_key_is_configured()) + + def test_empty_env_var(self): + with mock.patch.dict( + os.environ, + {auth_context.ADMIN_KEY_ENV_VAR: ""}, + clear=False, + ): + self.assertFalse(auth_context.admin_key_is_configured()) + + def test_whitespace_only_env_var(self): + with mock.patch.dict( + os.environ, + {auth_context.ADMIN_KEY_ENV_VAR: " \n\t "}, + clear=False, + ): + self.assertFalse(auth_context.admin_key_is_configured()) + + def test_real_value(self): + with mock.patch.dict( + os.environ, + {auth_context.ADMIN_KEY_ENV_VAR: "s3cret"}, + clear=False, + ): + self.assertTrue(auth_context.admin_key_is_configured()) + + +class AdminKeyMatchesTests(unittest.TestCase): + """``admin_key_matches(provided)`` is the auth decision.""" + + def test_returns_false_when_env_unset(self): + with mock.patch.dict(os.environ, {}, clear=False): + os.environ.pop(auth_context.ADMIN_KEY_ENV_VAR, None) + self.assertFalse(auth_context.admin_key_matches("anything")) + self.assertFalse(auth_context.admin_key_matches(None)) + + def test_returns_false_for_none_provided(self): + with mock.patch.dict( + os.environ, + {auth_context.ADMIN_KEY_ENV_VAR: "s3cret"}, + clear=False, + ): + self.assertFalse(auth_context.admin_key_matches(None)) + + def test_returns_false_for_empty_provided(self): + with mock.patch.dict( + os.environ, + {auth_context.ADMIN_KEY_ENV_VAR: "s3cret"}, + clear=False, + ): + self.assertFalse(auth_context.admin_key_matches("")) + self.assertFalse(auth_context.admin_key_matches(" ")) + + def test_returns_true_for_exact_match(self): + with mock.patch.dict( + os.environ, + {auth_context.ADMIN_KEY_ENV_VAR: "s3cret"}, + clear=False, + ): + self.assertTrue(auth_context.admin_key_matches("s3cret")) + + def test_returns_false_for_wrong_value(self): + with mock.patch.dict( + os.environ, + {auth_context.ADMIN_KEY_ENV_VAR: "s3cret"}, + clear=False, + ): + self.assertFalse(auth_context.admin_key_matches("wrong")) + # Case-sensitive: rejected even with the right letters. + self.assertFalse(auth_context.admin_key_matches("S3CRET")) + # Partial / prefix matches don't satisfy. + self.assertFalse(auth_context.admin_key_matches("s3cre")) + self.assertFalse(auth_context.admin_key_matches("s3cretX")) + + def test_strips_whitespace_on_both_sides(self): + """Both the configured env value and the supplied header are + stripped — matters for secret managers that append a newline.""" + with mock.patch.dict( + os.environ, + {auth_context.ADMIN_KEY_ENV_VAR: " s3cret\n"}, + clear=False, + ): + self.assertTrue(auth_context.admin_key_matches("s3cret")) + self.assertTrue(auth_context.admin_key_matches(" s3cret ")) + + def test_constant_time_comparison_is_used(self): + """We use hmac.compare_digest under the hood to resist timing + oracle attacks. Verifying the symbol is wired in is a cheap + regression gate against someone 'simplifying' it to ==. + """ + import hmac + + with mock.patch.object( + hmac, "compare_digest", wraps=hmac.compare_digest + ) as wrapped: + with mock.patch.dict( + os.environ, + {auth_context.ADMIN_KEY_ENV_VAR: "s3cret"}, + clear=False, + ): + auth_context.admin_key_matches("s3cret") + wrapped.assert_called_once() + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/backend/tests/test_server.py b/apps/backend/tests/test_server.py index d917a43a..5d212b27 100644 --- a/apps/backend/tests/test_server.py +++ b/apps/backend/tests/test_server.py @@ -1361,6 +1361,198 @@ def test_delete_analysis_run_removes_owned_run(self) -> None: payload = self._decode_json_response(response) self.assertEqual(response.status_code, 202) self.assertTrue(payload["deleted"]) + # New: every deletion records who performed it. + self.assertEqual(payload["deletedBy"], "owner") + + def test_delete_analysis_run_admin_key_can_delete_other_users_run(self) -> None: + """With SONIC_ANALYZER_ADMIN_KEY set and the matching header, + an operator can delete a run owned by a different user.""" + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_server_runtime_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + created = runtime.create_run( + filename="track.mp3", + content=b"fake-audio", + mime_type="audio/mpeg", + owner_user_id="some_other_user", + analysis_mode="full", + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + ) + with ( + patch.object(server, "get_analysis_runtime", return_value=runtime), + patch.object( + server, + "_interrupt_active_child_processes", + return_value=[], + ), + patch.dict( + server.os.environ, + { + "SONIC_ANALYZER_RUNTIME_PROFILE": "hosted", + "SONIC_ANALYZER_ADMIN_KEY": "s3cret", + }, + clear=False, + ), + ): + response = asyncio.run( + server.delete_analysis_run( + created["runId"], + # Note: a *different* user makes the request, + # OR no user header at all — the admin key + # bypasses the ownership check entirely. + x_asa_user_id="someone_else", + x_admin_key="s3cret", + ) + ) + + payload = self._decode_json_response(response) + self.assertEqual(response.status_code, 202) + self.assertTrue(payload["deleted"]) + self.assertEqual(payload["deletedBy"], "admin") + + def test_delete_analysis_run_admin_key_works_without_user_header(self) -> None: + """The PR description claims the admin key skips user-context + resolution entirely. This test pins that claim: a request that + omits X-ASA-User-Id (which would fail in hosted mode under + ownership rules) still succeeds when X-Admin-Key matches. + """ + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_server_runtime_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + created = runtime.create_run( + filename="track.mp3", + content=b"fake-audio", + mime_type="audio/mpeg", + owner_user_id="owner_user", + analysis_mode="full", + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + ) + with ( + patch.object(server, "get_analysis_runtime", return_value=runtime), + patch.object( + server, + "_interrupt_active_child_processes", + return_value=[], + ), + patch.dict( + server.os.environ, + { + "SONIC_ANALYZER_RUNTIME_PROFILE": "hosted", + "SONIC_ANALYZER_ADMIN_KEY": "s3cret", + }, + clear=False, + ), + ): + response = asyncio.run( + server.delete_analysis_run( + created["runId"], + # No X-ASA-User-Id header — admin path bypasses + # user-context resolution. Under ownership rules + # alone, this would 401 in hosted mode. + x_asa_user_id=None, + x_admin_key="s3cret", + ) + ) + + payload = self._decode_json_response(response) + self.assertEqual(response.status_code, 202) + self.assertTrue(payload["deleted"]) + self.assertEqual(payload["deletedBy"], "admin") + + def test_delete_analysis_run_wrong_admin_key_falls_back_to_ownership(self) -> None: + """Wrong X-Admin-Key value does not grant access; the request + is treated as a regular (non-admin) ownership-based delete. + A non-owner gets RUN_NOT_FOUND.""" + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_server_runtime_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + created = runtime.create_run( + filename="track.mp3", + content=b"fake-audio", + mime_type="audio/mpeg", + owner_user_id="owner_user", + analysis_mode="full", + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + ) + with ( + patch.object(server, "get_analysis_runtime", return_value=runtime), + patch.dict( + server.os.environ, + { + "SONIC_ANALYZER_RUNTIME_PROFILE": "hosted", + "SONIC_ANALYZER_ADMIN_KEY": "s3cret", + }, + clear=False, + ), + ): + response = asyncio.run( + server.delete_analysis_run( + created["runId"], + x_asa_user_id="not_the_owner", + x_admin_key="WRONG", + ) + ) + + payload = self._decode_json_response(response) + self.assertEqual(response.status_code, 404) + self.assertEqual(payload["error"]["code"], "RUN_NOT_FOUND") + + def test_delete_analysis_run_admin_key_unset_ignores_header(self) -> None: + """When SONIC_ANALYZER_ADMIN_KEY is unset, supplying any + X-Admin-Key header must NOT grant cross-user access.""" + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_server_runtime_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + created = runtime.create_run( + filename="track.mp3", + content=b"fake-audio", + mime_type="audio/mpeg", + owner_user_id="owner_user", + analysis_mode="full", + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + ) + with ( + patch.object(server, "get_analysis_runtime", return_value=runtime), + patch.dict( + server.os.environ, + {"SONIC_ANALYZER_RUNTIME_PROFILE": "hosted"}, + clear=False, + ), + ): + # Explicitly remove the admin env var to prove the + # admin path is closed. + server.os.environ.pop("SONIC_ANALYZER_ADMIN_KEY", None) + response = asyncio.run( + server.delete_analysis_run( + created["runId"], + x_asa_user_id="not_the_owner", + x_admin_key="anything-here", + ) + ) + + payload = self._decode_json_response(response) + self.assertEqual(response.status_code, 404) + self.assertEqual(payload["error"]["code"], "RUN_NOT_FOUND") @patch.object(server, "get_audio_duration_seconds", return_value=214.6, create=True) @patch.object(