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