From 13a8ac62a8a099c7f2ff25d79bb794fe447cfb16 Mon Sep 17 00:00:00 2001 From: 0xMarcio Date: Thu, 22 Oct 2026 22:01:28 +0200 Subject: [PATCH] Reuse NVD feeds and read public Gists without the Actions token --- .github/workflows/sync_metadata.yml | 8 +++----- scripts/sync_cve_metadata.py | 16 +++++++++++---- tests/test_sync_cve_metadata.py | 30 +++++++++++++++++++++++++++-- 3 files changed, 43 insertions(+), 11 deletions(-) diff --git a/.github/workflows/sync_metadata.yml b/.github/workflows/sync_metadata.yml index 0c25d13a467..a8e94933119 100644 --- a/.github/workflows/sync_metadata.yml +++ b/.github/workflows/sync_metadata.yml @@ -41,9 +41,9 @@ jobs: GITHUB_TOKEN: ${{ github.token }} run: | if [ "${{ github.event.schedule }}" = "10 3 * * 0" ] || [ "${{ inputs.full }}" = "true" ]; then - python3 scripts/sync_cve_metadata.py --full + python3 scripts/sync_cve_metadata.py --full --feed-cache-dir "$RUNNER_TEMP/nvd-feeds" else - python3 scripts/sync_cve_metadata.py + python3 scripts/sync_cve_metadata.py --feed-cache-dir "$RUNNER_TEMP/nvd-feeds" fi - name: Refresh CISA known-exploited status @@ -53,9 +53,7 @@ jobs: id: references continue-on-error: true timeout-minutes: 5 - env: - GITHUB_TOKEN: ${{ github.token }} - run: python3 scripts/sync_reference_pocs.py --limit 40 + run: python3 scripts/sync_reference_pocs.py --feed-dir "$RUNNER_TEMP/nvd-feeds" --limit 40 - name: Verify the generated index run: | diff --git a/scripts/sync_cve_metadata.py b/scripts/sync_cve_metadata.py index 46566ca2f60..f7ef3fd65c5 100644 --- a/scripts/sync_cve_metadata.py +++ b/scripts/sync_cve_metadata.py @@ -864,7 +864,7 @@ def feed_url(name: str) -> str: return f"{FEED_ROOT}/nvdcve-2.0-{name}.json.gz" -def load_feed(name: str) -> list[dict[str, Any]]: +def load_feed(name: str, *, cache_dir: Path | None = None) -> list[dict[str, Any]]: request = urllib.request.Request(feed_url(name), headers={"User-Agent": USER_AGENT}) for attempt in range(3): try: @@ -883,8 +883,15 @@ def load_feed(name: str) -> list[dict[str, Any]]: if payload.get("format") != "NVD_CVE" or payload.get("version") != "2.0": raise RuntimeError(f"{name}: unexpected NVD feed format") vulnerabilities = payload.get("vulnerabilities") - if not isinstance(vulnerabilities, list): - raise RuntimeError(f"{name}: NVD feed has no vulnerability list") + if not isinstance(vulnerabilities, list) or payload.get("totalResults") != len(vulnerabilities): + raise RuntimeError(f"{name}: incomplete NVD vulnerability list") + if cache_dir is not None: + cache_dir.mkdir(parents=True, exist_ok=True) + path = cache_dir / f"nvd-{name}.json.gz" + temporary = path.with_suffix(".tmp") + with gzip.open(temporary, "wt", encoding="utf-8") as handle: + json.dump(payload, handle, separators=(",", ":")) + temporary.replace(path) return vulnerabilities @@ -963,6 +970,7 @@ def parse_args() -> argparse.Namespace: help="rebuild the reviewed GitHub advisory cache without a full NVD sync", ) parser.add_argument("--year", action="append", type=int, default=[], help="process one NVD feed year") + parser.add_argument("--feed-cache-dir", type=Path, help="Save validated feeds for the reference refresh") parser.add_argument("--dry-run", action="store_true", help="report without writing files") args = parser.parse_args() if args.full and args.year: @@ -991,7 +999,7 @@ def main() -> int: seen: set[str] = set() processed = 0 for name in feed_names: - vulnerabilities = load_feed(name) + vulnerabilities = load_feed(name, cache_dir=None if args.dry_run else args.feed_cache_dir) matched = 0 for wrapper in vulnerabilities: cve = wrapper.get("cve") if isinstance(wrapper, dict) else None diff --git a/tests/test_sync_cve_metadata.py b/tests/test_sync_cve_metadata.py index d509e9aea5e..b1c9886148a 100644 --- a/tests/test_sync_cve_metadata.py +++ b/tests/test_sync_cve_metadata.py @@ -16,6 +16,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import build_site import sync_cve_metadata as metadata +import sync_reference_pocs as references import update_cves @@ -28,7 +29,32 @@ class FeedDownloadTests(unittest.TestCase): def setUp(self): self.feed = [{"cve": {"id": CVE}}] self.body = gzip.compress(json.dumps({"format": "NVD_CVE", "version": "2.0", - "vulnerabilities": self.feed}).encode()) + "totalResults": 1, "vulnerabilities": self.feed}).encode()) + + def test_cached_feed_is_reused_by_reference_discovery_without_network(self): + url = "https://gist.github.com/researcher/1fe3fe58dd275edb77dcbe890fce2f2c" + rows = [{"cve": {"id": CVE, "references": [{"url": url, "tags": ["Exploit"]}]}}] + body = gzip.compress(json.dumps({"format": "NVD_CVE", "version": "2.0", + "totalResults": 1, "vulnerabilities": rows}).encode()) + with tempfile.TemporaryDirectory() as directory: + cache = Path(directory) + with patch.object(metadata.urllib.request, "urlopen", return_value=io.BytesIO(body)) as fetch: + self.assertEqual(metadata.load_feed("modified", cache_dir=cache), rows) + fetch.assert_called_once() + with patch.object(references.Budget, "read", side_effect=AssertionError("duplicate download")): + self.assertEqual(references.load_feeds(cache), {CVE: [url]}) + + def test_incomplete_feed_does_not_replace_cached_feed(self): + body = gzip.compress(json.dumps({"format": "NVD_CVE", "version": "2.0", + "totalResults": 2, "vulnerabilities": self.feed}).encode()) + with tempfile.TemporaryDirectory() as directory: + cache = Path(directory) + path = cache / "nvd-modified.json.gz" + path.write_bytes(self.body) + with patch.object(metadata.urllib.request, "urlopen", return_value=io.BytesIO(body)): + with self.assertRaisesRegex(RuntimeError, "incomplete NVD"): + metadata.load_feed("modified", cache_dir=cache) + self.assertEqual(path.read_bytes(), self.body) def test_truncated_gzip_restarts_download_and_returns_complete_feed(self): with patch.object(metadata.urllib.request, "urlopen", side_effect=[ @@ -165,7 +191,7 @@ class AdvisoryLifecycleTests(unittest.TestCase): def run_sync(entries: dict, feed: list, *, full: bool = False, dry_run: bool = True, fallback: bool = True) -> dict: with patch.object(metadata, "parse_args", return_value=Namespace( - full=full, github_full=False, dry_run=dry_run, year=[] if full else [2026] + full=full, github_full=False, dry_run=dry_run, year=[] if full else [2026], feed_cache_dir=None )), patch.object(metadata, "sync_github_cache", return_value={}), patch.object( metadata, "sync_github_pocs" ), patch.object(metadata, "local_cves", return_value=set(entries)), patch.object(