Coverage for .github / scripts / check_npm_audit.py: 100.00%
137 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1#!/usr/bin/env python3
2"""Gate npm-audit JSON with exact, expiring suppressions.
4Suppression file format (one entry per non-comment line)::
6 package-dir|package|advisory|node-path|exp:YYYY-MM-DD
8A finding is suppressed only when all four identity fields match. Suppressions
9expire inclusively and stale entries fail, forcing their removal after the
10upstream dependency is fixed.
12When no fixed version of a package exists anywhere in range, npm also reports
13every dependent whose in-range versions all reach it ("depends on vulnerable
14versions of X") as a record of its own, with no advisory, naming the packages
15it reaches through. Such a chain record needs no entry: it clears once every
16package it reaches through has cleared, and fails closed otherwise.
17"""
19from __future__ import annotations
21import argparse
22import datetime as dt
23import json
24import re
25import sys
26from dataclasses import dataclass
27from pathlib import Path
28from typing import Any
30ADVISORY_RE = re.compile(r"^https://github\.com/advisories/(GHSA-[0-9a-z-]+)$")
31SEVERITY = {"info": 0, "low": 1, "moderate": 2, "high": 3, "critical": 4}
32# Findings at this severity or above fail the job unless suppressed. Moderate
33# is where npm's advisories stop being mostly ReDoS-in-a-dev-tool noise and
34# start describing things a proxy or CLI could actually be hit with; every
35# owned graph is clean here, so the gate costs nothing until an advisory lands.
36FAIL_AT = "moderate"
39@dataclass(frozen=True)
40class Suppression:
41 package_dir: str
42 package: str
43 advisory: str
44 node_path: str
45 expires: dt.date
46 line: int
48 @property
49 def identity(self) -> tuple[str, str, str, str]:
50 return (self.package_dir, self.package, self.advisory, self.node_path)
53def _load_suppressions(path: Path, today: dt.date) -> list[Suppression]:
54 if not path.is_file():
55 raise ValueError(f"suppression file not found: {path}")
57 suppressions: list[Suppression] = []
58 identities: set[tuple[str, str, str, str]] = set()
59 for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1):
60 line = raw_line.strip()
61 if not line or line.startswith("#"):
62 continue
64 fields = line.split("|")
65 if len(fields) != 5 or not fields[4].startswith("exp:"):
66 raise ValueError(
67 f"{path}:{line_number}: expected package-dir|package|advisory|"
68 "node-path|exp:YYYY-MM-DD"
69 )
70 package_dir, package, advisory, node_path, expiry_field = fields
71 if not all((package_dir, package, advisory, node_path)):
72 raise ValueError(f"{path}:{line_number}: suppression fields cannot be empty")
73 try:
74 expires = dt.date.fromisoformat(expiry_field.removeprefix("exp:"))
75 except ValueError as exc:
76 raise ValueError(f"{path}:{line_number}: invalid expiration date") from exc
77 if expires <= today:
78 raise ValueError(f"{path}:{line_number}: {advisory} expired on {expires.isoformat()}")
80 suppression = Suppression(
81 package_dir=package_dir,
82 package=package,
83 advisory=advisory,
84 node_path=node_path,
85 expires=expires,
86 line=line_number,
87 )
88 if suppression.identity in identities:
89 raise ValueError(f"{path}:{line_number}: duplicate suppression")
90 identities.add(suppression.identity)
91 suppressions.append(suppression)
92 return suppressions
95def _advisories(via: Any) -> set[str]:
96 advisories: set[str] = set()
97 if not isinstance(via, list):
98 return advisories
99 for item in via:
100 if not isinstance(item, dict):
101 continue
102 match = ADVISORY_RE.fullmatch(str(item.get("url", "")))
103 if match:
104 advisories.add(match.group(1))
105 return advisories
108def _reached_through(via: Any) -> set[str]:
109 """Packages a record is vulnerable through (npm lists them as bare names)."""
110 if not isinstance(via, list):
111 return set()
112 return {item for item in via if isinstance(item, str)}
115def _load_report(path: Path) -> dict[str, Any]:
116 try:
117 report = json.loads(path.read_text(encoding="utf-8"))
118 except (OSError, json.JSONDecodeError) as exc:
119 raise ValueError(f"invalid npm-audit JSON in {path}: {exc}") from exc
120 if not isinstance(report, dict):
121 raise ValueError("npm-audit report must be a JSON object")
122 if isinstance(report.get("error"), dict):
123 raise ValueError(f"npm audit returned an operational error: {report.get('error')!r}")
124 if not isinstance(report.get("vulnerabilities"), dict):
125 raise ValueError("npm-audit report is missing the vulnerabilities object")
126 return report
129def check_report(report: dict[str, Any], package_dir: str, suppressions: list[Suppression]) -> int:
130 scoped = {item.identity: item for item in suppressions if item.package_dir == package_dir}
131 used: set[tuple[str, str, str, str]] = set()
132 failures: list[str] = []
133 # Packages whose own advisories are all suppressed. A chain record clears
134 # only once every package it reaches through is in here.
135 cleared: set[str] = set()
136 # Chain records still waiting on the packages they reach through.
137 pending: dict[str, tuple[set[str], str, set[str]]] = {}
139 for package, finding in report["vulnerabilities"].items():
140 if not isinstance(finding, dict):
141 failures.append(f"{package}: malformed vulnerability record")
142 continue
143 severity = str(finding.get("severity", ""))
144 if SEVERITY.get(severity, -1) < SEVERITY[FAIL_AT]:
145 # Below the gate by policy. It still has to count as a cleared
146 # link: a dependent that npm marks high because of one suppressed
147 # chain can also be "vulnerable through" a low-severity package
148 # (markdownlint-cli2 via markdownlint via katex, 2026-10-05), and
149 # the gate that ignores the low record must not hold the chain
150 # record hostage to it.
151 cleared.add(package)
152 continue
154 advisories = _advisories(finding.get("via"))
155 reached = _reached_through(finding.get("via"))
156 nodes = set(finding.get("nodes", [])) if isinstance(finding.get("nodes"), list) else set()
157 matches = {
158 identity
159 for identity in scoped
160 if identity[1] == package and identity[2] in advisories and identity[3] in nodes
161 }
162 # Fail closed for compound records: every advisory and node must be
163 # represented, and no broader package-level suppression is accepted.
164 expected = {
165 (package_dir, package, advisory, node) for advisory in advisories for node in nodes
166 }
167 if advisories and nodes and matches == expected:
168 used.update(matches)
169 for identity in sorted(matches):
170 suppression = scoped[identity]
171 print(
172 "::warning::Temporarily suppressing "
173 f"{suppression.advisory} for {suppression.node_path} "
174 f"until {suppression.expires.isoformat()}"
175 )
176 elif advisories or not (reached and nodes):
177 failures.append(
178 f"{package}: unsuppressed {severity} finding "
179 f"(advisories={sorted(advisories)}, nodes={sorted(nodes)})"
180 )
181 continue
182 # The record's own advisories, if any, are suppressed; whether it clears
183 # now depends only on the packages it reaches through.
184 if reached:
185 pending[package] = (reached, severity, nodes)
186 else:
187 cleared.add(package)
189 # Chain records clear from the bottom up, one link per pass.
190 while pending:
191 resolved = sorted(
192 package for package, (reached, _, _) in pending.items() if reached <= cleared
193 )
194 if not resolved:
195 break
196 for package in resolved:
197 reached, _, nodes = pending.pop(package)
198 print(
199 f"::warning::Accepting {package} at {', '.join(sorted(nodes))}: "
200 "no advisory of its own, vulnerable only through "
201 f"{', '.join(sorted(reached))}"
202 )
203 cleared.add(package)
205 for package, (reached, severity, nodes) in sorted(pending.items()):
206 failures.append(
207 f"{package}: unsuppressed {severity} finding reached through "
208 f"{sorted(reached - cleared)} (nodes={sorted(nodes)})"
209 )
211 stale = set(scoped) - used
212 for identity in sorted(stale):
213 suppression = scoped[identity]
214 failures.append(
215 f"stale suppression at line {suppression.line}: {suppression.advisory} "
216 f"for {suppression.node_path} no longer matches npm audit"
217 )
219 for failure in failures:
220 print(f"ERROR: {failure}", file=sys.stderr)
221 return 1 if failures else 0
224def main(argv: list[str] | None = None) -> int:
225 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
226 parser.add_argument("--report", required=True, type=Path)
227 parser.add_argument("--package-dir", required=True)
228 parser.add_argument("--ignore-file", required=True, type=Path)
229 args = parser.parse_args(argv)
231 try:
232 suppressions = _load_suppressions(args.ignore_file, dt.datetime.now(dt.UTC).date())
233 report = _load_report(args.report)
234 except ValueError as exc:
235 print(f"ERROR: {exc}", file=sys.stderr)
236 return 1
237 return check_report(report, args.package_dir, suppressions)
240if __name__ == "__main__":
241 raise SystemExit(main())