"""Advisory refresh is metadata-only, bounded, and atomic; no network needed.""" import copy import json import tempfile import unittest from pathlib import Path from unittest.mock import patch from rpcscope.advisories import ADVISORIES, _download, sync_advisories from rpcscope.catalog import CHECKS MICROSOFT = "f38d906d-7342-40ea-92c1-6c4a2c6478c8" def record_for(advisory): return { "dataType": "CVE_RECORD", "cveMetadata": { "cveId": advisory.key, "state": "PUBLISHED", "assignerOrgId": MICROSOFT, "datePublished": "2026-05-12T16:58:50.091Z", "dateUpdated": "2026-08-10T15:12:59.779Z", }, "containers": {"cna": { "title": "Vendor advisory title", "providerMetadata": {"orgId": MICROSOFT, "shortName": "microsoft"}, "references": [ {"url": advisory.reference}, {"url": "https://example.com/poc.py"}, {"url": "https://msrc.microsoft.com.evil.example/metadata"}, ], "affected": [{"vendor": "Microsoft", "product": "Windows Example", "platforms": ["x64-based Systems"], "versions": [{"version": "10.0.0", "lessThan": "10.0.1", "versionType": "custom", "status": "affected"}]}], "untrusted_command": "do not carry executable fields into cache", }}, } class AdvisoryTests(unittest.TestCase): def test_registry_remains_separate_from_executable_checks(self): self.assertEqual(len(ADVISORIES), 7) self.assertTrue(all(item.status == "review-only" for item in ADVISORIES.values())) executable_cves = {cve for check in CHECKS.values() if check.kind == "vulnerability" for cve in check.cves} self.assertFalse(executable_cves.intersection(ADVISORIES)) self.assertEqual(CHECKS["longlogon"].kind, "precondition") self.assertEqual(CHECKS["longlogon"].positive_status, "precondition-met") self.assertIn("/2020/1xxx/", ADVISORIES["CVE-2020-1206"].cna_url) self.assertIn("/2026/41xxx/", ADVISORIES["CVE-2026-41089"].cna_url) def test_sync_copies_selected_metadata_and_never_fetches_references(self): fetched = [] by_url = {item.cna_url: item for item in ADVISORIES.values()} def fetch(url): fetched.append(url) return json.dumps(record_for(by_url[url])).encode() with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "nested" / "advisories.json" result = sync_advisories(path, fetch=fetch) dataset = json.loads(path.read_text(encoding="utf-8")) self.assertEqual(fetched, list(by_url)) self.assertEqual(dataset["advisories"], result) self.assertFalse(dataset["scanner_coverage"]) self.assertEqual(result[0]["title"], "Vendor advisory title") self.assertEqual(result[0]["vendorReferences"], [result[0]["reference"]]) self.assertEqual(result[0]["affected"][0]["versions"][0]["lessThan"], "10.0.1") self.assertNotIn("untrusted_command", json.dumps(result)) self.assertTrue(all(item["status"] == "review-only" for item in result)) def test_invalid_later_record_preserves_existing_destination(self): advisories = list(ADVISORIES.values()) index = 0 def fetch(url): nonlocal index record = record_for(advisories[index]) index += 1 if index == 2: record["cveMetadata"]["cveId"] = "CVE-2000-0001" return json.dumps(record).encode() with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "advisories.json" path.write_text("existing valid cache", encoding="utf-8") with self.assertRaisesRegex(ValueError, "CVE ID mismatch"): sync_advisories(path, fetch=fetch) self.assertEqual(path.read_text(encoding="utf-8"), "existing valid cache") def test_unpublished_untrusted_and_malformed_records_fail_closed(self): advisory = next(iter(ADVISORIES.values())) original = record_for(advisory) cases = [] for mutate in ( lambda record: record["cveMetadata"].update(state="REJECTED"), lambda record: record["cveMetadata"].update(assignerOrgId="other"), lambda record: record["containers"]["cna"]["providerMetadata"].update(orgId="other"), lambda record: record["containers"]["cna"]["providerMetadata"].update(shortName="other"), lambda record: record["containers"]["cna"].update(title="bad\x1b[31m"), lambda record: record["cveMetadata"].update(dateUpdated="yesterday"), lambda record: record["containers"]["cna"].update(references=[]), lambda record: record["containers"]["cna"].update(affected=[{}] * 513), ): record = copy.deepcopy(original) mutate(record) cases.append(json.dumps(record).encode()) cases.extend([b"not JSON", b"{}", b"x" * (2 * 1024 * 1024 + 1)]) with tempfile.TemporaryDirectory() as temporary: for payload in cases: with self.subTest(payload_size=len(payload)): path = Path(temporary) / "advisories.json" with self.assertRaises(ValueError): sync_advisories(path, fetch=lambda url: payload) self.assertFalse(path.exists()) def test_absent_cna_title_has_curated_fallback(self): by_url = {item.cna_url: item for item in ADVISORIES.values()} def fetch(url): record = record_for(by_url[url]) del record["containers"]["cna"]["title"] return json.dumps(record).encode() with tempfile.TemporaryDirectory() as temporary: snapshots = sync_advisories(Path(temporary) / "metadata.json", fetch=fetch) self.assertEqual(snapshots[0]["title"], next(iter(ADVISORIES.values())).title) def test_network_failure_preserves_existing_destination(self): def fetch(url): raise OSError("network unavailable") with tempfile.TemporaryDirectory() as temporary: path = Path(temporary) / "metadata.json" path.write_text("old cache", encoding="utf-8") with self.assertRaisesRegex(ValueError, "network unavailable"): sync_advisories(path, fetch=fetch) self.assertEqual(path.read_text(encoding="utf-8"), "old cache") def test_downloader_bounds_read_and_timeout_and_rejects_redirect(self): url = next(iter(ADVISORIES.values())).cna_url with patch("rpcscope.advisories.urlopen") as opening: response = opening.return_value.__enter__.return_value response.geturl.return_value = url response.read.return_value = b"{}" self.assertEqual(_download(url), b"{}") self.assertEqual(opening.call_args.kwargs["timeout"], 15) response.read.assert_called_once_with(2 * 1024 * 1024 + 1) response.geturl.return_value = "https://example.com/redirect" with self.assertRaisesRegex(ValueError, "redirected"): _download(url) if __name__ == "__main__": unittest.main()