#!/usr/bin/env python3 """Verify exact locked direct dependencies against a committed license baseline. The release gate is deterministic and offline: it evaluates only packages in project lockfiles, verifies their exact versions against a reviewed baseline, and rejects denied licenses. Unrelated packages installed in a shared runner do not affect the result. """ from __future__ import annotations import argparse import importlib.metadata import json import re from pathlib import Path from typing import Iterable DENIED_RE = re.compile( r"(?:^|\b)(?:AGPL(?:[- .]?v?\d|[- ]?\d)|GNU AFFERO GENERAL PUBLIC LICENSE|SSPL|SERVER SIDE PUBLIC LICENSE)(?:\b|$)", re.IGNORECASE, ) def _canonical(name: str) -> str: return re.sub(r"[-_.]+", "-", name).lower() def declared_requirements(paths: Iterable[Path]) -> dict[str, str]: requirements: dict[str, str] = {} for path in paths: for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or line.startswith("-"): continue match = re.fullmatch(r"([A-Za-z0-9_.-]+)(?:\[[^]]+\])?==([^;\s]+)", line) if not match: raise ValueError(f"dependency is not exactly pinned in {path}: {line}") name, version = _canonical(match.group(1)), match.group(2) previous = requirements.get(name) if previous is not None and previous != version: raise ValueError(f"conflicting locked versions for {name}: {previous} and {version}") requirements[name] = version return requirements def declared_names(paths: Iterable[Path]) -> set[str]: """Compatibility helper used by existing callers/tests.""" return set(declared_requirements(paths)) def load_baseline(path: Path) -> dict[str, dict[str, str]]: payload = json.loads(path.read_text(encoding="utf-8")) if payload.get("schemaVersion") != 1 or not isinstance(payload.get("packages"), list): raise ValueError("license baseline schema is invalid") baseline: dict[str, dict[str, str]] = {} for row in payload["packages"]: name = _canonical(str(row.get("name", ""))) version = str(row.get("version", "")) license_name = str(row.get("license", "")).strip() source = str(row.get("source", "")).strip() if not name or not version or not license_name or not source: raise ValueError("license baseline contains an incomplete package record") if name in baseline: raise ValueError(f"duplicate package in license baseline: {name}") baseline[name] = {"name": name, "version": version, "license": license_name, "source": source} return baseline def _requirement_name(requirement: str) -> str | None: """Extract a normalized distribution name from a PEP 508 requirement.""" match = re.match(r"\s*([A-Za-z0-9_.-]+)", requirement or "") return _canonical(match.group(1)) if match else None def dependency_closure( root_names: set[str], distributions: Iterable[object] | None = None ) -> tuple[set[str], list[str]]: """Resolve installed dependency closure for diagnostics without host noise.""" installed = list(distributions) if distributions is not None else list(importlib.metadata.distributions()) by_name = { _canonical(str(dist.metadata.get("Name") or "")): dist for dist in installed if dist.metadata.get("Name") } closure = {_canonical(name) for name in root_names} errors: list[str] = [] queue = list(sorted(closure)) visited: set[str] = set() while queue: name = queue.pop(0) if name in visited: continue visited.add(name) dist = by_name.get(name) if dist is None: errors.append(f"declared dependency is not installed: {name}") continue for requirement in getattr(dist, "requires", None) or []: dependency = _requirement_name(requirement) if dependency and dependency not in closure: closure.add(dependency) queue.append(dependency) return closure, errors def inventory( allowed_names: set[str] | None = None, distributions: Iterable[object] | None = None, *, require_installed: bool = False, declared_versions: dict[str, str] | None = None, baseline: dict[str, dict[str, str]] | None = None, ) -> tuple[list[dict[str, str]], list[str]]: errors: list[str] = [] if declared_versions is not None and baseline is not None: declared_set = set(declared_versions) baseline_set = set(baseline) for missing in sorted(declared_set - baseline_set): errors.append(f"declared dependency missing from license baseline: {missing}") for stale in sorted(baseline_set - declared_set): errors.append(f"license baseline contains undeclared dependency: {stale}") rows: list[dict[str, str]] = [] for name in sorted(declared_set & baseline_set): row = baseline[name] if row["version"] != declared_versions[name]: errors.append( f"license baseline version mismatch for {name}: " f"locked {declared_versions[name]}, baseline {row['version']}" ) if DENIED_RE.search(row["license"]): errors.append(f"denied dependency license: {name} {row['version']}: {row['license']}") rows.append(dict(row)) return rows, errors # Compatibility path for unit-level callers. It is intentionally scoped to # allowed_names and never evaluates unrelated installed distributions. rows = [] installed_names: set[str] = set() installed = list(distributions) if distributions is not None else list(importlib.metadata.distributions()) installed = sorted(installed, key=lambda d: (d.metadata.get("Name") or "").lower()) for dist in installed: name = dist.metadata.get("Name") or "unknown" canonical_name = _canonical(name) if allowed_names is not None and canonical_name not in allowed_names: continue installed_names.add(canonical_name) version = dist.version or "unknown" license_text = (dist.metadata.get("License-Expression") or dist.metadata.get("License") or "").strip() if not license_text: classifiers = dist.metadata.get_all("Classifier") or [] license_text = "; ".join(c for c in classifiers if c.startswith("License ::")) or "UNKNOWN" summary = " ".join(license_text.split())[:500] if DENIED_RE.search(summary): errors.append(f"denied dependency license: {name} {version}: {summary}") rows.append({"name": name, "version": version, "license": summary}) if require_installed and allowed_names is not None: for missing in sorted(allowed_names - installed_names): errors.append(f"declared dependency is not installed: {missing}") return rows, errors def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser() parser.add_argument("--output", default="dependency-licenses.json") parser.add_argument("--declared-from", nargs="+", required=True) parser.add_argument("--baseline", default="docs/dependency_licenses.json") parser.add_argument("--require-installed", action="store_true", help="legacy local diagnostic only") args = parser.parse_args(argv) try: declared = declared_requirements(Path(item) for item in args.declared_from) baseline = load_baseline(Path(args.baseline)) rows, errors = inventory(declared_versions=declared, baseline=baseline) except (OSError, ValueError, json.JSONDecodeError) as exc: rows, errors = [], [str(exc)] Path(args.output).write_text(json.dumps(rows, indent=2, sort_keys=True) + "\n", encoding="utf-8") for error in errors: print(f"ERROR: {error}") print(f"dependency_license_scan={'PASS' if not errors else 'FAIL'} packages={len(rows)}") return 1 if errors else 0 if __name__ == "__main__": raise SystemExit(main())