repo-manager/src/repo_manager/sbom.py

"""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,
    }