Integrate reviewed bounded npm advisory diagnostics

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/deployment-installability.md

@@ -4645,7 +4645,16 @@ no JavaScript. It separates the two and nothing else. A conclusive result is
 acted on immediately and any vulnerability at any severity still fails, even
 if the same response also carries a transport error, so a severity threshold
 must never be introduced; only an unreachable endpoint is
-retried. Retrying is bounded by wall clock and not by attempt count alone,
+retried. Finding diagnostics must use that same captured JSON response, never
+a second registry request: such a request could hang outside the watchdog or
+describe a different verdict. Logs retain the affected package, severity,
+range, locked paths, advisory source/title/link and fix availability through
+allowlisted, JSON-escaped fields, not arbitrary registry text or transport-error
+details. Missing package-level detail leaves the vulnerability failure intact;
+it must not trigger another request. `scripts/tests/test_npm_audit_retry.py`
+executes changed-second-response, missing-detail, escaping and single-request
+fixtures, including production argument forwarding and findings accompanied by
+transport errors. Retrying is bounded by wall clock and not by attempt count alone,
 because npm's own `fetch-timeout` defaults to five minutes and it retries
 internally: on 2026-09-04 three attempts against a hanging endpoint ran for
 10m56s and cancelled the Frontend job at its own timeout with every test

scripts/npm-audit-retry.sh

@@ -111,7 +111,8 @@ run_audit() {
   return 0
 }
 
-# Classify one audit run. Prints a verdict word on stdout:
+# Classify one audit run. Prints a verdict word, summary and (for a finding)
+# allowlisted, JSON-escaped package/advisory details from that same response:
 #   clean          — audit completed, no vulnerabilities
 #   vulnerable     — audit completed, vulnerabilities present
 #   unreachable    — npm could not get an answer from the advisory endpoint
@@ -141,6 +142,42 @@ if isinstance(vulns, dict) and "total" in vulns:
     )
     print("vulnerable" if total else "clean")
     print(f"total={total} {detail}")
+    if total:
+        # Do not query npm again for human-readable detail. A second request
+        # can hang outside the watchdog or return different advisory evidence.
+        # JSON encoding keeps registry text from becoming terminal escapes or
+        # GitHub workflow commands; transport errors and unknown fields stay out.
+        findings = report.get("vulnerabilities")
+        emitted = False
+        if isinstance(findings, dict):
+            for name, finding in sorted(findings.items()):
+                if not isinstance(finding, dict):
+                    continue
+                projected = {"name": name}
+                for key in ("name", "severity", "isDirect", "range", "nodes"):
+                    if key in finding:
+                        projected[key] = finding[key]
+                via = finding.get("via")
+                if isinstance(via, list):
+                    projected["via"] = [
+                        {key: advisory[key] for key in
+                         ("source", "name", "dependency", "title", "url", "severity", "range")
+                         if key in advisory}
+                        if isinstance(advisory, dict) else advisory
+                        for advisory in via if isinstance(advisory, (dict, str))
+                    ]
+                fix = finding.get("fixAvailable")
+                if isinstance(fix, bool):
+                    projected["fixAvailable"] = fix
+                elif isinstance(fix, dict):
+                    projected["fixAvailable"] = {
+                        key: fix[key] for key in ("name", "version", "isSemVerMajor")
+                        if key in fix
+                    }
+                print("audit finding " + json.dumps(projected, ensure_ascii=True, sort_keys=True))
+                emitted = True
+        if not emitted:
+            print("npm audit: package-level detail unavailable in captured verdict")
     sys.exit(0)
 
 if isinstance(report, dict) and report.get("error"):
@@ -194,9 +231,9 @@ while [ "${attempt}" -le "${ATTEMPTS}" ]; do
     vulnerable)
       echo "npm audit (${SCOPE}): vulnerabilities present (${summary})"
       echo "::error::npm audit (${SCOPE}) found vulnerabilities: ${summary}"
-      # Re-run without --json so the log carries the human-readable advisory
-      # detail a maintainer needs to act on.
-      "${NPM_BIN}" audit "${AUDIT_ARGS[@]}" "${SCOPE_ARGS[@]}" || true
+      # Keep diagnostics bound to the conclusive response, with no extra
+      # registry request outside the attempt/total wall-clock limits.
+      printf '%s\n' "${verdict_output}" | tail -n +3
       exit 1
       ;;
     *)

scripts/tests/test_npm_audit_retry.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 import os
 from pathlib import Path
 import subprocess
@@ -17,11 +18,15 @@ SCRIPT = ROOT / "scripts" / "npm-audit-retry.sh"
 
 
 class NpmAuditRetryTest(unittest.TestCase):
-    def run_check(self, mode: str, *arguments: str, require: str = "true"):
+    def run_check(
+        self, mode: str, *arguments: str, require: str = "true", report=None
+    ):
         with tempfile.TemporaryDirectory() as directory:
             fake_bin = Path(directory)
             count = fake_bin / "count"
             calls = fake_bin / "calls"
+            report_path = fake_bin / "report.json"
+            report_path.write_text(json.dumps(report), encoding="utf-8")
             count.write_text("0\n", encoding="utf-8")
             fake_npm = fake_bin / "npm"
             fake_npm.write_text(
@@ -61,6 +66,15 @@ class NpmAuditRetryTest(unittest.TestCase):
                         printf '%s\n' "$vulnerable_with_error"
                         exit 1
                         ;;
+                      captured-vulnerability)
+                        if [ "$count" -eq 1 ]; then
+                          cat "$FAKE_NPM_REPORT"
+                        else
+                          # A diagnostic re-query could return a different verdict.
+                          printf '%s\n' "$clean"
+                        fi
+                        exit 1
+                        ;;
                       garbage)
                         echo 'not json'
                         exit 1
@@ -78,6 +92,7 @@ class NpmAuditRetryTest(unittest.TestCase):
                     "FAKE_NPM_CALLS": str(calls),
                     "FAKE_NPM_COUNT": str(count),
                     "FAKE_NPM_MODE": mode,
+                    "FAKE_NPM_REPORT": str(report_path),
                     "NPM_AUDIT_ATTEMPTS": "3",
                     "NPM_AUDIT_CMD": str(fake_npm),
                     "NPM_AUDIT_REQUIRE_RESULT": require,
@@ -123,10 +138,7 @@ class NpmAuditRetryTest(unittest.TestCase):
         self.assertEqual(result.returncode, 1)
         self.assertEqual(
             calls,
-            [
-                "audit --json",
-                "audit",
-            ],
+            ["audit --json"],
         )
         self.assertIn("vulnerabilities present", result.stdout)
         self.assertNotIn("retrying", result.stdout)
@@ -134,10 +146,96 @@ class NpmAuditRetryTest(unittest.TestCase):
     def test_vulnerability_verdict_precedes_a_transport_error(self) -> None:
         result, calls = self.run_check("vulnerability-with-error", "all")
         self.assertEqual(result.returncode, 1)
-        self.assertEqual(len(calls), 2)
+        self.assertEqual(calls, ["audit --json"])
         self.assertIn("vulnerabilities present", result.stdout)
         self.assertNotIn("retrying", result.stdout)
 
+    def test_reports_the_original_advisory_without_an_unbounded_requery(self) -> None:
+        nodes = [
+            "node_modules/@eslint/config-array/node_modules/brace-expansion",
+            "node_modules/@eslint/eslintrc/node_modules/brace-expansion",
+            "node_modules/brace-expansion",
+            "node_modules/eslint/node_modules/brace-expansion",
+        ]
+        report = {
+            "metadata": {"vulnerabilities": {"high": 1, "total": 1}},
+            "vulnerabilities": {
+                "brace-expansion": {
+                    "name": "brace-expansion",
+                    "severity": "high",
+                    "isDirect": False,
+                    "range": "<=1.1.20 || 4.0.0 - 5.0.11",
+                    "nodes": nodes,
+                    "fixAvailable": True,
+                    "via": [
+                        {
+                            "source": 123456,
+                            "name": "brace-expansion",
+                            "title": "Quadratic-time expansion causes CPU denial of service",
+                            "url": "https://github.com/advisories/GHSA-q2hr-2g5m-vwhr",
+                            "severity": "high",
+                            "range": "<=1.1.20",
+                            "unrecognised": "must not be printed",
+                        },
+                        "transitive-dependency",
+                    ],
+                }
+            },
+            "error": {"detail": "must not be printed"},
+        }
+        result, calls = self.run_check(
+            "captured-vulnerability", "all", require="false", report=report
+        )
+        self.assertEqual(result.returncode, 1)
+        self.assertEqual(calls, ["audit --json"])
+        finding_lines = [
+            line.removeprefix("audit finding ")
+            for line in result.stdout.splitlines()
+            if line.startswith("audit finding ")
+        ]
+        self.assertEqual(len(finding_lines), 1, result.stdout)
+        finding = json.loads(finding_lines[0])
+        self.assertEqual(finding["nodes"], nodes)
+        self.assertIs(finding["fixAvailable"], True)
+        self.assertEqual(finding["name"], "brace-expansion")
+        self.assertEqual(finding["range"], "<=1.1.20 || 4.0.0 - 5.0.11")
+        expected_advisory = report["vulnerabilities"]["brace-expansion"]["via"][0]
+        self.assertEqual(
+            finding["via"][0],
+            {key: value for key, value in expected_advisory.items() if key != "unrecognised"},
+        )
+        self.assertEqual(finding["via"][1], "transitive-dependency")
+        self.assertNotIn("must not be printed", result.stdout)
+        self.assertNotIn("no vulnerabilities", result.stdout)
+
+    def test_missing_advisory_detail_cannot_trigger_a_requery_or_clear_the_failure(self) -> None:
+        result, calls = self.run_check("vulnerability", "all")
+        self.assertEqual(result.returncode, 1)
+        self.assertEqual(calls, ["audit --json"])
+        self.assertIn("package-level detail unavailable", result.stdout)
+
+    def test_escapes_advisory_text_instead_of_emitting_workflow_commands(self) -> None:
+        title = "unsafe title\n::error::injected annotation\x1b[31m"
+        report = {
+            "metadata": {"vulnerabilities": {"high": 1, "total": 1}},
+            "vulnerabilities": {
+                "dependency": {"name": "dependency", "via": [{"title": title}]}
+            },
+        }
+        result, calls = self.run_check(
+            "captured-vulnerability", "production", "--package-lock-only", report=report
+        )
+        self.assertEqual(result.returncode, 1)
+        self.assertEqual(calls, ["audit --json --package-lock-only --omit=dev"])
+        finding_line = next(
+            (line for line in result.stdout.splitlines() if line.startswith("audit finding ")),
+            "audit finding {}",
+        )
+        finding = json.loads(finding_line.removeprefix("audit finding "))
+        self.assertEqual(finding.get("via"), [{"title": title}], result.stdout)
+        self.assertNotIn("\n::error::injected", result.stdout)
+        self.assertNotIn("\x1b", result.stdout)
+
     def test_persistent_outage_fails_when_a_result_is_required(self) -> None:
         result, calls = self.run_check("transient-failure", "all")
         self.assertEqual(result.returncode, 1)