mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-23 20:25:56 +00:00
Watch main and active release lines daily for new npm advisories
On 30 Sep-1 Oct 2026 GHSA-q2hr-2g5m-vwhr (brace-expansion) and GHSA-p98j-92pf-mc4p (DOMPurify) failed the required "Audit complete frontend dependency graph" step on every pull request to main, release/v6.4 and release/v6.5. That held the v6.4.6 preparation PR and the v6.5 RC for days with no owner. The scheduled audit in security-scan.yml only informs and runs on the default branch, so release lines were never checked. dependency-advisory-watch.yml runs daily and on dispatch. It lists main plus every release/v<major>.<minor> line at or newer than the latest stable's line (all lines if that lookup fails), then audits each in a fail-fast-free matrix. Each job fetches only that line's frontend-modern/package.json and package-lock.json with git and runs scripts/npm-audit-retry.sh all on them under the same Node.js pin that build-and-test requires. npm audit reads the lockfile, so nothing is installed. The job never checks out or runs the audited branch's code, because a scheduled run holds the default branch's cache scope and CodeQL flagged running npm ci there as cache poisoning. A job fails when the audit fails, and its step summary and annotation name the branch, the GHSA ids and the fix (npm audit fix --package-lock-only plus raised floors in dependencySecurity.test.ts). Read-only, hosted-only, no secrets, no uploads.
This commit is contained in:
committed by
rcourtman
parent
e5082c5fb1
commit
64d8dbc7fc
@@ -0,0 +1,173 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Helpers for the daily dependency advisory watch workflow.
|
||||
|
||||
`lines` turns `git ls-remote --heads` output and the newest stable release tag
|
||||
into the JSON list of branches the watch audits: `main` plus every
|
||||
`release/v<major>.<minor>` line at or newer than the newest stable's line.
|
||||
Without a usable stable tag every release line is kept, so a lookup failure
|
||||
widens the watch instead of silently narrowing it.
|
||||
|
||||
`summary` reads the output of `scripts/npm-audit-retry.sh` for one branch and
|
||||
writes the step summary plus one failure annotation naming the branch, the
|
||||
advisories and the fix. Registry-sourced text never reaches the annotation:
|
||||
only the validated branch name, package names and GHSA identifiers do.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
LINE_RE = re.compile(r"^release/v(\d+)\.(\d+)$")
|
||||
STABLE_RE = re.compile(r"^v(\d+)\.(\d+)\.(\d+)$")
|
||||
GHSA_RE = re.compile(r"GHSA(?:-[23456789cfghjmpqrvwx]{4}){3}")
|
||||
PACKAGE_RE = re.compile(r"^(?:@[a-z0-9][a-z0-9._~-]*/)?[a-z0-9][a-z0-9._~-]*$")
|
||||
FINDING_PREFIX = "audit finding "
|
||||
UNREACHABLE_MARKER = "could not reach the advisory endpoint"
|
||||
FIX_STEPS = (
|
||||
"In frontend-modern on that branch run `npm audit fix --package-lock-only`, "
|
||||
"then raise the matching floors in "
|
||||
"`frontend-modern/src/security/__tests__/dependencySecurity.test.ts` "
|
||||
"and land both on the branch."
|
||||
)
|
||||
|
||||
|
||||
def release_lines(ls_remote: str, latest_stable: str) -> list[str]:
|
||||
lines: dict[tuple[int, int], str] = {}
|
||||
for row in ls_remote.splitlines():
|
||||
parts = row.split()
|
||||
if len(parts) != 2 or not parts[1].startswith("refs/heads/"):
|
||||
continue
|
||||
branch = parts[1][len("refs/heads/"):]
|
||||
match = LINE_RE.fullmatch(branch)
|
||||
if match:
|
||||
lines[(int(match.group(1)), int(match.group(2)))] = branch
|
||||
stable = STABLE_RE.fullmatch(latest_stable.strip())
|
||||
floor = (int(stable.group(1)), int(stable.group(2))) if stable else (0, 0)
|
||||
active = [lines[key] for key in sorted(lines) if key >= floor]
|
||||
return ["main", *active]
|
||||
|
||||
|
||||
def parse_findings(log: str) -> list[dict]:
|
||||
findings = []
|
||||
for row in log.splitlines():
|
||||
row = row.strip()
|
||||
if not row.startswith(FINDING_PREFIX):
|
||||
continue
|
||||
try:
|
||||
finding = json.loads(row[len(FINDING_PREFIX):])
|
||||
except ValueError:
|
||||
continue
|
||||
if isinstance(finding, dict):
|
||||
findings.append(finding)
|
||||
return findings
|
||||
|
||||
|
||||
def advisory_ids(finding: dict) -> list[str]:
|
||||
found = set()
|
||||
for item in finding.get("via") or []:
|
||||
if isinstance(item, dict):
|
||||
found.update(GHSA_RE.findall(str(item.get("url") or "")))
|
||||
return sorted(found)
|
||||
|
||||
|
||||
def safe_package(name: object) -> str:
|
||||
text = str(name or "")
|
||||
return text if PACKAGE_RE.fullmatch(text) else "unnamed-package"
|
||||
|
||||
|
||||
def markdown_cell(value: object) -> str:
|
||||
text = " ".join(str(value or "").split())[:160]
|
||||
for character in "\\|`*_[]<>":
|
||||
text = text.replace(character, "\\" + character)
|
||||
return text
|
||||
|
||||
|
||||
def render(branch: str, outcome: str, log: str) -> tuple[str, str | None]:
|
||||
"""Return the step summary and the annotation (None when clean)."""
|
||||
findings = parse_findings(log)
|
||||
header = f"## Frontend dependency audit: `{branch}`\n\n"
|
||||
if outcome == "success":
|
||||
note = ""
|
||||
if UNREACHABLE_MARKER in log:
|
||||
note = " The advisory endpoint did not answer, so this run proves nothing new."
|
||||
return header + f"No advisories reported.{note}\n", None
|
||||
if not findings:
|
||||
if UNREACHABLE_MARKER in log:
|
||||
text = (f"The npm advisory endpoint could not be reached for {branch}. "
|
||||
"No advisory was observed; rerun the watch once npm recovers.")
|
||||
else:
|
||||
text = (f"The frontend dependency audit for {branch} failed without an advisory finding. "
|
||||
"Read the job log: install or audit tooling broke before a verdict.")
|
||||
return header + text + "\n", text
|
||||
rows = ["| Package | Severity | Advisories | Fix available |", "| --- | --- | --- | --- |"]
|
||||
identifiers: set[str] = set()
|
||||
packages = []
|
||||
for finding in findings:
|
||||
package = safe_package(finding.get("name"))
|
||||
packages.append(package)
|
||||
ids = advisory_ids(finding)
|
||||
identifiers.update(ids)
|
||||
fix = finding.get("fixAvailable")
|
||||
if isinstance(fix, dict):
|
||||
fix_text = f"{fix.get('name', '')} {fix.get('version', '')}".strip()
|
||||
if fix.get("isSemVerMajor"):
|
||||
fix_text += " (major)"
|
||||
else:
|
||||
fix_text = "yes" if fix is True else "no"
|
||||
titles = [str(item.get("title") or "") for item in finding.get("via") or [] if isinstance(item, dict)]
|
||||
rows.append("| " + " | ".join((
|
||||
markdown_cell(package), markdown_cell(finding.get("severity")),
|
||||
markdown_cell(", ".join(ids) or "; ".join(titles) or "via dependency"),
|
||||
markdown_cell(fix_text))) + " |")
|
||||
advisories = ", ".join(sorted(identifiers)) or "see job log"
|
||||
body = (header
|
||||
+ f"New npm advisories block every pull request and release on `{branch}` through the "
|
||||
+ "required *Audit complete frontend dependency graph* check in build-and-test.yml.\n\n"
|
||||
+ "\n".join(rows)
|
||||
+ "\n\n**Suggested fix.** " + FIX_STEPS + "\n")
|
||||
annotation = (f"Frontend dependency advisory on {branch}: {advisories} in "
|
||||
f"{', '.join(sorted(set(packages)))}. This blocks every PR and release on {branch}. "
|
||||
"The fix is a lockfile advisory floor bump on that line: npm audit fix --package-lock-only "
|
||||
"in frontend-modern, then raise the floors in dependencySecurity.test.ts.")
|
||||
return body, annotation
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
commands = parser.add_subparsers(dest="command", required=True)
|
||||
lines = commands.add_parser("lines")
|
||||
lines.add_argument("--heads", required=True, type=Path, help="git ls-remote --heads output")
|
||||
lines.add_argument("--latest-stable", default="", help="newest stable release tag, if known")
|
||||
summary = commands.add_parser("summary")
|
||||
summary.add_argument("--branch", required=True)
|
||||
summary.add_argument("--outcome", required=True)
|
||||
summary.add_argument("--log", required=True, type=Path)
|
||||
summary.add_argument("--summary-file", required=True, type=Path)
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
if args.command == "lines":
|
||||
print(json.dumps(release_lines(args.heads.read_text(encoding="utf-8"), args.latest_stable)))
|
||||
return 0
|
||||
|
||||
if args.branch != "main" and not LINE_RE.fullmatch(args.branch):
|
||||
print(f"refusing unexpected branch name {args.branch!r}", file=sys.stderr)
|
||||
return 2
|
||||
try:
|
||||
log = args.log.read_text(encoding="utf-8", errors="replace")
|
||||
except OSError:
|
||||
log = ""
|
||||
body, annotation = render(args.branch, args.outcome, log)
|
||||
with args.summary_file.open("a", encoding="utf-8") as stream:
|
||||
stream.write(body)
|
||||
if annotation:
|
||||
print(f"::error title=Dependency advisory watch ({args.branch})::{annotation}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,215 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
import yaml
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
SCRIPT = ROOT / "scripts" / "dependency_advisory_watch.py"
|
||||
WORKFLOW_PATH = ROOT / ".github" / "workflows" / "dependency-advisory-watch.yml"
|
||||
BUILD_AND_TEST = ROOT / ".github" / "workflows" / "build-and-test.yml"
|
||||
SPEC = importlib.util.spec_from_file_location("dependency_advisory_watch", SCRIPT)
|
||||
watch = importlib.util.module_from_spec(SPEC)
|
||||
assert SPEC.loader is not None
|
||||
SPEC.loader.exec_module(watch)
|
||||
|
||||
HEADS = "\n".join((
|
||||
"29069d4139ef908862a179e1b0dd05364d895776\trefs/heads/release/5.1",
|
||||
"e261cd3eb8c2abed21357721df6506d30e2a0ec8\trefs/heads/release/v6.3.2",
|
||||
"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa\trefs/heads/release/v6.3",
|
||||
"b50aeb6a8d9e38ab362b23f4ed15a010f675a90f\trefs/heads/release/v6.4",
|
||||
"26476f317a68746fd353967470cd8be3bc6b7670\trefs/heads/release/v6.4.2",
|
||||
"5f88b2002d1e1085053bf7bf3969aa61cdb0f637\trefs/heads/release/v6.5",
|
||||
"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb\trefs/heads/release/v6.10",
|
||||
"cccccccccccccccccccccccccccccccccccccccc\trefs/heads/release/v6.5-hotfix",
|
||||
))
|
||||
|
||||
FINDING = {
|
||||
"name": "dompurify", "severity": "moderate", "isDirect": True, "range": "<3.2.8",
|
||||
"via": [{"source": 1, "name": "dompurify", "title": "DOMPurify `bypass` | <x>",
|
||||
"url": "https://github.com/advisories/GHSA-p98j-92pf-mc4p", "severity": "moderate"}],
|
||||
"fixAvailable": {"name": "dompurify", "version": "3.2.8", "isSemVerMajor": False},
|
||||
}
|
||||
TRANSITIVE = {
|
||||
"name": "brace-expansion", "severity": "low",
|
||||
"via": [{"url": "https://github.com/advisories/GHSA-q2hr-2g5m-vwhr", "title": "ReDoS"}, "minimatch"],
|
||||
"fixAvailable": True,
|
||||
}
|
||||
|
||||
|
||||
def audit_log(*findings: dict) -> str:
|
||||
rows = ["npm audit (all) attempt 1/3 (limit 60s, 240s of budget left)",
|
||||
"npm audit (all): vulnerabilities present (total=2)",
|
||||
"::error::npm audit (all) found vulnerabilities: total=2"]
|
||||
rows += ["audit finding " + json.dumps(f, sort_keys=True) for f in findings]
|
||||
return "\n".join(rows) + "\n"
|
||||
|
||||
|
||||
class ReleaseLinesTest(unittest.TestCase):
|
||||
def test_newest_stable_line_and_newer_plus_main(self) -> None:
|
||||
self.assertEqual(watch.release_lines(HEADS, "v6.4.5"),
|
||||
["main", "release/v6.4", "release/v6.5", "release/v6.10"])
|
||||
|
||||
def test_patch_and_legacy_branches_are_never_lines(self) -> None:
|
||||
lines = watch.release_lines(HEADS, "")
|
||||
for branch in ("release/5.1", "release/v6.3.2", "release/v6.4.2", "release/v6.5-hotfix"):
|
||||
self.assertNotIn(branch, lines)
|
||||
|
||||
def test_unknown_or_prerelease_stable_keeps_every_line(self) -> None:
|
||||
for latest in ("", "v6.5.0-rc.1", "garbage"):
|
||||
with self.subTest(latest=latest):
|
||||
self.assertEqual(watch.release_lines(HEADS, latest),
|
||||
["main", "release/v6.3", "release/v6.4", "release/v6.5", "release/v6.10"])
|
||||
|
||||
def test_main_is_audited_even_without_release_lines(self) -> None:
|
||||
self.assertEqual(watch.release_lines("", "v6.4.5"), ["main"])
|
||||
|
||||
def test_cli_emits_json_list(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
heads = Path(directory) / "heads"
|
||||
heads.write_text(HEADS, encoding="utf-8")
|
||||
result = subprocess.run([sys.executable, str(SCRIPT), "lines", "--heads", str(heads),
|
||||
"--latest-stable", "v6.5.0"], capture_output=True, text=True, check=True)
|
||||
self.assertEqual(json.loads(result.stdout), ["main", "release/v6.5", "release/v6.10"])
|
||||
|
||||
|
||||
class SummaryTest(unittest.TestCase):
|
||||
def run_summary(self, branch: str, outcome: str, log: str) -> tuple[subprocess.CompletedProcess, str]:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
log_path = Path(directory) / "audit.log"
|
||||
summary = Path(directory) / "summary.md"
|
||||
log_path.write_text(log, encoding="utf-8")
|
||||
result = subprocess.run([sys.executable, str(SCRIPT), "summary", "--branch", branch,
|
||||
"--outcome", outcome, "--log", str(log_path), "--summary-file", str(summary)],
|
||||
capture_output=True, text=True, check=False)
|
||||
return result, summary.read_text(encoding="utf-8") if summary.exists() else ""
|
||||
|
||||
def test_advisory_names_branch_advisories_and_fix(self) -> None:
|
||||
result, summary = self.run_summary("release/v6.4", "failure", audit_log(FINDING, TRANSITIVE))
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertIn("`release/v6.4`", summary)
|
||||
self.assertIn("GHSA-p98j-92pf-mc4p", summary)
|
||||
self.assertIn("GHSA-q2hr-2g5m-vwhr", summary)
|
||||
self.assertIn("npm audit fix --package-lock-only", summary)
|
||||
self.assertIn("dependencySecurity.test.ts", summary)
|
||||
annotations = [line for line in result.stdout.splitlines() if line.startswith("::error")]
|
||||
self.assertEqual(len(annotations), 1)
|
||||
self.assertIn("release/v6.4", annotations[0])
|
||||
self.assertIn("GHSA-p98j-92pf-mc4p, GHSA-q2hr-2g5m-vwhr", annotations[0])
|
||||
self.assertIn("lockfile advisory floor bump", annotations[0])
|
||||
|
||||
def test_registry_text_is_escaped_and_kept_out_of_annotation(self) -> None:
|
||||
hostile = dict(FINDING, name="evil\n::warning::x", via=[{"title": "x | <script>", "url": "javascript:"}])
|
||||
result, summary = self.run_summary("main", "failure", audit_log(hostile))
|
||||
self.assertNotIn("<script>", summary)
|
||||
self.assertIn("\\<script\\>", summary)
|
||||
self.assertIn("unnamed-package", result.stdout)
|
||||
self.assertEqual(sum(line.startswith("::") for line in result.stdout.splitlines()), 1)
|
||||
self.assertNotIn("script", result.stdout)
|
||||
|
||||
def test_clean_run_has_no_annotation(self) -> None:
|
||||
result, summary = self.run_summary("release/v6.5", "success", "npm audit (all): no vulnerabilities\n")
|
||||
self.assertNotIn("::error", result.stdout)
|
||||
self.assertIn("No advisories reported.", summary)
|
||||
|
||||
def test_unreachable_endpoint_is_not_called_an_advisory(self) -> None:
|
||||
log = "::error::npm audit (all) could not reach the advisory endpoint after 3 attempts\n"
|
||||
result, summary = self.run_summary("main", "failure", log)
|
||||
self.assertIn("could not be reached for main", result.stdout)
|
||||
self.assertNotIn("floor bump", result.stdout)
|
||||
self.assertIn("No advisory was observed", summary)
|
||||
|
||||
def test_install_failure_without_audit_is_reported(self) -> None:
|
||||
result, _ = self.run_summary("release/v6.5", "skipped", "")
|
||||
self.assertIn("failed without an advisory finding", result.stdout)
|
||||
|
||||
def test_unexpected_branch_name_is_refused(self) -> None:
|
||||
result, summary = self.run_summary("feature/x;rm", "failure", audit_log(FINDING))
|
||||
self.assertEqual(result.returncode, 2)
|
||||
self.assertEqual(summary, "")
|
||||
|
||||
|
||||
class WorkflowContractTest(unittest.TestCase):
|
||||
text = WORKFLOW_PATH.read_text(encoding="utf-8")
|
||||
workflow = yaml.safe_load(text)
|
||||
|
||||
def test_triggers_are_daily_schedule_and_dispatch_only(self) -> None:
|
||||
triggers = self.workflow[True]
|
||||
self.assertEqual(set(triggers), {"schedule", "workflow_dispatch"})
|
||||
self.assertRegex(triggers["schedule"][0]["cron"], r"^\d+ \d+ \* \* \*$")
|
||||
|
||||
def test_read_only_hosted_without_secrets_or_uploads(self) -> None:
|
||||
self.assertEqual(self.workflow["permissions"], {"contents": "read"})
|
||||
self.assertNotIn(": write", self.text)
|
||||
self.assertNotIn("secrets.", self.text)
|
||||
self.assertNotIn("upload-artifact", self.text)
|
||||
for job in self.workflow["jobs"].values():
|
||||
self.assertEqual(job["runs-on"], "ubuntu-24.04")
|
||||
for step in job["steps"]:
|
||||
if str(step.get("uses", "")).startswith("actions/checkout@"):
|
||||
self.assertIs(step["with"]["persist-credentials"], False)
|
||||
|
||||
def test_matrix_covers_every_listed_branch_without_fail_fast(self) -> None:
|
||||
audit = self.workflow["jobs"]["audit"]
|
||||
self.assertFalse(audit["strategy"]["fail-fast"])
|
||||
self.assertEqual(audit["strategy"]["matrix"]["branch"], "${{ fromJSON(needs.lines.outputs.branches) }}")
|
||||
read = next(step for step in audit["steps"] if step.get("name") == "Read the audited branch's lockfile inputs")
|
||||
self.assertEqual(read["env"]["BRANCH"], "${{ matrix.branch }}")
|
||||
self.assertIn('git fetch --no-tags --depth=1 origin "+refs/heads/${BRANCH}:refs/remotes/audited/head"', read["run"])
|
||||
self.assertIn("for file in package.json package-lock.json", read["run"])
|
||||
|
||||
def test_never_checks_out_installs_or_runs_another_branchs_code(self) -> None:
|
||||
# A scheduled job runs in the default branch's cache and token scope,
|
||||
# so it reads only the audited line's two lockfile inputs as data:
|
||||
# no checkout of that branch, no install, no dependency cache.
|
||||
audit = self.workflow["jobs"]["audit"]["steps"]
|
||||
checkouts = [step for step in audit if str(step.get("uses", "")).startswith("actions/checkout@")]
|
||||
self.assertEqual(1, len(checkouts))
|
||||
self.assertNotIn("ref", checkouts[0].get("with", {}))
|
||||
self.assertNotIn("npm ci", self.text)
|
||||
self.assertNotIn("npm install", self.text)
|
||||
self.assertNotIn("actions/cache", self.text)
|
||||
self.assertNotRegex(self.text, r"(?m)^\s+cache(-dependency-path)?:")
|
||||
|
||||
def test_audit_matches_the_required_build_and_test_audit(self) -> None:
|
||||
build = yaml.safe_load(BUILD_AND_TEST.read_text(encoding="utf-8"))
|
||||
frontend = build["jobs"]["frontend"]["steps"]
|
||||
audit = self.workflow["jobs"]["audit"]["steps"]
|
||||
|
||||
def step(steps: list[dict], name: str) -> dict:
|
||||
return next(item for item in steps if item.get("name") == name)
|
||||
for name in ("Set up Node.js",):
|
||||
self.assertEqual(step(audit, name)["uses"], step(frontend, name)["uses"])
|
||||
self.assertEqual(step(audit, name)["with"]["node-version"], step(frontend, name)["with"]["node-version"])
|
||||
required = step(frontend, "Audit complete frontend dependency graph")["run"]
|
||||
self.assertIn(required, step(audit, "Audit complete frontend dependency graph")["run"])
|
||||
self.assertNotIn("continue-on-error", step(audit, "Audit complete frontend dependency graph"))
|
||||
self.assertEqual(step(audit, "Audit complete frontend dependency graph")["env"]["NPM_AUDIT_REQUIRE_RESULT"],
|
||||
"true")
|
||||
|
||||
def test_checkout_pins_match_build_and_test(self) -> None:
|
||||
pins = set(re.findall(r"uses: (\S+@[0-9a-f]{40})", BUILD_AND_TEST.read_text(encoding="utf-8")))
|
||||
for pin in re.findall(r"uses: (\S+@\S+)", self.text):
|
||||
self.assertIn(pin, pins)
|
||||
|
||||
def test_matrix_values_reach_shell_only_through_env(self) -> None:
|
||||
for line in self.text.splitlines():
|
||||
if "${{" in line and ("matrix." in line or "steps." in line or "github." in line):
|
||||
stripped = line.strip()
|
||||
if stripped.startswith(("name:", "ref:", "branch:", "branches:")):
|
||||
continue
|
||||
self.assertRegex(stripped, r"^[A-Z_]+: \$\{\{ [a-z_.]+ \}\}$")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user