refactor: delegate SBOM scans to Nexus
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834
This commit is contained in:
parent
6d57d1c1b2
commit
ad0ba6f2ba
8 changed files with 277 additions and 428 deletions
|
|
@ -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 <checkout> --slug <slug>
|
||||
rmgr sbom licence-report --path <checkout> --slug <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.
|
||||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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_.-]+)(?:[>=<!~^]+([^\s;]+))?", line)
|
||||
if match:
|
||||
entries.append(_entry(match.group(1), match.group(2), "python", is_direct=True))
|
||||
return entries
|
||||
|
||||
|
||||
def parse_package_lock(path: Path) -> 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,
|
||||
}
|
||||
86
src/repo_manager/sbom_client.py
Normal file
86
src/repo_manager/sbom_client.py
Normal file
|
|
@ -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",
|
||||
}
|
||||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue