From ad0ba6f2ba30f8438dee38d49cfb5db47efb2975 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 22 Aug 2026 20:22:45 +0200 Subject: [PATCH] refactor: delegate SBOM scans to Nexus Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834 --- ...-0008-sbom-authority-handoff-2026-08-22.md | 46 +++ .../RMGR-WP-0008-sbom-cutover-2026-08-21.md | 4 + docs/repository-registers_v0.1.md | 8 + src/repo_manager/cli.py | 26 +- src/repo_manager/sbom.py | 326 ------------------ src/repo_manager/sbom_client.py | 86 +++++ tests/test_sbom.py | 202 ++++++----- ...k-record-and-register-receiving-surface.md | 7 + 8 files changed, 277 insertions(+), 428 deletions(-) create mode 100644 docs/evidence/RMGR-WP-0008-sbom-authority-handoff-2026-08-22.md delete mode 100644 src/repo_manager/sbom.py create mode 100644 src/repo_manager/sbom_client.py diff --git a/docs/evidence/RMGR-WP-0008-sbom-authority-handoff-2026-08-22.md b/docs/evidence/RMGR-WP-0008-sbom-authority-handoff-2026-08-22.md new file mode 100644 index 0000000..69da432 --- /dev/null +++ b/docs/evidence/RMGR-WP-0008-sbom-authority-handoff-2026-08-22.md @@ -0,0 +1,46 @@ +# RMGR-WP-0008 SBOM authority handoff — 2026-08-22 + +Repo Manager no longer contains or claims an SBOM scanner implementation. +`src/repo_manager/sbom.py` and its parser-specific tests were removed after the +production State Hub façade moved read/write authority to SBOM Nexus. + +Existing operator commands remain usable as deprecated compatibility aliases: + +```text +rmgr sbom scan --path --slug +rmgr sbom licence-report --path --slug +``` + +Both invoke the `sbom-nexus` executable directly with `shell=False`. The command +is discovered on `PATH` or supplied through `SBOM_NEXUS_CLI`. Scan JSON now uses +`sbom-nexus.snapshot.v1` and adds: + +```json +{ + "delegated_by": "repo-manager", + "product_owner": "sbom-nexus" +} +``` + +The `licence-report` alias derives only the legacy report-shaped view from the +Nexus snapshot response; it contains no parser or licence-evaluation logic. +Output refusal/`--force` behavior remains in Repo Manager for CLI compatibility. + +Verification: + +- `uv run ruff check src tests`: passed; +- complete Repo Manager suite: 87 passed; +- end-to-end alias using the real Nexus CLI: `ok=true`, Nexus schema, + `product_owner=sbom-nexus`, 39 entries, Git revision present, zero errors; +- source search found no remaining `repo_manager.sbom`, + `repo-manager.sbom-snapshot.v1`, `scan_repository`, or `detect_sources` + implementation reference outside the new delegation adapter/test names. + +The direct command is now documented first: + +```text +sbom-nexus scan . --output sbom-snapshot.json +``` + +This preserves operator continuity without leaving competing permanent product +authority in Repo Manager. diff --git a/docs/evidence/RMGR-WP-0008-sbom-cutover-2026-08-21.md b/docs/evidence/RMGR-WP-0008-sbom-cutover-2026-08-21.md index c9ce933..45b62ad 100644 --- a/docs/evidence/RMGR-WP-0008-sbom-cutover-2026-08-21.md +++ b/docs/evidence/RMGR-WP-0008-sbom-cutover-2026-08-21.md @@ -1,5 +1,9 @@ # RMGR-WP-0008 SBOM receiving and cutover evidence — 2026-08-21 +> Superseded ownership note (2026-08-22): this evidence records the interim +> Repo Manager scanner. SBOM Nexus is now the sole scanner/snapshot product +> owner; see `RMGR-WP-0008-sbom-authority-handoff-2026-08-22.md`. + Repo Manager now derives an SBOM snapshot from repository-owned lockfiles and the reviewed `sbom-tools.yaml` escape hatch. The snapshot includes source path and SHA-256 provenance, Git revision, canonical UTC generation time, ecosystem, diff --git a/docs/repository-registers_v0.1.md b/docs/repository-registers_v0.1.md index 77e0011..7c0e7f8 100644 --- a/docs/repository-registers_v0.1.md +++ b/docs/repository-registers_v0.1.md @@ -63,10 +63,18 @@ replaceable snapshot and licence/copyleft report without copying State Hub rows back into a new source-of-truth file. ```bash +sbom-nexus scan . --output sbom-snapshot.json + +# Deprecated compatibility aliases; both delegate to the sbom-nexus executable. rmgr sbom scan --path . --output sbom-snapshot.json rmgr sbom licence-report --path . ``` +SBOM Nexus is the sole scanner and snapshot product owner. Repo Manager retains +these aliases only so existing repository workflows do not break; set +`SBOM_NEXUS_CLI` when `sbom-nexus` is not on `PATH`. Their JSON includes +`product_owner: sbom-nexus` and the Nexus `sbom-nexus.snapshot.v1` schema. + Detection covers `uv.lock`, `requirements.txt`, `package-lock.json`, `yarn.lock`, `Cargo.lock`, `go.sum`, `.terraform.lock.hcl`, Ansible requirements under `ansible/`, and root `sbom-tools.yaml`. Each source carries a SHA-256 and diff --git a/src/repo_manager/cli.py b/src/repo_manager/cli.py index da603f6..3bdf6d4 100644 --- a/src/repo_manager/cli.py +++ b/src/repo_manager/cli.py @@ -342,7 +342,10 @@ def main(argv: list[str] | None = None) -> int: help="Write files; without this flag only validate and report", ) - p_sbom = sub.add_parser("sbom", help="Derive SBOM snapshots and licence reports from repository files") + p_sbom = sub.add_parser( + "sbom", + help="Deprecated compatibility commands delegated to SBOM Nexus", + ) sbom_sub = p_sbom.add_subparsers(dest="sbom_command") p_sbom_scan = sbom_sub.add_parser("scan", help="Scan recognised lockfiles and tool manifests") p_sbom_scan.add_argument("--path", default=".") @@ -802,18 +805,17 @@ def main(argv: list[str] | None = None) -> int: if not args.sbom_command: p_sbom.print_help() return 2 - from repo_manager.sbom import scan_repository + from repo_manager.sbom_client import ( + licence_report_from_snapshot, + scan_repository_via_nexus, + ) - snapshot = scan_repository(Path(args.path), slug=args.slug) - result = snapshot if args.sbom_command == "scan" else { - "ok": snapshot["ok"], - "repo_slug": snapshot["repo_slug"], - "source_revision": snapshot["source_revision"], - "generated_at": snapshot["generated_at"], - "entry_count": snapshot["entry_count"], - "licence_report": snapshot["licence_report"], - "errors": snapshot["errors"], - } + snapshot = scan_repository_via_nexus(Path(args.path), slug=args.slug) + result = ( + snapshot + if args.sbom_command == "scan" + else licence_report_from_snapshot(snapshot) + ) if args.sbom_command == "scan" and args.output: output = Path(args.output) if output.exists() and not args.force: diff --git a/src/repo_manager/sbom.py b/src/repo_manager/sbom.py deleted file mode 100644 index 2519c8c..0000000 --- a/src/repo_manager/sbom.py +++ /dev/null @@ -1,326 +0,0 @@ -"""Repository-derived SBOM scanning and licence reporting.""" - -from __future__ import annotations - -import hashlib -import json -import os -import re -import tomllib -from collections import Counter -from collections.abc import Callable -from pathlib import Path -from typing import Any - -import yaml - -from repo_manager.gitops import head_sha -from repo_manager.time import utc_now_text - -SBOM_SCHEMA = "repo-manager.sbom-snapshot.v1" -COPYLEFT_MARKERS = frozenset({"GPL", "AGPL", "LGPL", "EUPL", "CDDL", "MPL"}) -VALID_ECOSYSTEMS = frozenset( - {"python", "node", "rust", "go", "java", "terraform", "ansible", "tool", "other"} -) -SKIP_DIRECTORIES = frozenset( - { - ".git", - ".hg", - ".svn", - ".venv", - "venv", - ".env", - "node_modules", - "__pycache__", - ".mypy_cache", - ".pytest_cache", - ".ruff_cache", - "dist", - "build", - ".build", - "target", - ".tox", - ".nox", - } -) - -Entry = dict[str, Any] -Parser = Callable[[Path], list[Entry]] - - -def _entry( - name: str, - version: str | None, - ecosystem: str, - *, - license_spdx: str | None = None, - is_direct: bool = False, - is_dev: bool = False, -) -> Entry: - return { - "package_name": name, - "package_version": version, - "ecosystem": ecosystem, - "license_spdx": license_spdx, - "is_direct": is_direct, - "is_dev": is_dev, - } - - -def _parse_toml_packages(path: Path, ecosystem: str) -> list[Entry]: - data = tomllib.loads(path.read_text(encoding="utf-8")) - return [ - _entry(str(item["name"]), str(item["version"]) if item.get("version") else None, ecosystem) - for item in data.get("package", []) - if isinstance(item, dict) and item.get("name") - ] - - -def parse_uv_lock(path: Path) -> list[Entry]: - return _parse_toml_packages(path, "python") - - -def parse_cargo_lock(path: Path) -> list[Entry]: - return _parse_toml_packages(path, "rust") - - -def parse_requirements(path: Path) -> list[Entry]: - entries: list[Entry] = [] - for raw in path.read_text(encoding="utf-8").splitlines(): - line = raw.strip() - if not line or line.startswith(("#", "-")): - continue - match = re.match(r"^([A-Za-z0-9_.-]+)(?:[>= list[Entry]: - data = json.loads(path.read_text(encoding="utf-8")) - entries: list[Entry] = [] - for package_path, item in (data.get("packages") or {}).items(): - if not package_path or not isinstance(item, dict): - continue - name = item.get("name") or package_path.rsplit("node_modules/", 1)[-1] - entries.append( - _entry( - str(name), - str(item["version"]) if item.get("version") else None, - "node", - license_spdx=str(item["license"]) if item.get("license") else None, - is_direct=not bool(item.get("indirect", False)), - is_dev=bool(item.get("dev", False)), - ) - ) - return entries - - -def parse_yarn_lock(path: Path) -> list[Entry]: - entries: list[Entry] = [] - names: list[str] = [] - for raw in [*path.read_text(encoding="utf-8").splitlines(), ""]: - stripped = raw.strip() - if raw and not raw.startswith((" ", "\t")) and stripped.endswith(":"): - names = [] - for specifier in stripped.rstrip(":").split(","): - match = re.match(r'"?((?:@[^/" ]+/)?[^@" ]+)@', specifier.strip()) - if match: - names.append(match.group(1)) - elif stripped.startswith("version ") and names: - version_match = re.search(r'"([^"]+)"', stripped) - version = version_match.group(1) if version_match else None - entries.extend(_entry(name, version, "node") for name in names) - names = [] - return entries - - -def _go_direct_modules(directory: Path) -> set[str]: - path = directory / "go.mod" - if not path.is_file(): - return set() - direct: set[str] = set() - in_block = False - for raw in path.read_text(encoding="utf-8").splitlines(): - line = raw.strip() - if line == "require (": - in_block = True - continue - if in_block and line == ")": - in_block = False - continue - candidate = line if in_block else line.removeprefix("require ") if line.startswith("require ") else "" - if candidate and "// indirect" not in candidate: - direct.add(candidate.split()[0]) - return direct - - -def parse_go_sum(path: Path) -> list[Entry]: - direct = _go_direct_modules(path.parent) - seen: set[tuple[str, str]] = set() - entries: list[Entry] = [] - for raw in path.read_text(encoding="utf-8").splitlines(): - parts = raw.split() - if len(parts) < 2 or parts[1].endswith("/go.mod"): - continue - key = (parts[0], parts[1]) - if key in seen: - continue - seen.add(key) - entries.append(_entry(key[0], key[1], "go", is_direct=parts[0] in direct)) - return entries - - -def parse_terraform_lock(path: Path) -> list[Entry]: - entries: list[Entry] = [] - provider: str | None = None - version: str | None = None - for raw in path.read_text(encoding="utf-8").splitlines(): - line = raw.strip() - match = re.match(r'^provider\s+"([^"]+)"\s*\{', line) - if match: - provider, version = match.group(1), None - elif provider: - version_match = re.match(r'version\s*=\s*"([^"]+)"', line) - if version_match: - version = version_match.group(1) - elif line == "}": - entries.append(_entry(provider, version, "terraform", is_direct=True)) - provider, version = None, None - return entries - - -def parse_ansible_requirements(path: Path) -> list[Entry]: - data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} - if not isinstance(data, dict): - return [] - entries: list[Entry] = [] - for kind in ("collections", "roles"): - for item in data.get(kind, []) or []: - if isinstance(item, str): - name, version = item, None - elif isinstance(item, dict): - name = item.get("name") or item.get("src") - version = str(item["version"]) if item.get("version") else None - else: - continue - if name: - entries.append(_entry(str(name), version, "ansible", is_direct=True)) - return entries - - -def parse_tools_manifest(path: Path) -> list[Entry]: - data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} - if not isinstance(data, dict): - return [] - entries: list[Entry] = [] - for item in data.get("tools", []) or []: - if not isinstance(item, dict) or not item.get("name"): - continue - ecosystem = str(item.get("ecosystem") or "tool") - if ecosystem not in VALID_ECOSYSTEMS: - ecosystem = "tool" - version = str(item["version"]) if item.get("version") not in {None, "unknown"} else None - entries.append( - _entry( - str(item["name"]), - version, - ecosystem, - license_spdx=str(item["license_spdx"]) if item.get("license_spdx") else None, - is_direct=bool(item.get("is_direct", True)), - is_dev=bool(item.get("is_dev", False)), - ) - ) - return entries - - -LOCKFILE_PARSERS: dict[str, Parser] = { - "uv.lock": parse_uv_lock, - "requirements.txt": parse_requirements, - "package-lock.json": parse_package_lock, - "yarn.lock": parse_yarn_lock, - "Cargo.lock": parse_cargo_lock, - ".terraform.lock.hcl": parse_terraform_lock, - "go.sum": parse_go_sum, -} - - -def detect_sources(repo_root: Path) -> list[tuple[Path, Parser]]: - found: list[tuple[Path, Parser]] = [] - seen: set[Path] = set() - for directory, directories, filenames in os.walk(repo_root): - directories[:] = sorted(name for name in directories if name not in SKIP_DIRECTORIES) - current = Path(directory) - for filename, parser in LOCKFILE_PARSERS.items(): - if filename in filenames: - path = current / filename - found.append((path, parser)) - seen.add(path) - if current.name == "ansible": - for filename in ("requirements.yml", "requirements.yaml"): - if filename in filenames: - path = current / filename - found.append((path, parse_ansible_requirements)) - seen.add(path) - tools = repo_root / "sbom-tools.yaml" - if tools.is_file() and tools not in seen: - found.append((tools, parse_tools_manifest)) - return sorted(found, key=lambda item: str(item[0])) - - -def is_copyleft(spdx: str | None) -> bool: - upper = (spdx or "").upper() - return any(marker in upper for marker in COPYLEFT_MARKERS) - - -def licence_report(entries: list[Entry]) -> dict[str, Any]: - counts = Counter(entry.get("license_spdx") for entry in entries) - groups = [ - {"license_spdx": license_id, "count": count, "is_copyleft": is_copyleft(license_id)} - for license_id, count in sorted(counts.items(), key=lambda item: (-item[1], item[0] or "")) - ] - risks = [ - { - "package_name": entry["package_name"], - "package_version": entry.get("package_version"), - "license_spdx": entry.get("license_spdx"), - "source_path": entry["source_path"], - } - for entry in entries - if is_copyleft(entry.get("license_spdx")) - and entry.get("is_direct") - and not entry.get("is_dev") - ] - return {"groups": groups, "copyleft_direct_prod": risks, "copyleft_direct_count": len(risks)} - - -def scan_repository(repo_root: Path, *, slug: str | None = None) -> dict[str, Any]: - repo_root = repo_root.resolve() - sources: list[dict[str, Any]] = [] - entries: list[Entry] = [] - errors: list[dict[str, str]] = [] - for path, parser in detect_sources(repo_root): - relative = str(path.relative_to(repo_root)) - try: - parsed = parser(path) - except (OSError, ValueError, TypeError, json.JSONDecodeError, tomllib.TOMLDecodeError, yaml.YAMLError) as exc: - errors.append({"source_path": relative, "error": str(exc)}) - continue - digest = hashlib.sha256(path.read_bytes()).hexdigest() - sources.append({"path": relative, "sha256": digest, "entry_count": len(parsed)}) - for item in parsed: - entries.append({**item, "source_path": relative}) - return { - "schema": SBOM_SCHEMA, - "ok": not errors, - "repo_slug": slug or repo_root.name, - "repo_path": str(repo_root), - "source_revision": head_sha(repo_root), - "generated_at": utc_now_text(), - "authority": "detected lockfiles and reviewed sbom-tools.yaml", - "sources": sources, - "entry_count": len(entries), - "entries": entries, - "licence_report": licence_report(entries), - "errors": errors, - } diff --git a/src/repo_manager/sbom_client.py b/src/repo_manager/sbom_client.py new file mode 100644 index 0000000..425113d --- /dev/null +++ b/src/repo_manager/sbom_client.py @@ -0,0 +1,86 @@ +"""Deprecated Repo Manager CLI adapter for the SBOM Nexus product owner.""" +from __future__ import annotations + +import json +import os +import shutil +import subprocess +from pathlib import Path +from typing import Any + + +def scan_repository_via_nexus(repo_root: Path, *, slug: str | None = None) -> dict[str, Any]: + """Delegate a local source scan to SBOM Nexus without owning scanner logic.""" + executable = os.getenv("SBOM_NEXUS_CLI") or shutil.which("sbom-nexus") + if not executable: + return _error( + "sbom-nexus CLI not found; install the sbom-nexus package or set " + "SBOM_NEXUS_CLI to its executable path" + ) + + command = [executable, "scan", str(repo_root)] + if slug: + command.extend(["--slug", slug]) + try: + completed = subprocess.run( + command, + capture_output=True, + text=True, + check=False, + ) + except OSError as exc: + return _error(f"could not execute SBOM Nexus CLI: {exc}") + + try: + payload = json.loads(completed.stdout) + except json.JSONDecodeError: + detail = completed.stderr.strip() or completed.stdout.strip() or "no output" + return _error( + f"SBOM Nexus CLI returned invalid JSON (exit {completed.returncode}): {detail}" + ) + if not isinstance(payload, dict): + return _error("SBOM Nexus CLI returned a non-object JSON response") + + payload["delegated_by"] = "repo-manager" + payload["product_owner"] = "sbom-nexus" + if completed.returncode and payload.get("ok", True): + payload["ok"] = False + payload.setdefault("errors", []).append( + { + "reason": "sbom-nexus-exit", + "detail": completed.stderr.strip() or f"exit {completed.returncode}", + } + ) + return payload + + +def licence_report_from_snapshot(snapshot: dict[str, Any]) -> dict[str, Any]: + """Preserve the old rmgr report-shaped view over a Nexus snapshot.""" + return { + "ok": bool(snapshot.get("ok")), + "repo_slug": snapshot.get("repo_slug"), + "source_revision": snapshot.get("source_revision"), + "generated_at": snapshot.get("generated_at"), + "entry_count": int(snapshot.get("entry_count") or 0), + "licence_report": snapshot.get("licence_report") or { + "groups": [], + "copyleft_direct_prod": [], + "copyleft_direct_count": 0, + }, + "errors": snapshot.get("errors") or [], + "delegated_by": "repo-manager", + "product_owner": "sbom-nexus", + } + + +def _error(detail: str) -> dict[str, Any]: + return { + "schema": "sbom-nexus.snapshot.v1", + "ok": False, + "entry_count": 0, + "entries": [], + "sources": [], + "errors": [{"reason": "sbom-nexus-delegation", "detail": detail}], + "delegated_by": "repo-manager", + "product_owner": "sbom-nexus", + } diff --git a/tests/test_sbom.py b/tests/test_sbom.py index be16774..4d5f757 100644 --- a/tests/test_sbom.py +++ b/tests/test_sbom.py @@ -4,108 +4,130 @@ import json import subprocess from pathlib import Path -from repo_manager.sbom import detect_sources, scan_repository +from repo_manager.cli import main +from repo_manager.sbom_client import ( + licence_report_from_snapshot, + scan_repository_via_nexus, +) -def _git(repo: Path, *args: str) -> None: - subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True) - - -def test_scan_derives_snapshot_and_copyleft_report(tmp_path: Path) -> None: - repo = tmp_path / "example" - repo.mkdir() - _git(repo, "init") - _git(repo, "config", "user.email", "test@example.com") - _git(repo, "config", "user.name", "Test") - (repo / "uv.lock").write_text( - 'version = 1\n[[package]]\nname = "pyyaml"\nversion = "6.0.2"\n', - encoding="utf-8", - ) - (repo / "package-lock.json").write_text( - json.dumps( +def _snapshot() -> dict: + return { + "schema": "sbom-nexus.snapshot.v1", + "ok": True, + "repo_slug": "example", + "source_revision": "abc123", + "generated_at": "2026-08-22T18:00:00Z", + "entry_count": 1, + "entries": [ { - "packages": { - "": {"name": "root", "version": "1.0.0"}, - "node_modules/copyleft": { - "name": "copyleft", - "version": "2.0.0", - "license": "GPL-3.0-only", - }, - "node_modules/dev-only": { - "name": "dev-only", - "version": "3.0.0", - "license": "AGPL-3.0-only", - "dev": True, - }, - } + "package_name": "pyyaml", + "package_version": "6.0.3", + "ecosystem": "python", + "license_spdx": "MIT", + "is_direct": True, + "is_dev": False, + "source_path": "uv.lock", } - ), - encoding="utf-8", - ) - _git(repo, "add", ".") - _git(repo, "commit", "-m", "seed") - - result = scan_repository(repo) - - assert result["ok"] is True - assert result["schema"] == "repo-manager.sbom-snapshot.v1" - assert result["source_revision"] - assert result["generated_at"].endswith("Z") - assert result["entry_count"] == 3 - assert result["licence_report"]["copyleft_direct_count"] == 1 - assert result["licence_report"]["copyleft_direct_prod"][0]["package_name"] == "copyleft" - - -def test_detection_covers_sources_and_skips_dependency_directories(tmp_path: Path) -> None: - (tmp_path / "go.sum").write_text("example.test/mod v1.2.3 h1:abc\n", encoding="utf-8") - terraform = tmp_path / "infra" - terraform.mkdir() - (terraform / ".terraform.lock.hcl").write_text( - 'provider "registry.terraform.io/hashicorp/null" {\n version = "3.2.3"\n}\n', - encoding="utf-8", - ) - ansible = tmp_path / "deploy" / "ansible" - ansible.mkdir(parents=True) - (ansible / "requirements.yml").write_text("collections:\n - community.general\n", encoding="utf-8") - ignored = tmp_path / "node_modules" - ignored.mkdir() - (ignored / "package-lock.json").write_text("{}", encoding="utf-8") - - sources = {str(path.relative_to(tmp_path)) for path, _parser in detect_sources(tmp_path)} - - assert sources == { - "deploy/ansible/requirements.yml", - "go.sum", - "infra/.terraform.lock.hcl", + ], + "sources": [{"path": "uv.lock", "entry_count": 1, "sha256": "abc"}], + "licence_report": { + "groups": [ + { + "license_spdx": "MIT", + "count": 1, + "is_copyleft": False, + } + ], + "copyleft_direct_prod": [], + "copyleft_direct_count": 0, + }, + "errors": [], } -def test_go_sum_marks_modules_declared_in_go_mod_as_direct(tmp_path: Path) -> None: - (tmp_path / "go.mod").write_text( - "module example.test/app\n\nrequire example.test/direct v1.2.3\n", - encoding="utf-8", - ) - (tmp_path / "go.sum").write_text( - "example.test/direct v1.2.3 h1:abc\n" - "example.test/direct v1.2.3/go.mod h1:def\n" - "example.test/transitive v2.0.0 h1:ghi\n", - encoding="utf-8", - ) +def test_scan_delegates_to_sbom_nexus_without_shell(monkeypatch, tmp_path: Path) -> None: + observed = {} - result = scan_repository(tmp_path) + def fake_run(command, **kwargs): + observed["command"] = command + observed["kwargs"] = kwargs + return subprocess.CompletedProcess(command, 0, json.dumps(_snapshot()), "") - assert result["ok"] is True - assert [(entry["package_name"], entry["is_direct"]) for entry in result["entries"]] == [ - ("example.test/direct", True), - ("example.test/transitive", False), + monkeypatch.setenv("SBOM_NEXUS_CLI", "/opt/sbom-nexus/bin/sbom-nexus") + monkeypatch.setattr(subprocess, "run", fake_run) + + result = scan_repository_via_nexus(tmp_path, slug="example") + + assert observed["command"] == [ + "/opt/sbom-nexus/bin/sbom-nexus", + "scan", + str(tmp_path), + "--slug", + "example", ] + assert observed["kwargs"] == { + "capture_output": True, + "text": True, + "check": False, + } + assert result["schema"] == "sbom-nexus.snapshot.v1" + assert result["product_owner"] == "sbom-nexus" + assert result["delegated_by"] == "repo-manager" -def test_invalid_source_is_reported_without_partial_failure(tmp_path: Path) -> None: - (tmp_path / "uv.lock").write_text("not = [valid", encoding="utf-8") +def test_missing_nexus_cli_returns_actionable_error(monkeypatch, tmp_path: Path) -> None: + monkeypatch.delenv("SBOM_NEXUS_CLI", raising=False) + monkeypatch.setattr("repo_manager.sbom_client.shutil.which", lambda _name: None) - result = scan_repository(tmp_path) + result = scan_repository_via_nexus(tmp_path) assert result["ok"] is False - assert result["entry_count"] == 0 - assert result["errors"][0]["source_path"] == "uv.lock" + assert result["schema"] == "sbom-nexus.snapshot.v1" + assert "SBOM_NEXUS_CLI" in result["errors"][0]["detail"] + + +def test_licence_report_alias_preserves_shape() -> None: + result = licence_report_from_snapshot(_snapshot()) + + assert set(result) == { + "ok", + "repo_slug", + "source_revision", + "generated_at", + "entry_count", + "licence_report", + "errors", + "delegated_by", + "product_owner", + } + assert result["licence_report"]["copyleft_direct_count"] == 0 + + +def test_cli_scan_preserves_output_file_behavior(monkeypatch, tmp_path: Path, capsys) -> None: + monkeypatch.setattr( + "repo_manager.sbom_client.scan_repository_via_nexus", + lambda path, slug=None: { + **_snapshot(), + "delegated_by": "repo-manager", + "product_owner": "sbom-nexus", + }, + ) + output = tmp_path / "snapshot.json" + + exit_code = main( + [ + "sbom", + "scan", + "--path", + str(tmp_path), + "--slug", + "example", + "--output", + str(output), + ] + ) + + assert exit_code == 0 + assert json.loads(output.read_text())["schema"] == "sbom-nexus.snapshot.v1" + assert json.loads(capsys.readouterr().out)["product_owner"] == "sbom-nexus" diff --git a/workplans/RMGR-WP-0008-work-record-and-register-receiving-surface.md b/workplans/RMGR-WP-0008-work-record-and-register-receiving-surface.md index 4c4d6d6..d66012f 100644 --- a/workplans/RMGR-WP-0008-work-record-and-register-receiving-surface.md +++ b/workplans/RMGR-WP-0008-work-record-and-register-receiving-surface.md @@ -197,6 +197,13 @@ would invert repository authority, so cutover rebuilds from source and retains old snapshot identifiers only as retirement provenance. Evidence: `docs/evidence/RMGR-WP-0008-sbom-cutover-2026-08-21.md`. +Authority correction (2026-08-22): SBOM Nexus is now the single scanner and +snapshot product owner. Repo Manager's duplicate parser module was removed; +the existing `rmgr sbom scan|licence-report` commands remain only as thin, +deprecated delegates to the `sbom-nexus` executable and identify Nexus as the +owner in their JSON. Evidence: +`docs/evidence/RMGR-WP-0008-sbom-authority-handoff-2026-08-22.md`. + ## Topic and classification contract ```task