diff --git a/backend/server/api/app.py b/backend/server/api/app.py index 8534c2c8c..47da13d1e 100644 --- a/backend/server/api/app.py +++ b/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 diff --git a/backend/server/api/routes/captures.py b/backend/server/api/routes/captures.py index b2a9adfc0..1ddd988c9 100644 --- a/backend/server/api/routes/captures.py +++ b/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) diff --git a/backend/server/api/routes/uploads.py b/backend/server/api/routes/uploads.py index 59884e711..64eeb1120 100644 --- a/backend/server/api/routes/uploads.py +++ b/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) diff --git a/backend/server/api/tests/test_uploads_token_gate.py b/backend/server/api/tests/test_uploads_token_gate.py new file mode 100644 index 000000000..28e59ab11 --- /dev/null +++ b/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")) diff --git a/backend/server/common/backends/storage_volume.py b/backend/server/common/backends/storage_volume.py index 9365bff5e..0ef222ce1 100644 --- a/backend/server/common/backends/storage_volume.py +++ b/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 diff --git a/backend/server/common/config.py b/backend/server/common/config.py index 523b15762..07d7b492d 100644 --- a/backend/server/common/config.py +++ b/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( diff --git a/backend/server/orchestrator/artifacts.py b/backend/server/orchestrator/artifacts.py index 92084cd82..668c42a88 100644 --- a/backend/server/orchestrator/artifacts.py +++ b/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) diff --git a/backend/server/orchestrator/tests/test_hydrate_inputs.py b/backend/server/orchestrator/tests/test_hydrate_inputs.py index b1471bb84..c049b0392 100644 --- a/backend/server/orchestrator/tests/test_hydrate_inputs.py +++ b/backend/server/orchestrator/tests/test_hydrate_inputs.py @@ -2,12 +2,18 @@ from __future__ import annotations +import shutil import threading import time from pathlib import Path from server.contracts.jobs import FileRecord, RunRecord, StageRecord -from server.orchestrator.artifacts import hydrate_artifacts, hydrate_packet, packet_cache_dir +from server.orchestrator.artifacts import ( + hydrate_artifacts, + hydrate_packet, + packet_cache_dir, + purge_packet_cache, +) from server.orchestrator.graph import STAGE_NAMES, select_stage_artifacts, stage_needs_packet_files from server.orchestrator.services import bind from server.orchestrator.tests.conftest import build_services @@ -161,6 +167,127 @@ def test_packet_cache_is_keyed_by_capture_and_sha(tmp_path: Path, monkeypatch) - assert not (amended / "packet.json").exists() +def test_purge_packet_cache_removes_the_whole_capture_slot(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("HSS_PACKET_CACHE", str(tmp_path / "cache")) + base = packet_cache_dir("cap_test", "a" * 64) + assert base is not None + (base / "photos").mkdir(parents=True) + (base / "packet.json").write_bytes(b"{}") + (base / ".complete").write_bytes(b"ok") + (base / "photos" / "k0.jpg").write_bytes(b"jpeg") + other = packet_cache_dir("cap_other") + assert other is not None + other.mkdir(parents=True) + (other / "keep.txt").write_bytes(b"keep") + + removed, failed = purge_packet_cache("cap_test") + + assert (removed, failed) == (3, []) + assert not (tmp_path / "cache" / "cap_test").exists() + assert (other / "keep.txt").is_file() + + +def test_purge_packet_cache_reports_paths_it_could_not_remove(tmp_path: Path, monkeypatch) -> None: + monkeypatch.setenv("HSS_PACKET_CACHE", str(tmp_path / "cache")) + base = packet_cache_dir("cap_test") + assert base is not None + base.mkdir(parents=True) + (base / "packet.json").write_bytes(b"{}") + + def broken_rmtree(path, *args, **kwargs): + raise OSError("volume busy") + + monkeypatch.setattr(shutil, "rmtree", broken_rmtree) + removed, failed = purge_packet_cache("cap_test") + assert removed == 0 + assert failed == [str(base / "packet.json")] + + +def test_hydrate_skips_a_tombstoned_capture(tmp_path: Path) -> None: + """A worker mid-run must not restore the cache DELETE just purged.""" + + cache = tmp_path / "cache" / "cap_test" + cache.mkdir(parents=True) + (cache / "packet.json").write_bytes(b"{}") + (cache / ".complete").write_bytes(b"ok") + + class _TombstonedState: + def __init__(self) -> None: + self.asked: list[str] = [] + + def get_kv(self, key: str): + self.asked.append(key) + return {"deleted_at": "2026-10-15T00:00:00Z"} + + state = _TombstonedState() + dest = tmp_path / "packet" + hydrate_packet(object(), state, "cap_test", dest, include_files=True, cache_dir=cache) + assert not dest.exists() + assert state.asked == ["tomb:cap_test"] + + +def test_delete_mid_hydrate_does_not_republish_the_cache(tmp_path: Path, monkeypatch) -> None: + """Codex P1 on PR #250 (artifacts.py publish step): a hydrate that passed + the entry tombstone check while DELETE was still running used to publish + the cache slot back after DELETE purged it. Here DELETE lands mid-download + -- purge, then tombstone -- so the finishing hydrate must leave the cache + gone.""" + + monkeypatch.setenv("HSS_PACKET_CACHE", str(tmp_path / "cache")) + services, _uri, _digest = build_services(tmp_path, "memory") + services.storage.put_bytes("captures/cap_test/keyframes/k0.jpg", b"jpeg", "image/jpeg") + services.state.upsert_file( + FileRecord( + capture_id="cap_test", + path="keyframes/k0.jpg", + role="keyframe", + bytes=4, + sha256="cd" * 32, + md5="0" * 32, + content_type="image/jpeg", + state="committed", + key="captures/cap_test/keyframes/k0.jpg", + ) + ) + + class _DeleteMidDownload: + """DELETE purges the cache and tombstones the capture while the hydrate + is downloading its files.""" + + def __init__(self, storage, state) -> None: + self._storage = storage + self._state = state + + def get_bytes(self, key: str) -> bytes: + data = self._storage.get_bytes(key) + if key == "captures/cap_test/packet.json": + purge_packet_cache("cap_test") + self._state.put_kv("tomb:cap_test", {"deleted_at": "2026-10-16T00:00:00Z"}) + return data + + def exists(self, key: str) -> bool: + return self._storage.exists(key) + + def list_prefix(self, prefix: str) -> list[str]: + return self._storage.list_prefix(prefix) + + cache = packet_cache_dir("cap_test") + assert cache is not None + dest = tmp_path / "packet" + hydrate_packet( + _DeleteMidDownload(services.storage, services.state), + services.state, + "cap_test", + dest, + include_files=True, + cache_dir=cache, + ) + + # The run's own working copy is fine; the shared cache must stay gone. + assert (dest / "keyframes" / "k0.jpg").is_file() + assert not cache.exists() + + def test_artifact_downloads_overlap(tmp_path: Path) -> None: class _Slow: def __init__(self) -> None: