From 1bfe6f9c96f9ba69bb5c21246000dce1386a2b24 Mon Sep 17 00:00:00 2001 From: 0xMarcio Date: Wed, 7 Oct 2026 02:03:44 +0200 Subject: [PATCH] Keep rejected commit clocks and replace superseded README links --- .github/getTrending.py | 4 ++ scripts/releases.py | 26 +++++++++++- tests/test_releases.py | 67 ++++++++++++++++++++++++++++++ tests/test_reviewed_publication.py | 49 ++++++++++++++++++++++ 4 files changed, 145 insertions(+), 1 deletion(-) diff --git a/.github/getTrending.py b/.github/getTrending.py index 991d8b8b671..c8ede4c7d26 100644 --- a/.github/getTrending.py +++ b/.github/getTrending.py @@ -291,6 +291,10 @@ def ledger_landed(ledger: dict, candidates: list[dict]) -> list[dict]: current = ledger.get(identity, {}) if any(current.get(flag) for flag in ("copy", "gone", "bulk")): continue + if not curated: + rows = {other: item for other, item in rows.items() + if other == identity or cve_of(item) != cve + or source_key(item.get("_artifact_url") or item["html_url"]) != source_key(url)} rows[identity] = repo return sorted(rows.values(), key=lambda repo: repo["_released"], reverse=True)[:LANDED_ROWS] diff --git a/scripts/releases.py b/scripts/releases.py index f7b10ce205a..a940f63ec0c 100644 --- a/scripts/releases.py +++ b/scripts/releases.py @@ -143,6 +143,8 @@ def date_release(evidence: dict, *, now: str | None = None) -> tuple[str | None, moment = timestamp(value) if moment is None or moment > (ceiling if name in {"commit", "merged"} else observed_ceiling): continue + if name == "commit" and (observed := timestamp(evidence.get("history_observed"))) and moment > observed: + continue if name in {"commit", "merged"} and evidence.get(f"{name}_verified") is not True: continue bounds.append((moment, DATE_ORDER[name], value, name)) @@ -235,6 +237,26 @@ def record(ledger: dict[str, dict], cve: str, url: str, evidence: dict | None = if evidence.get("error") and row.get("released"): # A failed refresh may advance retry bookkeeping, never valid evidence. evidence = {name: value for name, value in evidence.items() if name in {"error", "checked", "head"}} + future_shas = set(row.get("future_shas", [])) + observed = min(timestamp(value) for value in ( + observed_at, evidence.get("history_observed") or evidence.get("checked") + ) if timestamp(value)) + # A stale pushed_at is retryable; a SHA already seen before its clock is not. + for proof, checked in ((evidence, observed), + (row, timestamp(row.get("checked")) if not row.get("history_version") else None)): + clock = timestamp(proof.get("commit")) + if proof.get("sha") and proof.get("commit_verified") is True and clock and checked and clock > checked: + future_shas.add(proof["sha"]) + if future_shas: + row["future_shas"] = sorted(future_shas) + clock_fields = {"commit", "sha", "commit_verified", "history_version"} + if row.get("sha") in future_shas: + for name in clock_fields: + row.pop(name, None) + if row.get("basis") == "commit": + row.update(released=None, basis="unknown") + if evidence.get("sha") in future_shas: + evidence = {name: value for name, value in evidence.items() if name not in clock_fields} method = evidence.get("history_version", 0) method = method if type(method) is int and method > 0 else 0 old_method = row.get("history_version", 0) @@ -610,6 +632,7 @@ def current_repository_evidence(repo: dict, paths: list[str], headers: dict, *, if cache_path.exists(): cached = json.loads(cache_path.read_text()) if cached.get("history_version") == HISTORY_VERSION and cached.get("commit_verified") is True: + cached.setdefault("history_observed", cached.get("checked")) return {**cached, **evidence} entries = history.qualifying(name, revision, paths, cve) if not entries: @@ -621,7 +644,8 @@ def current_repository_evidence(repo: dict, paths: list[str], headers: dict, *, identity = artifact_identity(entries, [entry["path"] for entry in entries]) introductions = [history.introduction(name, revision, entry["path"], cve) for entry in entries] first = min(introductions, key=lambda row: timestamp(row["commit"])) - result = {**evidence, **identity, **first, "history_version": HISTORY_VERSION} + result = {**evidence, **identity, **first, "history_version": HISTORY_VERSION, + "history_observed": evidence["checked"]} cache_dir.mkdir(parents=True, exist_ok=True) temporary = cache_path.with_suffix(".tmp") temporary.write_text(json.dumps(result, sort_keys=True) + "\n") diff --git a/tests/test_releases.py b/tests/test_releases.py index af1bd962019..2a56ee581c6 100644 --- a/tests/test_releases.py +++ b/tests/test_releases.py @@ -157,6 +157,52 @@ class ReleaseLedgerTests(unittest.TestCase): "previous_head": "a" * 40, "absence_verified": True} self.assertEqual(releases.date_release(evidence, now=NOW), (None, "unknown")) + def test_future_commit_clock_stays_rejected_after_later_pushes(self): + ledger = releases.Ledger() + proof = {"created": "2026-10-01T00:00:00Z", "pushed_at": "2026-10-01T01:00:00Z", + "commit": "2026-10-03T00:00:00Z", "sha": "a" * 40, "commit_verified": True, + "history_version": releases.HISTORY_VERSION} + releases.record(ledger, CVE, URL, proof, observed_at="2026-10-01T02:00:00Z") + row = releases.record(ledger, CVE, URL, {**proof, "pushed_at": "2026-10-05T00:00:00Z"}, observed_at=NOW) + self.assertEqual((row["released"], row["basis"]), (proof["created"], "created")) + self.assertFalse(any(field in row for field in ("commit", "sha", "commit_verified", "history_version"))) + # Encountering another bad clock must not forget the first rejected SHA. + releases.record(ledger, CVE, URL, {**proof, "sha": "b" * 40, "commit": "2030-01-01T00:00:00Z"}, observed_at=NOW) + row = releases.record(ledger, CVE, URL, {**proof, "pushed_at": NOW}, observed_at=NOW) + self.assertEqual((row["released"], row["basis"]), (proof["created"], "created")) + self.assertEqual(row["seen"], "2026-10-01T02:00:00Z") + # A different qualifying commit can still introduce the actual PoC later. + real = {**proof, "sha": "c" * 40, "commit": "2026-10-05T00:00:00Z", "pushed_at": NOW} + row = releases.record(ledger, CVE, URL, real, observed_at=NOW) + self.assertEqual((row["released"], row["history_version"]), (real["commit"], releases.HISTORY_VERSION)) + self.assertEqual(row["seen"], "2026-10-01T02:00:00Z") + + def test_preexisting_future_commit_fields_are_removed_on_reobservation(self): + proof = {"created": "2026-10-01T00:00:00Z", "pushed_at": "2026-10-01T01:00:00Z", + "checked": "2026-10-01T02:00:00Z", "commit": "2026-10-03T00:00:00Z", + "sha": "a" * 40, "commit_verified": True} + identity = releases.key(CVE, URL) + ledger = releases.Ledger({identity: {**proof, "cve": CVE, "url": URL, "imported": True, + "released": proof["created"], "basis": "created"}}) + row = releases.record(ledger, CVE, URL, {**proof, "pushed_at": NOW, "checked": NOW, + "history_version": releases.HISTORY_VERSION}, observed_at=NOW) + self.assertEqual((row["released"], row["basis"]), (proof["created"], "created")) + self.assertFalse(any(field in row for field in ("commit", "sha", "commit_verified", "history_version"))) + self.assertTrue(row["imported"]) + self.assertNotIn("seen", row) + + def test_stale_push_date_can_catch_up_without_rejecting_observed_commit(self): + ledger = releases.Ledger() + proof = {"created": "2026-10-01T00:00:00Z", "pushed_at": "2026-10-01T01:00:00Z", + "checked": "2026-10-04T00:00:00Z", "commit": "2026-10-03T00:00:00Z", + "sha": "a" * 40, "commit_verified": True, "history_version": releases.HISTORY_VERSION} + first = releases.record(ledger, CVE, URL, proof, observed_at=proof["checked"]) + self.assertEqual(first["released"], proof["created"]) + row = releases.record(ledger, CVE, URL, {**proof, "pushed_at": NOW}, observed_at=NOW) + self.assertEqual((row["released"], row["basis"], row["history_version"]), + (proof["commit"], "commit", releases.HISTORY_VERSION)) + self.assertEqual(row["seen"], proof["checked"]) + def test_imported_history_uses_repository_creation_lower_bound(self): self.assertEqual(releases.date_release({"created": "2026-10-06T00:00:00Z", "commit": "2023-01-01T00:00:00Z", "commit_verified": True}, now=NOW), @@ -408,6 +454,27 @@ class PinnedHistoryTests(unittest.TestCase): self.assertEqual(second["history_version"], first["history_version"] + 1) self.assertEqual(len(list(self.directory.glob("*.json"))), 2) + def test_cached_history_retains_first_observation_for_clock_rejection(self): + self.histories["exploit.py"][0]["commit"]["committer"]["date"] = "2026-10-09T00:00:00Z" + first = self.evidence() + self.assertNotIn("error", first) + calls = len(self.calls) + later = "2026-10-10T00:00:00Z" + for legacy in (False, True): + with self.subTest(legacy=legacy): + if legacy: + path = next(self.directory.glob("*.json")) + cached = json.loads(path.read_text()) + cached.pop("history_observed", None) + path.write_text(json.dumps(cached)) + reused = releases.current_repository_evidence({**self.repo, "pushed_at": later}, ["exploit.py"], {}, + cve=CVE, cache_dir=self.directory, observed_at=later, history=self.history) + self.assertEqual(reused.get("history_observed"), NOW) + self.assertEqual(releases.date_release(reused, now=later), (self.repo["created_at"], "created")) + row = releases.record(releases.Ledger(), CVE, URL, reused, observed_at=later) + self.assertEqual((row["released"], row["basis"]), (self.repo["created_at"], "created")) + self.assertEqual(len(self.calls), calls) + def test_historical_function_stub_is_not_the_current_poc_introduction(self): self.sources[self.stub_sha]["exploit.py"] = b"# CVE-2026-12345 proof of concept placeholder\nimport requests\ndef exploit():\n pass\n" result = self.evidence() diff --git a/tests/test_reviewed_publication.py b/tests/test_reviewed_publication.py index eae06ad3a54..11d7229c518 100644 --- a/tests/test_reviewed_publication.py +++ b/tests/test_reviewed_publication.py @@ -76,6 +76,55 @@ class ReviewedPublicationTests(unittest.TestCase): patch.object(trending, 'known_exploited', return_value=set()): self.assertEqual(trending.ledger_landed({}, [candidate]), [candidate]) + def test_landed_switch_replaces_dated_root_without_removing_other_sources_or_cves(self): + row = source_artifacts.load_reviews()[0] + artifact = source_artifacts.ArtifactEvidence(row['cve'], row['repository'], row['path'], + row['revision'], row['sha256'], row['id']) + root = f"https://github.com/{row['repository']}" + other_cve = 'CVE-2024-1234' + other_source = 'https://github.com/another/poc' + released = trending.releases.utcnow() + ledger = trending.releases.Ledger() + entries = [{'cve': row['cve'], 'poc': [root, other_source]}, {'cve': other_cve, 'poc': [root]}] + for entry in entries: + for url in entry['poc']: + trending.releases.record(ledger, entry['cve'], url, + {'commit': released, 'commit_verified': True, + 'history_version': trending.releases.HISTORY_VERSION}) + candidate = {'cve': row['cve'], 'html_url': root, '_artifact_url': artifact.url, '_released': released} + with tempfile.TemporaryDirectory() as temp: + metadata = Path(temp) / 'metadata.json' + metadata.write_text('{}') + with patch.object(build_site, 'REPO_META', metadata), \ + patch.object(build_site, 'build_cve_list', return_value=(entries, len(entries))), \ + patch.object(trending, 'known_exploited', return_value=set()): + merged = trending.ledger_landed(ledger, [candidate]) + actual = {(trending.cve_of(item), item.get('_artifact_url') or item['html_url']) for item in merged} + self.assertEqual(actual, {(row['cve'], artifact.url), (row['cve'], other_source), (other_cve, root)}) + self.assertIn(candidate, merged) + + def test_landed_merge_preserves_dated_curated_sibling_variants(self): + cve = 'CVE-2024-1234' + urls = [f'https://github.com/google/security-research/blob/master/pocs/{cve}/variant-{n}.c' + for n in range(3)] + released = trending.releases.utcnow() + ledger = trending.releases.Ledger() + for url in urls: + trending.releases.record(ledger, cve, url, {'commit': released, 'commit_verified': True, + 'history_version': trending.releases.HISTORY_VERSION}) + candidate = {'cve': cve, 'html_url': 'https://github.com/google/security-research', + '_artifact_url': urls[0], '_released': released} + entries = [{'cve': cve, 'poc': [], 'collections': urls}] + with tempfile.TemporaryDirectory() as temp: + metadata = Path(temp) / 'metadata.json' + metadata.write_text('{}') + with patch.object(build_site, 'REPO_META', metadata), \ + patch.object(build_site, 'build_cve_list', return_value=(entries, 1)), \ + patch.object(trending, 'known_exploited', return_value=set()): + merged = trending.ledger_landed(ledger, [candidate]) + self.assertEqual({item.get('_artifact_url') or item['html_url'] for item in merged}, set(urls)) + self.assertIn(candidate, merged) + def test_curated_variants_keep_independent_paths(self): urls = [f'https://github.com/google/security-research/blob/master/pocs/CVE-2024-1234/variant-{n}.c' for n in range(116)]