From 121d538d55549019a0bfb0fce9d8d265ee5bc42c Mon Sep 17 00:00:00 2001 From: Efthimios Fousekis Date: Wed, 29 Jul 2026 17:26:47 +0300 Subject: [PATCH] feat(backend): Firebase token verification + async render jobs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports two already-merged Cinemory capabilities (same owner, MIT, shared B2/ provenance foundation) into ClaimScene, laying the foundation for a later multitenancy PR: 1. claimscene.auth — standalone RS256 Firebase ID-token verification against Google's public x509 certs (no Admin SDK, no service-account secret). Hardcoded RS256 allow-list checked before any key lookup blocks alg:none and the RS256/HS256 confusion attack. Not wired into api.py yet. 2. claimscene.jobs + async render endpoints — a storage-backed submit+poll job store (jobs//status.json, same StorageBackend the app already uses) plus POST /cases/render/jobs (202 + job_id) and GET /cases/render/jobs/{job_id} alongside the existing synchronous POST /cases/render. The two-step illustration generation (still, then image-to-video clip) can take minutes; async submit+poll avoids the Firebase Hosting proxy's ~60s cap. The worker thread gets an isolated storage instance in live mode (own B2Storage per job) so it never shares mutable adapter state with a request thread. _run_render and render_case gain a pure extract-method refactor (_prepare_render, an optional storage param) — behavior-preserving for the existing synchronous caller. deploy-cloudrun.sh gains --no-cpu-throttling so the background worker gets CPU between polls; CLOUDRUN.md documents the honest limitation (in-process worker, no external queue). Co-Authored-By: Claude Opus 5 --- deploy/CLOUDRUN.md | 37 +++ deploy/deploy-cloudrun.sh | 6 + pyproject.toml | 5 + requirements.txt | 5 + src/claimscene/api.py | 256 +++++++++++++++--- src/claimscene/auth.py | 288 ++++++++++++++++++++ src/claimscene/jobs.py | 156 +++++++++++ tests/e2e/test_render_jobs.py | 206 ++++++++++++++ tests/unit/test_auth.py | 490 ++++++++++++++++++++++++++++++++++ tests/unit/test_jobs.py | 113 ++++++++ 10 files changed, 1530 insertions(+), 32 deletions(-) create mode 100644 src/claimscene/auth.py create mode 100644 src/claimscene/jobs.py create mode 100644 tests/e2e/test_render_jobs.py create mode 100644 tests/unit/test_auth.py create mode 100644 tests/unit/test_jobs.py diff --git a/deploy/CLOUDRUN.md b/deploy/CLOUDRUN.md index 350c573..199086a 100644 --- a/deploy/CLOUDRUN.md +++ b/deploy/CLOUDRUN.md @@ -14,6 +14,7 @@ files — the Dockerfile builds it into the image). Cloud Run scales it to | Port | `8000` | | Auth | public (`--allow-unauthenticated`) | | Request timeout | `600s` (`--timeout 600` — see note) | +| CPU allocation | always-on (`--no-cpu-throttling` — see async-job note below) | > **Why 600s:** a live illustration is a two-step generation (an establish-shot > still, then an image-to-video clip chained from it) that runs for minutes. @@ -22,6 +23,15 @@ files — the Dockerfile builds it into the image). Cloud Run scales it to > so a synchronous `POST /cases/render` outlives the real generation path. > (In offline mode render is near-instant.) +> **Why `--no-cpu-throttling`:** by default Cloud Run only allocates CPU to an +> instance while it has a request in flight — the moment a response is sent, +> CPU is throttled to near-zero. `POST /cases/render/jobs` (async submit + +> poll — see below) returns its `202` immediately and keeps rendering in a +> **background thread** after that response has already gone out. Without +> `--no-cpu-throttling` that thread would barely progress between polls. This +> only affects billing while a job is actually in flight — Cloud Run still +> scales to zero (and bills nothing) when idle. + ## Prerequisites (one-time) ```bash @@ -52,6 +62,33 @@ curl -s "$URL/scenarios" | head -c 200 # committed sample scenarios curl -s -o /dev/null -w '%{http_code}\n' "$URL/" # 200 (React SPA index) ``` +## Async job submission (submit + poll) + +`POST /cases/render/jobs` / `GET /cases/render/jobs/{job_id}` exist alongside +the synchronous `POST /cases/render` for the same reason `--timeout 600` +does: edge proxies in front of Cloud Run — Firebase Hosting's rewrite proxy in +particular — cap a single request at **~60s**, well under a real two-step +generation's several minutes. The async pair splits that into a fast submit +(`202` + a job id) and a cheap poll (`GET /cases/render/jobs/{job_id}` → +`queued` / `running` / `done` / `failed`), so no single HTTP request needs to +stay open for the full render. See `src/claimscene/jobs.py` for the job-store +design (status objects live in the same B2/fake storage backend the rest of +the app already uses, under `jobs//status.json` — not a separate +database). + +**Honest limitation:** the background worker runs **in-process, on the Cloud +Run instance that accepted the submit** — there is no external queue and no +separate worker fleet. A client that keeps polling keeps that instance warm +(a request in flight resets Cloud Run's idle-scale-down clock), so in +practice the job completes. But if that specific instance is scaled down +mid-job, the in-flight generation is lost with no automatic retry — the +stored status simply stops advancing. This is a deliberate, acceptable +tradeoff for a demo, not a production job queue; a production version would +hand the work to a durable queue (e.g. Cloud Tasks) consumed by a Cloud Run +**Job** (not a request-serving instance), so the work outlives any one +instance. `--no-cpu-throttling` (above) is required even for this demo +version — without it, the worker thread barely progresses between polls. + ## Deploy — LIVE cutover (secrets via Secret Manager) The live path extracts scenes with the VLM ladder, renders real illustrations diff --git a/deploy/deploy-cloudrun.sh b/deploy/deploy-cloudrun.sh index 8bc5202..907fbaa 100644 --- a/deploy/deploy-cloudrun.sh +++ b/deploy/deploy-cloudrun.sh @@ -126,6 +126,11 @@ fi # --timeout 600 from the start: a live illustration (still + image-to-video # clip) runs for minutes; the default 300s edge deadline would 504 the # synchronous POST /cases/render while the case completes server-side. +# --no-cpu-throttling: POST /cases/render/jobs runs the actual two-step +# generation in a background thread AFTER the request that submitted it has +# already returned (see src/claimscene/jobs.py) — by default Cloud Run only +# allocates CPU while a request is in flight, which would starve that thread +# between polls. See deploy/CLOUDRUN.md for the full async-job note. gcloud run deploy "${SERVICE}" \ --image "${IMAGE}" \ --region "${REGION}" \ @@ -134,6 +139,7 @@ gcloud run deploy "${SERVICE}" \ --port 8000 \ --timeout 600 \ --cpu 1 --memory 1Gi \ + --no-cpu-throttling \ --min-instances 0 --max-instances 4 \ --set-env-vars "${ENV_VARS}" \ ${SET_SECRETS_ARGS[@]+"${SET_SECRETS_ARGS[@]}"} \ diff --git a/pyproject.toml b/pyproject.toml index b6af1ca..3926477 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,11 @@ authors = [{ name = "upgradedev" }] dependencies = [ "pydantic>=2.6", "pillow>=10.2", + # Firebase ID-token verification (claimscene.auth) — RS256 JWT decode + + # Google's x509 signing certs. No Firebase Admin SDK, no service-account + # secret. + "pyjwt>=2.8", + "cryptography>=42", ] [project.optional-dependencies] diff --git a/requirements.txt b/requirements.txt index 6be42ed..04f5038 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,3 +3,8 @@ # extra — see pyproject [live]. Keep this list pip-audit --strict clean. pydantic>=2.6 pillow>=10.2 +# Firebase ID-token verification (claimscene.auth) — RS256 JWT decode + +# Google's x509 signing certs. No Firebase Admin SDK, no service-account +# secret. +pyjwt>=2.8 +cryptography>=42 diff --git a/src/claimscene/api.py b/src/claimscene/api.py index 6602b65..1a65b4b 100644 --- a/src/claimscene/api.py +++ b/src/claimscene/api.py @@ -20,6 +20,11 @@ POST /cases/extract photos|scenario → constrained SceneGraph POST /cases/preview-schematic SceneGraph → static schematic (live review) POST /cases/render reviewed SceneGraph → sealed case +POST /cases/render/jobs submit a render as a background job (202 + + job_id) — does not block for the full + two-step generation +GET /cases/render/jobs/{job_id} poll a submitted render job's status + (queued/running/done/failed) GET /cases/{id} the sealed manifest (raw, for in-browser verify) GET /cases/{id}/verify server-side re-verification (named-check receipt) GET /cases/{id}/receipt the detached, self-sealed receipt (raw bytes) @@ -31,24 +36,37 @@ their credentials are present, and a live-provider failure degrades *this request* to the offline provider (storage untouched) — the response says so honestly and the sealed manifest records the provider that actually ran. + +Async submit + poll (``POST /cases/render/jobs`` / ``GET +/cases/render/jobs/{job_id}``): the synchronous ``POST /cases/render`` above +runs a real two-step generation (an establish-shot still, then an +image-to-video clip chained from it) that can take minutes — see +``deploy/CLOUDRUN.md``'s ``--timeout 600`` note — longer than the Firebase +Hosting proxy cap (~60s), so a caller that cannot hold one request open that +long submits a job and polls it instead. The async route runs the exact same +ingest validation and render path as the synchronous one (see +``_run_render_job`` below and the ``claimscene.jobs`` module for the job-store +design and its honest limitation — the worker runs in-process on the instance +that accepted the submit). """ from __future__ import annotations import json import logging import os +import threading import uuid from collections.abc import Callable from pathlib import Path -from typing import Annotated +from typing import Annotated, NamedTuple from urllib.parse import quote from fastapi import FastAPI, File, Form, HTTPException, UploadFile from fastapi.responses import RedirectResponse, Response from pydantic import ValidationError -from . import config, scenarios -from .adapters import FakeMediaProvider, FakeVisionExtractor +from . import config, jobs, scenarios +from .adapters import FakeMediaProvider, FakeVisionExtractor, InMemoryStorage from .case import ( CLIENT_REVIEW_CLASSIFICATIONS, CasePhoto, @@ -61,6 +79,7 @@ from .keys import safe_component from .layout import LayoutEngine from .pipeline import CasePipeline, CaseResult +from .ports import StorageBackend from .provenance import input_record, verify_all from .scene import SceneGraph, scene_to_json, semantic_warnings from .schematic import build_static_svg @@ -317,7 +336,8 @@ def _case_body(result: CaseResult, *, degraded_request: bool) -> dict: def _run_render(scene: SceneGraph, photos: list[CasePhoto], case_id: str, *, proposed: SceneGraph | None = None, reviewer_id: str | None = None, classification: ReviewClassification = ReviewClassification.interactive_demo, - prior_notes: list[str] | None = None) -> dict: + prior_notes: list[str] | None = None, + storage: StorageBackend | None = None) -> dict: """Seal the reviewed scene; degrade THIS request honestly on live failure. On a live media-provider failure the same reviewed scene + inputs are @@ -326,14 +346,21 @@ def _run_render(scene: SceneGraph, photos: list[CasePhoto], case_id: str, *, of the offline provider itself is a genuine bug and propagates. The sealed approval receipt (proposed→confirmed diff, reviewer, classification) rides on the ``CaseSpec`` so both the primary run and the degrade re-run seal it. + + ``storage`` defaults to the module-level ``_storage``. The async job + worker (``_run_render_job``) passes its own isolated storage instead (see + ``_job_storage``), so a background thread never shares mutable adapter + state with a request thread — the honest-degrade fallback below then also + writes to THAT storage, not necessarily the module-level one. """ + storage = storage if storage is not None else _storage spec = CaseSpec(case_id=case_id, photos=photos, proposed_scene=proposed, reviewer_id=reviewer_id, review_classification=classification, prior_confidence_notes=prior_notes or []) extractor = FixedSceneExtractor(scene) provider = config.build_provider() try: - result = CasePipeline(extractor, provider, _storage).run(spec) + result = CasePipeline(extractor, provider, storage).run(spec) return _case_body(result, degraded_request=False) except Exception as exc: if isinstance(provider, FakeMediaProvider): @@ -343,7 +370,7 @@ def _run_render(scene: SceneGraph, photos: list[CasePhoto], case_id: str, *, "request with the offline provider (storage unchanged)", getattr(provider, "name", "?"), case_id, ) - result = CasePipeline(extractor, FakeMediaProvider(), _storage).run(spec) + result = CasePipeline(extractor, FakeMediaProvider(), storage).run(spec) body = _case_body(result, degraded_request=True) body["degrade_reason"] = type(exc).__name__ return body @@ -457,36 +484,40 @@ def preview_schematic(scene: SceneGraph) -> dict: } -@app.post("/cases/render") -async def render_case( - scene: Annotated[str, Form()], - case_id: Annotated[str, Form()] = "case", - scenario_id: Annotated[str | None, Form()] = None, - roles: Annotated[str | None, Form()] = None, - proposed_scene: Annotated[str | None, Form()] = None, - reviewer_id: Annotated[str | None, Form()] = None, - review_classification: Annotated[str | None, Form()] = None, - files: Annotated[list[UploadFile] | None, File()] = None, -) -> dict: - """Seal a human-reviewed SceneGraph into a full, verifiable case. +class _RenderInputs(NamedTuple): + """Parsed + validated render inputs — shared by the sync (``POST + /cases/render``) and async (``POST /cases/render/jobs``) submit paths, so + both run the identical ingest + 422-vocabulary checks before any + generation starts.""" + + scene: SceneGraph + photos: list[CasePhoto] + case_id: str + proposed: SceneGraph | None + reviewer_id: str | None + classification: ReviewClassification + prior_notes: list[str] + + +async def _prepare_render( + scene: str, case_id: str, *, scenario_id: str | None, roles: str | None, + proposed_scene: str | None, reviewer_id: str | None, + review_classification: str | None, files: list[UploadFile], +) -> _RenderInputs: + """Ingest validation shared by every render path — sync AND async. Inputs are resolved (in order): a ``scenario_id`` → the committed sample views; else uploaded ``files``; else a single staged placeholder so a type-a-scene flow still seals honestly. The reviewed scene's ``confidence_notes`` are reset server-side to a human-in-the-loop note, so - only server-derived text is sealed. - - When an AI-``proposed_scene`` is supplied, the server seals a signed - approval receipt — the proposed→confirmed field diff it computes itself - (never the client's), the ``reviewer_id``, and an honesty-gated - ``review_classification``. Absent one, an honest ``unverified_no_baseline`` - receipt is sealed. The AI's real notes are carried into the receipt before - the confirmed scene's notes are reset (non-destructive). + only server-derived text is sealed. The AI's real notes are captured into + ``prior_notes`` BEFORE that reset, so an approval receipt can carry them + forward non-destructively. Raises the identical ``HTTPException`` (422 for + a hallucinated/malformed scene, 404 for an unknown scenario) either path + would raise — this runs entirely before any job is created or any + generation starts. """ - files = files or [] reviewed = _parse_scene(scene) - # Capture the AI's real notes BEFORE the human-in-the-loop reset, so the - # approval receipt can carry them forward non-destructively. prior_notes = list(reviewed.confidence_notes) reviewed.confidence_notes = [ "scene reviewed and confirmed by a human operator before rendering " @@ -513,9 +544,170 @@ async def render_case( # A unique, url-safe id per render so each sealed case is unambiguously # addressable (content-addressed keys otherwise collide on the case prefix). effective_id = f"{safe_component(case_id) or 'case'}-{uuid.uuid4().hex[:8]}" - return _run_render(reviewed, photos, effective_id, proposed=proposed, - reviewer_id=reviewer_id, classification=classification, - prior_notes=prior_notes) + return _RenderInputs(reviewed, photos, effective_id, proposed, reviewer_id, + classification, prior_notes) + + +@app.post("/cases/render") +async def render_case( + scene: Annotated[str, Form()], + case_id: Annotated[str, Form()] = "case", + scenario_id: Annotated[str | None, Form()] = None, + roles: Annotated[str | None, Form()] = None, + proposed_scene: Annotated[str | None, Form()] = None, + reviewer_id: Annotated[str | None, Form()] = None, + review_classification: Annotated[str | None, Form()] = None, + files: Annotated[list[UploadFile] | None, File()] = None, +) -> dict: + """Seal a human-reviewed SceneGraph into a full, verifiable case. + + Inputs are resolved (in order): a ``scenario_id`` → the committed sample + views; else uploaded ``files``; else a single staged placeholder so a + type-a-scene flow still seals honestly. The reviewed scene's + ``confidence_notes`` are reset server-side to a human-in-the-loop note, so + only server-derived text is sealed. + + When an AI-``proposed_scene`` is supplied, the server seals a signed + approval receipt — the proposed→confirmed field diff it computes itself + (never the client's), the ``reviewer_id``, and an honesty-gated + ``review_classification``. Absent one, an honest ``unverified_no_baseline`` + receipt is sealed. The AI's real notes are carried into the receipt before + the confirmed scene's notes are reset (non-destructive). + """ + ri = await _prepare_render( + scene, case_id, scenario_id=scenario_id, roles=roles, + proposed_scene=proposed_scene, reviewer_id=reviewer_id, + review_classification=review_classification, files=files or [], + ) + return _run_render(ri.scene, ri.photos, ri.case_id, proposed=ri.proposed, + reviewer_id=ri.reviewer_id, classification=ri.classification, + prior_notes=ri.prior_notes) + + +# ── Async render submission (submit + poll) ───────────────────────────────── +# See the ``claimscene.jobs`` module for the job-store design and its honest +# limitation (the worker below runs in-process on the accepting instance). + + +def _job_storage() -> StorageBackend: + """Storage instance a background job thread may safely read/write. + + ``InMemoryStorage`` (offline/tests) is pure in-memory with no remote + substrate — a FRESH instance here would be invisible to request threads + (including this same process's own ``GET /cases/render/jobs/{job_id}`` + reads), so the shared module-level ``_storage`` is reused; its plain + dict/list mutations are safe to share across threads under the GIL (no + read-modify-write cycle). + + ``B2Storage`` (live) is different: it keeps a mutable local ``index`` list + that a concurrent ``put`` from another thread could also be mutating. + Sharing the SAME Python object — and its local ``index`` list — between + this background thread and request-serving threads would add an avoidable + race (a shared list can be reassigned/appended by one thread while another + thread is mid-iteration over the old reference — a lost update). So live + mode gets its own fresh ``B2Storage`` per job; its constructor re-reads the + bucket's ``index.jsonl``, so nothing already-durable is lost — the bucket, + not the Python object, is the source of truth. + + Adapted from Cinemory's identical helper with one noted difference: on + Cinemory, this isolation sits on top of a ``B2Storage`` whose + ``_persist_index`` does a remote read-modify-write MERGE (by key) before + every write, so two independent instances writing around the same time + still converge. ClaimScene's ``B2Storage._persist_index`` (this repo, + ``adapters/b2_storage.py``) currently persists a blind snapshot of its own + local ``index`` with no such re-read/merge step — so two independent + ``B2Storage`` instances (e.g. a job's and a concurrent request's) writing + around the same time can still last-writer-wins clobber each other's index + rows. Giving the job its own instance still helps (its writes are no + longer interleaved with a request thread sharing the exact same mutable + object), but does not eliminate that pre-existing gap. Porting Cinemory's + merge-on-write fix into ClaimScene's adapter is tracked as a separate, + out-of-scope hardening follow-up. + """ + if isinstance(_storage, InMemoryStorage): + return _storage + return config.build_storage() # pragma: no cover - exercised only in live mode + + +def _run_render_job( + job_id: str, scene: SceneGraph, photos: list[CasePhoto], case_id: str, *, + proposed: SceneGraph | None, reviewer_id: str | None, + classification: ReviewClassification, prior_notes: list[str], +) -> None: + """Background worker thread: run the render, updating the stored status. + + Runs the SAME honest-degrade path the synchronous endpoint uses + (``_run_render``), against an isolated storage instance (see + ``_job_storage``) so this thread never mutates state a request thread also + touches. Never raises out of the thread and never produces an HTTP 500 for + a generation failure: caught here and recorded as status ``failed`` with + the exception CLASS NAME only (mirrors ``_run_render``'s + ``degrade_reason``) — the full traceback goes to the log. + """ + storage = _job_storage() + try: + jobs.mark_running(storage, job_id) + result = _run_render(scene, photos, case_id, proposed=proposed, + reviewer_id=reviewer_id, classification=classification, + prior_notes=prior_notes, storage=storage) + jobs.mark_done(storage, job_id, result) + except Exception as exc: + _log.exception("background render job %r failed", job_id) + jobs.mark_failed(storage, job_id, exc) + + +@app.post("/cases/render/jobs", status_code=202) +async def create_render_job( + scene: Annotated[str, Form()], + case_id: Annotated[str, Form()] = "case", + scenario_id: Annotated[str | None, Form()] = None, + roles: Annotated[str | None, Form()] = None, + proposed_scene: Annotated[str | None, Form()] = None, + reviewer_id: Annotated[str | None, Form()] = None, + review_classification: Annotated[str | None, Form()] = None, + files: Annotated[list[UploadFile] | None, File()] = None, +) -> dict: + """Submit a render as a background job; poll ``GET /cases/render/jobs/{id}``. + + Accepts the IDENTICAL body as ``POST /cases/render`` — same ingest + validation, same 422 constrained-vocabulary rejection, same 404 for an + unknown scenario — but returns immediately (``202``) with a job id instead + of blocking for the full two-step generation. See the module docstring and + ``claimscene.jobs`` for why (edge proxy timeouts) and for the honest + limitation of the in-process worker. + """ + ri = await _prepare_render( + scene, case_id, scenario_id=scenario_id, roles=roles, + proposed_scene=proposed_scene, reviewer_id=reviewer_id, + review_classification=review_classification, files=files or [], + ) + job_id = jobs.new_job_id() + jobs.create(_job_storage(), job_id) + threading.Thread( + target=_run_render_job, + args=(job_id, ri.scene, ri.photos, ri.case_id), + kwargs={ + "proposed": ri.proposed, "reviewer_id": ri.reviewer_id, + "classification": ri.classification, "prior_notes": ri.prior_notes, + }, + daemon=True, name=f"claimscene-job-{job_id}", + ).start() + return {"job_id": job_id, "status": "queued"} + + +@app.get("/cases/render/jobs/{job_id}") +def get_render_job(job_id: str) -> dict: + """Poll a submitted render job's status. + + 404 for an unknown job id. A stored-but-unreadable status object degrades + to the identical 404 (see ``claimscene.jobs.read``) rather than a 500 — + from the caller's side an unreadable job and a nonexistent one are the + same "nothing to show" answer. + """ + status = jobs.read(_storage, job_id) + if status is None: + raise HTTPException(404, f"no job {job_id!r}") + return status @app.get("/cases/{case_id}") diff --git a/src/claimscene/auth.py b/src/claimscene/auth.py new file mode 100644 index 0000000..9e51f3c --- /dev/null +++ b/src/claimscene/auth.py @@ -0,0 +1,288 @@ +"""Firebase ID-token verification — no Firebase Admin SDK, no service-account +secret. + +A Firebase ID token is an RS256 JWT signed by Google. It can be verified with +nothing more than Google's *public* signing certificates plus the standard +``iss``/``aud``/``exp`` claim checks — no Admin SDK, no service-account key, +nothing new to keep secret. This module is the security root of ClaimScene's +coming multitenancy: the tenant id (the token's ``sub``, Firebase's ``uid``) +must be derived ONLY from a cryptographically verified token, never trusted +from a client-supplied field (that would be an IDOR). + +Ported from our other MIT entry, Cinemory, which pioneered this pattern (and +shares the B2 storage + provenance foundation this app also builds on — see +``keys.py`` / ``adapters/b2_storage.py``). This module is intentionally +standalone — nothing else in ``claimscene`` imports it yet, and it does not +touch ``api.py``. A later change wires ``uid_from_authorization`` into the API +so an optional ``Authorization: Bearer `` header resolves a tenant; +this PR builds and adversarially tests the primitive in isolation. + +Algorithm safety (the classic RS256/HS256 "algorithm confusion" attack): the +token's own ``alg`` header is NEVER trusted to pick the verification +algorithm. ``algorithms=["RS256"]`` is hardcoded into the ``jwt.decode`` call, +and the header's ``alg`` is independently checked against that same hardcoded +value *before* any key material is even looked up — so neither ``alg: none`` +nor an HS256 token forged with the RSA public key as an HMAC "secret" can ever +reach the signature check. + +``FIREBASE_PROJECT_ID`` (the env var ``project_id`` defaults from) is a +PUBLIC value — the same Firebase/GCP project id baked into client-side +Firebase config and visible in any browser's network tab. It is not a secret, +unlike a service-account key. +""" +from __future__ import annotations + +import json +import os +import threading +import time +import urllib.request +from collections.abc import Callable, Mapping + +import jwt +from cryptography.x509 import load_pem_x509_certificate + +#: Google's public signing certificates for Firebase/Google Identity Platform +#: ID tokens — keyed by ``kid``, PEM-encoded x509 certs (NOT a JWKS document). +#: See https://firebase.google.com/docs/auth/admin/verify-id-tokens +#: ("Retrieve the current set of public keys ..."). +GOOGLE_CERTS_URL = ( + "https://www.googleapis.com/robot/v1/metadata/x509/" + "securetoken@system.gserviceaccount.com" +) + +#: The only algorithm a Firebase ID token is ever signed with. Hardcoded and +#: never derived from caller input or the token itself — see the module +#: docstring's algorithm-confusion note. +ALLOWED_ALGORITHMS = ("RS256",) + +#: Tolerance for clock skew between this host and Google's token issuer, +#: applied to every time-based claim: ``exp``/``nbf`` via PyJWT's ``leeway`` +#: kwarg, ``iat``/``auth_time`` manually (PyJWT does not validate either). +CLOCK_SKEW_LEEWAY_SECONDS = 60 + +#: Env var ``project_id`` defaults from when not passed explicitly. +PROJECT_ID_ENV_VAR = "FIREBASE_PROJECT_ID" + + +class AuthError(ValueError): + """Raised when a Firebase ID token fails verification, for any reason.""" + + +# ── Google certs (default key_source) — cached, network only when stale ──── +# The test suite never exercises the live fetch: it injects its own +# ``key_source`` backed by a locally generated RSA keypair, so verification +# logic runs fully offline in CI (see tests/unit/test_auth.py). The fetch +# itself mirrors this codebase's other lazy-import / injectable-transport +# seams for the same reason (e.g. ``claimscene.adapters.b2_storage``'s boto3 +# client, ``claimscene.adapters.vlm_extractor``'s HTTP client) — its +# real-network lines are excluded from the coverage gate the same way those +# real-path-only branches are. + +_certs_cache_lock = threading.Lock() +_certs_cache: dict[str, object] = {"certs": None, "expires_at": 0.0} +_DEFAULT_CACHE_TTL_SECONDS = 3600.0 + + +def _parse_cache_max_age(cache_control: str | None) -> float | None: + """Extract ``max-age`` (seconds) from a ``Cache-Control`` header value. + + Returns ``None`` when absent, malformed, or negative — callers then fall + back to a fixed default TTL. + """ + if not cache_control: + return None + for directive in cache_control.split(","): + name, _, value = directive.strip().partition("=") + value = value.strip() + if name.strip().lower() == "max-age" and value.lstrip("-").isdigit(): + age = float(value) + return age if age >= 0 else None + return None + + +def _default_key_source() -> Mapping[str, str]: + """Return Google's current certs as ``{kid: pem_cert}``, cached in-process. + + Honours the live response's ``Cache-Control: max-age`` when present + (falls back to a fixed TTL otherwise) so a long-running process does not + refetch on every call. + """ + now = time.time() + with _certs_cache_lock: + cached = _certs_cache["certs"] + fresh = cached is not None and now < _certs_cache["expires_at"] + if fresh: + return cached # type: ignore[return-value] + return _refresh_certs_cache(now) + + +def _refresh_certs_cache(now: float) -> Mapping[str, str]: # pragma: no cover - real network path + """Fetch Google's cert bundle over HTTPS and repopulate the cache. + + Never exercised by the test suite (which always injects an offline + ``key_source``) — this is the one real-network function in the module, + excluded from the coverage gate the same way this codebase's other + real-path-only branches are. + """ + request = urllib.request.Request(GOOGLE_CERTS_URL, headers={"Accept": "application/json"}) + with urllib.request.urlopen(request, timeout=10) as response: + body = response.read() + ttl = _parse_cache_max_age(response.headers.get("Cache-Control")) + certs = json.loads(body) + with _certs_cache_lock: + _certs_cache["certs"] = certs + _certs_cache["expires_at"] = now + (ttl or _DEFAULT_CACHE_TTL_SECONDS) + return certs + + +# ── Verification ───────────────────────────────────────────────────────── + + +def verify_firebase_id_token( + token: str, + *, + project_id: str | None = None, + key_source: Callable[[], Mapping[str, str]] | None = None, +) -> str: + """Verify a Firebase ID token and return its ``uid`` (the ``sub`` claim). + + Every check below is mandatory; ANY failure raises :class:`AuthError` — + there is no partial-success result. + + - Signature: RS256 against Google's public cert for the token's ``kid``. + - ``iss == "https://securetoken.google.com/"``. + - ``aud == project_id``. + - ``exp`` in the future, ``iat`` in the past, ``auth_time`` present and in + the past (a small clock-skew allowance applies to all three — see + :data:`CLOCK_SKEW_LEEWAY_SECONDS`). + - ``sub`` present and non-empty (this is the returned ``uid``). + - Algorithm: only ``RS256`` is ever accepted — see the module docstring. + + Args: + token: The raw JWT (the ``Authorization: Bearer `` value minus + the ``Bearer`` scheme). + project_id: Your Firebase/GCP project id — a PUBLIC value, not a + secret. Defaults to the ``FIREBASE_PROJECT_ID`` env var + (:data:`PROJECT_ID_ENV_VAR`) when omitted. + key_source: Zero-arg callable returning Google's current certs as + ``{kid: pem_cert}``. Defaults to :func:`_default_key_source` (a + live, cached fetch). Inject a fake here to verify tokens signed by + a locally generated keypair with no network — see + ``tests/unit/test_auth.py``. + + Returns: + The verified ``uid``. + + Raises: + AuthError: ``project_id`` cannot be resolved, or the token is + missing/malformed/expired/mis-signed/mis-issued/mis-audienced, or + any required claim is missing or fails its check. + """ + resolved_project_id = project_id or os.environ.get(PROJECT_ID_ENV_VAR) + if not resolved_project_id: + raise AuthError(f"no project_id given and {PROJECT_ID_ENV_VAR} is not set") + if not token or not isinstance(token, str): + raise AuthError("token is empty") + + source = key_source or _default_key_source + + try: + header = jwt.get_unverified_header(token) + except jwt.PyJWTError as exc: + raise AuthError(f"malformed token: {exc}") from exc + + # Algorithm-confusion defence: the token's ``alg`` is compared against a + # hardcoded allow-list *before* any key material is looked up, and the + # same hardcoded list (never the header's value) is what gets passed to + # jwt.decode below. This blocks both `alg: none` and an HS256 token + # forged using the RSA public key bytes as an HMAC secret. + alg = header.get("alg") + if alg not in ALLOWED_ALGORITHMS: + raise AuthError(f"unsupported algorithm {alg!r} (only RS256 is accepted)") + + kid = header.get("kid") + if not kid: + raise AuthError("token header is missing kid") + + try: + certs = source() + except Exception as exc: + raise AuthError(f"could not obtain signing keys: {exc}") from exc + + cert_pem = certs.get(kid) if certs else None + if not cert_pem: + raise AuthError(f"unknown signing key id {kid!r}") + + try: + public_key = load_pem_x509_certificate(cert_pem.encode("utf-8")).public_key() + except Exception as exc: + raise AuthError(f"invalid signing certificate for kid {kid!r}: {exc}") from exc + + issuer = f"https://securetoken.google.com/{resolved_project_id}" + try: + claims = jwt.decode( + token, + key=public_key, + algorithms=list(ALLOWED_ALGORITHMS), # hardcoded, never caller/token-derived + audience=resolved_project_id, + issuer=issuer, + leeway=CLOCK_SKEW_LEEWAY_SECONDS, + options={"require": ["exp", "iat", "sub", "auth_time"]}, + ) + except jwt.PyJWTError as exc: + raise AuthError(f"token verification failed: {exc}") from exc + + now = time.time() + _require_past_timestamp(claims, "iat", now) + _require_past_timestamp(claims, "auth_time", now) + + sub = claims.get("sub") + if not isinstance(sub, str) or not sub: + raise AuthError("token 'sub' claim is missing or empty") + return sub + + +def _require_past_timestamp(claims: Mapping[str, object], field: str, now: float) -> None: + """Raise :class:`AuthError` unless ``claims[field]`` is numeric and at or + before ``now`` (within :data:`CLOCK_SKEW_LEEWAY_SECONDS`).""" + value = claims.get(field) + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise AuthError(f"token {field!r} claim is missing or not numeric") + if value > now + CLOCK_SKEW_LEEWAY_SECONDS: + raise AuthError(f"token {field!r} claim is in the future") + + +def uid_from_authorization( + header: str | None, + *, + project_id: str | None = None, + key_source: Callable[[], Mapping[str, str]] | None = None, +) -> str | None: + """Verified ``uid`` from an ``Authorization`` header value, or ``None``. + + A missing or empty header is the clean GUEST case and returns ``None`` — + that is not an error. A header that IS present must be a well-formed + ``Bearer ``; anything else (wrong scheme, no token after the + scheme, or a present-but-invalid/expired token) raises :class:`AuthError` + so callers turn it into a 401 rather than silently downgrading a bad + credential to "anonymous". + + Args: + header: The raw ``Authorization`` header value, or ``None``. + project_id: See :func:`verify_firebase_id_token`. + key_source: See :func:`verify_firebase_id_token`. + + Returns: + The verified ``uid``, or ``None`` when ``header`` is absent/empty. + + Raises: + AuthError: ``header`` is present but not a well-formed bearer token, + or the token fails verification. + """ + if not header: + return None + parts = header.split() + if len(parts) != 2 or parts[0].lower() != "bearer" or not parts[1]: + raise AuthError("Authorization header must be 'Bearer '") + return verify_firebase_id_token(parts[1], project_id=project_id, key_source=key_source) diff --git a/src/claimscene/jobs.py b/src/claimscene/jobs.py new file mode 100644 index 0000000..e0bf7dd --- /dev/null +++ b/src/claimscene/jobs.py @@ -0,0 +1,156 @@ +"""Storage-backed job store for async render generation (the submit + poll +pattern). + +A case render runs a real two-step generation (an establish-shot still, then +an image-to-video clip chained from it) that can take minutes — see +``deploy/CLOUDRUN.md``'s ``--timeout 600`` note — which is longer than the +Firebase Hosting proxy cap (~60s) and comparable to Cloud Run's own request +ceiling. ``POST /cases/render/jobs`` (see ``claimscene.api``) avoids blocking +on that: it validates the request, writes a "queued" status object, starts the +real render in a background thread, and returns immediately with a job id. +``GET /cases/render/jobs/{job_id}`` polls that status back. + +A job's state is a small JSON object stored through the SAME +:class:`~claimscene.ports.StorageBackend` the rest of the app already uses (B2 +in live mode, the in-memory fake offline), under ``jobs//status.json`` +— NOT a separate in-process registry. That is what makes status pollable +across Cloud Run instances: the object store, not any one process's memory, is +the source of truth, so a scale-to-zero instance that never saw the submit can +still answer the poll by re-reading the key. The job id itself is a +:func:`secrets.token_urlsafe` token (unguessable, not a sequential id or a +``uuid1`` that would leak MAC address / timestamp bits). + +Ported from our other MIT entry, Cinemory, which pioneered this pattern for +its own async reel generation. + +Honest limitation — read before assuming this is durable +---------------------------------------------------------- +The background worker runs **in-process, on the Cloud Run instance that +accepted the submit** (see ``claimscene.api._run_render_job``). There is no +external queue and no separate worker fleet. A client that keeps polling keeps +that instance warm (a request in flight resets Cloud Run's idle-scale-down +clock), so in practice the job completes. But if that specific instance is +scaled down mid-job — e.g. nobody polls for a while, or Cloud Run reclaims it +for another reason — the in-flight generation is lost with no automatic retry; +the stored status simply stops advancing past whatever it last reached. This +is an acceptable, deliberate tradeoff for a demo, not a production job queue. A +production version would hand the work to a durable queue (e.g. Cloud Tasks) +consumed by a Cloud Run *Job* (not a request-serving instance), so the work +outlives any one instance. See ``deploy/CLOUDRUN.md`` for the Cloud Run +``--no-cpu-throttling`` flag this needs even for the demo version (without it, +Cloud Run throttles CPU to near-zero between requests, and the background +thread would barely make progress between polls). +""" +from __future__ import annotations + +import json +import logging +import secrets +from datetime import datetime, timezone +from typing import Any + +from .keys import safe_component +from .ports import StorageBackend + +_log = logging.getLogger("claimscene.jobs") + +#: Every job status object lives under this prefix, in its own id-named folder +#: — a namespace that can never collide with a case's own +#: ``////`` keys (different segment count; +#: see ``claimscene.keys.make_key``). +_KEY_PREFIX = "jobs" + +#: Terminal statuses a poller should stop on. +TERMINAL_STATUSES = ("done", "failed") + + +def new_job_id() -> str: + """A fresh, unguessable job id. + + ``secrets.token_urlsafe`` (a CSPRNG), never ``uuid1`` (leaks MAC address + + timestamp bits) and never a sequential counter (guessable/enumerable). + """ + return secrets.token_urlsafe(18) + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _key(job_id: str) -> str: + # job_id is server-generated (url-safe alphabet, no '/'), but a lookup by + # job_id also flows in from the URL path on GET — sanitise defensively the + # same way every other user-reachable key segment in this codebase is + # (see claimscene.keys.safe_component), so a hostile id can never be + # shaped into a key outside this job's own prefix. A legitimate id is + # untouched by this (it is already within the safe charset). + return f"{_KEY_PREFIX}/{safe_component(job_id)}/status.json" + + +def _write(storage: StorageBackend, job_id: str, status: dict) -> dict: + storage.put( + _key(job_id), + json.dumps(status, indent=2).encode("utf-8"), + content_type="application/json", + ) + return status + + +def create(storage: StorageBackend, job_id: str) -> dict: + """Seed a new job as ``queued`` and persist it. The first write for this id.""" + now = _now() + return _write(storage, job_id, { + "job_id": job_id, + "status": "queued", + "created_at": now, + "updated_at": now, + }) + + +def read(storage: StorageBackend, job_id: str) -> dict | None: + """The stored status dict, or ``None`` if unknown/unreadable. + + Honest-degrade, mirroring ``claimscene.api.verify_case``'s "unreadable + bytes never 500" contract: a job id that was never created, a storage read + error, corrupt JSON, or (defensively) JSON that isn't an object all read + the same way — ``None`` — so the caller (``GET /cases/render/jobs/{job_id}``) + can turn any of them into one honest 404 rather than leaking a stack trace. + """ + try: + parsed: Any = json.loads(storage.get(_key(job_id))) + except Exception: + return None + return parsed if isinstance(parsed, dict) else None + + +def _update(storage: StorageBackend, job_id: str, **fields: Any) -> dict: + """Read-modify-write the stored status: apply ``fields``, stamp ``updated_at``. + + Falls back to a fresh ``queued`` base if nothing was stored yet (defensive + only — the normal flow always calls :func:`create` first), so a transition + call can never itself raise for a missing predecessor object. + """ + current = read(storage, job_id) or { + "job_id": job_id, "status": "queued", "created_at": _now(), + } + current.update(fields) + current["updated_at"] = _now() + return _write(storage, job_id, current) + + +def mark_running(storage: StorageBackend, job_id: str) -> dict: + return _update(storage, job_id, status="running") + + +def mark_done(storage: StorageBackend, job_id: str, result: dict) -> dict: + return _update(storage, job_id, status="done", result=result) + + +def mark_failed(storage: StorageBackend, job_id: str, exc: BaseException) -> dict: + # Full detail goes to the log only; the stored/returned status carries the + # exception CLASS NAME alone — exception text may embed URLs/identifiers + # that don't belong in an API-readable object. Mirrors + # claimscene.api._run_render's ``degrade_reason`` (same reasoning, same + # shape). + _log.warning("job %r failed: %s: %s", job_id, type(exc).__name__, exc) + return _update(storage, job_id, status="failed", error=type(exc).__name__) diff --git a/tests/e2e/test_render_jobs.py b/tests/e2e/test_render_jobs.py new file mode 100644 index 0000000..a5556a8 --- /dev/null +++ b/tests/e2e/test_render_jobs.py @@ -0,0 +1,206 @@ +"""Integration tests for async render generation (submit + poll). + +``POST /cases/render/jobs`` accepts the identical body as ``POST +/cases/render`` but returns ``202`` + a job id immediately instead of blocking +for the full two-step generation (establish-shot still, then image-to-video +clip); ``GET /cases/render/jobs/{job_id}`` polls the stored status back. +Driven through the real FastAPI app, offline (InMemoryStorage + +FakeMediaProvider), so "background" generation is effectively instant — the +tests still poll with a short bounded timeout rather than assume one check +suffices, since the work genuinely runs on a separate thread. + +The existing synchronous endpoint (``tests/e2e/test_api.py`` and friends) is +untouched by this file and by the refactor that backs it — see +``src/claimscene/api.py``'s ``_prepare_render``/``_run_render``. + +Ported from Cinemory's equivalent suite (``tests/integration/test_api_jobs.py`` +there) for the identical submit+poll design. +""" +from __future__ import annotations + +import json +import time + +import pytest + +pytest.importorskip("fastapi") +pytest.importorskip("multipart") # python-multipart, needed for the form routes + +from fastapi.testclient import TestClient # noqa: E402 + +import claimscene.api as api # noqa: E402 +from claimscene import jobs # noqa: E402 +from claimscene.adapters import FakeMediaProvider, InMemoryStorage # noqa: E402 + +client = TestClient(api.app) + + +@pytest.fixture(autouse=True) +def _isolated_storage(monkeypatch): + """Give this file's tests their own fresh, private ``InMemoryStorage``. + + Job-status keys (``jobs//status.json`` — 3 segments, no 64-char SHA + anchor) would otherwise land in the SAME shared module-level index other + test files scan when asserting the "every stored key is content-addressed" + invariant (see ``tests/security/test_injection_path_traversal.py`` and + ``tests/e2e/test_api.py``). Those existing scans already scope by a + case-id prefix so they are not actually at risk, but isolating this file's + own submit/poll/corrupt-status assertions against a storage instance no + other test can see makes them fully order-independent from whatever any + other test file already wrote (or writes later) into the shared store. + Every function under test reads the module attribute ``api._storage`` + fresh at call time (``_job_storage``, ``_run_render``'s default, + ``_case_index_match``, the job GET handler) rather than caching it, so + patching the module attribute for the duration of each test is + sufficient — no other seam is needed. + """ + monkeypatch.setattr(api, "_storage", InMemoryStorage()) + + +def _scene(scenario_id: str = "s02_left_cross") -> dict: + return client.post("/cases/extract", data={"scenario_id": scenario_id}).json()["scene"] + + +def _poll_until_terminal(c: TestClient, job_id: str, *, timeout: float = 30.0) -> dict: + """Poll GET /cases/render/jobs/{job_id} until it reaches a terminal status. + + Unlike Cinemory's fully in-memory offline fakes, ClaimScene's DEFAULT + schematic renderer shells out to a real `ffmpeg` subprocess for the + animated schematic whenever ffmpeg is on PATH — offline still means "no + credentials", not "no subprocess". That happens for every render + (sync or async), so this is not new latency introduced by the async path; + it just means the bound here has to tolerate a real (if small) amount of + wall-clock work — including a slower first-ever ffmpeg spawn in the test + process — rather than assume the near-instant, pure-in-memory generation + Cinemory's own offline fakes give it. + """ + deadline = time.monotonic() + timeout + body: dict = {} + while time.monotonic() < deadline: + r = c.get(f"/cases/render/jobs/{job_id}") + assert r.status_code == 200 + body = r.json() + if body.get("status") in jobs.TERMINAL_STATUSES: + return body + time.sleep(0.02) + raise AssertionError( + f"job {job_id} did not reach a terminal status within {timeout}s: {body}" + ) + + +# ── submit ─────────────────────────────────────────────────────────────────── +def test_submit_returns_202_with_job_id_and_queued_status(): + scene = _scene() + r = client.post("/cases/render/jobs", + data={"scene": json.dumps(scene), "case_id": "asyncjob-submit"}) + assert r.status_code == 202 + body = r.json() + assert body["status"] == "queued" + assert isinstance(body["job_id"], str) and body["job_id"] + + +def test_submit_invalid_scene_is_422_synchronously(): + """The constrained-vocabulary gate runs BEFORE a job is created — same 422 + as the sync endpoint, no job id handed out for a request that will never + render.""" + scene = _scene() + scene["vehicles"] = [{"id": "v", "kind": "hovercraft", "color": "red"}] + r = client.post("/cases/render/jobs", data={"scene": json.dumps(scene)}) + assert r.status_code == 422 + assert "job_id" not in r.json() + + +def test_submit_malformed_scene_json_is_422_synchronously(): + r = client.post("/cases/render/jobs", data={"scene": "not json at all"}) + assert r.status_code == 422 + + +def test_submit_unknown_scenario_is_404_synchronously(): + scene = _scene() + r = client.post("/cases/render/jobs", + data={"scene": json.dumps(scene), "scenario_id": "does-not-exist"}) + assert r.status_code == 404 + + +# ── poll → done ────────────────────────────────────────────────────────────── +def test_poll_reaches_done_with_real_sealed_case_result(): + scene = _scene() + r = client.post("/cases/render/jobs", + data={"scene": json.dumps(scene), "case_id": "asyncjob-done", + "scenario_id": "s02_left_cross"}) + assert r.status_code == 202 + job_id = r.json()["job_id"] + + status = _poll_until_terminal(client, job_id) + assert status["status"] == "done" + assert status["job_id"] == job_id + assert status["created_at"] and status["updated_at"] + + result = status["result"] + assert result["case_id"].startswith("asyncjob-done-") + assert result["manifest_hash"] + assert result["degraded"] is True # offline illustration is deterministic bytes + assert "manifest_url" in result and "schematic_url" in result + + # The job's render really landed in storage — fetchable the normal way. + manifest = client.get(result["manifest_url"]).json() + assert manifest["case_id"] == result["case_id"] + + +def test_poll_with_uploaded_photos_matches_sync_shape(): + """The async path accepts real photo uploads too, not just scenario ids — + same ingest as the synchronous endpoint.""" + scene = _scene() + files = [("files", ("scene.png", api._PNG_1x1 + b"a", "image/png"))] + r = client.post("/cases/render/jobs", + data={"scene": json.dumps(scene), "case_id": "asyncjob-upload"}, + files=files) + assert r.status_code == 202 + status = _poll_until_terminal(client, r.json()["job_id"]) + assert status["status"] == "done" + manifest = client.get(status["result"]["manifest_url"]).json() + assert manifest["inputs"][0]["source"] == "user_upload" + + +# ── poll → unknown / corrupt / failed ─────────────────────────────────────── +def test_unknown_job_id_is_404(): + r = client.get("/cases/render/jobs/does-not-exist") + assert r.status_code == 404 + + +def test_corrupt_stored_status_degrades_to_404_not_500(): + """A status object that exists but doesn't parse degrades the same way an + unknown job id does (see claimscene.jobs.read) — never a 500.""" + api._storage.put("jobs/corrupt-job/status.json", b"not valid json") + r = client.get("/cases/render/jobs/corrupt-job") + assert r.status_code == 404 + + +def test_job_forced_failure_marks_status_failed_not_500(monkeypatch): + """A generation failure surfaces ONLY through the polled status — never as + an HTTP error anywhere in the submit/poll flow. Mirrors the sync path's + ``test_render_propagates_a_genuine_offline_bug`` fixture (a + FakeMediaProvider subclass that raises), except here the offline failure + has nowhere to surface as a request 500, because the generation never + happens on a request thread at all.""" + + class _BrokenFake(FakeMediaProvider): + def generate(self, **_kwargs) -> bytes: + raise RuntimeError("offline provider bug (forced for this test)") + + monkeypatch.setattr(api.config, "build_provider", lambda: _BrokenFake()) + + scene = _scene() + r = client.post("/cases/render/jobs", + data={"scene": json.dumps(scene), "case_id": "asyncjob-forced-fail"}) + # The submit itself never sees the failure — it happens later, in the + # background thread. + assert r.status_code == 202 + job_id = r.json()["job_id"] + + status = _poll_until_terminal(client, job_id) + assert status["status"] == "failed" + assert status["error"] == "RuntimeError" + assert "result" not in status + # Class name only — the raw exception text never reaches the response. + assert "forced for this test" not in json.dumps(status) diff --git a/tests/unit/test_auth.py b/tests/unit/test_auth.py new file mode 100644 index 0000000..bbea8d5 --- /dev/null +++ b/tests/unit/test_auth.py @@ -0,0 +1,490 @@ +"""Adversarial, fully-offline tests for :mod:`claimscene.auth`. + +Every token here is signed with a locally generated RSA keypair — no network, +no real Firebase project, no Google connectivity. The ``key_source`` +injection point is what makes this possible: production wires the live +Google cert fetch (``claimscene.auth._default_key_source``); tests wire a fake +exposing a self-signed x509 cert built for exactly this purpose, the same +PEM-x509-per-``kid`` shape Google's own endpoint returns. + +Ported from Cinemory's adversarial suite for the identical module (this repo +shares that auth primitive verbatim — see ``src/claimscene/auth.py``). +""" +from __future__ import annotations + +import base64 +import hashlib +import hmac +import json +import time +from datetime import datetime, timedelta, timezone + +import jwt +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.x509.oid import NameOID + +from claimscene import auth +from claimscene.auth import AuthError, uid_from_authorization, verify_firebase_id_token + +PROJECT_ID = "claimscene-test-project" +KID = "test-signing-key-1" + + +# ── Local RSA keypair + self-signed cert fixtures ─────────────────────────── +# Google's certs endpoint returns a PEM x509 certificate per kid (not a raw +# JWK) — these fixtures build the same shape locally so the suite never +# touches the network. + + +def _generate_keypair_and_cert() -> tuple[str, str]: + """Return ``(private_key_pem, public_cert_pem)`` for a fresh 2048-bit RSA + keypair wrapped in a self-signed x509 cert.""" + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "claimscene-test")]) + now = datetime.now(timezone.utc).replace(tzinfo=None) + cert = ( + x509.CertificateBuilder() + .subject_name(name) + .issuer_name(name) + .public_key(private_key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(days=1)) + .not_valid_after(now + timedelta(days=1)) + .sign(private_key, hashes.SHA256()) + ) + private_pem = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ).decode("ascii") + cert_pem = cert.public_bytes(serialization.Encoding.PEM).decode("ascii") + return private_pem, cert_pem + + +@pytest.fixture(scope="module") +def keypair() -> tuple[str, str]: + """The "legitimate" signing key — its cert is what ``key_source`` exposes + under :data:`KID`.""" + return _generate_keypair_and_cert() + + +@pytest.fixture(scope="module") +def attacker_keypair() -> tuple[str, str]: + """A second, unrelated keypair whose cert is never exposed via + ``key_source`` under any kid. Proves a signature made with the WRONG + private key is rejected even when the token claims the legitimate kid.""" + return _generate_keypair_and_cert() + + +@pytest.fixture +def key_source(keypair): + _private_pem, cert_pem = keypair + + def _source(): + return {KID: cert_pem} + + return _source + + +# ── Token construction helpers ────────────────────────────────────────────── + + +def _valid_claims(*, now: float | None = None, **overrides) -> dict: + now = time.time() if now is None else now + claims = { + "iss": f"https://securetoken.google.com/{PROJECT_ID}", + "aud": PROJECT_ID, + "sub": "firebase-uid-abc123", + "iat": int(now) - 30, + "exp": int(now) + 3600, + "auth_time": int(now) - 30, + } + claims.update(overrides) + return claims + + +def _sign(private_key_pem: str, claims: dict, *, kid: str = KID) -> str: + return jwt.encode(claims, private_key_pem, algorithm="RS256", headers={"kid": kid}) + + +def _b64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _b64url_decode(value: str) -> bytes: + padding = "=" * (-len(value) % 4) + return base64.urlsafe_b64decode(value + padding) + + +def _unverified_segments(header: dict, payload: dict) -> tuple[str, str]: + h = _b64url(json.dumps(header, separators=(",", ":")).encode("utf-8")) + p = _b64url(json.dumps(payload, separators=(",", ":")).encode("utf-8")) + return h, p + + +def _tamper_payload(token: str, **overrides) -> str: + """Re-encode the payload segment with ``overrides`` applied, WITHOUT + re-signing — the original signature no longer matches the new payload.""" + h, p, s = token.split(".") + payload = json.loads(_b64url_decode(p)) + payload.update(overrides) + new_p = _b64url(json.dumps(payload, separators=(",", ":")).encode("utf-8")) + return f"{h}.{new_p}.{s}" + + +# ── (a) Happy path ─────────────────────────────────────────────────────── + + +def test_valid_token_returns_uid(keypair, key_source): + private_pem, _cert_pem = keypair + token = _sign(private_pem, _valid_claims()) + uid = verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + assert uid == "firebase-uid-abc123" + + +# ── (b) Time-based claims: exp / iat / auth_time ───────────────────────── + + +def test_expired_token_rejected(keypair, key_source): + private_pem, _ = keypair + token = _sign( + private_pem, + _valid_claims(iat=int(time.time()) - 7200, exp=int(time.time()) - 3600), + ) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_future_iat_rejected(keypair, key_source): + private_pem, _ = keypair + far_future = int(time.time()) + 10_000 + token = _sign(private_pem, _valid_claims(iat=far_future, auth_time=far_future)) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_future_auth_time_rejected(keypair, key_source): + private_pem, _ = keypair + token = _sign(private_pem, _valid_claims(auth_time=int(time.time()) + 10_000)) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_missing_auth_time_rejected(keypair, key_source): + private_pem, _ = keypair + claims = _valid_claims() + del claims["auth_time"] + token = _sign(private_pem, claims) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_non_numeric_auth_time_rejected(keypair, key_source): + private_pem, _ = keypair + token = _sign(private_pem, _valid_claims(auth_time="not-a-number")) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_clock_skew_within_leeway_accepted(keypair, key_source): + """A few seconds of future iat/auth_time (ordinary clock skew) is + tolerated, not rejected outright.""" + private_pem, _ = keypair + now = time.time() + token = _sign(private_pem, _valid_claims(iat=int(now) + 5, auth_time=int(now) + 5)) + uid = verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + assert uid == "firebase-uid-abc123" + + +# ── (c) Audience / issuer ──────────────────────────────────────────────── + + +def test_wrong_audience_rejected(keypair, key_source): + private_pem, _ = keypair + token = _sign(private_pem, _valid_claims(aud="some-other-project")) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_wrong_issuer_rejected(keypair, key_source): + private_pem, _ = keypair + token = _sign( + private_pem, + _valid_claims(iss="https://securetoken.google.com/some-other-project"), + ) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_missing_audience_rejected(keypair, key_source): + private_pem, _ = keypair + claims = _valid_claims() + del claims["aud"] + token = _sign(private_pem, claims) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +# ── (d) Signature integrity ────────────────────────────────────────────── + + +def test_tampered_payload_rejected(keypair, key_source): + private_pem, _ = keypair + token = _sign(private_pem, _valid_claims()) + tampered = _tamper_payload(token, sub="someone-elses-uid") + with pytest.raises(AuthError): + verify_firebase_id_token(tampered, project_id=PROJECT_ID, key_source=key_source) + + +def test_signed_by_different_key_rejected(attacker_keypair, key_source): + """Signed with an attacker-controlled private key, but the token claims + the LEGITIMATE kid: key_source resolves the real public cert for that + kid, so the signature check fails even though the kid lookup succeeds.""" + attacker_private_pem, _attacker_cert_pem = attacker_keypair + token = _sign(attacker_private_pem, _valid_claims(), kid=KID) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +# ── (e) Algorithm safety: alg:none and RS256/HS256 confusion ──────────── +# +# Both hand-craft the token bytes directly instead of going through +# jwt.encode(): PyJWT (>=2.4) itself refuses to encode an HS256 token using +# an asymmetric key as the HMAC secret (raises InvalidKeyError), so the +# attack has to be built at the wire level to even exist as a token — exactly +# what a real attacker would do. Everything else about these tokens (a `kid` +# key_source actually knows about, otherwise-valid claims) is kept valid, so +# the ONLY thing that can trigger rejection is the algorithm gate itself, not +# an unrelated malformed-token/unknown-kid path. + + +def test_alg_none_rejected(key_source): + header, payload = _unverified_segments( + {"alg": "none", "typ": "JWT", "kid": KID}, _valid_claims() + ) + token = f"{header}.{payload}." # empty signature segment — the classic alg:none shape + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_hs256_confusion_with_public_key_as_secret_rejected(keypair, key_source): + """The textbook RS256->HS256 downgrade: the attacker knows the RSA + PUBLIC cert (it's public by definition) and HMAC-signs a forged token + using its PEM bytes as the HMAC "secret". A verifier that let the + token's own `alg` header pick the algorithm, or that allowed HS256 + alongside RS256, would wrongly accept this. Ours rejects it before even + reading the signature bytes, because `alg` is checked against a + hardcoded allow-list first.""" + _private_pem, cert_pem = keypair + header, payload = _unverified_segments( + {"alg": "HS256", "typ": "JWT", "kid": KID}, _valid_claims() + ) + signing_input = f"{header}.{payload}".encode("ascii") + forged_sig = hmac.new(cert_pem.encode("utf-8"), signing_input, hashlib.sha256).digest() + token = f"{header}.{payload}.{_b64url(forged_sig)}" + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_unsupported_algorithm_rejected(key_source): + header, payload = _unverified_segments( + {"alg": "ES256", "typ": "JWT", "kid": KID}, _valid_claims() + ) + token = f"{header}.{payload}.{_b64url(b'not-a-real-signature')}" + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +# ── (f) sub / kid ───────────────────────────────────────────────────────── + + +def test_missing_sub_rejected(keypair, key_source): + private_pem, _ = keypair + claims = _valid_claims() + del claims["sub"] + token = _sign(private_pem, claims) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_empty_sub_rejected(keypair, key_source): + private_pem, _ = keypair + token = _sign(private_pem, _valid_claims(sub="")) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_missing_kid_rejected(keypair, key_source): + private_pem, _ = keypair + token = jwt.encode(_valid_claims(), private_pem, algorithm="RS256") # no `kid` header + assert "kid" not in jwt.get_unverified_header(token) + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +def test_unknown_kid_rejected(keypair, key_source): + private_pem, _ = keypair + token = _sign(private_pem, _valid_claims(), kid="a-kid-key-source-has-never-heard-of") + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + + +# ── (g) Malformed input / misconfiguration ──────────────────────────────── + + +def test_garbage_token_rejected(key_source): + with pytest.raises(AuthError): + verify_firebase_id_token("not-a-jwt-at-all", project_id=PROJECT_ID, key_source=key_source) + + +def test_empty_token_rejected(key_source): + with pytest.raises(AuthError): + verify_firebase_id_token("", project_id=PROJECT_ID, key_source=key_source) + + +def test_missing_project_id_raises(monkeypatch, key_source): + monkeypatch.delenv("FIREBASE_PROJECT_ID", raising=False) + with pytest.raises(AuthError): + verify_firebase_id_token("irrelevant", project_id=None, key_source=key_source) + + +def test_project_id_defaults_from_env(monkeypatch, keypair, key_source): + private_pem, _ = keypair + monkeypatch.setenv("FIREBASE_PROJECT_ID", PROJECT_ID) + token = _sign(private_pem, _valid_claims()) + uid = verify_firebase_id_token(token, key_source=key_source) # no project_id kwarg + assert uid == "firebase-uid-abc123" + + +def test_explicit_project_id_overrides_env(monkeypatch, keypair, key_source): + private_pem, _ = keypair + monkeypatch.setenv("FIREBASE_PROJECT_ID", "a-totally-different-project") + token = _sign(private_pem, _valid_claims()) # issued for PROJECT_ID, not the env value + uid = verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=key_source) + assert uid == "firebase-uid-abc123" + + +def test_key_source_failure_becomes_autherror(keypair): + private_pem, _ = keypair + token = _sign(private_pem, _valid_claims()) + + def _broken_source(): + raise RuntimeError("network is down") + + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=_broken_source) + + +def test_invalid_certificate_becomes_autherror(keypair): + private_pem, _ = keypair + token = _sign(private_pem, _valid_claims()) + + def _bad_cert_source(): + return {KID: "not a real PEM certificate"} + + with pytest.raises(AuthError): + verify_firebase_id_token(token, project_id=PROJECT_ID, key_source=_bad_cert_source) + + +# ── uid_from_authorization ──────────────────────────────────────────────── + + +def test_uid_from_authorization_no_header_is_guest(key_source): + assert uid_from_authorization(None, project_id=PROJECT_ID, key_source=key_source) is None + + +def test_uid_from_authorization_empty_header_is_guest(key_source): + assert uid_from_authorization("", project_id=PROJECT_ID, key_source=key_source) is None + + +def test_uid_from_authorization_valid_bearer_returns_uid(keypair, key_source): + private_pem, _ = keypair + token = _sign(private_pem, _valid_claims()) + uid = uid_from_authorization(f"Bearer {token}", project_id=PROJECT_ID, key_source=key_source) + assert uid == "firebase-uid-abc123" + + +def test_uid_from_authorization_scheme_is_case_insensitive(keypair, key_source): + private_pem, _ = keypair + token = _sign(private_pem, _valid_claims()) + uid = uid_from_authorization(f"bearer {token}", project_id=PROJECT_ID, key_source=key_source) + assert uid == "firebase-uid-abc123" + + +@pytest.mark.parametrize( + "header", + [ + "Basic dXNlcjpwYXNz", + "Bearer", + "Bearer ", + "BearerNoSpaceToken", + "Bearer token extra-garbage", + " ", + ], +) +def test_uid_from_authorization_malformed_header_raises(header, key_source): + with pytest.raises(AuthError): + uid_from_authorization(header, project_id=PROJECT_ID, key_source=key_source) + + +def test_uid_from_authorization_invalid_token_raises(key_source): + with pytest.raises(AuthError): + uid_from_authorization( + "Bearer not-a-valid-jwt", project_id=PROJECT_ID, key_source=key_source + ) + + +def test_uid_from_authorization_expired_token_raises(keypair, key_source): + private_pem, _ = keypair + token = _sign( + private_pem, + _valid_claims(iat=int(time.time()) - 7200, exp=int(time.time()) - 3600), + ) + with pytest.raises(AuthError): + uid_from_authorization(f"Bearer {token}", project_id=PROJECT_ID, key_source=key_source) + + +# ── Pure helpers: Cache-Control parsing + default key_source caching ────── + + +@pytest.mark.parametrize( + ("header_value", "expected"), + [ + ("public, max-age=3600", 3600.0), + ("max-age=0", 0.0), + ("no-cache", None), + (None, None), + ("", None), + ("public, max-age=not-a-number", None), + ("max-age=-5", None), + ("private, max-age=120, must-revalidate", 120.0), + ], +) +def test_parse_cache_max_age(header_value, expected): + assert auth._parse_cache_max_age(header_value) == expected + + +def test_default_key_source_cache_hit_returns_cached_value_with_no_network(monkeypatch): + sentinel = {"cached-kid": "cached-cert-pem"} + monkeypatch.setitem(auth._certs_cache, "certs", sentinel) + monkeypatch.setitem(auth._certs_cache, "expires_at", time.time() + 3600) + + def _fail_if_called(_now): + raise AssertionError("must not hit the network on a cache hit") + + monkeypatch.setattr(auth, "_refresh_certs_cache", _fail_if_called) + + assert auth._default_key_source() is sentinel + + +def test_default_key_source_cache_miss_delegates_to_refresh(monkeypatch): + monkeypatch.setitem(auth._certs_cache, "certs", None) + monkeypatch.setitem(auth._certs_cache, "expires_at", 0.0) + + sentinel = {"fresh-kid": "fresh-cert-pem"} + monkeypatch.setattr(auth, "_refresh_certs_cache", lambda now: sentinel) + + assert auth._default_key_source() is sentinel diff --git a/tests/unit/test_jobs.py b/tests/unit/test_jobs.py new file mode 100644 index 0000000..da9b7b8 --- /dev/null +++ b/tests/unit/test_jobs.py @@ -0,0 +1,113 @@ +"""Unit tests for the async job store (``claimscene.jobs``). + +Driven directly against a :class:`~claimscene.adapters.InMemoryStorage` (the +same ``StorageBackend`` port the real B2 adapter implements) — no HTTP, no +threading, so these pin the module's own contract in isolation. The full +submit -> background-thread -> poll flow is covered separately in +``tests/e2e/test_render_jobs.py``. + +Ported from Cinemory's equivalent suite for the identical job-store design. +""" +from __future__ import annotations + +import json + +from claimscene import jobs +from claimscene.adapters import InMemoryStorage + + +def test_new_job_id_is_url_safe_and_unique(): + a = jobs.new_job_id() + b = jobs.new_job_id() + assert a != b + assert len(a) >= 16 + allowed = set("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789_-") + assert set(a) <= allowed + + +def test_create_writes_queued_status_and_is_readable_back(): + storage = InMemoryStorage() + status = jobs.create(storage, "job1") + assert status["job_id"] == "job1" + assert status["status"] == "queued" + assert status["created_at"] == status["updated_at"] + + read_back = jobs.read(storage, "job1") + assert read_back == status + # Stored through the same StorageBackend port the rest of the app uses, + # under the documented key — not a separate in-process registry. + assert storage.get("jobs/job1/status.json") + + +def test_read_unknown_job_is_none(): + storage = InMemoryStorage() + assert jobs.read(storage, "does-not-exist") is None + + +def test_read_corrupt_bytes_is_none(): + storage = InMemoryStorage() + storage.put("jobs/broken/status.json", b"not json at all") + assert jobs.read(storage, "broken") is None + + +def test_read_non_object_json_is_none(): + """Defensive: valid JSON that isn't an object (e.g. a bare number) still + degrades to None rather than being handed back as a "status".""" + storage = InMemoryStorage() + storage.put("jobs/weird/status.json", b"42") + assert jobs.read(storage, "weird") is None + + +def test_mark_running_then_done_preserves_created_at_and_advances_updated_at(): + storage = InMemoryStorage() + created = jobs.create(storage, "job2") + running = jobs.mark_running(storage, "job2") + assert running["status"] == "running" + assert running["created_at"] == created["created_at"] + + result = {"case_id": "job2", "manifest_hash": "a" * 64, "provider": "fake-media"} + done = jobs.mark_done(storage, "job2", result) + assert done["status"] == "done" + assert done["result"] == result + assert done["created_at"] == created["created_at"] + + # The poll path (GET /cases/render/jobs/{id}) reads back exactly this. + assert jobs.read(storage, "job2") == done + + +def test_mark_failed_records_exception_class_name_only(): + storage = InMemoryStorage() + jobs.create(storage, "job3") + failed = jobs.mark_failed(storage, "job3", RuntimeError("some sensitive internal detail")) + assert failed["status"] == "failed" + assert failed["error"] == "RuntimeError" + assert "result" not in failed + assert "sensitive internal detail" not in json.dumps(failed) + + +def test_mark_running_without_prior_create_seeds_a_base_status(): + """Defensive fallback: a transition call never raises for a missing + predecessor object (the normal flow always calls create() first, via + claimscene.api.create_render_job before the background thread starts).""" + storage = InMemoryStorage() + running = jobs.mark_running(storage, "orphan") + assert running["job_id"] == "orphan" + assert running["status"] == "running" + assert running["created_at"] + + +def test_hostile_job_id_key_stays_sanitised_and_namespaced(): + """A job id can never be shaped into a key outside jobs//... — the id + is sanitised the same way every other user-reachable key segment in this + codebase is (claimscene.keys.safe_component), even though in the real flow + job ids are always server-generated (secrets.token_urlsafe), never + attacker-supplied on write.""" + storage = InMemoryStorage() + jobs.create(storage, "../../evil") + keys = {row["key"] for row in storage.index} + assert keys # something was written + for key in keys: + segments = key.split("/") + assert segments[0] == "jobs" + assert ".." not in segments + assert not any(seg.startswith(".") for seg in segments)