Spaces:
Sleeping
Sleeping
File size: 8,206 Bytes
2e658e7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 | #!/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())
|