Spaces:
Sleeping
Sleeping
| #!/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()) | |