mirror of
https://github.com/norandom/OpenUltraSAST.git
synced 2026-10-16 03:49:26 +00:00
Marius CiepluchandClaude Opus 4.8
90e2400e6f
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]>
328 lines
12 KiB
Python
328 lines
12 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:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def get(self, url, headers):
|
|
self.calls.append((url, headers))
|
|
parsed = urlsplit(url)
|
|
if parsed.path == "/advisories":
|
|
page = int(parse_qs(parsed.query)["page"][0])
|
|
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 {
|
|
"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:
|
|
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:
|
|
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")
|