[obvious-fleet] Backend: fail closed on missing upload-token secret; purge packet cache on delete; close delete/hydrate cache race

backend/server/api/app.py

@@ -229,7 +229,7 @@ async def validation_error(_request: Request, exc: RequestValidationError):
     amendments.register(app)
     confirmations.register(app)
     devices.register(app)
-    uploads.register(app)
+    uploads.register(app, settings)
     return app
 
 

backend/server/api/routes/captures.py

@@ -19,6 +19,7 @@
 from server.common.logging import log_event
 from server.contracts.api import CreateCaptureRequest, CreateCaptureResponse, UploadLimits
 from server.contracts.jobs import CaptureRecord
+from server.orchestrator.artifacts import purge_packet_cache
 
 
 def _body_bytes(payload: dict) -> bytes:
@@ -217,10 +218,35 @@ async def delete_capture(capture_id: str, request: Request):
             packet_id = record.packet_id
             deps.state.delete_capture(capture_id)
             deps.state.put_kv(f"tomb:{capture_id}", {"deleted_at": isoformat(deps.now()), "packet_id": packet_id})
+            # Tombstone before purging the cache: a hydrate that passed the
+            # entry tombstone check while this delete was running can only be
+            # stopped from republishing once the tombstone is visible, so every
+            # publish it attempts after that must land inside this purge.
+            # Object storage is clean, but workers publish packet copies to the
+            # shared volume too. Purge them; report what could not be removed
+            # instead of claiming a clean deletion.
+            cache_removed, cache_failed = purge_packet_cache(capture_id)
+            if cache_failed:
+                log_event(
+                    "storage.warning",
+                    op="purge_packet_cache",
+                    capture_id=capture_id,
+                    level="warning",
+                    msg=(
+                        f"packet cache: {len(cache_failed)} path(s) could not be removed: "
+                        + ", ".join(cache_failed[:20])
+                    ),
+                )
             idem = deps.state.get_kv(f"idem:create:{packet_id}")
             if idem:
                 idem["tomb"] = True
                 idem["ref"] = f"tomb:{capture_id}"
                 deps.state.put_kv(f"idem:create:{packet_id}", idem)
-            log_event("capture.deleted", objects_deleted=deleted, capture_id=capture_id, principal=principal.label)
+            log_event(
+                "capture.deleted",
+                objects_deleted=deleted,
+                capture_id=capture_id,
+                principal=principal.label,
+                msg=f"packet cache: removed {cache_removed} file(s)",
+            )
             return Response(status_code=204)

backend/server/api/routes/uploads.py

@@ -12,14 +12,36 @@
 from server.api.deps import get_deps
 from server.api.errors import fail
 from server.common.backends.storage_volume import read_token
+from server.common.logging import log_event
 
 
 class _TooLarge(Exception):
     pass
 
 
+def routes_enabled(settings) -> bool:
+    """Signed upload URLs are HMAC tokens, so only a deployment with a secret serves them.
+
+    Volume deployments reject a missing secret at startup (`resolve_storage`).
+    Outside volume storage the routes are disabled outright: tokens are never
+    minted there and an empty-key verifier would accept forgeries. A test that
+    wants the routes sets UPLOAD_URL_HMAC_SECRET to a throwaway value.
+    """
+
+    return bool(settings.UPLOAD_URL_HMAC_SECRET)
+
+
 def _secret(deps) -> bytes:
     raw = deps.settings.UPLOAD_URL_HMAC_SECRET or ""
+    if not raw:
+        # Fail closed even if the routes were registered another way: an empty
+        # HMAC key would let anyone who knows a storage key forge a token.
+        raise fail(
+            503,
+            "uploads_disabled",
+            "/",
+            "upload token routes are disabled: UPLOAD_URL_HMAC_SECRET is not configured",
+        )
     return raw.encode()
 
 
@@ -44,7 +66,14 @@ def _registered_bound(deps, key: str) -> int:
     return int(record.bytes)
 
 
-def register(app: FastAPI) -> None:
+def register(app: FastAPI, settings) -> None:
+    if not routes_enabled(settings):
+        log_event(
+            "api.startup",
+            msg=("upload token routes disabled: UPLOAD_URL_HMAC_SECRET is not configured"),
+        )
+        return
+
     @app.put("/v1/uploads/{token}")
     async def put_upload(token: str, request: Request):
         deps = get_deps(request)

backend/server/api/tests/test_uploads_token_gate.py

@@ -0,0 +1,153 @@
+"""Upload token routes fail closed without a signing secret.
+
+Greptile finding on PR #247 (backend/server/api/routes/uploads.py:23): with
+UPLOAD_URL_HMAC_SECRET unset on a gcs, r2, or local deployment, the token-only
+routes verified tokens against an empty HMAC key, so anyone who knew a storage
+key could forge a GET token and read protected capture files.
+"""
+
+from __future__ import annotations
+
+import time
+
+import pytest
+from fastapi.testclient import TestClient
+
+from server.api.app import create_app
+from server.api.routes import captures as captures_routes
+from server.api.tests.test_ingest import _create
+from server.common.backends.storage_volume import mint_token
+from server.common.config import Settings
+
+
+def _local_settings(tmp_path, **overrides) -> Settings:
+    values = dict(
+        SMP_ENVIRONMENT="dev",
+        STORAGE_BACKEND="local",
+        STATE_BACKEND="memory",
+        VOLUME_ROOT=str(tmp_path / "store"),
+        HSS_INGEST_HOOKS="null",
+        HSS_AUTH_MODE="static",
+        CAPTURE_API_TOKENS=f"device:{'d' * 40}",
+    )
+    values.update(overrides)
+    return Settings(**values)
+
+
+def _forged_get_token(key: str) -> str:
+    """A token an attacker mints with the empty key they know the verifier uses."""
+
+    return mint_token(
+        {"op": "get", "key": key, "md5": "", "ct": "", "exp": int(time.time()) + 600},
+        b"",
+    )
+
+
+def test_token_routes_are_absent_without_a_secret(tmp_path) -> None:
+    app = create_app(settings=_local_settings(tmp_path))
+    paths = {getattr(route, "path", "") for route in app.routes}
+    assert "/v1/uploads/{token}" not in paths
+
+
+def test_forged_empty_key_token_cannot_read_files(tmp_path) -> None:
+    """The regression test for the finding: fails against the old code, which
+    served the forged token the file's bytes with a 200."""
+
+    app = create_app(settings=_local_settings(tmp_path))
+    app.state.deps.storage.put_bytes("captures/cap_x/packet.json", b"private-bytes", "application/json")
+    response = TestClient(app).get(f"/v1/uploads/{_forged_get_token('captures/cap_x/packet.json')}")
+    assert response.status_code == 404
+    assert b"private-bytes" not in response.content
+
+
+def test_local_test_secret_serves_token_routes(tmp_path) -> None:
+    """A configured secret (the way every test opts in, per conftest.py) serves
+    the routes; a token minted with that same key verifies. There is no
+    empty-key path left in the code."""
+
+    test_secret = "h" * 32
+    settings = _local_settings(tmp_path, UPLOAD_URL_HMAC_SECRET=test_secret)
+    app = create_app(settings=settings)
+    app.state.deps.storage.put_bytes("captures/cap_x/packet.json", b"shared-bytes", "application/json")
+
+    def mint(key: str) -> str:
+        return mint_token(
+            {"op": "get", "key": key, "md5": "", "ct": "", "exp": int(time.time()) + 600},
+            test_secret.encode(),
+        )
+
+    response = TestClient(app).get(f"/v1/uploads/{mint('captures/cap_x/packet.json')}")
+    assert response.status_code == 200
+    assert response.content == b"shared-bytes"
+
+
+def test_volume_without_a_secret_refuses_to_start(tmp_path) -> None:
+    settings = Settings(
+        STORAGE_BACKEND="volume",
+        STATE_BACKEND="memory",
+        VOLUME_ROOT=str(tmp_path / "vol"),
+        CAPTURE_API_PUBLIC_URL="http://testserver",
+    )
+    with pytest.raises(RuntimeError, match="UPLOAD_URL_HMAC_SECRET"):
+        create_app(settings=settings)
+
+
+def test_delete_purges_the_shared_packet_cache(client, monkeypatch, tmp_path) -> None:
+    """Greptile finding on PR #247 (backend/server/api/routes/captures.py:213):
+    workers copy the packet to /vol/packet-cache/{capture_id}/{sha} and DELETE
+    only removed object-storage prefixes, so photo copies survived a 204."""
+
+    capture_id = _create(client, "packet-cache-purge")
+    cache_root = tmp_path / "packet-cache"
+    slot = cache_root / capture_id / ("a" * 64)
+    (slot / "photos").mkdir(parents=True)
+    (slot / "packet.json").write_bytes(b"{}")
+    (slot / ".complete").write_bytes(b"ok")
+    (slot / "photos" / "k0.jpg").write_bytes(b"jpeg")
+    monkeypatch.setenv("HSS_PACKET_CACHE", str(cache_root))
+
+    assert client.delete(f"/v1/captures/{capture_id}").status_code == 204
+    assert not (cache_root / capture_id).exists()
+
+
+def test_delete_tombstones_before_purging_the_cache(client, monkeypatch) -> None:
+    """The delete half of the publish race (Codex P1 on PR #250): a hydrate
+    that checked the tombstone before DELETE wrote it can only be stopped from
+    republishing once the tombstone is visible, so every publish it attempts
+    afterwards must land inside the cache purge. Under the old order -- purge
+    first, tombstone second -- a publish that raced the purge survived the
+    delete. Recording the tombstone at purge time pins the order."""
+
+    capture_id = _create(client, "tomb-before-purge")
+    seen: dict = {}
+    real_purge = captures_routes.purge_packet_cache
+
+    def purge_and_note(cid: str):
+        seen["tomb_at_purge"] = client.app_state.state.get_kv(f"tomb:{cid}")
+        return real_purge(cid)
+
+    monkeypatch.setattr("server.api.routes.captures.purge_packet_cache", purge_and_note)
+
+    assert client.delete(f"/v1/captures/{capture_id}").status_code == 204
+    assert seen["tomb_at_purge"], "the tombstone must already be visible when the cache purge runs"
+
+
+def test_delete_reports_a_failed_packet_cache_purge(client, monkeypatch) -> None:
+    capture_id = _create(client, "packet-cache-stuck")
+    survivors = [f"/vol/packet-cache/{capture_id}/{'a' * 64}/packet.json"]
+    events = []
+
+    monkeypatch.setattr(
+        "server.api.routes.captures.purge_packet_cache",
+        lambda cid: (0, survivors) if cid == capture_id else (0, []),
+    )
+    monkeypatch.setattr(
+        "server.api.routes.captures.log_event",
+        lambda event, **fields: events.append((event, fields)),
+    )
+
+    assert client.delete(f"/v1/captures/{capture_id}").status_code == 204
+    warnings = [fields for event, fields in events if event == "storage.warning"]
+    assert any("could not be removed" in str(fields.get("msg")) for fields in warnings)
+    deleted = [fields for event, fields in events if event == "capture.deleted"]
+    assert deleted and "removed 0 file(s)" in str(deleted[0].get("msg"))

backend/server/common/backends/storage_volume.py

@@ -52,6 +52,10 @@ def read_token(token: str, secret: bytes) -> dict | None:
 
 class VolumeStorage(LocalFsStorage):
     def __init__(self, root: str | Path, public_url: str, hmac_secret: bytes) -> None:
+        if not hmac_secret:
+            # Refuse to sign tokens with an empty key: anyone who knows a storage
+            # key could forge one.
+            raise ValueError("VolumeStorage needs a non-empty UPLOAD_URL_HMAC_SECRET")
         super().__init__(Path(root))
         self.public_url = public_url.rstrip("/")
         self.hmac_secret = hmac_secret

backend/server/common/config.py

@@ -121,7 +121,11 @@ def resolve_storage(settings: Settings) -> Storage:
         return LocalFsStorage(root)
     if settings.STORAGE_BACKEND == "volume":
         if not settings.VOLUME_ROOT or not settings.CAPTURE_API_PUBLIC_URL or not settings.UPLOAD_URL_HMAC_SECRET:
-            raise RuntimeError("volume storage needs VOLUME_ROOT, CAPTURE_API_PUBLIC_URL, and UPLOAD_URL_HMAC_SECRET")
+            raise RuntimeError(
+                "volume storage needs VOLUME_ROOT, CAPTURE_API_PUBLIC_URL, and UPLOAD_URL_HMAC_SECRET; "
+                "UPLOAD_URL_HMAC_SECRET signs the upload tokens, and a missing key lets anyone who "
+                "knows a storage key forge one."
+            )
         from server.common.backends.storage_volume import VolumeStorage
 
         return VolumeStorage(

backend/server/orchestrator/artifacts.py

@@ -182,6 +182,41 @@ def packet_cache_dir(capture_id: str, packet_sha256: str = "") -> Path | None:
     return Path("/vol/packet-cache") / slot
 
 
+def purge_packet_cache(capture_id: str) -> tuple[int, list[str]]:
+    """Remove every cached packet copy for a capture from the shared volume.
+
+    DELETE calls this because object-storage prefixes do not cover the packet
+    cache workers publish under /vol/packet-cache. Returns the number of files
+    removed and the paths that survived, so the caller can report a partial
+    purge instead of claiming a clean deletion.
+    """
+
+    base = packet_cache_dir(capture_id)
+    if base is None or not base.is_dir():
+        return 0, []
+    files = [path for path in base.rglob("*") if path.is_file() and not path.is_symlink()]
+    try:
+        shutil.rmtree(base)
+        return len(files), []
+    except OSError:
+        survivors = sorted(base.rglob("*"))
+        kept = sum(1 for path in survivors if path.is_file() and not path.is_symlink())
+        return len(files) - kept, [str(path) for path in survivors]
+
+
+def _capture_tombstoned(state: object, capture_id: str) -> bool:
+    """True once DELETE has removed the capture. States without kv storage never
+    tombstone, and a failing lookup must not block a live capture's stage."""
+
+    get_kv = getattr(state, "get_kv", None)
+    if not callable(get_kv):
+        return False
+    try:
+        return bool(get_kv(f"tomb:{capture_id}"))
+    except Exception:
+        return False
+
+
 def hydrate_packet(
     storage: object,
     state: object,
@@ -197,9 +232,13 @@ def hydrate_packet(
     ``include_files`` is false for stages that only read ``packet.json``. A warm
     volume cache skips GCS for the frames a later stage does need.
     ``packet_key`` is the manifest's storage key: an amendment run passes its
-    own ``captures/{capture_id}/amendments/{amendment_id}/packet.json``.
+    own ``captures/{capture_id}/amendments/{amendment_id}/packet.json``. A
+    capture the API has deleted must not hydrate at all: restoring it would
+    re-publish the packet cache DELETE just purged.
     """
 
+    if _capture_tombstoned(state, capture_id):
+        return
     packet_dir.mkdir(parents=True, exist_ok=True)
     if _copy_packet_cache(cache_dir, packet_dir, include_files=include_files):
         return
@@ -208,7 +247,10 @@ def hydrate_packet(
     if not include_files:
         return
     _download_all(storage, _packet_file_jobs(state, capture_id, packet_dir, manifest_key))
-    if cache_dir is not None and (packet_dir / "packet.json").is_file():
+    # A delete that lands mid-hydrate must be the last word: the entry check
+    # above ran before the tombstone existed, so re-check immediately before
+    # publishing, or this hydrate restores the cache the delete just purged.
+    if cache_dir is not None and (packet_dir / "packet.json").is_file() and not _capture_tombstoned(state, capture_id):
         _publish_packet_cache(packet_dir, cache_dir)