From 879012c77677d471af9462f1ae8d0c125c339aa8 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 22 Aug 2026 23:57:37 +0200 Subject: [PATCH] feat: add controlled source ingestion and replay Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b22-9638-76d2-bbff-b7ea1770b118 --- README.md | 8 + WORK-RECORDS.md | 5 + docs/operator-guide.md | 44 ++- ...ontrolled_source_and_operation_receipts.py | 47 +++ src/sbom_nexus/api.py | 320 ++++++++++++++--- src/sbom_nexus/cli.py | 5 +- src/sbom_nexus/database.py | 17 + src/sbom_nexus/projection.py | 39 ++- src/sbom_nexus/scanner.py | 9 +- src/sbom_nexus/source_fetch.py | 234 +++++++++++++ src/sbom_nexus/storage.py | 329 +++++++++++++----- tests/test_api.py | 95 +++++ tests/test_migrations.py | 7 + tests/test_projection.py | 6 + tests/test_source_fetch.py | 58 +++ ...OM-WP-0003-controlled-source-and-replay.md | 87 +++++ 16 files changed, 1156 insertions(+), 154 deletions(-) create mode 100644 migrations/versions/0002_controlled_source_and_operation_receipts.py create mode 100644 src/sbom_nexus/source_fetch.py create mode 100644 tests/test_source_fetch.py create mode 100644 workplans/SBOM-WP-0003-controlled-source-and-replay.md diff --git a/README.md b/README.md index 5518bc2..107c1a6 100644 --- a/README.md +++ b/README.md @@ -31,5 +31,13 @@ development environments; mounted secret files are preferred for production. - `POST /sbom/{repo_slug}/ingest` - State Hub-compatible `/sbom/` snapshot, entry, repository, and licence routes +Authoritative automation may ingest a controlled `forgejo-archive-v1` +`source_ref` pinned to a full commit SHA. Nexus constructs the Forgejo URL, +streams and safely extracts the archive within configured limits, records the +archive and manifest provenance, and removes the transient directory. It does +not accept arbitrary source URLs. `Idempotency-Key` is durably enforced on +repository ingest and skip operations; a matching retry replays the original +snapshot outcome and conflicting key reuse returns HTTP 409. + See [docs/state-hub-sbom-extraction-review.md](docs/state-hub-sbom-extraction-review.md) for the extraction inventory and cutover dispositions. diff --git a/WORK-RECORDS.md b/WORK-RECORDS.md index a285482..c16e3fc 100644 --- a/WORK-RECORDS.md +++ b/WORK-RECORDS.md @@ -9,8 +9,13 @@ | Kind | ID | Status | Lane | Source | | --- | --- | --- | --- | --- | | workplan | SBOM-WP-0001 | finished | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md | +| workplan | SBOM-WP-0003 | active | — | workplans/SBOM-WP-0003-controlled-source-and-replay.md | | task | SBOM-WP-0001-T01 | done | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md | | task | SBOM-WP-0001-T02 | done | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md | | task | SBOM-WP-0001-T03 | done | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md | | task | SBOM-WP-0001-T04 | done | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md | | task | SBOM-WP-0001-T05 | done | — | workplans/SBOM-WP-0001-bootstrap-and-state-hub-extraction.md | +| task | SBOM-WP-0003-T01 | done | — | workplans/SBOM-WP-0003-controlled-source-and-replay.md | +| task | SBOM-WP-0003-T02 | done | — | workplans/SBOM-WP-0003-controlled-source-and-replay.md | +| task | SBOM-WP-0003-T03 | done | — | workplans/SBOM-WP-0003-controlled-source-and-replay.md | +| task | SBOM-WP-0003-T04 | wait | — | workplans/SBOM-WP-0003-controlled-source-and-replay.md | diff --git a/docs/operator-guide.md b/docs/operator-guide.md index bce6007..d5d76ca 100644 --- a/docs/operator-guide.md +++ b/docs/operator-guide.md @@ -42,8 +42,48 @@ curl -X POST http://127.0.0.1:8010/sbom/example/ingest ``` The ingest response is terminal: `ingested`, or `skipped` with one of -`no-checkout`, `no-manifest`, or `ingest-error`. A skip advances queue fairness -but does not advance `last_success_at`. +`no-checkout`, `no-manifest`, `ingest-error`, `source-unavailable`, or +`source-rejected`. A skip advances queue fairness but does not advance +`last_success_at`. + +## Controlled Forgejo source + +Production automation uses the contract selected by `CUST-WP-0064`: Repo +Manager projects a canonical Coulomb repository and full commit SHA, Activity +Core freezes that `source_ref` with its bounded target set, and Nexus alone +fetches and scans the source. Example projection: + +```json +{ + "source_ref": { + "kind": "forgejo-archive-v1", + "repository": "coulomb/example", + "revision": "0123456789abcdef0123456789abcdef01234567", + "observed_ref": "refs/heads/main", + "observed_at": "2026-08-22T20:00:00Z" + } +} +``` + +Enable only after the package has migration `0002`, bounded ephemeral storage, +and Forgejo/DNS egress: + +```sh +export SBOM_NEXUS_CONTROLLED_SOURCE_ENABLED=true +export SBOM_NEXUS_FORGEJO_BASE_URL=http://forgejo-gitea-http.forgejo.svc.cluster.local:3000 +export SBOM_NEXUS_SOURCE_TMP=/var/run/sbom-sources +``` + +The runtime defaults are one concurrent scan, 120 seconds each for fetch and +scan, 100 MiB compressed, 512 MiB expanded, and 100,000 archive members. The +source endpoint accepts no arbitrary URL and requires no credential for public +Coulomb repositories. Never substitute a Forgejo administrator token. + +Automated ingest and skip requests send the same value in `Idempotency-Key` +and `X-Activity-Core-Operation-ID`. Nexus persists that operation identity in +the snapshot transaction. A retry with the same request replays the original +terminal outcome; reuse against a different route, repository, reason, or +source reference returns HTTP 409. ## Inspect catch-up diff --git a/migrations/versions/0002_controlled_source_and_operation_receipts.py b/migrations/versions/0002_controlled_source_and_operation_receipts.py new file mode 100644 index 0000000..2448278 --- /dev/null +++ b/migrations/versions/0002_controlled_source_and_operation_receipts.py @@ -0,0 +1,47 @@ +"""Controlled source provenance and durable operation receipts. + +Revision ID: 0002 +Revises: 0001 +Create Date: 2026-08-22 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op + +revision = "0002" +down_revision = "0001" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column("repositories", sa.Column("source_ref_json", sa.JSON(), nullable=True)) + op.add_column("snapshots", sa.Column("source_provenance_json", sa.JSON(), nullable=True)) + op.create_table( + "operation_receipts", + sa.Column("operation_key", sa.String(length=200), nullable=False), + sa.Column("request_fingerprint", sa.String(length=64), nullable=False), + sa.Column("snapshot_id", sa.String(length=36), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["snapshot_id"], + ["snapshots.id"], + name=op.f("fk_operation_receipts_snapshot_id_snapshots"), + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("operation_key", name=op.f("pk_operation_receipts")), + ) + op.create_index( + "ix_operation_receipts_snapshot", + "operation_receipts", + ["snapshot_id"], + ) + + +def downgrade() -> None: + op.drop_index("ix_operation_receipts_snapshot", table_name="operation_receipts") + op.drop_table("operation_receipts") + op.drop_column("snapshots", "source_provenance_json") + op.drop_column("repositories", "source_ref_json") diff --git a/src/sbom_nexus/api.py b/src/sbom_nexus/api.py index 2d0c206..dd14bbc 100644 --- a/src/sbom_nexus/api.py +++ b/src/sbom_nexus/api.py @@ -1,23 +1,43 @@ """FastAPI product and State Hub compatibility surface.""" from __future__ import annotations +import hashlib +import json import os +import subprocess +import sys from datetime import datetime from pathlib import Path from typing import Any, Literal -from fastapi import FastAPI, HTTPException, Query, Request +from fastapi import FastAPI, Header, HTTPException, Query, Request from pydantic import BaseModel, Field from sbom_nexus.config import database_target from sbom_nexus.scanner import VALID_ECOSYSTEMS, scan_repository -from sbom_nexus.storage import Store +from sbom_nexus.source_fetch import ( + ControlledSourceDisabled, + SourceRejected, + SourceUnavailable, + controlled_source_enabled, + fetch_controlled_source, +) +from sbom_nexus.storage import OperationKeyConflict, Store DEFAULT_STALE_DAYS = int(os.getenv("SBOM_NEXUS_STALE_DAYS", "30")) +class SourceRef(BaseModel): + kind: Literal["forgejo-archive-v1"] + repository: str = Field(min_length=1, max_length=400) + revision: str = Field(pattern=r"^[0-9a-f]{40}$") + observed_ref: str | None = Field(default=None, max_length=300) + observed_at: datetime | None = None + + class RepositoryUpsert(BaseModel): checkout_path: str | None = None + source_ref: SourceRef | None = None active: bool = True last_sbom_at: datetime | None = None last_attempt_at: datetime | None = None @@ -50,10 +70,20 @@ class HistoricalImport(BaseModel): class SkipRequest(BaseModel): - reason: Literal["no-checkout", "no-manifest", "ingest-error"] + reason: Literal[ + "no-checkout", + "no-manifest", + "ingest-error", + "source-unavailable", + "source-rejected", + ] detail: str | None = None +class IngestRequest(BaseModel): + source_ref: SourceRef | None = None + + def _store(request: Request) -> Store: return request.app.state.store @@ -81,6 +111,96 @@ def _auto_create(store: Store) -> bool: return store.dialect == "sqlite" +def _operation_key(idempotency_key: str | None, operation_id: str | None) -> str | None: + if idempotency_key and operation_id and idempotency_key != operation_id: + raise HTTPException(status_code=400, detail="operation identity headers disagree") + key = idempotency_key or operation_id + if key and len(key) > 200: + raise HTTPException(status_code=400, detail="operation identity is too long") + return key + + +def _request_fingerprint(route: str, repo_slug: str, body: dict[str, Any]) -> str: + encoded = json.dumps( + {"route": route, "repo_slug": repo_slug, "body": body}, + sort_keys=True, + separators=(",", ":"), + ).encode() + return hashlib.sha256(encoded).hexdigest() + + +def _outcome(repo_slug: str, snapshot: dict[str, Any]) -> dict[str, Any]: + status = snapshot["status"] + if status == "ingested": + return { + "repo_slug": repo_slug, + "snapshot_id": snapshot["id"], + "status": "ingested", + "entry_count": snapshot["entry_count"], + "snapshot_at": snapshot["snapshot_at"], + "source_revision": snapshot["source_revision"], + } + result = { + "repo_slug": repo_slug, + "snapshot_id": snapshot["id"], + "status": "skipped", + "reason": status, + "entry_count": snapshot["entry_count"], + "snapshot_at": snapshot["snapshot_at"], + } + if status == "ingest-error": + result["errors"] = snapshot["errors"] + return result + + +def _replay_or_none( + store: Store, + repo_slug: str, + operation_key: str | None, + fingerprint: str, +) -> dict[str, Any] | None: + try: + snapshot = store.replay_operation(operation_key, fingerprint) + except OperationKeyConflict as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + return _outcome(repo_slug, snapshot) if snapshot else None + + +def _scan_controlled_source( + source_root: Path, repo_slug: str, revision: str +) -> dict[str, Any]: + timeout = int(os.getenv("SBOM_NEXUS_SOURCE_SCAN_TIMEOUT_SECONDS", "120")) + try: + completed = subprocess.run( + [ + sys.executable, + "-m", + "sbom_nexus.cli", + "scan", + str(source_root), + "--slug", + repo_slug, + "--source-revision", + revision, + ], + check=False, + capture_output=True, + text=True, + timeout=timeout, + ) + except subprocess.TimeoutExpired as exc: + raise SourceRejected("source scan exceeded its time limit") from exc + if completed.returncode not in {0, 1}: + raise SourceRejected("source scanner process failed") + try: + result = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise SourceRejected("source scanner returned malformed output") from exc + if not isinstance(result, dict) or result.get("source_revision") != revision: + raise SourceRejected("source scanner returned invalid provenance") + return result + + def create_app(database_path: str | Path | None = None) -> FastAPI: application = FastAPI( title="SBOM Nexus", @@ -107,6 +227,11 @@ def create_app(database_path: str | Path | None = None) -> FastAPI: return _store(request).upsert_repository( repo_slug, checkout_path=body.checkout_path, + source_ref=( + body.source_ref.model_dump(mode="json", exclude_none=True) + if body.source_ref + else None + ), active=body.active, last_attempt_at=body.last_attempt_at or legacy_time, last_success_at=body.last_success_at or legacy_time, @@ -210,63 +335,140 @@ def create_app(database_path: str | Path | None = None) -> FastAPI: ) @application.post("/sbom/{repo_slug}/ingest") - def ingest_repository(repo_slug: str, request: Request) -> dict[str, Any]: + def ingest_repository( + repo_slug: str, + request: Request, + body: IngestRequest | None = None, + idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"), + operation_id: str | None = Header( + default=None, alias="X-Activity-Core-Operation-ID" + ), + ) -> dict[str, Any]: store = _store(request) repo = store.get_repository(repo_slug) if repo is None: raise _not_found(repo_slug) - checkout_path = repo["checkout_path"] - if not checkout_path or not Path(checkout_path).is_dir(): - return _record_skip(store, repo_slug, "no-checkout") + key = _operation_key(idempotency_key, operation_id) + selected_ref = body.source_ref if body and body.source_ref else None + source_ref = ( + selected_ref.model_dump(mode="json", exclude_none=True) + if selected_ref + else repo.get("source_ref") + ) + fingerprint = _request_fingerprint( + "ingest", repo_slug, {"source_ref": source_ref} + ) + replay = _replay_or_none(store, repo_slug, key, fingerprint) + if replay: + return replay - scan = scan_repository(Path(checkout_path), slug=repo_slug) + if source_ref: + try: + with fetch_controlled_source(repo_slug, source_ref) as controlled: + scan = _scan_controlled_source( + controlled.root, + repo_slug, + source_ref["revision"], + ) + provenance = controlled.provenance + except ControlledSourceDisabled as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc + except SourceUnavailable as exc: + return _record_skip( + store, + repo_slug, + "source-unavailable", + str(exc), + operation_key=key, + request_fingerprint=fingerprint, + source_revision=source_ref["revision"], + source_provenance=source_ref, + ) + except SourceRejected as exc: + return _record_skip( + store, + repo_slug, + "source-rejected", + str(exc), + operation_key=key, + request_fingerprint=fingerprint, + source_revision=source_ref["revision"], + source_provenance=source_ref, + ) + source = "forgejo-archive-v1" + else: + provenance = None + checkout_path = repo["checkout_path"] + if not checkout_path or not Path(checkout_path).is_dir(): + reason = "source-unavailable" if controlled_source_enabled() else "no-checkout" + return _record_skip( + store, + repo_slug, + reason, + operation_key=key, + request_fingerprint=fingerprint, + ) + scan = scan_repository(Path(checkout_path), slug=repo_slug) + source = "repository-scan" if not scan["sources"] and not scan["errors"]: - return _record_skip(store, repo_slug, "no-manifest") + return _record_skip( + store, + repo_slug, + "no-manifest", + operation_key=key, + request_fingerprint=fingerprint, + source_revision=scan["source_revision"], + source_provenance=provenance, + ) if scan["errors"]: + status = "ingest-error" + else: + status = "ingested" + try: snapshot, _created = store.record_snapshot( repo_slug, entries=scan["entries"], - status="ingest-error", - source="ingest-error", + status=status, + source=source if status == "ingested" else "ingest-error", source_revision=scan["source_revision"], + source_provenance=provenance, sources=scan["sources"], errors=scan["errors"], + operation_key=key, + request_fingerprint=fingerprint if key else None, ) - return { - "repo_slug": repo_slug, - "snapshot_id": snapshot["id"], - "status": "skipped", - "reason": "ingest-error", - "entry_count": snapshot["entry_count"], - "errors": scan["errors"], - "snapshot_at": snapshot["snapshot_at"], - } - - snapshot, _created = store.record_snapshot( - repo_slug, - entries=scan["entries"], - status="ingested", - source="repository-scan", - source_revision=scan["source_revision"], - sources=scan["sources"], - ) - return { - "repo_slug": repo_slug, - "snapshot_id": snapshot["id"], - "status": "ingested", - "entry_count": snapshot["entry_count"], - "snapshot_at": snapshot["snapshot_at"], - "source_revision": snapshot["source_revision"], - } + except OperationKeyConflict as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + return _outcome(repo_slug, snapshot) @application.post("/sbom/{repo_slug}/skip") def skip_repository( - repo_slug: str, body: SkipRequest, request: Request + repo_slug: str, + body: SkipRequest, + request: Request, + idempotency_key: str | None = Header(default=None, alias="Idempotency-Key"), + operation_id: str | None = Header( + default=None, alias="X-Activity-Core-Operation-ID" + ), ) -> dict[str, Any]: store = _store(request) if store.get_repository(repo_slug) is None: raise _not_found(repo_slug) - return _record_skip(store, repo_slug, body.reason, body.detail) + key = _operation_key(idempotency_key, operation_id) + fingerprint = _request_fingerprint( + "skip", repo_slug, {"reason": body.reason, "detail": body.detail} + ) + replay = _replay_or_none(store, repo_slug, key, fingerprint) + if replay: + return replay + return _record_skip( + store, + repo_slug, + body.reason, + body.detail, + operation_key=key, + request_fingerprint=fingerprint, + ) @application.get("/sbom/{repo_slug}") def get_repo_sbom(repo_slug: str, request: Request) -> dict[str, Any]: @@ -279,21 +481,29 @@ def create_app(database_path: str | Path | None = None) -> FastAPI: def _record_skip( - store: Store, repo_slug: str, reason: str, detail: str | None = None + store: Store, + repo_slug: str, + reason: str, + detail: str | None = None, + *, + operation_key: str | None = None, + request_fingerprint: str | None = None, + source_revision: str | None = None, + source_provenance: dict[str, Any] | None = None, ) -> dict[str, Any]: errors = [{"reason": reason, "detail": detail}] if detail else [] - snapshot, _created = store.record_snapshot( - repo_slug, - entries=[], - status=reason, - source=reason, - errors=errors, - ) - return { - "repo_slug": repo_slug, - "snapshot_id": snapshot["id"], - "status": "skipped", - "reason": reason, - "entry_count": 0, - "snapshot_at": snapshot["snapshot_at"], - } + try: + snapshot, _created = store.record_snapshot( + repo_slug, + entries=[], + status=reason, + source=reason, + source_revision=source_revision, + source_provenance=source_provenance, + errors=errors, + operation_key=operation_key, + request_fingerprint=request_fingerprint if operation_key else None, + ) + except OperationKeyConflict as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + return _outcome(repo_slug, snapshot) diff --git a/src/sbom_nexus/cli.py b/src/sbom_nexus/cli.py index 3b76407..87a850c 100644 --- a/src/sbom_nexus/cli.py +++ b/src/sbom_nexus/cli.py @@ -27,7 +27,9 @@ def _serve(args: argparse.Namespace) -> None: def _scan(args: argparse.Namespace) -> None: - result = scan_repository(Path(args.path), slug=args.slug) + result = scan_repository( + Path(args.path), slug=args.slug, source_revision=args.source_revision + ) rendered = json.dumps(result, indent=2, sort_keys=True) if args.output: output = Path(args.output) @@ -55,6 +57,7 @@ def build_parser() -> argparse.ArgumentParser: scan = subparsers.add_parser("scan", help="Derive a snapshot from repository sources") scan.add_argument("path", nargs="?", default=".") scan.add_argument("--slug") + scan.add_argument("--source-revision") scan.add_argument("--output") scan.add_argument("--force", action="store_true") scan.set_defaults(handler=_scan) diff --git a/src/sbom_nexus/database.py b/src/sbom_nexus/database.py index 03de638..c7e4b08 100644 --- a/src/sbom_nexus/database.py +++ b/src/sbom_nexus/database.py @@ -37,6 +37,7 @@ repositories = Table( Column("id", String(36), primary_key=True), Column("slug", String(200), nullable=False, unique=True), Column("checkout_path", Text, nullable=True), + Column("source_ref_json", JSON, nullable=True), Column("active", Boolean, nullable=False), Column("last_attempt_at", DateTime(timezone=True), nullable=True), Column("last_success_at", DateTime(timezone=True), nullable=True), @@ -61,12 +62,27 @@ snapshots = Table( Column("status", String(50), nullable=False), Column("entry_count", Integer, nullable=False), Column("source_revision", String(200), nullable=True), + Column("source_provenance_json", JSON, nullable=True), Column("sources_json", JSON, nullable=False), Column("errors_json", JSON, nullable=False), Column("legacy_id", String(100), nullable=True, unique=True), Column("created_at", DateTime(timezone=True), nullable=False), ) +operation_receipts = Table( + "operation_receipts", + metadata, + Column("operation_key", String(200), primary_key=True), + Column("request_fingerprint", String(64), nullable=False), + Column( + "snapshot_id", + String(36), + ForeignKey("snapshots.id", ondelete="RESTRICT"), + nullable=False, + ), + Column("created_at", DateTime(timezone=True), nullable=False), +) + entries = Table( "entries", metadata, @@ -97,6 +113,7 @@ Index("ix_snapshots_repo_time", snapshots.c.repo_id, snapshots.c.snapshot_at) Index("ix_entries_snapshot", entries.c.snapshot_id) Index("ix_entries_repo", entries.c.repo_id) Index("ix_entries_license", entries.c.license_spdx) +Index("ix_operation_receipts_snapshot", operation_receipts.c.snapshot_id) def database_url(value: str | Path) -> str: diff --git a/src/sbom_nexus/projection.py b/src/sbom_nexus/projection.py index deca0dd..6256735 100644 --- a/src/sbom_nexus/projection.py +++ b/src/sbom_nexus/projection.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re import socket import urllib.parse from collections import Counter @@ -16,6 +17,38 @@ def _checkout_path(repo: dict[str, Any], host_id: str | None) -> str | None: return host_paths.get(selected_host) or repo.get("local_path") +def _forgejo_repository(remote_url: str | None) -> str | None: + if not remote_url: + return None + patterns = ( + r"^forgejo-remote:(coulomb/[a-z0-9._-]+?)(?:\.git)?$", + r"^https://forgejo\.coulomb\.social/(coulomb/[a-z0-9._-]+?)(?:\.git)?$", + r"^ssh://git@forgejo\.coulomb\.social(?::\d+)?/(coulomb/[a-z0-9._-]+?)(?:\.git)?$", + ) + for pattern in patterns: + match = re.fullmatch(pattern, remote_url) + if match: + return match.group(1) + return None + + +def _source_ref(repo: dict[str, Any]) -> dict[str, Any] | None: + repository = _forgejo_repository(repo.get("remote_url")) + revision = repo.get("git_fingerprint") + if repository != f"coulomb/{repo['slug']}": + return None + if not isinstance(revision, str) or not re.fullmatch(r"[0-9a-f]{40}", revision): + return None + observed_at = repo.get("last_state_synced_at") or repo.get("updated_at") + return { + "kind": "forgejo-archive-v1", + "repository": repository, + "revision": revision, + "observed_ref": "repo-manager:git_fingerprint", + **({"observed_at": observed_at} if observed_at else {}), + } + + def sync_repository_projections( source_url: str, target_url: str, @@ -38,13 +71,16 @@ def sync_repository_projections( } active = repo.get("status", "active") == "active" checkout_path = _checkout_path(repo, host_id) + source_ref = _source_ref(repo) projections[slug] = { "slug": slug, "active": active, "checkout_path": checkout_path, + "source_ref": source_ref, } statuses["active" if active else "inactive"] += 1 statuses["with_checkout_path" if checkout_path else "without_checkout_path"] += 1 + statuses["with_source_ref" if source_ref else "without_source_ref"] += 1 if dry_run: return { @@ -65,6 +101,7 @@ def sync_repository_projections( body={ "active": projection["active"], "checkout_path": projection["checkout_path"], + "source_ref": projection["source_ref"], }, ) @@ -78,7 +115,7 @@ def sync_repository_projections( actual = target_repositories[slug] differences = { field: {"source": expected[field], "target": actual.get(field)} - for field in ("active", "checkout_path") + for field in ("active", "checkout_path", "source_ref") if expected[field] != actual.get(field) } if differences: diff --git a/src/sbom_nexus/scanner.py b/src/sbom_nexus/scanner.py index 6dbeae2..5d6a02b 100644 --- a/src/sbom_nexus/scanner.py +++ b/src/sbom_nexus/scanner.py @@ -325,7 +325,12 @@ def _utc_now_text() -> str: return datetime.now(UTC).isoformat().replace("+00:00", "Z") -def scan_repository(repo_root: Path, *, slug: str | None = None) -> dict[str, Any]: +def scan_repository( + repo_root: Path, + *, + slug: str | None = None, + source_revision: str | None = None, +) -> dict[str, Any]: repo_root = repo_root.resolve() sources: list[dict[str, Any]] = [] entries: list[Entry] = [] @@ -353,7 +358,7 @@ def scan_repository(repo_root: Path, *, slug: str | None = None) -> dict[str, An "ok": not errors, "repo_slug": slug or repo_root.name, "repo_path": str(repo_root), - "source_revision": _head_sha(repo_root), + "source_revision": source_revision or _head_sha(repo_root), "generated_at": _utc_now_text(), "authority": "detected lockfiles and reviewed sbom-tools.yaml", "sources": sources, diff --git a/src/sbom_nexus/source_fetch.py b/src/sbom_nexus/source_fetch.py new file mode 100644 index 0000000..69ba59a --- /dev/null +++ b/src/sbom_nexus/source_fetch.py @@ -0,0 +1,234 @@ +"""Bounded retrieval and extraction for controlled Forgejo source archives.""" + +from __future__ import annotations + +import hashlib +import os +import re +import shutil +import tarfile +import tempfile +import threading +import time +import urllib.error +import urllib.parse +import urllib.request +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from pathlib import Path, PurePosixPath +from typing import Any + +SOURCE_KIND = "forgejo-archive-v1" +REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$") +SLUG_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{0,199}$") +_SOURCE_SCAN_SLOT = threading.BoundedSemaphore(value=1) + + +class SourceUnavailable(RuntimeError): + """The selected immutable source cannot currently be retrieved.""" + + +class SourceRejected(RuntimeError): + """The source reference or archive violates the controlled-input policy.""" + + +class ControlledSourceDisabled(RuntimeError): + """The controlled-source feature flag is disabled.""" + + +@dataclass(frozen=True) +class ControlledSource: + root: Path + provenance: dict[str, Any] + + +def controlled_source_enabled() -> bool: + return os.getenv("SBOM_NEXUS_CONTROLLED_SOURCE_ENABLED", "false").lower() in { + "1", + "true", + "yes", + } + + +def _positive_int(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None: + return default + try: + value = int(raw) + except ValueError as exc: + raise RuntimeError(f"{name} must be an integer") from exc + if value < 1: + raise RuntimeError(f"{name} must be positive") + return value + + +def validate_source_ref(repo_slug: str, source_ref: dict[str, Any]) -> dict[str, Any]: + if not SLUG_PATTERN.fullmatch(repo_slug): + raise SourceRejected("invalid repository slug") + kind = source_ref.get("kind") + repository = source_ref.get("repository") + revision = source_ref.get("revision") + if kind != SOURCE_KIND: + raise SourceRejected("unsupported source kind") + if repository != f"coulomb/{repo_slug}": + raise SourceRejected("source repository does not match repository slug") + if not isinstance(revision, str) or not REVISION_PATTERN.fullmatch(revision): + raise SourceRejected("source revision must be a full lowercase Git SHA") + return { + "kind": SOURCE_KIND, + "repository": repository, + "revision": revision, + **( + {"observed_ref": str(source_ref["observed_ref"])} + if source_ref.get("observed_ref") + else {} + ), + **( + {"observed_at": str(source_ref["observed_at"])} + if source_ref.get("observed_at") + else {} + ), + } + + +class _SameHostRedirect(urllib.request.HTTPRedirectHandler): + def __init__(self, allowed_scheme: str, allowed_netloc: str) -> None: + super().__init__() + self.allowed_scheme = allowed_scheme + self.allowed_netloc = allowed_netloc + + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 + target = urllib.parse.urlsplit(newurl) + if ( + target.scheme != self.allowed_scheme + or target.netloc != self.allowed_netloc + ): + raise SourceRejected("source redirect left the configured Forgejo host") + return super().redirect_request(req, fp, code, msg, headers, newurl) + + +def _download(url: str, destination: Path) -> tuple[str, int]: + max_bytes = _positive_int("SBOM_NEXUS_SOURCE_MAX_ARCHIVE_BYTES", 100 * 1024 * 1024) + timeout = _positive_int("SBOM_NEXUS_SOURCE_FETCH_TIMEOUT_SECONDS", 120) + parsed_url = urllib.parse.urlsplit(url) + if not parsed_url.hostname: + raise SourceRejected("configured Forgejo base URL has no host") + request = urllib.request.Request(url, headers={"User-Agent": "sbom-nexus/controlled-source-v1"}) + opener = urllib.request.build_opener( + _SameHostRedirect(parsed_url.scheme, parsed_url.netloc) + ) + digest = hashlib.sha256() + size = 0 + started = time.monotonic() + try: + with opener.open(request, timeout=timeout) as response, destination.open("wb") as out: + while chunk := response.read(1024 * 1024): + if time.monotonic() - started > timeout: + raise SourceRejected("source fetch exceeded its time limit") + size += len(chunk) + if size > max_bytes: + raise SourceRejected("source archive exceeded its compressed size limit") + digest.update(chunk) + out.write(chunk) + except SourceRejected: + raise + except (OSError, urllib.error.HTTPError, urllib.error.URLError) as exc: + raise SourceUnavailable("pinned Forgejo source archive is unavailable") from exc + return digest.hexdigest(), size + + +def _safe_member_path(name: str) -> PurePosixPath: + path = PurePosixPath(name) + if not name or path.is_absolute() or ".." in path.parts: + raise SourceRejected("source archive contains an unsafe path") + return path + + +def _extract(archive: Path, destination: Path) -> Path: + max_members = _positive_int("SBOM_NEXUS_SOURCE_MAX_MEMBERS", 100_000) + max_expanded = _positive_int("SBOM_NEXUS_SOURCE_MAX_EXPANDED_BYTES", 512 * 1024 * 1024) + try: + bundle = tarfile.open(archive, mode="r:gz") + except (OSError, tarfile.TarError) as exc: + raise SourceRejected("source archive is not a valid gzip-compressed tar") from exc + with bundle: + members = bundle.getmembers() + if not members or len(members) > max_members: + raise SourceRejected("source archive member count is outside the allowed range") + expanded = 0 + roots: set[str] = set() + checked: list[tuple[tarfile.TarInfo, PurePosixPath]] = [] + for member in members: + path = _safe_member_path(member.name) + roots.add(path.parts[0]) + if not (member.isdir() or member.isfile()): + raise SourceRejected("source archive contains a link or special file") + if member.isfile(): + expanded += member.size + if expanded > max_expanded: + raise SourceRejected("source archive exceeded its expanded size limit") + checked.append((member, path)) + if len(roots) != 1: + raise SourceRejected("source archive must have exactly one top-level directory") + + for member, relative in checked: + target = destination.joinpath(*relative.parts) + if member.isdir(): + target.mkdir(parents=True, exist_ok=True) + continue + target.parent.mkdir(parents=True, exist_ok=True) + source = bundle.extractfile(member) + if source is None: + raise SourceRejected("source archive member could not be read") + with source, target.open("wb") as out: + shutil.copyfileobj(source, out, length=1024 * 1024) + return destination / next(iter(roots)) + + +@contextmanager +def fetch_controlled_source( + repo_slug: str, source_ref: dict[str, Any] +) -> Iterator[ControlledSource]: + if not controlled_source_enabled(): + raise ControlledSourceDisabled("controlled source ingestion is disabled") + validated = validate_source_ref(repo_slug, source_ref) + base = os.getenv( + "SBOM_NEXUS_FORGEJO_BASE_URL", + "https://forgejo.coulomb.social", + ).rstrip("/") + parsed = urllib.parse.urlsplit(base) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise SourceRejected("configured Forgejo base URL is invalid") + repository = validated["repository"] + revision = validated["revision"] + url = f"{base}/{repository}/archive/{revision}.tar.gz" + temp_parent = Path(os.getenv("SBOM_NEXUS_SOURCE_TMP", "/tmp/sbom-nexus-sources")) + temp_parent.mkdir(parents=True, exist_ok=True) + wait_seconds = _positive_int("SBOM_NEXUS_SOURCE_SLOT_TIMEOUT_SECONDS", 240) + if not _SOURCE_SCAN_SLOT.acquire(timeout=wait_seconds): + raise SourceRejected("controlled source scan slot wait exceeded its time limit") + try: + with tempfile.TemporaryDirectory(prefix="scan-", dir=temp_parent) as raw_temp: + temp = Path(raw_temp) + archive = temp / "source.tar.gz" + extracted = temp / "extracted" + extracted.mkdir() + archive_sha256, archive_bytes = _download(url, archive) + try: + root = _extract(archive, extracted) + except SourceRejected: + raise + except OSError as exc: + raise SourceRejected("source archive could not be safely extracted") from exc + yield ControlledSource( + root=root, + provenance={ + **validated, + "archive_sha256": archive_sha256, + "archive_bytes": archive_bytes, + }, + ) + finally: + _SOURCE_SCAN_SLOT.release() diff --git a/src/sbom_nexus/storage.py b/src/sbom_nexus/storage.py index d45ddec..011b49d 100644 --- a/src/sbom_nexus/storage.py +++ b/src/sbom_nexus/storage.py @@ -2,6 +2,7 @@ from __future__ import annotations +import time import uuid from collections import defaultdict from datetime import UTC, datetime, timedelta @@ -10,10 +11,12 @@ from typing import Any from sqlalchemy import func, insert, select, update from sqlalchemy.engine import RowMapping +from sqlalchemy.exc import IntegrityError from sbom_nexus.database import ( create_database_engine, metadata, + operation_receipts, repositories, snapshots, ) @@ -23,15 +26,17 @@ from sbom_nexus.scanner import is_copyleft SUCCESS_STATUSES = frozenset({"ingested", "imported"}) +class OperationKeyConflict(ValueError): + """An operation key was reused for a different mutation request.""" + + def utc_now() -> datetime: return datetime.now(UTC) def parse_datetime(value: datetime | str) -> datetime: parsed = ( - datetime.fromisoformat(value.replace("Z", "+00:00")) - if isinstance(value, str) - else value + datetime.fromisoformat(value.replace("Z", "+00:00")) if isinstance(value, str) else value ) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=UTC) @@ -64,6 +69,7 @@ class Store: slug: str, *, checkout_path: str | None = None, + source_ref: dict[str, Any] | None = None, active: bool = True, last_attempt_at: datetime | str | None = None, last_success_at: datetime | str | None = None, @@ -72,15 +78,18 @@ class Store: attempt = parse_datetime(last_attempt_at) if last_attempt_at else None success = parse_datetime(last_success_at) if last_success_at else None with self.engine.begin() as connection: - existing = connection.execute( - select(repositories).where(repositories.c.slug == slug) - ).mappings().one_or_none() + existing = ( + connection.execute(select(repositories).where(repositories.c.slug == slug)) + .mappings() + .one_or_none() + ) if existing: connection.execute( update(repositories) .where(repositories.c.slug == slug) .values( checkout_path=checkout_path, + source_ref_json=source_ref, active=active, last_attempt_at=attempt or existing["last_attempt_at"], last_success_at=success or existing["last_success_at"], @@ -93,6 +102,7 @@ class Store: id=str(uuid.uuid4()), slug=slug, checkout_path=checkout_path, + source_ref_json=source_ref, active=active, last_attempt_at=attempt, last_success_at=success, @@ -102,25 +112,61 @@ class Store: updated_at=now, ) ) - row = connection.execute( - select(repositories).where(repositories.c.slug == slug) - ).mappings().one() + row = ( + connection.execute(select(repositories).where(repositories.c.slug == slug)) + .mappings() + .one() + ) return self._repository_dict(row) def get_repository(self, slug: str) -> dict[str, Any] | None: with self.engine.connect() as connection: - row = connection.execute( - select(repositories).where(repositories.c.slug == slug) - ).mappings().one_or_none() + row = ( + connection.execute(select(repositories).where(repositories.c.slug == slug)) + .mappings() + .one_or_none() + ) return self._repository_dict(row) if row else None def list_repositories(self) -> list[dict[str, Any]]: with self.engine.connect() as connection: - rows = connection.execute( - select(repositories).order_by(repositories.c.slug) - ).mappings().all() + rows = ( + connection.execute(select(repositories).order_by(repositories.c.slug)) + .mappings() + .all() + ) return [self._repository_dict(row) for row in rows] + def replay_operation( + self, operation_key: str | None, request_fingerprint: str | None + ) -> dict[str, Any] | None: + if operation_key is None: + return None + if request_fingerprint is None: + raise ValueError("request_fingerprint is required with operation_key") + with self.engine.connect() as connection: + receipt = ( + connection.execute( + select(operation_receipts).where( + operation_receipts.c.operation_key == operation_key + ) + ) + .mappings() + .one_or_none() + ) + if receipt is None: + return None + if receipt["request_fingerprint"] != request_fingerprint: + raise OperationKeyConflict("operation key was already used for a different request") + snapshot = ( + connection.execute( + select(snapshots).where(snapshots.c.id == receipt["snapshot_id"]) + ) + .mappings() + .one() + ) + return self._snapshot_dict(snapshot) + def record_snapshot( self, repo_slug: str, @@ -130,83 +176,174 @@ class Store: source: str, snapshot_at: datetime | str | None = None, source_revision: str | None = None, + source_provenance: dict[str, Any] | None = None, sources: list[dict[str, Any]] | None = None, errors: list[dict[str, Any]] | None = None, snapshot_id: str | None = None, legacy_id: str | None = None, + operation_key: str | None = None, + request_fingerprint: str | None = None, ) -> tuple[dict[str, Any], bool]: + if bool(operation_key) != bool(request_fingerprint): + raise ValueError("operation_key and request_fingerprint must be supplied together") timestamp = parse_datetime(snapshot_at or utc_now()) created_at = utc_now() - with self.engine.begin() as connection: - repo = connection.execute( - select(repositories).where(repositories.c.slug == repo_slug) - ).mappings().one_or_none() - if repo is None: - raise KeyError(repo_slug) - if legacy_id: - existing = connection.execute( - select(snapshots).where(snapshots.c.legacy_id == legacy_id) - ).mappings().one_or_none() - if existing: - return self._snapshot_dict(existing), False - - new_snapshot_id = snapshot_id or str(uuid.uuid4()) - connection.execute( - insert(snapshots).values( - id=new_snapshot_id, - repo_id=repo["id"], - snapshot_at=timestamp, - source=source, + try: + with self.engine.begin() as connection: + return self._record_snapshot_transaction( + connection, + repo_slug=repo_slug, + entries=entries, status=status, - entry_count=len(entries), + source=source, + timestamp=timestamp, + created_at=created_at, source_revision=source_revision, - sources_json=sources or [], - errors_json=errors or [], + source_provenance=source_provenance, + sources=sources, + errors=errors, + snapshot_id=snapshot_id, legacy_id=legacy_id, + operation_key=operation_key, + request_fingerprint=request_fingerprint, + ) + except IntegrityError: + if not operation_key or not request_fingerprint: + raise + for _attempt in range(20): + replay = self.replay_operation(operation_key, request_fingerprint) + if replay is not None: + return replay, False + time.sleep(0.05) + raise + + def _record_snapshot_transaction( + self, + connection, + *, + repo_slug: str, + entries: list[dict[str, Any]], + status: str, + source: str, + timestamp: datetime, + created_at: datetime, + source_revision: str | None, + source_provenance: dict[str, Any] | None, + sources: list[dict[str, Any]] | None, + errors: list[dict[str, Any]] | None, + snapshot_id: str | None, + legacy_id: str | None, + operation_key: str | None, + request_fingerprint: str | None, + ) -> tuple[dict[str, Any], bool]: + repo = ( + connection.execute(select(repositories).where(repositories.c.slug == repo_slug)) + .mappings() + .one_or_none() + ) + if repo is None: + raise KeyError(repo_slug) + if operation_key: + receipt = ( + connection.execute( + select(operation_receipts).where( + operation_receipts.c.operation_key == operation_key + ) + ) + .mappings() + .one_or_none() + ) + if receipt: + if receipt["request_fingerprint"] != request_fingerprint: + raise OperationKeyConflict( + "operation key was already used for a different request" + ) + existing = ( + connection.execute( + select(snapshots).where(snapshots.c.id == receipt["snapshot_id"]) + ) + .mappings() + .one() + ) + return self._snapshot_dict(existing), False + if legacy_id: + existing = ( + connection.execute(select(snapshots).where(snapshots.c.legacy_id == legacy_id)) + .mappings() + .one_or_none() + ) + if existing: + return self._snapshot_dict(existing), False + + new_snapshot_id = snapshot_id or str(uuid.uuid4()) + connection.execute( + insert(snapshots).values( + id=new_snapshot_id, + repo_id=repo["id"], + snapshot_at=timestamp, + source=source, + status=status, + entry_count=len(entries), + source_revision=source_revision, + source_provenance_json=source_provenance, + sources_json=sources or [], + errors_json=errors or [], + legacy_id=legacy_id, + created_at=created_at, + ) + ) + if entries: + connection.execute( + insert(entry_table), + [ + { + "id": str(uuid.uuid4()), + "repo_id": repo["id"], + "snapshot_id": new_snapshot_id, + "package_name": entry["package_name"], + "package_version": entry.get("package_version"), + "ecosystem": entry["ecosystem"], + "license_spdx": entry.get("license_spdx"), + "is_direct": bool(entry.get("is_direct", True)), + "is_dev": bool(entry.get("is_dev", False)), + "source_path": entry.get("source_path"), + "created_at": created_at, + } + for entry in entries + ], + ) + + previous_attempt = repo["last_attempt_at"] + is_latest = previous_attempt is None or timestamp >= parse_datetime(previous_attempt) + if is_latest: + last_success = repo["last_success_at"] + if status in SUCCESS_STATUSES: + last_success = timestamp + connection.execute( + update(repositories) + .where(repositories.c.id == repo["id"]) + .values( + last_attempt_at=timestamp, + last_success_at=last_success, + last_status=status, + last_source=source, + updated_at=created_at, + ) + ) + if operation_key: + connection.execute( + insert(operation_receipts).values( + operation_key=operation_key, + request_fingerprint=request_fingerprint, + snapshot_id=new_snapshot_id, created_at=created_at, ) ) - if entries: - connection.execute( - insert(entry_table), - [ - { - "id": str(uuid.uuid4()), - "repo_id": repo["id"], - "snapshot_id": new_snapshot_id, - "package_name": entry["package_name"], - "package_version": entry.get("package_version"), - "ecosystem": entry["ecosystem"], - "license_spdx": entry.get("license_spdx"), - "is_direct": bool(entry.get("is_direct", True)), - "is_dev": bool(entry.get("is_dev", False)), - "source_path": entry.get("source_path"), - "created_at": created_at, - } - for entry in entries - ], - ) - - previous_attempt = repo["last_attempt_at"] - is_latest = previous_attempt is None or timestamp >= parse_datetime(previous_attempt) - if is_latest: - last_success = repo["last_success_at"] - if status in SUCCESS_STATUSES: - last_success = timestamp - connection.execute( - update(repositories) - .where(repositories.c.id == repo["id"]) - .values( - last_attempt_at=timestamp, - last_success_at=last_success, - last_status=status, - last_source=source, - updated_at=created_at, - ) - ) - row = connection.execute( - select(snapshots).where(snapshots.c.id == new_snapshot_id) - ).mappings().one() + row = ( + connection.execute(select(snapshots).where(snapshots.c.id == new_snapshot_id)) + .mappings() + .one() + ) return self._snapshot_dict(row), True def list_snapshots(self, repo_slug: str | None = None) -> list[dict[str, Any]]: @@ -231,11 +368,15 @@ class Store: row = connection.execute(statement).mappings().one_or_none() if row is None: return None - entry_rows = connection.execute( - select(entry_table) - .where(entry_table.c.snapshot_id == snapshot_id) - .order_by(entry_table.c.package_name) - ).mappings().all() + entry_rows = ( + connection.execute( + select(entry_table) + .where(entry_table.c.snapshot_id == snapshot_id) + .order_by(entry_table.c.package_name) + ) + .mappings() + .all() + ) result = self._snapshot_dict(row) result["entries"] = [] for entry in entry_rows: @@ -299,9 +440,7 @@ class Store: def licence_report(self) -> dict[str, Any]: current_entries = self.latest_entries() - groups: dict[str | None, dict[str, Any]] = defaultdict( - lambda: {"count": 0, "repos": set()} - ) + groups: dict[str | None, dict[str, Any]] = defaultdict(lambda: {"count": 0, "repos": set()}) risks: list[dict[str, Any]] = [] for entry in current_entries: license_id = entry.get("license_spdx") @@ -317,9 +456,7 @@ class Store: "source_path": entry.get("source_path"), } ) - sorted_groups = sorted( - groups.items(), key=lambda item: (-item[1]["count"], item[0] or "") - ) + sorted_groups = sorted(groups.items(), key=lambda item: (-item[1]["count"], item[0] or "")) return { "groups": [ { @@ -350,8 +487,7 @@ class Store: stale = [ repo for repo in active_repositories - if repo["last_attempt_at"] is None - or parse_datetime(repo["last_attempt_at"]) < cutoff + if repo["last_attempt_at"] is None or parse_datetime(repo["last_attempt_at"]) < cutoff ] stale.sort( key=lambda repo: ( @@ -390,6 +526,7 @@ class Store: "id": row["id"], "slug": row["slug"], "checkout_path": row["checkout_path"], + "source_ref": row["source_ref_json"], "active": bool(row["active"]), "last_attempt_at": datetime_text(row["last_attempt_at"]), "last_success_at": datetime_text(row["last_success_at"]), @@ -409,6 +546,7 @@ class Store: "status": row["status"], "entry_count": row["entry_count"], "source_revision": row["source_revision"], + "source_provenance": row["source_provenance_json"], "sources": row["sources_json"] or [], "errors": row["errors_json"] or [], "legacy_id": row["legacy_id"], @@ -444,6 +582,7 @@ class Store: attempt = repo["last_attempt_at"] age_days = 9999 if attempt is None else max(0, (now - parse_datetime(attempt)).days) checkout_path = repo["checkout_path"] + source_ref = repo.get("source_ref") return { "repo_slug": repo["slug"], "last_sbom_at": attempt, @@ -451,6 +590,10 @@ class Store: "last_success_at": repo["last_success_at"], "sbom_age_days": age_days, "has_sbom": repo["last_success_at"] is not None, - "checkout_available": bool(checkout_path and Path(checkout_path).is_dir()), + "checkout_available": bool( + source_ref or (checkout_path and Path(checkout_path).is_dir()) + ), + "source_available": source_ref is not None, + "source_ref": source_ref, "last_status": repo["last_status"], } diff --git a/tests/test_api.py b/tests/test_api.py index 18ecb92..50b1d37 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,11 +1,14 @@ from __future__ import annotations +from contextlib import contextmanager from datetime import UTC, datetime, timedelta from pathlib import Path from fastapi.testclient import TestClient +from sbom_nexus import api from sbom_nexus.api import create_app +from sbom_nexus.source_fetch import ControlledSource def client_for(tmp_path: Path) -> TestClient: @@ -19,6 +22,7 @@ def register( checkout_path: str | None = None, last_sbom_at: datetime | None = None, active: bool = True, + source_ref: dict | None = None, ) -> None: response = client.put( f"/repositories/{slug}", @@ -26,6 +30,7 @@ def register( "checkout_path": checkout_path, "last_sbom_at": last_sbom_at.isoformat() if last_sbom_at else None, "active": active, + "source_ref": source_ref, }, ) assert response.status_code == 200 @@ -135,6 +140,96 @@ def test_checkout_scan_records_provenance_and_success(tmp_path: Path) -> None: assert detail["entries"][0]["snapshot_at"] == detail["snapshot_at"] +def test_ingest_operation_key_replays_without_a_second_snapshot(tmp_path: Path) -> None: + repo = tmp_path / "idempotent" + repo.mkdir() + (repo / "requirements.txt").write_text("fastapi==0.136.1\n", encoding="utf-8") + client = client_for(tmp_path) + register(client, "idempotent", checkout_path=str(repo)) + headers = { + "Idempotency-Key": "activity-operation-1", + "X-Activity-Core-Operation-ID": "activity-operation-1", + } + + first = client.post("/sbom/idempotent/ingest", headers=headers) + second = client.post("/sbom/idempotent/ingest", headers=headers) + + assert first.status_code == 200 + assert second.json() == first.json() + assert len(client.get("/sbom/snapshots/?repo_slug=idempotent").json()) == 1 + conflict = client.post( + "/sbom/idempotent/skip", + headers=headers, + json={"reason": "no-checkout"}, + ) + assert conflict.status_code == 409 + + +def test_skip_operation_key_replays_original_outcome(tmp_path: Path) -> None: + client = client_for(tmp_path) + register(client, "skip-replay") + headers = {"Idempotency-Key": "skip-operation-1"} + + first = client.post( + "/sbom/skip-replay/skip", + headers=headers, + json={"reason": "source-unavailable", "detail": "not projected"}, + ) + second = client.post( + "/sbom/skip-replay/skip", + headers=headers, + json={"reason": "source-unavailable", "detail": "not projected"}, + ) + + assert first.status_code == 200 + assert second.json() == first.json() + assert len(client.get("/sbom/snapshots/?repo_slug=skip-replay").json()) == 1 + + +def test_controlled_source_records_explicit_revision_and_archive_provenance( + tmp_path: Path, monkeypatch +) -> None: + extracted = tmp_path / "extracted" + extracted.mkdir() + (extracted / "requirements.txt").write_text( + "fastapi==0.136.1\n", encoding="utf-8" + ) + revision = "a" * 40 + source_ref = { + "kind": "forgejo-archive-v1", + "repository": "coulomb/controlled", + "revision": revision, + "observed_ref": "refs/heads/main", + "observed_at": "2026-08-22T20:00:00Z", + } + + @contextmanager + def fake_fetch(repo_slug: str, selected: dict): + assert repo_slug == "controlled" + assert selected["revision"] == revision + yield ControlledSource( + root=extracted, + provenance={**selected, "archive_sha256": "b" * 64, "archive_bytes": 42}, + ) + + monkeypatch.setattr(api, "fetch_controlled_source", fake_fetch) + client = client_for(tmp_path) + register(client, "controlled", source_ref=source_ref) + + result = client.post( + "/sbom/controlled/ingest", + json={"source_ref": source_ref}, + headers={"Idempotency-Key": "controlled-operation-1"}, + ) + + assert result.status_code == 200 + assert result.json()["source_revision"] == revision + detail = client.get(f"/sbom/snapshots/{result.json()['snapshot_id']}").json() + assert detail["source"] == "forgejo-archive-v1" + assert detail["source_provenance"]["archive_sha256"] == "b" * 64 + assert detail["sources"][0]["path"] == "requirements.txt" + + def test_historical_import_is_idempotent(tmp_path: Path) -> None: client = client_for(tmp_path) payload = { diff --git a/tests/test_migrations.py b/tests/test_migrations.py index ff6f949..c40ec5b 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -23,9 +23,16 @@ def test_initial_migration_upgrades_and_downgrades(tmp_path: Path) -> None: assert set(inspect(engine).get_table_names()) == { "alembic_version", "entries", + "operation_receipts", "repositories", "snapshots", } + assert "source_ref_json" in { + column["name"] for column in inspect(engine).get_columns("repositories") + } + assert "source_provenance_json" in { + column["name"] for column in inspect(engine).get_columns("snapshots") + } assert {index["name"] for index in inspect(engine).get_indexes("entries")} == { "ix_entries_license", "ix_entries_repo", diff --git a/tests/test_projection.py b/tests/test_projection.py index 874664a..362bdc1 100644 --- a/tests/test_projection.py +++ b/tests/test_projection.py @@ -12,6 +12,9 @@ def test_projection_sync_is_projection_only_and_reconciles(monkeypatch) -> None: "status": "active", "local_path": "/fallback/active", "host_paths": {"build-host": "/repos/active"}, + "remote_url": "forgejo-remote:coulomb/active-repo.git", + "git_fingerprint": "a" * 40, + "last_state_synced_at": "2026-08-22T20:00:00Z", }, { "slug": "retired-repo", @@ -51,10 +54,13 @@ def test_projection_sync_is_projection_only_and_reconciles(monkeypatch) -> None: assert result["counts"] == { "active": 1, "with_checkout_path": 2, + "with_source_ref": 1, "inactive": 1, + "without_source_ref": 1, } assert result["reconciliation"]["matched_repo_count"] == 2 assert target["active-repo"]["checkout_path"] == "/repos/active" + assert target["active-repo"]["source_ref"]["revision"] == "a" * 40 assert target["retired-repo"]["active"] is False assert all("/sbom/" not in path for _, _, path in calls) diff --git a/tests/test_source_fetch.py b/tests/test_source_fetch.py new file mode 100644 index 0000000..756b64a --- /dev/null +++ b/tests/test_source_fetch.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +import io +import tarfile +from pathlib import Path + +import pytest + +from sbom_nexus.source_fetch import SourceRejected, _extract, validate_source_ref + + +def _archive(path: Path, members: dict[str, bytes]) -> None: + with tarfile.open(path, "w:gz") as bundle: + for name, content in members.items(): + info = tarfile.TarInfo(name) + info.size = len(content) + bundle.addfile(info, io.BytesIO(content)) + + +def test_validate_source_ref_binds_repository_to_slug() -> None: + revision = "a" * 40 + assert validate_source_ref( + "demo", + { + "kind": "forgejo-archive-v1", + "repository": "coulomb/demo", + "revision": revision, + }, + )["revision"] == revision + with pytest.raises(SourceRejected, match="does not match"): + validate_source_ref( + "demo", + { + "kind": "forgejo-archive-v1", + "repository": "coulomb/other", + "revision": revision, + }, + ) + + +def test_extract_rejects_path_traversal(tmp_path: Path) -> None: + archive = tmp_path / "unsafe.tar.gz" + _archive(archive, {"repo/../../escaped": b"nope"}) + + with pytest.raises(SourceRejected, match="unsafe path"): + _extract(archive, tmp_path / "out") + + +def test_extract_accepts_one_regular_root(tmp_path: Path) -> None: + archive = tmp_path / "safe.tar.gz" + _archive(archive, {"repo/requirements.txt": b"fastapi==0.136.1\n"}) + destination = tmp_path / "out" + destination.mkdir() + + root = _extract(archive, destination) + + assert root == destination / "repo" + assert (root / "requirements.txt").read_text() == "fastapi==0.136.1\n" diff --git a/workplans/SBOM-WP-0003-controlled-source-and-replay.md b/workplans/SBOM-WP-0003-controlled-source-and-replay.md new file mode 100644 index 0000000..0d633e7 --- /dev/null +++ b/workplans/SBOM-WP-0003-controlled-source-and-replay.md @@ -0,0 +1,87 @@ +--- +id: SBOM-WP-0003 +type: workplan +title: "Controlled Forgejo source ingestion and durable operation replay" +domain: infotech +repo: sbom-nexus +status: active +owner: codex +topic_slug: infotech +created: "2026-08-22" +updated: "2026-08-22" +quality_dor: DoR-Ok +quality_dor_at: "2026-08-22" +quality_dor_by: codex +quality_dor_note: "CUST-WP-0064 selected a full-SHA public Forgejo archive contract with bounded extraction, explicit provenance, owner handoffs, idempotency, failure semantics, acceptance evidence, and rollback." +parent_workplan: CUST-WP-0064 +related: + - CUST-IN-0013 + - ACTIVITY-WP-0033 + - RMGR-WP-0011 +--- + +# Controlled Forgejo source ingestion and durable operation replay + +## Implement durable operation receipts + +```task +id: SBOM-WP-0003-T01 +status: done +priority: high +``` + +Enforce supplied `Idempotency-Key` / `X-Activity-Core-Operation-ID` values on +repository ingest and skip. Persist a request fingerprint and snapshot link in +the same transaction, replay the original terminal outcome, and reject key +reuse for a different operation. + +Completed with migration `0002`, transactional operation receipts, early +replay before source work, request-conflict HTTP 409 behavior, and ingest/skip +tests proving one snapshot across duplicate requests. + +## Add controlled full-SHA source ingestion + +```task +id: SBOM-WP-0003-T02 +status: done +priority: high +``` + +Consume the `forgejo-archive-v1` source reference selected in +`the-custodian/docs/sbom-controlled-scan-input-contract-v1.md`. Validate the +identity, stream and safely extract within fixed limits, pass the explicit +revision into the scanner, persist archive provenance, and always clean up. + +Completed with strict Coulomb identity/full-SHA validation, same-host fetches, +streaming compressed limits, safe regular-file-only extraction, one scan slot, +subprocess scan timeout, explicit revision override, archive provenance, and +temporary-directory cleanup. A real Forgejo archive scan produced 33 entries +from one manifest with zero errors. + +## Extend repository projection and outcomes + +```task +id: SBOM-WP-0003-T03 +status: done +priority: high +``` + +Store and return source references in repository/catch-up projections. Add +terminal `source-unavailable` and `source-rejected` outcomes without changing +oldest-N ranking or success-time semantics. + +Completed in the repository projection, API model, storage schema, catch-up +response, and additive skip handling. Legacy checkout scanning remains +available for local/operator compatibility while the production flag is dark. + +## Prove package integration and production behavior + +```task +id: SBOM-WP-0003-T04 +status: wait +priority: high +``` + +Coordinate the schema migration, ephemeral volume, Forgejo-only egress, and +feature flag with `rapp-sbom-nexus`; then pass unit/integration tests and the +attended plus scheduled production proof owned by CUST-WP-0064.