From 562016554081b5c9698f65e1d52b1aa001fd1b78 Mon Sep 17 00:00:00 2001 From: Marius Ciepluch <11855163+norandom@users.noreply.github.com> Date: Thu, 8 Oct 2026 10:26:11 +0200 Subject: [PATCH] unseen: cursor pagination for advisory enumeration The GitHub /advisories endpoint ignores the `page` query parameter: it paginates with the Link header's rel="next" cursor. The page-number pager re-fetched the same first 100 rows forever and never terminated, so enumeration never completed and no draft converged (a run fetched 5137 "pip pages" holding only 100 unique advisories). Surface the successful response headers from JSONTransport, and walk the rel="next" cursor per ecosystem until it is absent. The ecosystem / published>=2019 / bare-number cwes filters are preserved on the first request. Journal keys stay ecosystem:n (now cursor-walk positions) with the saved next URL for resume; journal_version 4 refuses older journals. Live smoke via the production pager: 300 unique advisories pulled in 3 API calls, advancing past the first page. Offline: 170 unseen tests pass, ruff clean. Co-Authored-By: Claude Opus 4.8 --- benchmarks/unseen/draft.py | 12 ++++-- benchmarks/unseen/source.py | 25 ++++++++----- tests/test_unseen_draft.py | 69 +++++++++++++++++++--------------- tests/test_unseen_source.py | 75 ++++++++++++++++++++++++++++++++++--- 4 files changed, 131 insertions(+), 50 deletions(-) diff --git a/benchmarks/unseen/draft.py b/benchmarks/unseen/draft.py index cc8eaf4..46c28cd 100644 --- a/benchmarks/unseen/draft.py +++ b/benchmarks/unseen/draft.py @@ -219,7 +219,7 @@ def inputs(used, seed, known_empty, clones): "used_set_digest": digest({label: sorted(names) for label, names in used.sources.items()}), "known_empty": sorted(known_empty), "clone_limit_bytes": getattr(clones, "limit_bytes", 500 * 1024 * 1024), - "journal_version": 3, + "journal_version": 4, } # Hardcoded design thresholds and algorithms are covered as well as taxonomy data. for name in ("draft", "source", "extract", "eligibility", "journal"): @@ -232,15 +232,19 @@ def enumerate_advisories(github, journal): records = {} for ecosystem in source.ECOSYSTEMS: page = 1 + next_url = None while True: key = f"{ecosystem}:{page}" if key not in journal.rows["advisories"]: journal.boundary() - result = github.advisory_page(page, ecosystem=ecosystem) - journal.put("advisories", key, {"rows": [source.reduce_advisory(row) for row in result["rows"]], "done": result["done"]}) + result = github.advisory_page(next_url, ecosystem=ecosystem) + journal.put( + "advisories", key, {"rows": [source.reduce_advisory(row) for row in result["rows"]], "next_url": result["next_url"]} + ) result = journal.rows["advisories"][key] records.update((row["ghsa_id"], row) for row in result["rows"]) - if result["done"]: + next_url = result["next_url"] + if next_url is None: break page += 1 if not records: diff --git a/benchmarks/unseen/source.py b/benchmarks/unseen/source.py index 9c700a6..3b1f118 100644 --- a/benchmarks/unseen/source.py +++ b/benchmarks/unseen/source.py @@ -181,6 +181,7 @@ class JSONTransport: self.clock = clock or time.time self.attempts = 0 self.rate_limit = {} + self.response_headers = {} def get(self, url: str, headers: dict) -> object: from http.client import IncompleteRead @@ -194,7 +195,9 @@ class JSONTransport: with self.opener(Request(url, headers=headers), timeout=60) as response: retry_headers = getattr(response, "headers", {}) self.record_rate(retry_headers) - return json.load(response) + body = json.load(response) + self.response_headers = retry_headers + return body except HTTPError as exc: self.record_rate(exc.headers) if exc.code == 404: @@ -247,7 +250,8 @@ class GitHub: def get(self, path: str): self.calls += 1 - return self.transport.get("https://api.github.com" + path, self.headers) + url = path if path.startswith("https://api.github.com/") else "https://api.github.com" + path + return self.transport.get(url, self.headers) def get_repo(self, name: str) -> dict: name = normalize(name) @@ -269,14 +273,13 @@ class GitHub: "text": base64.b64decode(row.get("content", "")).decode("utf-8", errors="replace"), } - def advisory_page(self, page: int, *, ecosystem: str) -> dict: + def advisory_page(self, next_url: str | None = None, *, ecosystem: str) -> dict: if ecosystem not in ECOSYSTEMS: raise ValueError("invalid_ecosystem") # Oldest first: newly published advisories do not shift already persisted pages. parameters = { "type": "reviewed", "per_page": 100, - "page": page, "sort": "published", "direction": "asc", "ecosystem": ecosystem, @@ -290,25 +293,27 @@ class GitHub: # GitHub's advisories `cwes` filter expects bare numbers (89), not "CWE-89". if cwes and all(re.fullmatch(r"CWE-[1-9][0-9]*", cwe) for cwe in cwes): parameters["cwes"] = ",".join(sorted((cwe.removeprefix("CWE-") for cwe in cwes), key=int)) - rows = self.get("/advisories?" + urlencode(parameters)) + rows = self.get(next_url or "/advisories?" + urlencode(parameters)) if not isinstance(rows, list): raise ValueError("invalid_advisory_page") - return {"rows": rows, "done": len(rows) < 100} + link = self.transport.response_headers.get("Link", "") + match = re.search(r'<([^>]+)>;\s*rel="next"', link) + return {"rows": rows, "next_url": match[1] if match else None} def advisories(self): """Enumerate the candidate population, deduplicating multi-ecosystem advisories.""" seen = set() for ecosystem in ECOSYSTEMS: - page = 1 + next_url = None while True: - result = self.advisory_page(page, ecosystem=ecosystem) + result = self.advisory_page(next_url, ecosystem=ecosystem) for row in result["rows"]: if row["ghsa_id"] not in seen: seen.add(row["ghsa_id"]) yield row - if result["done"]: + next_url = result["next_url"] + if next_url is None: break - page += 1 class OSV: diff --git a/tests/test_unseen_draft.py b/tests/test_unseen_draft.py index 835b56e..f59db83 100644 --- a/tests/test_unseen_draft.py +++ b/tests/test_unseen_draft.py @@ -139,12 +139,14 @@ class FakeAPI: # P is MIT but actual introducing head is GPL: public output must use V. return {"text": "", "spdx_id": "MIT" if ref == "b" * 40 else "GPL-3.0", "path": "LICENSE"} - def advisory_page(self, page, *, ecosystem): + def advisory_page(self, next_url=None, *, ecosystem): self.calls += 1 - self.advisory_pages.append((ecosystem, page)) + self.advisory_pages.append((ecosystem, next_url)) population = list(self.advisories()) if ecosystem == "pip" else [] - rows = population[(page - 1) * 100 : page * 100] - return {"rows": rows, "done": len(rows) < 100} + offset = int(next_url.rsplit("=", 1)[1]) if next_url else 0 + rows = population[offset : offset + 100] + following = f"https://api.github.com/advisories?after={offset + 100}" if offset + 100 < len(population) else None + return {"rows": rows, "next_url": following} def advisories(self): for i in range(self.count): @@ -460,8 +462,8 @@ def test_max_hours_stops_after_completed_page(tmp_path): now = [0.0] class TimedAPI(FakeAPI): - def advisory_page(self, page, *, ecosystem): - result = super().advisory_page(page, ecosystem=ecosystem) + def advisory_page(self, next_url=None, *, ecosystem): + result = super().advisory_page(next_url, ecosystem=ecosystem) now[0] += 3601 return result @@ -544,9 +546,9 @@ def test_advisory_journal_reduces_nested_blobs_and_resumes_by_page(tmp_path): class Pages: calls = [] - def advisory_page(self, page, *, ecosystem): - self.calls.append((ecosystem, page)) - return {"rows": [raw], "done": True, "unused": "raw page blob"} + def advisory_page(self, next_url=None, *, ecosystem): + self.calls.append((ecosystem, next_url)) + return {"rows": [raw], "next_url": None, "unused": "raw page blob"} api, used = Pages(), e.UsedSet() parameters = d.inputs(used, 1, set(used.sources), FakeClones()) @@ -556,13 +558,13 @@ def test_advisory_journal_reduces_nested_blobs_and_resumes_by_page(tmp_path): assert rows == [expected] saved = [json.loads(line) for line in (path / "advisories.jsonl").read_bytes().splitlines()] assert [row["id"] for row in saved] == [f"{eco}:1" for eco in s.ECOSYSTEMS] - assert all(row["value"] == {"rows": [expected], "done": True} for row in saved) + assert all(row["value"] == {"rows": [expected], "next_url": None} for row in saved) assert b"unused" not in (path / "advisories.jsonl").read_bytes() assert b"raw advisory blob" not in (path / "advisories.jsonl").read_bytes() assert b"raw page blob" not in (path / "advisories.jsonl").read_bytes() with d.Journal(path).open(parameters) as journal: resumed = d.enumerate_advisories(api, journal) - assert api.calls == [(eco, 1) for eco in s.ECOSYSTEMS] and resumed == rows + assert api.calls == [(eco, None) for eco in s.ECOSYSTEMS] and resumed == rows class Bracket(FakeOSV): def get(self, identifier): @@ -603,12 +605,12 @@ def test_reduced_advisories_preserve_all_fix_exclusions_and_selection(): s.candidate(s.reduce_advisory(raw), FakeAPI(), FakeOSV()) -@pytest.mark.parametrize("old_version", [1, 2]) +@pytest.mark.parametrize("old_version", [1, 2, 3]) def test_old_advisory_journal_version_refused_before_calls(tmp_path, old_version): used, api, osv, clones = e.UsedSet(), FakeAPI(1), FakeOSV(), FakeClones() path = tmp_path / "journal" parameters = d.inputs(used, 7, set(used.sources), clones) - assert parameters["journal_version"] == 3 + assert parameters["journal_version"] == 4 with d.Journal(path).open({**parameters, "journal_version": old_version}) as journal: journal.put("advisories", "1", {"rows": list(api.advisories()), "done": True}) report = d.run( @@ -620,29 +622,30 @@ def test_old_advisory_journal_version_refused_before_calls(tmp_path, old_version def test_enumerate_ecosystem_pages_dedupes_and_resumes_partial_stream(tmp_path): rows = list(FakeAPI(3).advisories()) + cursor = "https://api.github.com/advisories?ecosystem=pip&after=opaque-cursor" pages = { - ("pip", 1): {"rows": [rows[0]], "done": False}, - ("pip", 2): {"rows": [rows[1]], "done": True}, - ("npm", 1): {"rows": [rows[0], rows[2]], "done": True}, - ("maven", 1): {"rows": [], "done": True}, - ("composer", 1): {"rows": [], "done": True}, + ("pip", None): {"rows": [rows[0]], "next_url": cursor}, + ("pip", cursor): {"rows": [rows[0], rows[1]], "next_url": None}, + ("npm", None): {"rows": [rows[0], rows[2]], "next_url": None}, + ("maven", None): {"rows": [], "next_url": None}, + ("composer", None): {"rows": [], "next_url": None}, } calls = [] class Pages: - def advisory_page(self, page, *, ecosystem): - calls.append((ecosystem, page)) - return pages[ecosystem, page] + def advisory_page(self, next_url=None, *, ecosystem): + calls.append((ecosystem, next_url)) + return pages[ecosystem, next_url] path = tmp_path / "journal" - with d.Journal(path).open({"journal_version": 3}) as journal: - journal.put("advisories", "pip:1", pages["pip", 1]) - with d.Journal(path).open({"journal_version": 3}) as journal: + with d.Journal(path).open({"journal_version": 4}) as journal: + journal.put("advisories", "pip:1", pages["pip", None]) + with d.Journal(path).open({"journal_version": 4}) as journal: result = d.enumerate_advisories(Pages(), journal) assert [row["ghsa_id"] for row in result] == [row["ghsa_id"] for row in rows] - assert list(journal.rows["advisories"]) == [f"{eco}:{page}" for eco, page in pages] + assert list(journal.rows["advisories"]) == ["pip:1", "pip:2", "npm:1", "maven:1", "composer:1"] assert calls == list(pages)[1:] - with d.Journal(path).open({"journal_version": 3}) as journal: + with d.Journal(path).open({"journal_version": 4}) as journal: assert d.enumerate_advisories(Pages(), journal) == result assert calls == list(pages)[1:] @@ -660,11 +663,15 @@ def test_filtered_http_population_reaches_draft_floor_with_bounded_pages(tmp_pat assert query["ecosystem"][0] in s.ECOSYSTEMS assert query["published"] == [">=2019-01-01"] assert "89" in query["cwes"][0].split(",") - eco, page = query["ecosystem"][0], int(query["page"][0]) - pages.append((eco, page)) - assert len(pages) <= 7 + eco, offset = query["ecosystem"][0], int(query.get("after", ["0"])[0]) + assert "page" not in query + pages.append((eco, offset)) + assert len(pages) <= 6 rows = population if eco == "pip" else [] - return rows[(page - 1) * 100 : page * 100] + self.response_headers = {} + if offset + 100 < len(rows): + self.response_headers = {"Link": f'<{url.split("&after=")[0]}&after={offset + 100}>; rel="next"'} + return rows[offset : offset + 100] api, used = FakeAPI(), e.UsedSet() http = s.GitHub("fake", FilteredTransport()) @@ -674,4 +681,4 @@ def test_filtered_http_population_reaches_draft_floor_with_bounded_pages(tmp_pat report = d.run(tmp_path, github=api, osv=FakeOSV(), clones=FakeClones(), used=used, seed=7, known_empty=set(used.sources)) assert report["status"] == "draft" and report["repositories"] == 300 assert report["advisories"] == 300 - assert pages == [("pip", n) for n in range(1, 5)] + [(eco, 1) for eco in s.ECOSYSTEMS[1:]] + assert pages == [("pip", n) for n in (0, 100, 200)] + [(eco, 0) for eco in s.ECOSYSTEMS[1:]] diff --git a/tests/test_unseen_source.py b/tests/test_unseen_source.py index 49b77bb..afa868f 100644 --- a/tests/test_unseen_source.py +++ b/tests/test_unseen_source.py @@ -114,6 +114,8 @@ def test_http_adapters_with_fake_transport(): from urllib.parse import parse_qs, urlsplit class Transport: + response_headers = {} + def __init__(self): self.calls = [] @@ -121,9 +123,10 @@ def test_http_adapters_with_fake_transport(): self.calls.append((url, headers)) parsed = urlsplit(url) if parsed.path == "/advisories": - page = int(parse_qs(parsed.query)["page"][0]) + after = parse_qs(parsed.query).get("after") + self.response_headers = {} if after else {"Link": f'<{url}&after=cursor>; rel="next"'} 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) + {**advisory(), "ghsa_id": f"GHSA-test-test-{i:04d}"} for i in range(0 if not after else 100, 100 if not after else 101) ] if "/license" in parsed.path: assert parse_qs(parsed.query)["ref"] == ["b" * 40] @@ -150,6 +153,8 @@ def test_http_adapters_with_fake_transport(): def test_http_failures_are_not_empty_results(): class Transport: + response_headers = {} + def get(self, url, headers): raise ValueError("http_failure") @@ -283,12 +288,13 @@ def test_advisory_page_filters_match_candidate_population(ecosystem): expected = {c for f in taxonomy.families if f.id in s.FAMILIES for c in f.cwes} class Transport: + response_headers = {} + 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], @@ -297,7 +303,7 @@ def test_advisory_page_filters_match_candidate_population(ecosystem): } return [] - assert s.GitHub("fake", Transport()).advisory_page(2, ecosystem=ecosystem) == {"rows": [], "done": True} + assert s.GitHub("fake", Transport()).advisory_page(ecosystem=ecosystem) == {"rows": [], "next_url": None} # Every enumerated CWE is selectable at the inclusive date boundary. for cwe in expected: row = advisory() @@ -317,6 +323,8 @@ def test_advisory_page_omits_entire_cwe_filter_for_non_exact_taxonomy(monkeypatc monkeypatch.setattr(s, "load_families", lambda: taxonomy) class Transport: + response_headers = {} + def get(self, url, headers): query = parse_qs(urlsplit(url).query) assert "cwes" not in query @@ -324,4 +332,61 @@ def test_advisory_page_omits_entire_cwe_filter_for_non_exact_taxonomy(monkeypatc assert query["ecosystem"] == ["pip"] return [] - s.GitHub("fake", Transport()).advisory_page(1, ecosystem="pip") + s.GitHub("fake", Transport()).advisory_page(ecosystem="pip") + + +@pytest.mark.parametrize("journaled", [False, True]) +def test_cursor_walk_ignores_page_numbers_and_stops_without_next(tmp_path, journaled): + import io + from email.message import Message + from urllib.parse import parse_qs, urlencode, urlsplit + + calls = [] + + def opener(request, timeout): + query = parse_qs(urlsplit(request.full_url).query) + eco = query["ecosystem"][0] + after = query.get("after", [None])[0] + calls.append((eco, after)) + assert len(calls) <= 12 # Fail loudly if a page-number loop repeats forever. + # Like /advisories, ignore any page parameter. Only the opaque cursor advances. + offset = 100 if after == "opaque+/=" else 0 + rows = [{**advisory(), "ghsa_id": f"GHSA-test-test-{i:04d}"} for i in range(offset, offset + 100)] + response = io.BytesIO(json.dumps(rows).encode()) + response.headers = Message() + if after is None: + following = "https://api.github.com/advisories?" + urlencode({**query, "after": ["opaque+/="]}, doseq=True) + response.headers["link"] = f'; rel="prev", <{following}>; rel="next"' + else: + # A full last page with a Link header, but no next relation, is exhausted. + response.headers["link"] = '; rel="prev"' + return response + + transport = s.JSONTransport(opener=opener) + first = "https://api.github.com/advisories?ecosystem=pip" + assert transport.get(first + "&page=1", {}) == transport.get(first + "&page=100", {}) + calls.clear() + api = s.GitHub("fake", transport) + if journaled: + draft = importlib.import_module("benchmarks.unseen.draft") + with draft.Journal(tmp_path / "journal").open({"journal_version": 4}) as journal: + rows = draft.enumerate_advisories(api, journal) + else: + rows = list(api.advisories()) + assert [row["ghsa_id"] for row in rows] == [f"GHSA-test-test-{i:04d}" for i in range(200)] + assert calls == [(eco, cursor) for eco in s.ECOSYSTEMS for cursor in (None, "opaque+/=")] + + +def test_transport_replaces_last_successful_response_headers(): + import io + + responses = [io.BytesIO(b"{}"), io.BytesIO(b'{"broken":'), io.BytesIO(b"[]")] + responses[0].headers = {"Link": '; rel="next"'} + responses[1].headers = {"Link": "failed response"} + waits = [] + transport = s.JSONTransport(opener=lambda *a, **kw: responses.pop(0), sleeper=waits.append) + assert transport.get("https://example.invalid", {}) == {} + assert "cursor" in transport.response_headers["Link"] + assert transport.get("https://example.invalid", {}) == [] + assert transport.response_headers == {} + assert waits == [1]