/
/
/
1"""
2CI check: decide whether pip-audit findings are ones a pull request introduces.
3
4pip-audit scans the whole installed environment, so on any pull request it also reports
5vulnerabilities that already exist on the branch it targets. Those must not gate it: the
6author cannot act on them, and the resulting status is not overridable by a maintainer.
7
8Findings are compared against the resolved dependency set of the target branch, so one
9that reaches the environment through a transitive dependency gates as well. Without that
10set the comparison falls back to the requirement lines the pull request changes, which
11covers direct dependencies only.
12
13Prints the status the reporting workflow gates on:
14 pass no known vulnerabilities at all
15 preexisting vulnerabilities exist, but the target branch has all of them too
16 fail vulnerabilities this pull request introduces
17
18Usage:
19 python3 scripts/audit_changed_packages.py audit.json new_deps.txt [base_closure.txt]
20"""
21
22from __future__ import annotations
23
24import argparse
25import json
26import re
27from pathlib import Path
28
29# ruff: noqa: T201
30
31# Package name of a requirement line, plus the version when the line pins one exactly
32REQUIREMENT = re.compile(r"^([A-Za-z0-9._-]+)\s*(?:\[[^\]]*\])?\s*(?:==\s*([^\s;]+))?")
33
34
35def normalize(name: str) -> str:
36 """
37 Return the PEP 503 normalized form of a package name.
38
39 :param name: Package name as written in a requirement line or an audit report.
40 """
41 return re.sub(r"[-_.]+", "-", name).lower()
42
43
44def requirement_versions(requirements: str) -> dict[str, set[str]]:
45 """
46 Return the versions the given requirement lines pin, keyed on normalized package name.
47
48 A package maps to an empty set when no line pins it to an exact version, as a URL
49 or a range requirement does not.
50
51 :param requirements: Requirement lines, one per line.
52 """
53 versions: dict[str, set[str]] = {}
54 for raw in requirements.splitlines():
55 line = raw.strip()
56 # Skip comments and pip options such as --index-url
57 if not line or line.startswith(("#", "-")):
58 continue
59 if match := REQUIREMENT.match(line):
60 pinned = versions.setdefault(normalize(match.group(1)), set())
61 if match.group(2):
62 pinned.add(match.group(2))
63 return versions
64
65
66def changed_packages(requirements: str) -> set[str]:
67 """
68 Return the normalized names of the packages in the given requirement lines.
69
70 :param requirements: Requirement lines a pull request adds or changes, one per line.
71 """
72 return set(requirement_versions(requirements))
73
74
75def vulnerable_packages(audit: str) -> set[tuple[str, str]]:
76 """
77 Return the packages pip-audit reported vulnerabilities for, as (name, version) pairs.
78
79 :param audit: A pip-audit report in JSON format.
80 """
81 # Indexed rather than fetched with a default: pip-audit is installed unpinned, so a
82 # schema change has to surface as an error instead of an empty, passing result set
83 report = json.loads(audit)
84 return {
85 (normalize(dep["name"]), dep["version"])
86 for dep in report["dependencies"]
87 if dep.get("vulns")
88 }
89
90
91def introduced_packages(
92 findings: set[tuple[str, str]],
93 resolved: dict[str, set[str]],
94 changed: frozenset[str] | set[str] = frozenset(),
95) -> set[tuple[str, str]]:
96 """
97 Return the findings the given resolved dependency set does not already contain.
98
99 :param findings: Vulnerable packages as (name, version) pairs.
100 :param resolved: Versions the target branch resolves to, keyed on package name.
101 :param changed: Names of the packages whose requirement the pull request changes.
102 """
103 introduced = set()
104 for name, version in findings:
105 pinned = resolved.get(name)
106 if pinned is None:
107 introduced.add((name, version))
108 elif pinned:
109 if version not in pinned:
110 introduced.add((name, version))
111 # A package the target branch pins by URL carries no version to compare against,
112 # so only a change to its requirement tells the two branches apart
113 elif name in changed:
114 introduced.add((name, version))
115 return introduced
116
117
118def audit_status(audit: str, requirements: str, base_closure: str = "") -> str:
119 """
120 Return `pass`, `preexisting` or `fail` for the given audit report.
121
122 :param audit: A pip-audit report in JSON format.
123 :param requirements: Requirement lines the pull request adds or changes, one per line.
124 :param base_closure: The target branch's resolved dependency set as pinned requirement
125 lines. When empty, only the changed requirement lines are compared.
126 """
127 findings = vulnerable_packages(audit)
128 if not findings:
129 return "pass"
130
131 if base_closure.strip():
132 resolved = requirement_versions(base_closure)
133 # A resolution that cannot be read would mark every finding as introduced
134 if not resolved:
135 raise ValueError("No packages found in the resolved dependency set")
136 changed = changed_packages(requirements)
137 return "fail" if introduced_packages(findings, resolved, changed) else "preexisting"
138
139 changed = changed_packages(requirements)
140 return "fail" if {name for name, _ in findings} & changed else "preexisting"
141
142
143def main(argv: list[str] | None = None) -> int:
144 """Print the audit status for the given report and dependency sets."""
145 parser = argparse.ArgumentParser(description=__doc__)
146 parser.add_argument("audit", type=Path, help="pip-audit report in JSON format.")
147 parser.add_argument(
148 "requirements", type=Path, help="Requirement lines the pull request adds or changes."
149 )
150 parser.add_argument(
151 "base_closure",
152 type=Path,
153 nargs="?",
154 help="Resolved dependency set of the branch the pull request targets.",
155 )
156 args = parser.parse_args(argv)
157
158 # The requirements file is only written when the pull request changes dependencies,
159 # the resolved set only when resolving the target branch succeeded
160 requirements = args.requirements.read_text() if args.requirements.is_file() else ""
161 base_closure = (
162 args.base_closure.read_text() if args.base_closure and args.base_closure.is_file() else ""
163 )
164 print(audit_status(args.audit.read_text(), requirements, base_closure))
165 return 0
166
167
168if __name__ == "__main__":
169 raise SystemExit(main())
170