mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Marius CiepluchandClaude Opus 4.8
5620165540
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 <[email protected]>
393 lines
15 KiB
Python
393 lines
15 KiB
Python
"""Sanitized REST-shaped advisory fixtures; every external client is a fake."""
|
|
|
|
import copy
|
|
import importlib
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
s = importlib.import_module("benchmarks.unseen.source")
|
|
|
|
|
|
def advisory():
|
|
path = Path(__file__).resolve().parents[1] / "benchmarks/unseen/fixtures/advisory.json"
|
|
return json.loads(path.read_bytes())
|
|
|
|
|
|
class API:
|
|
def __init__(self):
|
|
self.parents = ["b" * 40]
|
|
self.spdx = "MIT"
|
|
self.license_text = "SPDX-License-Identifier: MIT"
|
|
self.calls = []
|
|
self.repo = {"full_name": "fixture/project", "private": False, "size": 100, "clone_url": "https://github.com/fixture/project.git"}
|
|
|
|
def get_repo(self, name):
|
|
return copy.deepcopy(self.repo)
|
|
|
|
def get_commit(self, name, sha):
|
|
return {"sha": sha, "parents": [{"sha": p} for p in self.parents]}
|
|
|
|
def get_license(self, name, ref):
|
|
self.calls.append((name, ref))
|
|
return {"text": self.license_text, "spdx_id": self.spdx, "path": "LICENSE"}
|
|
|
|
|
|
class OSV:
|
|
def get(self, identifier):
|
|
return {
|
|
"affected": [
|
|
{
|
|
"package": {"ecosystem": "PyPI", "name": "fixture"},
|
|
"ranges": [{"type": "ECOSYSTEM", "events": [{"introduced": "1.2"}, {"fixed": "1.3"}]}],
|
|
}
|
|
]
|
|
}
|
|
|
|
|
|
def test_candidate_cwe_cutoff_bracket_and_pinned_license():
|
|
api = API()
|
|
candidate = s.candidate(advisory(), api, OSV())
|
|
assert candidate.family == "injection" and candidate.post_cutoff
|
|
assert candidate.first_affected == "1.2"
|
|
assert candidate.license_class == "permissive"
|
|
assert api.calls == [("fixture/project", "b" * 40)]
|
|
raw = advisory()
|
|
raw["published_at"] = "2026-06-01T12:00:00Z"
|
|
assert not s.candidate(raw, api, OSV()).post_cutoff
|
|
raw["published_at"] = "2018-12-31T00:00:00Z"
|
|
with pytest.raises(s.Rejected, match="date"):
|
|
s.candidate(raw, api, OSV())
|
|
|
|
|
|
@pytest.mark.parametrize("spdx", ["GPL-3.0", "AGPL-3.0-only", "LGPL-2.1", "NOASSERTION", "", "BUSL-1.1", "MIT OR GPL-3.0"])
|
|
def test_private_licenses(spdx):
|
|
api = API()
|
|
api.spdx = spdx
|
|
api.license_text = ""
|
|
assert s.candidate(advisory(), api, OSV()).license_class == "private"
|
|
|
|
|
|
def test_license_file_overrides_api_and_missing_file_is_private():
|
|
api = API()
|
|
api.license_text = "SPDX-License-Identifier: GPL-3.0-only"
|
|
assert s.candidate(advisory(), api, OSV()).license_class == "private"
|
|
api.license_text = ""
|
|
api.spdx = "Apache-2.0"
|
|
assert s.candidate(advisory(), api, OSV()).license_spdx == "Apache-2.0"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutation,reason",
|
|
[
|
|
(lambda a, api: api.parents.append("c" * 40), "parents"),
|
|
(lambda a, api: a["references"].append("https://github.com/fixture/other/commit/" + "d" * 40), "fix_links"),
|
|
(lambda a, api: a.update(cwes=[{"cwe_id": "CWE-99999"}]), "family"),
|
|
(lambda a, api: api.repo.update(size=512001), "repository"),
|
|
(lambda a, api: api.repo.update(private=True), "repository"),
|
|
(lambda a, api: api.repo.update(clone_url="ssh://fixture/project"), "repository"),
|
|
],
|
|
)
|
|
def test_rejections(mutation, reason):
|
|
api, raw = API(), advisory()
|
|
mutation(raw, api)
|
|
with pytest.raises(s.Rejected, match=reason):
|
|
s.candidate(raw, api, OSV())
|
|
|
|
|
|
def test_post_cutoff_first_within_ecosystem_is_seeded():
|
|
rows = [advisory() for _ in range(12)]
|
|
for i, row in enumerate(rows):
|
|
row["ghsa_id"] += str(i)
|
|
if i % 2:
|
|
row["published_at"] = "2025-01-01T00:00:00Z"
|
|
ordered = s.ordered_advisories(rows, 17)
|
|
assert ordered == s.ordered_advisories(list(reversed(rows)), 17)
|
|
assert all(s.post_cutoff(row) for row in ordered[:6])
|
|
|
|
|
|
def test_http_adapters_with_fake_transport():
|
|
import base64
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
class Transport:
|
|
response_headers = {}
|
|
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def get(self, url, headers):
|
|
self.calls.append((url, headers))
|
|
parsed = urlsplit(url)
|
|
if parsed.path == "/advisories":
|
|
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 not after else 100, 100 if not after else 101)
|
|
]
|
|
if "/license" in parsed.path:
|
|
assert parse_qs(parsed.query)["ref"] == ["b" * 40]
|
|
return {
|
|
"license": {"spdx_id": "MIT"},
|
|
"path": "LICENSE",
|
|
"content": base64.b64encode(b"SPDX-License-Identifier: MIT").decode(),
|
|
}
|
|
if "/commits/" in parsed.path:
|
|
return {"sha": "a" * 40, "parents": [{"sha": "b" * 40}]}
|
|
if parsed.hostname == "api.osv.dev":
|
|
return OSV().get("ignored")
|
|
return API().repo
|
|
|
|
transport = Transport()
|
|
api = s.GitHub("fake-token", transport)
|
|
assert len(list(api.advisories())) == 101
|
|
assert s.candidate(advisory(), api, s.OSV(transport)).first_affected == "1.2"
|
|
assert all(headers["Authorization"] == "Bearer fake-token" for url, headers in transport.calls if "api.github.com" in url)
|
|
count = api.calls
|
|
api.get_repo("fixture/project")
|
|
assert api.calls == count
|
|
|
|
|
|
def test_http_failures_are_not_empty_results():
|
|
class Transport:
|
|
response_headers = {}
|
|
|
|
def get(self, url, headers):
|
|
raise ValueError("http_failure")
|
|
|
|
with pytest.raises(ValueError, match="http_failure"):
|
|
list(s.GitHub("fake", Transport()).advisories())
|
|
with pytest.raises(ValueError, match="http_failure"):
|
|
s.OSV(Transport()).get("GHSA-test-test-test")
|
|
|
|
|
|
def test_transport_rate_limit_retry_and_sanitized_failure():
|
|
import io
|
|
from urllib.error import HTTPError
|
|
|
|
waits, attempts = [], []
|
|
|
|
def opener(request, timeout):
|
|
attempts.append(request)
|
|
if len(attempts) == 1:
|
|
raise HTTPError(request.full_url, 403, "private message", {"X-RateLimit-Remaining": "0", "X-RateLimit-Reset": "120"}, None)
|
|
return io.BytesIO(b'{"ok": true}')
|
|
|
|
transport = s.JSONTransport(opener=opener, sleeper=waits.append, clock=lambda: 100)
|
|
assert transport.get("https://example.invalid", {}) == {"ok": True}
|
|
assert waits == [60] and transport.attempts == 2
|
|
|
|
def down(request, timeout):
|
|
raise HTTPError(request.full_url, 503, "private name", {}, None)
|
|
|
|
transport = s.JSONTransport(opener=down, sleeper=waits.append)
|
|
with pytest.raises(ValueError, match="^http_failure$"):
|
|
transport.get("https://example.invalid", {})
|
|
assert transport.attempts == 5
|
|
|
|
|
|
@pytest.mark.parametrize("failures", [1, 3, 4, 5, 7])
|
|
@pytest.mark.parametrize("kind", ["url", "timeout", "connection", "reset", "incomplete", "json"])
|
|
def test_transport_connection_and_body_transients(kind, failures):
|
|
import io
|
|
from http.client import IncompleteRead
|
|
from urllib.error import URLError
|
|
|
|
errors = {
|
|
"url": URLError("private endpoint"),
|
|
"timeout": TimeoutError("private endpoint"),
|
|
"connection": ConnectionError("private endpoint"),
|
|
"reset": ConnectionResetError("private endpoint"),
|
|
"incomplete": IncompleteRead(b"private partial body", 100),
|
|
}
|
|
waits, calls = [], []
|
|
|
|
class BrokenBody(io.BytesIO):
|
|
def read(self, *args):
|
|
raise errors[kind]
|
|
|
|
def opener(request, timeout):
|
|
calls.append(request)
|
|
if len(calls) <= failures:
|
|
if kind == "json":
|
|
return io.BytesIO(b'{"private":') # exercise json.load, not just the opener
|
|
if kind == "incomplete":
|
|
return BrokenBody()
|
|
raise errors[kind]
|
|
return io.BytesIO(b'{"ok": true}')
|
|
|
|
transport = s.JSONTransport(opener=opener, sleeper=waits.append)
|
|
if failures >= 5:
|
|
with pytest.raises(ValueError, match="^http_failure$"):
|
|
transport.get("https://example.invalid", {})
|
|
else:
|
|
assert transport.get("https://example.invalid", {}) == {"ok": True}
|
|
assert transport.attempts == len(calls) == min(failures + 1, 5)
|
|
assert waits == [2**i for i in range(min(failures, 4))]
|
|
|
|
|
|
@pytest.mark.parametrize("code", [400, 401, 403, 404, 422])
|
|
def test_transport_nonretryable_http_status(code):
|
|
from urllib.error import HTTPError
|
|
|
|
waits = []
|
|
|
|
def opener(request, timeout):
|
|
raise HTTPError(request.full_url, code, "private endpoint", {}, None)
|
|
|
|
transport = s.JSONTransport(opener=opener, sleeper=waits.append)
|
|
error, message = (s.RepositoryNotFound, "not_found") if code == 404 else (ValueError, "http_failure")
|
|
with pytest.raises(error, match=f"^{message}$"):
|
|
transport.get("https://example.invalid", {})
|
|
assert transport.attempts == 1 and waits == []
|
|
|
|
|
|
@pytest.mark.parametrize("code", [403, 429, 500, 502, 503, 504])
|
|
@pytest.mark.parametrize("retry_after", ["7", "invalid", "3606"])
|
|
def test_transport_http_retry_after_and_window_guard(code, retry_after):
|
|
import io
|
|
from urllib.error import HTTPError
|
|
|
|
waits, calls = [], []
|
|
|
|
def opener(request, timeout):
|
|
calls.append(request)
|
|
if len(calls) == 1:
|
|
raise HTTPError(request.full_url, code, "private endpoint", {"Retry-After": retry_after}, None)
|
|
return io.BytesIO(b"{}")
|
|
|
|
transport = s.JSONTransport(opener=opener, sleeper=waits.append)
|
|
if retry_after == "7":
|
|
assert transport.get("https://example.invalid", {}) == {}
|
|
assert waits == [7] and transport.attempts == 2
|
|
else:
|
|
with pytest.raises(ValueError, match="^http_retry_failed$"):
|
|
transport.get("https://example.invalid", {})
|
|
assert waits == [] and transport.attempts == 1
|
|
|
|
|
|
def test_transport_decode_retry_respects_response_retry_after():
|
|
import io
|
|
|
|
waits = []
|
|
responses = [io.BytesIO(b'{"truncated":'), io.BytesIO(b"{}")]
|
|
responses[0].headers = {"Retry-After": "9"}
|
|
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:
|
|
response_headers = {}
|
|
|
|
def get(self, url, headers):
|
|
query = parse_qs(urlsplit(url).query)
|
|
assert query == {
|
|
"type": ["reviewed"],
|
|
"per_page": ["100"],
|
|
"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(ecosystem=ecosystem) == {"rows": [], "next_url": None}
|
|
# 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:
|
|
response_headers = {}
|
|
|
|
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(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'<https://api.github.com/advisories?before=other>; rel="prev", <{following}>; rel="next"'
|
|
else:
|
|
# A full last page with a Link header, but no next relation, is exhausted.
|
|
response.headers["link"] = '<https://api.github.com/advisories?before=other>; 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": '<https://api.github.com/advisories?after=cursor>; 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]
|