ModelForge/scripts/m16_sbom.py

"""Generate a CycloneDX SBOM and an image provenance record for the ModelForge v1 components.

The SBOM lists components and their versions, not their contents: a release inventory has to be
readable and diffable, so no dependency binaries are stored. Provenance binds each built image to
the exact source commit and image digest that produced it, which is what makes the SBOM verifiable
rather than merely present.

    python scripts/m16_sbom.py --output docs/security/sbom
"""

from __future__ import annotations

import argparse
import hashlib
import json
import subprocess
import sys
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]

PYTHON_IMAGES = {
    "modelforge-api": "backend",
    "modelforge-node-agent": "node-agent",
    "modelforge-runtime-worker": "runtime-worker",
}
NODE_COMPONENTS = {"modelforge-web": "frontend"}


def run(*args: str, timeout: int = 300) -> str:
    completed = subprocess.run(
        list(args),
        capture_output=True,
        text=True,
        timeout=timeout,
        check=False,
    )
    return (completed.stdout or "").strip()


def git(*args: str) -> str:
    return run("git", "-C", str(ROOT), *args)


def docker(*args: str, timeout: int = 300) -> str:
    return run("docker", *args, timeout=timeout)


def python_components(image: str) -> list[dict[str, Any]]:
    """Read the installed distributions from the image itself, not from a manifest."""

    raw = docker(
        "run",
        "--rm",
        "--entrypoint",
        "python",
        image,
        "-c",
        "import json,importlib.metadata as m;"
        "print(json.dumps(sorted(((d.metadata['Name'] or '?'), (d.version or '?')) "
        "for d in m.distributions())))",
        timeout=600,
    )
    try:
        entries = json.loads(raw)
    except json.JSONDecodeError:
        return []
    seen: set[tuple[str, str]] = set()
    components = []
    for name, version in entries:
        key = (str(name).lower(), str(version))
        if key in seen:
            continue
        seen.add(key)
        components.append(
            {
                "type": "library",
                "name": str(name),
                "version": str(version),
                "purl": f"pkg:pypi/{str(name).lower()}@{version}",
                "scope": "required",
            }
        )
    return components


def node_components(directory: Path) -> list[dict[str, Any]]:
    lock = directory / "package-lock.json"
    if not lock.is_file():
        return []
    data = json.loads(lock.read_text(encoding="utf-8"))
    components = []
    for path, entry in sorted(data.get("packages", {}).items()):
        if not path.startswith("node_modules/"):
            continue
        name = path.removeprefix("node_modules/")
        version = entry.get("version")
        if not version:
            continue
        components.append(
            {
                "type": "library",
                "name": name,
                "version": str(version),
                "purl": f"pkg:npm/{name}@{version}",
                "scope": "optional" if entry.get("dev") else "required",
            }
        )
    return components


def image_provenance(image: str, commit: str) -> dict[str, Any]:
    digest = docker("inspect", "--format", "{{index .RepoDigests 0}}", image) or None
    labels_raw = docker("inspect", "--format", "{{json .Config.Labels}}", image)
    try:
        labels = json.loads(labels_raw or "null") or {}
    except json.JSONDecodeError:
        labels = {}
    if not isinstance(labels, dict):
        labels = {}
    return {
        "image": image,
        "image_id": docker("inspect", "--format", "{{.Id}}", image),
        "repo_digest": digest,
        "created": docker("inspect", "--format", "{{.Created}}", image),
        "base_image": docker("inspect", "--format", "{{index .Config.Labels \"base\"}}", image)
        or None,
        "source_commit": commit,
        "source_repository": git("config", "--get", "remote.origin.url"),
        "oci": {
            key: value
            for key, value in labels.items()
            if key.startswith("org.opencontainers.image.")
        },
    }


def build(
    output: Path,
    *,
    version: str,
    source_commit: str | None = None,
    generated_at: str | None = None,
) -> dict[str, Any]:
    commit = source_commit or git("rev-parse", "HEAD")
    generated = generated_at or datetime.now(UTC).isoformat()
    components: list[dict[str, Any]] = []
    provenance: list[dict[str, Any]] = []

    for image_name in PYTHON_IMAGES:
        image = f"{image_name}:{version}"
        entries = python_components(image)
        for entry in entries:
            entry["properties"] = [{"name": "modelforge:image", "value": image}]
        components.extend(entries)
        provenance.append(image_provenance(image, commit))

    for image_name, directory in NODE_COMPONENTS.items():
        image = f"{image_name}:{version}"
        entries = node_components(ROOT / directory)
        for entry in entries:
            entry["properties"] = [{"name": "modelforge:image", "value": image}]
        components.extend(entries)
        provenance.append(image_provenance(image, commit))

    serial = hashlib.sha256(
        json.dumps([c["purl"] for c in components], sort_keys=True).encode()
    ).hexdigest()

    sbom = {
        "bomFormat": "CycloneDX",
        "specVersion": "1.5",
        "serialNumber": f"urn:uuid:{serial[:8]}-{serial[8:12]}-{serial[12:16]}-{serial[16:20]}-{serial[20:32]}",
        "version": 1,
        "metadata": {
            "timestamp": generated,
            "component": {
                "type": "application",
                "name": "itworx-modelforge",
                "version": version,
            },
            "properties": [
                {"name": "modelforge:source_commit", "value": commit},
                {"name": "modelforge:generated_by", "value": "scripts/m16_sbom.py"},
            ],
        },
        "components": components,
    }

    output.mkdir(parents=True, exist_ok=True)
    (output / "modelforge-cyclonedx.json").write_text(
        json.dumps(sbom, indent=2) + "\n", encoding="utf-8"
    )
    (output / "image-provenance.json").write_text(
        json.dumps(
            {"generated_at": generated, "source_commit": commit, "images": provenance},
            indent=2,
        )
        + "\n",
        encoding="utf-8",
    )
    return {
        "components": len(components),
        "images": len(provenance),
        "source_commit": commit,
        "output": str(output),
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output", default=str(ROOT / "docs" / "security" / "sbom"))
    parser.add_argument("--version", default=(ROOT / "VERSION").read_text("utf-8").strip())
    parser.add_argument("--source-commit", default=None)
    parser.add_argument("--generated-at", default=None)
    args = parser.parse_args(argv)
    summary = build(
        Path(args.output),
        version=args.version,
        source_commit=args.source_commit,
        generated_at=args.generated_at,
    )
    print(json.dumps(summary, indent=2))
    return 0 if summary["components"] else 1


if __name__ == "__main__":
    sys.exit(main())