coulomb/repo-manager
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(sbom): project immutable Forgejo source refs

Browse source 
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
This commit is contained in:
tegwick 2026-08-22 23:34:01 +02:00
parent b068e9da42
commit e6cc18bf18
6 changed files with 543 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

35
docs/sbom-nexus-client-contract_v1.md
View file

@ -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/<slug>` 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,

66
src/repo_manager/cli.py
View file

@ -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,

20
src/repo_manager/sbom_client.py
View file

@ -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,

174
src/repo_manager/source_ref.py Normal file
View file

@ -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<host>[^:]+):(?P<path>.+)$")
_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,
}

234
tests/test_source_ref.py Normal file
View file

@ -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

17
workplans/RMGR-WP-0011-sbom-nexus-production-client.md
View file

@ -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/<slug>` 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