unseen: filter advisory enumeration server-side (ecosystem/date/cwes)

enumerate_advisories paged the entire global GitHub reviewed-advisory
database (every ecosystem, every year, every CWE) before extraction, so
a 300-repo draft never converged -- a live run reached 21,962 advisories
in 4.5h with extraction never started.

Push the existing candidate() rejections into the query: page per
ecosystem, published>=2019-01-01, and cwes restricted to the seven
families. The cwes filter uses bare CWE numbers (89), which is what the
GitHub advisories API matches -- "CWE-89" silently returns zero. The
whole cwes filter is omitted if the taxonomy ever holds a non-exact
token, so a selectable advisory is never dropped. Journal version 3
(ecosystem:page keys) refuses incompatible older journals.

Live page-1 smoke (pip, npm): 100/100 rows carry a family CWE, none
published before 2019. Offline: 166 unseen tests pass, ruff clean.

Co-Authored-By: Claude Opus 4.8 <[email protected]>
This commit is contained in:
Marius Ciepluch
2026-10-08 08:19:30 +02:00
co-authored by Claude Opus 4.8
parent d7bf0ba750
commit 90e2400e6f
4 changed files with 181 additions and 39 deletions
+55 -1
View File
@@ -122,7 +122,9 @@ def test_http_adapters_with_fake_transport():
parsed = urlsplit(url)
if parsed.path == "/advisories":
page = int(parse_qs(parsed.query)["page"][0])
return [advisory()] * (100 if page == 1 else 1)
return [
{**advisory(), "ghsa_id": f"GHSA-test-test-{i:04d}"} for i in range(0 if page == 1 else 100, 100 if page == 1 else 101)
]
if "/license" in parsed.path:
assert parse_qs(parsed.query)["ref"] == ["b" * 40]
return {
@@ -271,3 +273,55 @@ def test_transport_decode_retry_respects_response_retry_after():
transport = s.JSONTransport(opener=lambda *a, **kw: responses.pop(0), sleeper=waits.append)
assert transport.get("https://example.invalid", {}) == {}
assert waits == [9]
@pytest.mark.parametrize("ecosystem", s.ECOSYSTEMS)
def test_advisory_page_filters_match_candidate_population(ecosystem):
from urllib.parse import parse_qs, urlsplit
taxonomy = s.load_families()
expected = {c for f in taxonomy.families if f.id in s.FAMILIES for c in f.cwes}
class Transport:
def get(self, url, headers):
query = parse_qs(urlsplit(url).query)
assert query == {
"type": ["reviewed"],
"per_page": ["100"],
"page": ["2"],
"sort": ["published"],
"direction": ["asc"],
"ecosystem": [ecosystem],
"published": [">=2019-01-01"],
"cwes": [",".join(sorted((c.removeprefix("CWE-") for c in expected), key=int))],
}
return []
assert s.GitHub("fake", Transport()).advisory_page(2, ecosystem=ecosystem) == {"rows": [], "done": True}
# Every enumerated CWE is selectable at the inclusive date boundary.
for cwe in expected:
row = advisory()
row.update(published_at="2019-01-01T00:00:00Z", cwes=[{"cwe_id": cwe}])
row["vulnerabilities"][0]["package"]["ecosystem"] = ecosystem
assert s.candidate(row, API(), OSV()).family == taxonomy.family_of_cwe(cwe).id
@pytest.mark.parametrize("token", ["CWE-89*", "CWE-", "89", "CWE-79..CWE-89"])
def test_advisory_page_omits_entire_cwe_filter_for_non_exact_taxonomy(monkeypatch, token):
from dataclasses import replace
from urllib.parse import parse_qs, urlsplit
taxonomy = s.load_families()
family = taxonomy.by_id("injection")
taxonomy = replace(taxonomy, families=tuple(replace(f, cwes=f.cwes | {token}) if f == family else f for f in taxonomy.families))
monkeypatch.setattr(s, "load_families", lambda: taxonomy)
class Transport:
def get(self, url, headers):
query = parse_qs(urlsplit(url).query)
assert "cwes" not in query
assert query["published"] == [">=2019-01-01"]
assert query["ecosystem"] == ["pip"]
return []
s.GitHub("fake", Transport()).advisory_page(1, ecosystem="pip")