feat(api): Track 3.2 — X-Admin-Key gates cross-user DELETE on /api/analysis-runs

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
 

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}`

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)

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,
             },

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