fix: prevent minimum refresh timeouts and handle bulletin scan exits

.github/workflows/minimum-versions-refresh.yml

@@ -106,7 +106,7 @@ jobs:
           # deterministic: it fails every attempt, the generator writes nothing,
           # and the final exit below makes the job red.
           for attempt in 1 2 3; do
-            if python3 -m cmax.minimum_refresh --write "$MINIMUMS_PATH" --generated "$generated"; then
+            if python3 -u -m cmax.minimum_refresh --verbose --write "$MINIMUMS_PATH" --generated "$generated"; then
               exit 0
             fi
             echo "::warning::the minimum refresh failed on attempt ${attempt}"
@@ -260,8 +260,13 @@ jobs:
           report="$RUNNER_TEMP/new-bulletins.txt"
           rc=1
           for attempt in 1 2 3; do
-            python3 -m cmax.minimum_refresh --detect-new-bulletins > "$report" 2>&1
-            rc=$?
+            # GitHub invokes bash with -e. Capture expected exit 3 inside an
+            # if statement so it reaches found=true instead of exiting early.
+            if python3 -u -m cmax.minimum_refresh --detect-new-bulletins > "$report" 2>&1; then
+              rc=0
+            else
+              rc=$?
+            fi
             if [ "$rc" -eq 0 ] || [ "$rc" -eq 3 ]; then
               break
             fi

cmax/minimum_refresh.py

@@ -58,12 +58,16 @@
 import copy
 import difflib
 import json
+import logging
 import os
 import re
 import sys
+import time
 import urllib.error
 import urllib.parse
 import urllib.request
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
 from datetime import datetime, timezone
 from html.parser import HTMLParser
 from pathlib import Path
@@ -415,6 +419,20 @@ class Fetcher:
 
     timeout: int = 60
 
+    @contextmanager
+    def _open(self, request):
+        """Log request progress without printing headers or credentials."""
+        logger = logging.getLogger(__name__)
+        started = time.monotonic()
+        logger.info("fetch %s %s", request.get_method(), request.full_url)
+        try:
+            with SAFE_OPENER.open(request, timeout=self.timeout) as response:
+                yield response
+        finally:
+            logger.info(
+                "finished %s after %.1fs", request.full_url, time.monotonic() - started
+            )
+
     def _headers(self, url: str, **extra: str) -> dict[str, str]:
         headers = {"User-Agent": USER_AGENT, **extra}
         token = os.environ.get("GITHUB_TOKEN")
@@ -425,7 +443,7 @@ def _headers(self, url: str, **extra: str) -> dict[str, str]:
     def get_json(self, url: str) -> Any:
         request = urllib.request.Request(url, headers=self._headers(url))
         try:
-            with SAFE_OPENER.open(request, timeout=self.timeout) as response:
+            with self._open(request) as response:
                 return json.load(response)
         except (urllib.error.URLError, OSError, ValueError) as exc:
             raise MinimumRefreshError(f"cannot read {url}: {exc}") from exc
@@ -440,7 +458,7 @@ def get_optional_json(self, url: str) -> Any | None:
         """
         request = urllib.request.Request(url, headers=self._headers(url))
         try:
-            with SAFE_OPENER.open(request, timeout=self.timeout) as response:
+            with self._open(request) as response:
                 return json.load(response)
         except urllib.error.HTTPError as exc:
             if exc.code == 404:
@@ -456,7 +474,7 @@ def post_json(self, url: str, body: dict) -> Any:
             headers=self._headers(url, **{"Content-Type": "application/json"}),
         )
         try:
-            with SAFE_OPENER.open(request, timeout=self.timeout) as response:
+            with self._open(request) as response:
                 return json.load(response)
         except (urllib.error.URLError, OSError, ValueError) as exc:
             raise MinimumRefreshError(f"cannot query {url}: {exc}") from exc
@@ -472,7 +490,7 @@ def get_text_head(self, url: str, max_bytes: int = 1200) -> str | None:
             headers=self._headers(url, Range=f"bytes=0-{max_bytes - 1}"),
         )
         try:
-            with SAFE_OPENER.open(request, timeout=self.timeout) as response:
+            with self._open(request) as response:
                 if response.status not in (200, 206):
                     return None
                 return response.read(max_bytes).decode("utf-8", "replace")
@@ -488,7 +506,7 @@ def get_text(self, url: str) -> str:
         """
         request = urllib.request.Request(url, headers=self._headers(url))
         try:
-            with SAFE_OPENER.open(request, timeout=self.timeout) as response:
+            with self._open(request) as response:
                 return response.read().decode("utf-8", "replace")
         except (urllib.error.URLError, OSError) as exc:
             raise MinimumRefreshError(f"cannot read {url}: {exc}") from exc
@@ -1215,16 +1233,21 @@ def ubuntu_minimums(
     specs: Sequence[dict] = UBUNTU_PACKAGES,
     fetch: Fetcher | None = None,
 ) -> dict:
-    packages: dict[str, dict] = {}
-    for spec in specs:
+    def resolve(spec: dict) -> tuple[str, dict]:
         entry = ubuntu_entry(spec["cve"], spec["package"], codename, fetch=fetch)
         if spec.get("relatedCves"):
             entry["relatedCves"] = list(spec["relatedCves"])
         if spec.get("abi"):
             abi = kernel_abi(entry.get("fixed"))
             if abi is not None:
                 entry["abi"] = abi
-        packages[spec["key"]] = entry
+        return spec["key"], entry
+
+    # Independent package lookups must not add every slow Ubuntu response to
+    # the refresh's wall time. map preserves the input order and propagates
+    # failures: no partial package table is returned.
+    with ThreadPoolExecutor(max_workers=4) as pool:
+        packages = dict(pool.map(resolve, specs))
     return {
         "kind": "distroPackages",
         "release": codename,
@@ -2404,8 +2427,12 @@ def main(argv: Sequence[str] | None = None) -> int:
     )
     parser.add_argument("--generated", metavar="ISO8601", help="pin the generated timestamp")
     parser.add_argument("--existing", metavar="PATH", help="read the existing table from PATH")
+    parser.add_argument("--verbose", action="store_true", help="log feed requests to stderr")
     args = parser.parse_args(argv)
 
+    if args.verbose:
+        logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
+
     try:
         if args.detect_new_bulletins:
             existing = _existing_table(args.existing, None)

tests/audit/test_minimum_refresh_fetch_logging.py

@@ -0,0 +1,101 @@
+"""Request diagnostics must preserve fail-closed behavior and keep secrets out."""
+
+import io
+import os
+import re
+import subprocess
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from cmax import minimum_refresh as fr
+
+
+class FetchLoggingTests(unittest.TestCase):
+    def test_json_fetch_logs_url_and_duration_without_credentials(self):
+        url = "https://api.github.com/repos/example/releases"
+        with (
+            mock.patch.dict(fr.os.environ, {"GITHUB_TOKEN": "private-test-token"}),
+            mock.patch.object(fr.SAFE_OPENER, "open", return_value=io.BytesIO(b"{}")),
+            self.assertLogs(fr.__name__, level="INFO") as logs,
+        ):
+            self.assertEqual(fr.Fetcher().get_json(url), {})
+        output = "\n".join(logs.output)
+        self.assertIn("fetch GET " + url, output)
+        self.assertIn("finished " + url + " after ", output)
+        self.assertNotIn("private-test-token", output)
+        self.assertNotIn("Authorization", output)
+
+    def test_failed_request_logs_duration_and_still_raises(self):
+        url = "https://example.com/feed"
+        with (
+            mock.patch.object(fr.SAFE_OPENER, "open", side_effect=TimeoutError("timed out")),
+            self.assertLogs(fr.__name__, level="INFO") as logs,
+            self.assertRaisesRegex(fr.MinimumRefreshError, "timed out"),
+        ):
+            fr.Fetcher().get_text(url)
+        self.assertIn("finished " + url, "\n".join(logs.output))
+
+
+class UbuntuConcurrencyTests(unittest.TestCase):
+    def test_packages_run_concurrently_and_keep_input_order(self):
+        barrier = threading.Barrier(4, timeout=5)
+        specs = [
+            {"key": str(n), "cve": str(n), "package": "package"}
+            for n in range(4)
+        ]
+
+        def entry(cve, *args, **kwargs):
+            barrier.wait()
+            return {"fixed": cve}
+
+        with mock.patch.object(fr, "ubuntu_entry", side_effect=entry):
+            result = fr.ubuntu_minimums(specs=specs)
+        self.assertEqual(list(result["packages"]), ["0", "1", "2", "3"])
+        self.assertEqual(result["packages"]["2"]["fixed"], "2")
+
+    def test_a_failed_package_still_stops_the_build(self):
+        with (
+            mock.patch.object(
+                fr, "ubuntu_entry", side_effect=fr.MinimumRefreshError("feed unavailable")
+            ),
+            self.assertRaisesRegex(fr.MinimumRefreshError, "feed unavailable"),
+        ):
+            fr.ubuntu_minimums(specs=[{"key": "x", "cve": "x", "package": "x"}])
+
+
+class BulletinShellTests(unittest.TestCase):
+    def test_scan_exit_codes_under_github_errexit(self):
+        import yaml
+
+        path = Path(__file__).resolve().parents[2] / ".github/workflows/minimum-versions-refresh.yml"
+        steps = yaml.safe_load(path.read_text())["jobs"]["refresh"]["steps"]
+        script = next(s["run"] for s in steps if s.get("id") == "bulletins")
+        for code, found, attempts in ((0, "false", 1), (3, "true", 1), (2, None, 3)):
+            with self.subTest(code=code), tempfile.TemporaryDirectory() as directory:
+                output = Path(directory) / "output"
+                stub = (
+                    '(printf "attempt\\n" >> "$RUNNER_TEMP/attempts"; '
+                    f'exit {code}) > "$report" 2>&1'
+                )
+                run, replacements = re.subn(
+                    r'python3[^\n]+> "\$report" 2>&1', lambda _: stub, script
+                )
+                self.assertEqual(replacements, 1)
+                result = subprocess.run(
+                    ["bash", "-e", "-c", "sleep() { :; }\n" + run],
+                    env={**os.environ, "RUNNER_TEMP": directory, "GITHUB_OUTPUT": str(output)},
+                    capture_output=True,
+                    text=True,
+                )
+                self.assertEqual(result.returncode, 0 if found else 1)
+                self.assertEqual(
+                    (Path(directory) / "attempts").read_text().splitlines(),
+                    ["attempt"] * attempts,
+                )
+                if found:
+                    self.assertEqual(output.read_text(), f"found={found}\n")
+                else:
+                    self.assertFalse(output.exists())