/
/
/
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. The pull request's own resolved set is taken into account
12as well: the environment is installed before either branch is resolved, so a release
13published in between must not read as a finding the pull request brings in.
14
15Prints the status the reporting workflow gates on:
16 pass no known vulnerabilities at all
17 preexisting vulnerabilities exist, but none of them are this pull request's doing
18 fail vulnerabilities this pull request introduces
19
20Findings set aside because the pull request's own resolved set does not carry them are
21named on stderr, so a maintainer can tell a timing artifact from a wrongly cleared finding.
22
23Usage:
24 python3 scripts/audit_changed_packages.py audit.json new_deps.txt \
25 [base_closure.txt] [head_closure.txt]
26"""
27
28from __future__ import annotations
29
30import argparse
31import json
32import re
33import sys
34from pathlib import Path
35
36# ruff: noqa: T201
37
38# Package name of a requirement line, plus the version when the line pins one exactly
39REQUIREMENT = re.compile(r"^([A-Za-z0-9._-]+)\s*(?:\[[^\]]*\])?\s*(?:==\s*([^\s;]+))?")
40
41
42def normalize(name: str) -> str:
43 """
44 Return the PEP 503 normalized form of a package name.
45
46 :param name: Package name as written in a requirement line or an audit report.
47 """
48 return re.sub(r"[-_.]+", "-", name).lower()
49
50
51def requirement_versions(requirements: str) -> dict[str, set[str]]:
52 """
53 Return the versions the given requirement lines pin, keyed on normalized package name.
54
55 A package maps to an empty set when no line pins it to an exact version, as a URL
56 or a range requirement does not.
57
58 :param requirements: Requirement lines, one per line.
59 """
60 versions: dict[str, set[str]] = {}
61 for raw in requirements.splitlines():
62 line = raw.strip()
63 # Skip comments and pip options such as --index-url
64 if not line or line.startswith(("#", "-")):
65 continue
66 if match := REQUIREMENT.match(line):
67 pinned = versions.setdefault(normalize(match.group(1)), set())
68 if match.group(2):
69 pinned.add(match.group(2))
70 return versions
71
72
73def changed_packages(requirements: str) -> set[str]:
74 """
75 Return the normalized names of the packages in the given requirement lines.
76
77 :param requirements: Requirement lines a pull request adds or changes, one per line.
78 """
79 return set(requirement_versions(requirements))
80
81
82def vulnerable_packages(audit: str) -> set[tuple[str, str]]:
83 """
84 Return the packages pip-audit reported vulnerabilities for, as (name, version) pairs.
85
86 :param audit: A pip-audit report in JSON format.
87 """
88 # Indexed rather than fetched with a default: pip-audit is installed unpinned, so a
89 # schema change has to surface as an error instead of an empty, passing result set
90 report = json.loads(audit)
91 return {
92 (normalize(dep["name"]), dep["version"])
93 for dep in report["dependencies"]
94 if dep.get("vulns")
95 }
96
97
98def introduced_packages(
99 findings: set[tuple[str, str]],
100 resolved: dict[str, set[str]],
101 changed: frozenset[str] | set[str] = frozenset(),
102) -> set[tuple[str, str]]:
103 """
104 Return the findings the given resolved dependency set does not already contain.
105
106 :param findings: Vulnerable packages as (name, version) pairs.
107 :param resolved: Versions the target branch resolves to, keyed on package name.
108 :param changed: Names of the packages whose requirement the pull request changes.
109 """
110 introduced = set()
111 for name, version in findings:
112 pinned = resolved.get(name)
113 if pinned is None:
114 introduced.add((name, version))
115 elif pinned:
116 if version not in pinned:
117 introduced.add((name, version))
118 # A package the target branch pins by URL carries no version to compare against,
119 # so only a change to its requirement tells the two branches apart
120 elif name in changed:
121 introduced.add((name, version))
122 return introduced
123
124
125def resolved_findings(
126 findings: set[tuple[str, str]], resolved: dict[str, set[str]]
127) -> set[tuple[str, str]]:
128 """
129 Return the findings the given resolved dependency set installs.
130
131 :param findings: Vulnerable packages as (name, version) pairs.
132 :param resolved: Versions the dependency set resolves to, keyed on package name.
133 """
134 return {
135 (name, version)
136 for name, version in findings
137 # A URL requirement carries no version to compare against, so its presence is
138 # all that can be matched on
139 if (pinned := resolved.get(name)) is not None and (not pinned or version in pinned)
140 }
141
142
143def audit_status(
144 audit: str, requirements: str, base_closure: str = "", head_closure: str = ""
145) -> str:
146 """
147 Return `pass`, `preexisting` or `fail` for the given audit report.
148
149 Findings the pull request's own resolution does not carry are named on stderr.
150
151 :param audit: A pip-audit report in JSON format.
152 :param requirements: Requirement lines the pull request adds or changes, one per line.
153 :param base_closure: The target branch's resolved dependency set as pinned requirement
154 lines. When empty, only the changed requirement lines are compared.
155 :param head_closure: The pull request's own resolved dependency set as pinned requirement
156 lines. When empty, every introduced finding gates.
157 """
158 findings = vulnerable_packages(audit)
159 if not findings:
160 return "pass"
161
162 if base_closure.strip():
163 changed = changed_packages(requirements)
164 introduced = introduced_packages(findings, _resolution(base_closure), changed)
165 # The environment is installed before either branch is resolved, so a release
166 # published in between reads as introduced. Gate only on the findings the pull
167 # request's own resolution carries as well.
168 if head_closure.strip():
169 head_resolution = _resolution(head_closure)
170 carried = resolved_findings(introduced, head_resolution)
171 _report_set_aside(introduced - carried, head_resolution)
172 introduced = carried
173 return "fail" if introduced else "preexisting"
174
175 changed = changed_packages(requirements)
176 return "fail" if {name for name, _ in findings} & changed else "preexisting"
177
178
179def main(argv: list[str] | None = None) -> int:
180 """Print the audit status for the given report and dependency sets."""
181 parser = argparse.ArgumentParser(description=__doc__)
182 parser.add_argument("audit", type=Path, help="pip-audit report in JSON format.")
183 parser.add_argument(
184 "requirements", type=Path, help="Requirement lines the pull request adds or changes."
185 )
186 parser.add_argument(
187 "base_closure",
188 type=Path,
189 nargs="?",
190 help="Resolved dependency set of the branch the pull request targets.",
191 )
192 parser.add_argument(
193 "head_closure",
194 type=Path,
195 nargs="?",
196 help="Resolved dependency set of the pull request itself.",
197 )
198 args = parser.parse_args(argv)
199
200 # The requirements file is only written when the pull request changes dependencies,
201 # the resolved sets only when resolving them succeeded
202 print(
203 audit_status(
204 args.audit.read_text(),
205 _read_optional(args.requirements),
206 _read_optional(args.base_closure),
207 _read_optional(args.head_closure),
208 )
209 )
210 return 0
211
212
213def _describe_resolution(pinned: set[str] | None) -> str:
214 """Describe what a resolved dependency set holds for a package."""
215 # A package the set carries without an exact version is never set aside, so the only
216 # findings described here are the ones it resolves to another version or not at all
217 if pinned is None:
218 return "not in the resolved set"
219 return "resolved to " + ", ".join(sorted(pinned))
220
221
222def _read_optional(path: Path | None) -> str:
223 """Return the contents of a file the workflow writes conditionally."""
224 return path.read_text() if path and path.is_file() else ""
225
226
227def _report_set_aside(set_aside: set[tuple[str, str]], resolved: dict[str, set[str]]) -> None:
228 """Name the findings that were not gated on, and what the resolution carries instead."""
229 if not set_aside:
230 return
231 described = ", ".join(
232 f"`{name} {version}` ({_describe_resolution(resolved.get(name))})"
233 for name, version in sorted(set_aside)
234 )
235 print(
236 "Not counted against this PR, because its own resolved dependencies do not"
237 f" carry them: {described}.",
238 file=sys.stderr,
239 )
240
241
242def _resolution(closure: str) -> dict[str, set[str]]:
243 """Return the versions the given resolved dependency set pins."""
244 resolved = requirement_versions(closure)
245 # A resolution that cannot be read would silently reclassify every finding
246 if not resolved:
247 raise ValueError("No packages found in the resolved dependency set")
248 return resolved
249
250
251if __name__ == "__main__":
252 raise SystemExit(main())
253