gsadmin/pulse
1
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-10-24 04:33:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Watch main and every active release line for new frontend advisories daily (#2366)

Browse Source 
On 30 Sep and 1 Oct two new advisories (brace-expansion
GHSA-q2hr-2g5m-vwhr, DOMPurify GHSA-p98j-92pf-mc4p) failed the required
frontend audit on every PR across main, release/v6.4 and release/v6.5,
holding the v6.5 RC for days and blocking the v6.4.6 preparation. The
existing scheduled npm-audit is informational and only sees main. This
adds a daily workflow that runs the same complete frontend audit as
Build and Test on main and every active release line, fails on any
finding, and names the branch, advisories and fix. The maintainer's
release-path health check raises it to Delivery immediately.
This commit is contained in:
rcourtman
2026-10-01 12:01:43 +01:00
committed by GitHub
parent 6507bbad8e 64d8dbc7fc
commit 4eeb297ca6
6 changed files with 590 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/dependency_advisory_watch.py
+173
View File
@@ -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())
scripts/tests/test_dependency_advisory_watch.py
+215
View File
@@ -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()