From e6cc18bf182d8730e6dcadec138b03c750c0d006 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 22 Aug 2026 23:34:01 +0200 Subject: [PATCH] feat(sbom): project immutable Forgejo source refs Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d --- docs/sbom-nexus-client-contract_v1.md | 35 +++ src/repo_manager/cli.py | 66 +++++ src/repo_manager/sbom_client.py | 20 +- src/repo_manager/source_ref.py | 174 +++++++++++++ tests/test_source_ref.py | 234 ++++++++++++++++++ ...GR-WP-0011-sbom-nexus-production-client.md | 17 ++ 6 files changed, 543 insertions(+), 3 deletions(-) create mode 100644 src/repo_manager/source_ref.py create mode 100644 tests/test_source_ref.py diff --git a/docs/sbom-nexus-client-contract_v1.md b/docs/sbom-nexus-client-contract_v1.md index 404f236..08a45f4 100644 --- a/docs/sbom-nexus-client-contract_v1.md +++ b/docs/sbom-nexus-client-contract_v1.md @@ -96,6 +96,41 @@ that the mutation may already have committed. The checkout path sent to Nexus is Nexus-local controlled-source identity, never authorization to mount a workstation path into the cluster. +## Forgejo archive source-reference projection + +The `CUST-WP-0064` v1 source mechanism is available through a read-only +resolution command: + +```bash +rmgr sbom source-ref --path /path/to/checkout --slug example +``` + +Repo Manager reads the selected Git remote only to normalize identity. It then +uses the anonymous Forgejo API—not local `HEAD`, the working tree, or a cached +remote-tracking ref—to read repository visibility, default branch, and its full +40-character commit SHA. The result contains `kind: forgejo-archive-v1`, exact +`coulomb/` identity, `revision`, `observed_ref`, and a canonical UTC +`observed_at`. + +V1 returns `supported: false` and no `source_ref` for a missing remote, +non-Coulomb identity, slug mismatch, private repository, unavailable Forgejo +record, invalid branch response, or unresolved full SHA. It never falls back +to a workstation path or local revision. + +Projection is an explicit mutation and requires both service configuration and +confirmation: + +```bash +export SBOM_NEXUS_URL=https://sbom-nexus.example +rmgr sbom source-ref --path /path/to/checkout --slug example \ + --project --confirm-authoritative +``` + +The projection sends `checkout_path: null` plus the structured source reference. +It succeeds only when Nexus returns the exact same `source_ref`; an older server +that silently ignores the additive field fails closed. This makes the command a +direct readiness probe for the SBOM Nexus side of `CUST-WP-0064`. + ## Repository source identity and provenance Repo Manager remains authoritative for repository slug, active state, diff --git a/src/repo_manager/cli.py b/src/repo_manager/cli.py index 1fd6aa5..25cbcf2 100644 --- a/src/repo_manager/cli.py +++ b/src/repo_manager/cli.py @@ -361,6 +361,18 @@ def main(argv: list[str] | None = None) -> int: ) p_sbom_report.add_argument("--path", default=".") p_sbom_report.add_argument("--slug", default=None) + p_sbom_source = sbom_sub.add_parser( + "source-ref", + help="Resolve a controlled Forgejo source reference and optionally project it", + ) + p_sbom_source.add_argument("--path", default=".") + p_sbom_source.add_argument("--slug", default=None) + p_sbom_source.add_argument("--remote", default="origin") + p_sbom_source.add_argument( + "--forgejo-base", default="https://forgejo.coulomb.social" + ) + p_sbom_source.add_argument("--project", action="store_true") + p_sbom_source.add_argument("--confirm-authoritative", action="store_true") p_authority = sub.add_parser("authority", help="Resolve the one authoritative record owner") authority_sub = p_authority.add_subparsers(dest="authority_command") @@ -811,6 +823,60 @@ def main(argv: list[str] | None = None) -> int: if not args.sbom_command: p_sbom.print_help() return 2 + if args.sbom_command == "source-ref": + from repo_manager.sbom_client import ( + SBOMNexusClient, + SBOMNexusConfig, + SBOMServiceError, + ) + from repo_manager.source_ref import ForgejoSourceResolver + + resolution = ForgejoSourceResolver(base_url=args.forgejo_base).resolve( + Path(args.path), + repo_slug=args.slug, + remote_name=args.remote, + ) + if not args.project: + print(json.dumps(resolution, indent=2)) + return 0 if resolution.get("supported") else 1 + if not args.confirm_authoritative: + print( + json.dumps( + { + "ok": False, + "error": "--project requires --confirm-authoritative", + "resolution": resolution, + }, + indent=2, + ) + ) + return 2 + if not resolution.get("supported"): + print(json.dumps(resolution, indent=2)) + return 1 + try: + client = SBOMNexusClient(SBOMNexusConfig.from_environment()) + projection = client.upsert_repository( + resolution["repo_slug"], + nexus_checkout_path=None, + source_ref=resolution["source_ref"], + ) + except (ValueError, SBOMServiceError) as exc: + error = exc.to_dict() if isinstance(exc, SBOMServiceError) else {"code": "config_error"} + print( + json.dumps( + {"ok": False, "error": error, "resolution": resolution}, + indent=2, + ) + ) + return 1 + print( + json.dumps( + {"ok": True, "resolution": resolution, "projection": projection}, + indent=2, + ) + ) + return 0 from repo_manager.sbom_client import ( licence_report_from_snapshot, scan_repository_via_nexus, diff --git a/src/repo_manager/sbom_client.py b/src/repo_manager/sbom_client.py index ba52c7d..d31447b 100644 --- a/src/repo_manager/sbom_client.py +++ b/src/repo_manager/sbom_client.py @@ -129,16 +129,30 @@ class SBOMNexusClient: *, nexus_checkout_path: str | None, active: bool = True, + source_ref: dict[str, Any] | None = None, ) -> dict[str, Any]: """Project identity/path; the path must resolve inside Nexus's controlled source plane.""" - return self._request_object( + body: dict[str, Any] = {"checkout_path": nexus_checkout_path, "active": active} + if source_ref is not None: + body["source_ref"] = source_ref + result = self._request_object( "PUT", f"/repositories/{quote(repo_slug, safe='')}", - body={"checkout_path": nexus_checkout_path, "active": active}, + body=body, operation="repository-projection", writes_state=True, - required_fields=frozenset({"slug", "active"}), + required_fields=frozenset( + {"slug", "active"} | ({"source_ref"} if source_ref is not None else set()) + ), ) + if source_ref is not None and result.get("source_ref") != source_ref: + raise SBOMServiceError( + "source_ref_projection_mismatch", + "PUT", + f"/repositories/{quote(repo_slug, safe='')}", + mutation_may_have_committed=True, + ) + return result def ingest_repository( self, diff --git a/src/repo_manager/source_ref.py b/src/repo_manager/source_ref.py new file mode 100644 index 0000000..7de1001 --- /dev/null +++ b/src/repo_manager/source_ref.py @@ -0,0 +1,174 @@ +"""Controlled Forgejo source references for authoritative SBOM ingestion.""" + +from __future__ import annotations + +import re +import subprocess +from collections.abc import Callable +from pathlib import Path +from typing import Any +from urllib.parse import quote, urlparse + +import httpx + +from repo_manager.time import utc_now_text + +DEFAULT_FORGEJO_BASE_URL = "https://forgejo.coulomb.social" +SOURCE_REF_KIND = "forgejo-archive-v1" +_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$") +_SHA_RE = re.compile(r"^[0-9a-f]{40}$") +_SCP_REMOTE_RE = re.compile(r"^(?:[^@]+@)?(?P[^:]+):(?P.+)$") +_ALLOWED_REMOTE_HOSTS = frozenset({"forgejo.coulomb.social", "forgejo-remote"}) + + +def canonical_coulomb_repository(remote_url: str, repo_slug: str) -> str | None: + """Normalize accepted Forgejo remotes to the strict public v1 identity.""" + if not _SLUG_RE.fullmatch(repo_slug): + return None + remote_url = remote_url.strip() + parsed = urlparse(remote_url) + if "://" in remote_url: + if parsed.scheme not in {"http", "https", "ssh"}: + return None + if ( + parsed.username not in {None, "git"} + or parsed.password + or parsed.query + or parsed.fragment + ): + return None + host = parsed.hostname + path = parsed.path + else: + match = _SCP_REMOTE_RE.fullmatch(remote_url) + if not match: + return None + host = match.group("host") + path = match.group("path") + if host not in _ALLOWED_REMOTE_HOSTS: + return None + parts = path.strip("/").removesuffix(".git").split("/") + if parts != ["coulomb", repo_slug]: + return None + return f"coulomb/{repo_slug}" + + +class ForgejoSourceResolver: + """Observe a public default-branch commit without using workstation Git state.""" + + def __init__( + self, + *, + base_url: str = DEFAULT_FORGEJO_BASE_URL, + timeout_seconds: float = 10.0, + transport: httpx.BaseTransport | None = None, + now: Callable[[], str] = utc_now_text, + ) -> None: + parsed = httpx.URL(base_url) + if parsed.scheme not in {"http", "https"} or not parsed.host: + raise ValueError("Forgejo base_url must be an absolute HTTP(S) URL") + if parsed.username or parsed.password or parsed.query or parsed.fragment: + raise ValueError("Forgejo base_url must not contain credentials, query, or fragment") + if not 0.1 <= timeout_seconds <= 60.0: + raise ValueError("Forgejo timeout_seconds must be between 0.1 and 60") + self.base_url = base_url.rstrip("/") + self.timeout_seconds = timeout_seconds + self._transport = transport + self._now = now + + def resolve( + self, + repo_root: Path, + *, + repo_slug: str | None = None, + remote_name: str = "origin", + ) -> dict[str, Any]: + repo_root = repo_root.expanduser().resolve() + slug = repo_slug or repo_root.name + if not _SLUG_RE.fullmatch(slug): + return self._unsupported(slug, "invalid-repo-slug") + + remote = subprocess.run( + ["git", "remote", "get-url", remote_name], + cwd=repo_root, + capture_output=True, + text=True, + check=False, + ) + if remote.returncode != 0: + return self._unsupported(slug, "missing-canonical-remote") + repository = canonical_coulomb_repository(remote.stdout, slug) + if repository is None: + return self._unsupported(slug, "non-coulomb-or-identity-mismatch") + + repo_path = f"/api/v1/repos/coulomb/{quote(slug, safe='')}" + metadata = self._get_object(repo_path) + if metadata is None: + return self._unsupported(slug, "source-unresolvable") + if metadata.get("private") is True: + return self._unsupported(slug, "private-repository") + if metadata.get("full_name") != repository: + return self._unsupported(slug, "forgejo-identity-mismatch") + branch = metadata.get("default_branch") + if not isinstance(branch, str) or not branch or branch.startswith("/"): + return self._unsupported(slug, "default-branch-unresolvable") + + observed_ref = f"refs/heads/{branch}" + ref_path = f"{repo_path}/git/refs/heads/{quote(branch, safe='')}" + refs = self._get_json(ref_path) + if isinstance(refs, dict): + refs = [refs] + if not isinstance(refs, list): + return self._unsupported(slug, "default-branch-unresolvable") + matching = next( + (item for item in refs if isinstance(item, dict) and item.get("ref") == observed_ref), + None, + ) + obj = matching.get("object") if matching else None + revision = obj.get("sha") if isinstance(obj, dict) else None + if not isinstance(revision, str) or not _SHA_RE.fullmatch(revision): + return self._unsupported(slug, "default-branch-unresolvable") + + return { + "ok": True, + "supported": True, + "repo_slug": slug, + "source_ref": { + "kind": SOURCE_REF_KIND, + "repository": repository, + "revision": revision, + "observed_ref": observed_ref, + "observed_at": self._now(), + }, + } + + def _get_object(self, path: str) -> dict[str, Any] | None: + payload = self._get_json(path) + return payload if isinstance(payload, dict) else None + + def _get_json(self, path: str) -> Any | None: + try: + with httpx.Client( + base_url=self.base_url, + timeout=self.timeout_seconds, + transport=self._transport, + ) as client: + response = client.get(path) + except httpx.RequestError: + return None + if response.status_code != 200: + return None + try: + return response.json() + except ValueError: + return None + + @staticmethod + def _unsupported(repo_slug: str, reason: str) -> dict[str, Any]: + return { + "ok": True, + "supported": False, + "repo_slug": repo_slug, + "reason": reason, + "source_ref": None, + } diff --git a/tests/test_source_ref.py b/tests/test_source_ref.py new file mode 100644 index 0000000..fd750cb --- /dev/null +++ b/tests/test_source_ref.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import json +import subprocess +from pathlib import Path + +import httpx +import pytest + +from repo_manager.cli import main +from repo_manager.sbom_client import SBOMNexusClient, SBOMNexusConfig, SBOMServiceError +from repo_manager.source_ref import ForgejoSourceResolver, canonical_coulomb_repository + + +def _git(repo: Path, *args: str) -> None: + subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True) + + +def _repo(tmp_path: Path, remote: str) -> Path: + repo = tmp_path / "demo" + repo.mkdir() + _git(repo, "init") + _git(repo, "remote", "add", "origin", remote) + return repo + + +@pytest.mark.parametrize( + "remote", + [ + "forgejo-remote:coulomb/demo.git", + "git@forgejo.coulomb.social:coulomb/demo.git", + "ssh://git@forgejo.coulomb.social/coulomb/demo.git", + "https://forgejo.coulomb.social/coulomb/demo.git", + ], +) +def test_normalizes_only_canonical_coulomb_repository(remote: str) -> None: + assert canonical_coulomb_repository(remote, "demo") == "coulomb/demo" + assert canonical_coulomb_repository(remote, "other") is None + + +def test_rejects_non_coulomb_and_untrusted_remote_hosts() -> None: + assert ( + canonical_coulomb_repository("https://forgejo.coulomb.social/other/demo.git", "demo") + is None + ) + assert canonical_coulomb_repository("https://github.com/coulomb/demo.git", "demo") is None + + +def test_resolves_public_default_branch_from_forgejo_not_local_head(tmp_path: Path) -> None: + repo = _repo(tmp_path, "forgejo-remote:coulomb/demo.git") + revision = "a" * 40 + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/api/v1/repos/coulomb/demo": + return httpx.Response( + 200, + json={ + "full_name": "coulomb/demo", + "private": False, + "default_branch": "main", + }, + ) + return httpx.Response( + 200, + json=[ + { + "ref": "refs/heads/main", + "object": {"type": "commit", "sha": revision}, + } + ], + ) + + result = ForgejoSourceResolver( + transport=httpx.MockTransport(handler), + now=lambda: "2026-08-22T22:00:00Z", + ).resolve(repo) + + assert result == { + "ok": True, + "supported": True, + "repo_slug": "demo", + "source_ref": { + "kind": "forgejo-archive-v1", + "repository": "coulomb/demo", + "revision": revision, + "observed_ref": "refs/heads/main", + "observed_at": "2026-08-22T22:00:00Z", + }, + } + + +def test_private_or_unresolvable_repository_has_no_v1_source_ref(tmp_path: Path) -> None: + repo = _repo(tmp_path, "https://forgejo.coulomb.social/coulomb/demo.git") + private = ForgejoSourceResolver( + transport=httpx.MockTransport( + lambda request: httpx.Response( + 200, + json={ + "full_name": "coulomb/demo", + "private": True, + "default_branch": "main", + }, + ) + ) + ).resolve(repo) + assert private["supported"] is False + assert private["reason"] == "private-repository" + assert private["source_ref"] is None + + unavailable = ForgejoSourceResolver( + transport=httpx.MockTransport(lambda request: httpx.Response(404)) + ).resolve(repo) + assert unavailable["supported"] is False + assert unavailable["reason"] == "source-unresolvable" + + +def test_projection_requires_nexus_to_echo_exact_source_ref() -> None: + source_ref = { + "kind": "forgejo-archive-v1", + "repository": "coulomb/demo", + "revision": "a" * 40, + "observed_ref": "refs/heads/main", + "observed_at": "2026-08-22T22:00:00Z", + } + bodies = [] + + def echo(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + bodies.append(body) + return httpx.Response( + 200, + json={"slug": "demo", "active": True, "source_ref": body["source_ref"]}, + ) + + client = SBOMNexusClient( + SBOMNexusConfig("https://nexus.example.test"), + transport=httpx.MockTransport(echo), + ) + result = client.upsert_repository( + "demo", + nexus_checkout_path=None, + source_ref=source_ref, + ) + + assert bodies == [{"checkout_path": None, "active": True, "source_ref": source_ref}] + assert result["source_ref"] == source_ref + + dropping_client = SBOMNexusClient( + SBOMNexusConfig("https://nexus.example.test"), + transport=httpx.MockTransport( + lambda request: httpx.Response(200, json={"slug": "demo", "active": True}) + ), + ) + with pytest.raises(SBOMServiceError) as failed: + dropping_client.upsert_repository( + "demo", + nexus_checkout_path=None, + source_ref=source_ref, + ) + assert failed.value.code == "contract_error" + assert failed.value.mutation_may_have_committed is True + + +def test_cli_source_ref_is_read_only_until_explicit_projection( + monkeypatch, tmp_path: Path, capsys +) -> None: + source_ref = { + "kind": "forgejo-archive-v1", + "repository": "coulomb/demo", + "revision": "a" * 40, + "observed_ref": "refs/heads/main", + "observed_at": "2026-08-22T22:00:00Z", + } + resolution = { + "ok": True, + "supported": True, + "repo_slug": "demo", + "source_ref": source_ref, + } + monkeypatch.setattr( + "repo_manager.source_ref.ForgejoSourceResolver.resolve", + lambda self, path, repo_slug=None, remote_name="origin": resolution, + ) + + assert main(["sbom", "source-ref", "--path", str(tmp_path)]) == 0 + assert json.loads(capsys.readouterr().out) == resolution + + assert main(["sbom", "source-ref", "--path", str(tmp_path), "--project"]) == 2 + rejected = json.loads(capsys.readouterr().out) + assert rejected["error"] == "--project requires --confirm-authoritative" + + +def test_cli_projects_source_ref_through_configured_nexus( + monkeypatch, tmp_path: Path, capsys +) -> None: + source_ref = { + "kind": "forgejo-archive-v1", + "repository": "coulomb/demo", + "revision": "a" * 40, + "observed_ref": "refs/heads/main", + "observed_at": "2026-08-22T22:00:00Z", + } + monkeypatch.setattr( + "repo_manager.source_ref.ForgejoSourceResolver.resolve", + lambda self, path, repo_slug=None, remote_name="origin": { + "ok": True, + "supported": True, + "repo_slug": "demo", + "source_ref": source_ref, + }, + ) + monkeypatch.setenv("SBOM_NEXUS_URL", "https://nexus.example.test") + monkeypatch.setattr( + "repo_manager.sbom_client.SBOMNexusClient.upsert_repository", + lambda self, repo_slug, nexus_checkout_path, active=True, source_ref=None: { + "slug": repo_slug, + "source_ref": source_ref, + }, + ) + + exit_code = main( + [ + "sbom", + "source-ref", + "--path", + str(tmp_path), + "--project", + "--confirm-authoritative", + ] + ) + + assert exit_code == 0 + result = json.loads(capsys.readouterr().out) + assert result["projection"]["source_ref"] == source_ref diff --git a/workplans/RMGR-WP-0011-sbom-nexus-production-client.md b/workplans/RMGR-WP-0011-sbom-nexus-production-client.md index 2225ea6..6a423e6 100644 --- a/workplans/RMGR-WP-0011-sbom-nexus-production-client.md +++ b/workplans/RMGR-WP-0011-sbom-nexus-production-client.md @@ -126,6 +126,23 @@ commit uncertainty, idempotency headers, and revision mismatch. Remaining is the production consumer handoff and final source/ownership inspection after the controlled source-input topology is available. +**Controlled-source projection (2026-08-22):** Custodian decision +`c67833d0-62a9-4d14-9d74-4693cc0c497d` selected the +`forgejo-archive-v1` contract. `rmgr sbom source-ref` now normalizes only the +canonical public `coulomb/` Forgejo identity, observes the default branch +and full SHA through the anonymous Forgejo API, records `observed_ref` and +canonical UTC `observed_at`, and returns no source reference for missing, +private, non-Coulomb, mismatched, or unresolvable sources. Local `HEAD` and +workstation paths never supply the production revision. + +The explicit `--project --confirm-authoritative` path submits +`checkout_path: null` and the structured reference through the T02 client. +Projection passes only when Nexus echoes the exact reference, so the current +pre-contract service cannot silently discard it. Live read-only proof resolved +`coulomb/repo-manager@b068e9da421332f99eaa24a811887f9a5d85a478` from +`refs/heads/main`. Remaining T04 work is the Nexus/package/Activity Core +implementation and attended production proof owned through `CUST-WP-0064`. + ## Acceptance - Authoritative mode talks to SBOM Nexus and returns its pinned snapshot