diff --git a/.github/scripts/osv_lockout_filter.py b/.github/scripts/osv_lockout_filter.py new file mode 100644 index 00000000000..f107f19731c --- /dev/null +++ b/.github/scripts/osv_lockout_filter.py @@ -0,0 +1,379 @@ +#!/usr/bin/env python3 +"""Turn osv-scanner JSON results into a CI verdict that respects the dependency lockout window. + +A finding is only actionable if the fix it points at is old enough for our resolvers to +install it (``[tool.uv] exclude-newer`` for PyPI, ``min-release-age`` in ``.npmrc`` for npm). +Findings whose fix is still inside that window are reported and deferred until it expires. +""" + +from __future__ import annotations + +import argparse +import functools +import json +import os +import re +import sys +import time +import tomllib +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Callable, Mapping, Sequence + +from packaging.version import InvalidVersion, Version + +PYPI = "PyPI" +NPM = "npm" + +PublishTimes = Callable[[str, str], Mapping[str, datetime] | None] + +_SEMVER = re.compile( + r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)" + r"(?:-((?:0|[1-9]\d*|\d*[A-Za-z-][0-9A-Za-z-]*)(?:\.(?:0|[1-9]\d*|\d*[A-Za-z-][0-9A-Za-z-]*))*))?" + r"(?:\+[0-9A-Za-z-]+(?:\.[0-9A-Za-z-]+)*)?$" +) +_PYPI_NAME = re.compile(r"[-_.]+") +_DAYS = re.compile(r"\s*(\d+)\s*days?\s*") +_MIN_RELEASE_AGE = re.compile(r"^\s*min-release-age\s*=\s*(\d+)\s*$", re.MULTILINE) + + +@dataclass(frozen=True, slots=True) +class Finding: + ecosystem: str + package: str + installed: str + vuln_id: str + severity: str + fixed_versions: tuple[str, ...] + source: str + + +@dataclass(frozen=True, slots=True) +class Deferred: + finding: Finding + target: str + published: datetime + unlocks_at: datetime + + +@dataclass(frozen=True, slots=True) +class Actionable: + finding: Finding + target: str | None + reason: str + + +@dataclass(frozen=True, slots=True) +class WindowError: + message: str + + +Verdict = Deferred | Actionable + + +def semver_key(version: str) -> tuple[tuple[int, int, int], tuple[int, ...], tuple[tuple[int, int, str], ...]] | None: + matched = _SEMVER.match(version.strip()) + if matched is None: + return None + core = (int(matched.group(1)), int(matched.group(2)), int(matched.group(3))) + prerelease = matched.group(4) + if prerelease is None: + return (core, (1,), ()) + return ( + core, + (0,), + tuple((0, int(part), "") if part.isdigit() else (1, 0, part) for part in prerelease.split(".")), + ) + + +def pypi_key(version: str) -> Version | None: + try: + return Version(version) + except InvalidVersion: + return None + + +def version_key(ecosystem: str, version: str) -> object | None: + if ecosystem == PYPI: + return pypi_key(version) + if ecosystem == NPM: + return semver_key(version) + return None + + +def _base_ecosystem(raw: str) -> str: + return raw.split(":", 1)[0] + + +def _canonical_name(ecosystem: str, name: str) -> str: + return _PYPI_NAME.sub("-", name).lower() if ecosystem == PYPI else name + + +def _fixed_versions(vulnerability: dict, ecosystem: str, package: str) -> tuple[str, ...]: + wanted = _canonical_name(ecosystem, package) + return tuple( + dict.fromkeys( + event["fixed"] + for affected in vulnerability.get("affected") or [] + for affected_package in ((affected.get("package") or {}),) + if _base_ecosystem(str(affected_package.get("ecosystem") or "")) == ecosystem + and _canonical_name(ecosystem, str(affected_package.get("name") or "")) == wanted + for version_range in affected.get("ranges") or [] + if version_range.get("type") in ("ECOSYSTEM", "SEMVER") + for event in version_range.get("events") or [] + if "fixed" in event + ) + ) + + +def _severity(groups: Sequence[dict], vuln_id: str) -> str: + matches = tuple( + str(group.get("max_severity") or "") for group in groups or [] if vuln_id in (group.get("ids") or []) + ) + return next((severity for severity in matches if severity), "-") + + +def _findings_for_package(scanned: dict, source: str) -> tuple[Finding, ...]: + package = scanned["package"] + ecosystem = _base_ecosystem(str(package["ecosystem"])) + name = str(package["name"]) + groups = scanned.get("groups") or [] + return tuple( + Finding( + ecosystem=ecosystem, + package=name, + installed=str(package["version"]), + vuln_id=str(vulnerability["id"]), + severity=_severity(groups, str(vulnerability["id"])), + fixed_versions=_fixed_versions(vulnerability, ecosystem, name), + source=source, + ) + for vulnerability in scanned.get("vulnerabilities") or [] + ) + + +def _relative_source(path: str, repo_root: Path) -> str: + try: + return str(Path(path).relative_to(repo_root)) + except ValueError: + return path + + +def parse_findings(payload: dict, repo_root: Path) -> tuple[Finding, ...]: + return tuple( + finding + for result in payload.get("results") or [] + for source in (_relative_source(str((result.get("source") or {}).get("path") or "?"), repo_root),) + for scanned in result.get("packages") or [] + for finding in _findings_for_package(scanned, source) + ) + + +def _fetch_json(url: str, attempts: int = 3) -> dict | None: + request = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": "litellm-osv-lockout"}) + for attempt in range(attempts): + try: + with urllib.request.urlopen(request, timeout=60) as response: + return json.loads(response.read()) + except (urllib.error.URLError, TimeoutError, OSError, json.JSONDecodeError) as exc: + print(f"osv-lockout: {url} lookup failed ({exc})", file=sys.stderr) + if attempt + 1 < attempts: + time.sleep(2 * (attempt + 1)) + return None + + +def _parse_timestamp(raw: str) -> datetime | None: + try: + parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=timezone.utc) + + +def _earliest_upload(files: Sequence[dict]) -> datetime | None: + stamps = tuple( + stamp + for entry in files or [] + for stamp in (_parse_timestamp(str(entry.get("upload_time_iso_8601") or "")),) + if stamp is not None + ) + return min(stamps) if stamps else None + + +def _pypi_publish_times(name: str) -> Mapping[str, datetime] | None: + payload = _fetch_json(f"https://pypi.org/pypi/{urllib.parse.quote(name, safe='')}/json") + if payload is None: + return None + return { + version: stamp + for version, files in (payload.get("releases") or {}).items() + for stamp in (_earliest_upload(files),) + if stamp is not None + } + + +def _npm_publish_times(name: str) -> Mapping[str, datetime] | None: + payload = _fetch_json(f"https://registry.npmjs.org/{urllib.parse.quote(name, safe='@')}") + if payload is None: + return None + return { + version: stamp + for version, raw in (payload.get("time") or {}).items() + if version not in ("created", "modified") + for stamp in (_parse_timestamp(str(raw)),) + if stamp is not None + } + + +@functools.lru_cache(maxsize=None) +def registry_publish_times(ecosystem: str, name: str) -> Mapping[str, datetime] | None: + if ecosystem == PYPI: + return _pypi_publish_times(name) + if ecosystem == NPM: + return _npm_publish_times(name) + return None + + +def _uv_window(pyproject: Path) -> timedelta | WindowError: + if not pyproject.is_file(): + return WindowError(f"{pyproject} not found; cannot determine the PyPI lockout window") + raw = ((tomllib.loads(pyproject.read_text(encoding="utf-8")).get("tool") or {}).get("uv") or {}).get( + "exclude-newer" + ) + if raw is None: + return timedelta(0) + matched = _DAYS.fullmatch(str(raw)) + if matched is None: + return WindowError( + f"[tool.uv] exclude-newer = {raw!r} is not an 'N days' window; teach osv_lockout_filter.py how to read it" + ) + return timedelta(days=int(matched.group(1))) + + +def _npm_window(npmrc: Path) -> timedelta: + if not npmrc.is_file(): + return timedelta(0) + matched = _MIN_RELEASE_AGE.search(npmrc.read_text(encoding="utf-8")) + return timedelta(days=int(matched.group(1))) if matched else timedelta(0) + + +def lockout_windows(repo_root: Path) -> Mapping[str, timedelta] | WindowError: + pypi = _uv_window(repo_root / "pyproject.toml") + if isinstance(pypi, WindowError): + return pypi + return {PYPI: pypi, NPM: _npm_window(repo_root / ".npmrc")} + + +def _publish_date(times: Mapping[str, datetime], ecosystem: str, version: str) -> datetime | None: + if version in times: + return times[version] + wanted = version_key(ecosystem, version) + if wanted is None: + return None + return next((stamp for other, stamp in times.items() if version_key(ecosystem, other) == wanted), None) + + +def evaluate( + finding: Finding, + windows: Mapping[str, timedelta], + publish_times: PublishTimes, + now: datetime, +) -> Verdict: + window = windows.get(finding.ecosystem) + if window is None: + return Actionable(finding, None, f"no lockout window is configured for the {finding.ecosystem} ecosystem") + installed = version_key(finding.ecosystem, finding.installed) + if installed is None: + return Actionable(finding, None, f"cannot parse the installed version {finding.installed!r}") + upgrades = tuple( + (key, version) + for version in finding.fixed_versions + for key in (version_key(finding.ecosystem, version),) + if key is not None and key > installed + ) + if not upgrades: + return Actionable(finding, None, "no published fix is newer than the installed version") + target = sorted(upgrades, key=lambda pair: pair[0])[0][1] + times = publish_times(finding.ecosystem, finding.package) + if times is None: + return Actionable(finding, target, "could not read publish dates from the package registry") + published = _publish_date(times, finding.ecosystem, target) + if published is None: + return Actionable(finding, target, f"the registry lists no publish date for {target}") + unlocks_at = published + window + if unlocks_at > now: + return Deferred(finding, target, published, unlocks_at) + return Actionable(finding, target, f"{target} has been installable since {_stamp(unlocks_at)}") + + +def _stamp(moment: datetime) -> str: + return moment.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + + +def _describe(finding: Finding, target: str | None) -> str: + upgrade = f"{finding.installed} -> {target}" if target else f"{finding.installed} (no upgrade)" + return f"{finding.ecosystem} {finding.package} {upgrade} {finding.vuln_id} (severity {finding.severity}) [{finding.source}]" + + +def render( + windows: Mapping[str, timedelta], + deferred: Sequence[Deferred], + actionable: Sequence[Actionable], +) -> str: + header = "Dependency lockout window: " + ", ".join( + f"{ecosystem} {window.days}d" for ecosystem, window in sorted(windows.items()) + ) + deferred_lines = tuple( + f" {_describe(item.finding, item.target)}\n" + f" published {_stamp(item.published)}; installable from {_stamp(item.unlocks_at)}" + for item in deferred + ) + actionable_lines = tuple(f" {_describe(item.finding, item.target)}\n {item.reason}" for item in actionable) + sections = ( + ( + f"Deferred ({len(deferred)}) - the fix is still inside the lockout window, so it cannot be pulled in yet:", + deferred_lines or (" none",), + ), + ( + f"Blocking ({len(actionable)}) - the fix can be pulled in now:", + actionable_lines or (" none",), + ), + ) + return "\n".join((header, "", *(line for title, lines in sections for line in (title, *lines, "")))) + + +def main(argv: Sequence[str] | None = None, publish_times: PublishTimes = registry_publish_times) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--results", required=True, type=Path, help="osv-scanner --format json output file") + parser.add_argument("--repo-root", type=Path, default=Path(__file__).resolve().parents[2]) + args = parser.parse_args(argv) + + windows = lockout_windows(args.repo_root) + if isinstance(windows, WindowError): + print(f"osv-lockout: {windows.message}", file=sys.stderr) + return 1 + if not args.results.is_file(): + print(f"osv-lockout: {args.results} not found", file=sys.stderr) + return 1 + + findings = parse_findings(json.loads(args.results.read_text(encoding="utf-8")), args.repo_root) + now = datetime.now(timezone.utc) + verdicts = tuple(evaluate(finding, windows, publish_times, now) for finding in findings) + deferred = tuple(verdict for verdict in verdicts if isinstance(verdict, Deferred)) + actionable = tuple(verdict for verdict in verdicts if isinstance(verdict, Actionable)) + + report = render(windows, deferred, actionable) + print(report) + summary_path = os.environ.get("GITHUB_STEP_SUMMARY") + if summary_path: + with open(summary_path, "a", encoding="utf-8") as summary: + summary.write(f"### OSV scan\n\n```\n{report}\n```\n") + return 1 if actionable else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/.github/workflows/osv-scan.yml b/.github/workflows/osv-scan.yml index 31104002dab..a81f578617a 100644 --- a/.github/workflows/osv-scan.yml +++ b/.github/workflows/osv-scan.yml @@ -36,9 +36,30 @@ jobs: echo "bc98e15319ed0d515e3f9235287ba53cdc5535d576d24fd573978ecfe9ab92dc $RUNNER_TEMP/osv-scanner" | sha256sum -c - chmod +x "$RUNNER_TEMP/osv-scanner" + - name: Set up Python + uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 + with: + python-version: "3.12" + + # Exit 1 means "something is affected" and is handed to the filter step; anything above + # 1 is a scanner failure and still fails the job. - name: Scan lockfiles run: | + set +e "$RUNNER_TEMP/osv-scanner" scan source \ --config osv-scanner.toml \ + --format json \ + --output-file "$RUNNER_TEMP/osv-results.json" \ -L uv.lock \ -L ui/litellm-dashboard/package-lock.json + scan_status=$? + set -e + if [ "$scan_status" -gt 1 ]; then + echo "osv-scanner exited with status $scan_status" + exit "$scan_status" + fi + + - name: Apply the dependency lockout window + run: | + python -m pip install --disable-pip-version-check "packaging==26.2" + python .github/scripts/osv_lockout_filter.py --results "$RUNNER_TEMP/osv-results.json" diff --git a/tests/test_litellm/test_github_osv_lockout_filter.py b/tests/test_litellm/test_github_osv_lockout_filter.py new file mode 100644 index 00000000000..d1a9b729196 --- /dev/null +++ b/tests/test_litellm/test_github_osv_lockout_filter.py @@ -0,0 +1,412 @@ +"""Unit tests for `.github/scripts/osv_lockout_filter.py`. + +The filter decides whether an osv-scanner finding should fail CI. It must fail only when the +fix is old enough for our resolvers to install it, and must stay quiet while the fix is still +inside the dependency lockout window (`[tool.uv] exclude-newer`, `.npmrc` min-release-age). +Registry lookups are dependency-injected, so nothing here touches the network. +""" + +from __future__ import annotations + +import importlib.util +import json +import sys +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT_PATH = REPO_ROOT / ".github" / "scripts" / "osv_lockout_filter.py" + +NOW = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc) + + +@pytest.fixture(scope="module") +def filter_module(): + spec = importlib.util.spec_from_file_location("osv_lockout_filter", SCRIPT_PATH) + assert spec and spec.loader, f"Could not load spec for {SCRIPT_PATH}" + module = importlib.util.module_from_spec(spec) + sys.modules["osv_lockout_filter"] = module + spec.loader.exec_module(module) + return module + + +@pytest.fixture +def windows(filter_module): + return {filter_module.PYPI: timedelta(days=3), filter_module.NPM: timedelta(days=3)} + + +def _osv_payload( + *, + path: str, + ecosystem: str, + name: str, + version: str, + vuln_id: str, + fixed: list[str], + severity: str = "7.5", +) -> dict: + return { + "results": [ + { + "source": {"path": path, "type": "lockfile"}, + "packages": [ + { + "package": {"name": name, "version": version, "ecosystem": ecosystem}, + "groups": [{"ids": [vuln_id], "aliases": [vuln_id], "max_severity": severity}], + "vulnerabilities": [ + { + "id": vuln_id, + "affected": [ + { + "package": {"ecosystem": ecosystem, "name": name}, + "ranges": [ + { + "type": "ECOSYSTEM" if ecosystem == "PyPI" else "SEMVER", + "events": [{"introduced": "0"}, {"fixed": f}], + } + for f in fixed + ], + } + ], + } + ], + } + ], + } + ] + } + + +def _finding(filter_module, **overrides): + defaults = dict( + ecosystem=filter_module.PYPI, + package="gitpython", + installed="3.1.54", + vuln_id="GHSA-94p4-4cq8-9g67", + severity="7.5", + fixed_versions=("3.1.55",), + source="uv.lock", + ) + return filter_module.Finding(**{**defaults, **overrides}) + + +def _times(mapping: dict[str, datetime]): + return lambda ecosystem, name: mapping + + +class TestParsing: + def test_extracts_every_field_from_a_real_scanner_payload(self, filter_module): + payload = _osv_payload( + path=str(REPO_ROOT / "ui" / "litellm-dashboard" / "package-lock.json"), + ecosystem="npm", + name="brace-expansion", + version="5.0.7", + vuln_id="GHSA-mh99-v99m-4gvg", + fixed=["5.0.8"], + ) + + findings = filter_module.parse_findings(payload, REPO_ROOT) + + assert len(findings) == 1 + assert findings[0] == filter_module.Finding( + ecosystem="npm", + package="brace-expansion", + installed="5.0.7", + vuln_id="GHSA-mh99-v99m-4gvg", + severity="7.5", + fixed_versions=("5.0.8",), + source="ui/litellm-dashboard/package-lock.json", + ) + + def test_ignores_fixed_events_for_a_different_package(self, filter_module): + payload = _osv_payload( + path="uv.lock", ecosystem="PyPI", name="gitpython", version="3.1.54", vuln_id="GHSA-x", fixed=["3.1.55"] + ) + affected = payload["results"][0]["packages"][0]["vulnerabilities"][0]["affected"] + affected.append( + { + "package": {"ecosystem": "PyPI", "name": "some-other-package"}, + "ranges": [{"type": "ECOSYSTEM", "events": [{"introduced": "0"}, {"fixed": "9.9.9"}]}], + } + ) + + assert filter_module.parse_findings(payload, REPO_ROOT)[0].fixed_versions == ("3.1.55",) + + def test_matches_pypi_names_across_normalization(self, filter_module): + payload = _osv_payload( + path="uv.lock", ecosystem="PyPI", name="Zope.Interface", version="5.0", vuln_id="GHSA-x", fixed=["5.1"] + ) + payload["results"][0]["packages"][0]["vulnerabilities"][0]["affected"][0]["package"]["name"] = "zope_interface" + + assert filter_module.parse_findings(payload, REPO_ROOT)[0].fixed_versions == ("5.1",) + + def test_skips_git_ranges_which_carry_commit_hashes_not_versions(self, filter_module): + payload = _osv_payload( + path="uv.lock", ecosystem="PyPI", name="gitpython", version="3.1.54", vuln_id="GHSA-x", fixed=["3.1.55"] + ) + payload["results"][0]["packages"][0]["vulnerabilities"][0]["affected"][0]["ranges"].append( + {"type": "GIT", "events": [{"introduced": "0"}, {"fixed": "deadbeef"}]} + ) + + assert filter_module.parse_findings(payload, REPO_ROOT)[0].fixed_versions == ("3.1.55",) + + +class TestVersionOrdering: + @pytest.mark.parametrize( + "lower, higher", + [ + ("1.0.0", "1.0.1"), + ("1.9.0", "1.10.0"), + ("1.0.0-rc.1", "1.0.0"), + ("1.0.0-alpha.1", "1.0.0-alpha.2"), + ("1.0.0-alpha.1", "1.0.0-alpha.beta"), + ("1.0.0-alpha", "1.0.0-alpha.1"), + ], + ) + def test_semver_precedence(self, filter_module, lower, higher): + assert filter_module.semver_key(lower) < filter_module.semver_key(higher) + + def test_semver_ignores_build_metadata(self, filter_module): + assert filter_module.semver_key("1.2.3+build.5") == filter_module.semver_key("1.2.3") + + def test_semver_rejects_non_semver(self, filter_module): + assert filter_module.semver_key("not-a-version") is None + assert filter_module.semver_key("1.2") is None + + +class TestEvaluate: + def test_defers_when_the_fix_is_inside_the_window(self, filter_module, windows): + published = NOW - timedelta(days=2) + + verdict = filter_module.evaluate( + _finding(filter_module), windows, _times({"3.1.55": published}), NOW + ) + + assert isinstance(verdict, filter_module.Deferred) + assert verdict.target == "3.1.55" + assert verdict.unlocks_at == published + timedelta(days=3) + + def test_blocks_when_the_fix_is_older_than_the_window(self, filter_module, windows): + verdict = filter_module.evaluate( + _finding(filter_module), windows, _times({"3.1.55": NOW - timedelta(days=4)}), NOW + ) + + assert isinstance(verdict, filter_module.Actionable) + assert verdict.target == "3.1.55" + + def test_blocks_the_moment_the_window_expires(self, filter_module, windows): + exactly_three_days_old = NOW - timedelta(days=3) + + verdict = filter_module.evaluate( + _finding(filter_module), windows, _times({"3.1.55": exactly_three_days_old}), NOW + ) + + assert isinstance(verdict, filter_module.Actionable) + + def test_defers_one_second_before_the_window_expires(self, filter_module, windows): + verdict = filter_module.evaluate( + _finding(filter_module), + windows, + _times({"3.1.55": NOW - timedelta(days=3) + timedelta(seconds=1)}), + NOW, + ) + + assert isinstance(verdict, filter_module.Deferred) + + def test_targets_the_lowest_fix_above_the_installed_version_not_an_old_branch_backport( + self, filter_module, windows + ): + finding = _finding(filter_module, package="django", installed="5.2.0", fixed_versions=("4.2.9", "5.2.1")) + + verdict = filter_module.evaluate( + finding, + windows, + _times({"4.2.9": NOW - timedelta(days=400), "5.2.1": NOW - timedelta(days=1)}), + NOW, + ) + + assert isinstance(verdict, filter_module.Deferred) + assert verdict.target == "5.2.1" + + def test_picks_the_lowest_of_several_reachable_fixes(self, filter_module, windows): + finding = _finding(filter_module, installed="1.0.0", fixed_versions=("3.0.0", "1.0.1", "2.0.0")) + + verdict = filter_module.evaluate( + finding, + windows, + _times({v: NOW - timedelta(days=10) for v in ("1.0.1", "2.0.0", "3.0.0")}), + NOW, + ) + + assert verdict.target == "1.0.1" + + def test_blocks_when_no_fix_is_newer_than_the_installed_version(self, filter_module, windows): + finding = _finding(filter_module, installed="4.0.0", fixed_versions=("3.1.55",)) + + verdict = filter_module.evaluate(finding, windows, _times({"3.1.55": NOW}), NOW) + + assert isinstance(verdict, filter_module.Actionable) + assert verdict.target is None + + def test_blocks_when_the_registry_lookup_fails(self, filter_module, windows): + verdict = filter_module.evaluate( + _finding(filter_module), windows, lambda ecosystem, name: None, NOW + ) + + assert isinstance(verdict, filter_module.Actionable) + assert "registry" in verdict.reason + + def test_blocks_when_the_registry_has_no_date_for_the_target(self, filter_module, windows): + verdict = filter_module.evaluate( + _finding(filter_module), windows, _times({"3.1.54": NOW - timedelta(days=9)}), NOW + ) + + assert isinstance(verdict, filter_module.Actionable) + + def test_blocks_for_an_ecosystem_with_no_configured_window(self, filter_module, windows): + finding = _finding(filter_module, ecosystem="Go", package="golang.org/x/net", installed="0.1.0") + + verdict = filter_module.evaluate(finding, windows, _times({"0.2.0": NOW - timedelta(days=9)}), NOW) + + assert isinstance(verdict, filter_module.Actionable) + + def test_uses_the_npm_window_for_npm_findings(self, filter_module): + finding = _finding( + filter_module, ecosystem="npm", package="brace-expansion", installed="5.0.7", fixed_versions=("5.0.8",) + ) + mixed = {filter_module.PYPI: timedelta(days=0), filter_module.NPM: timedelta(days=3)} + + verdict = filter_module.evaluate(finding, mixed, _times({"5.0.8": NOW - timedelta(days=2)}), NOW) + + assert isinstance(verdict, filter_module.Deferred) + + +class TestLockoutWindows: + def test_reads_the_windows_this_repo_actually_enforces(self, filter_module): + resolved = filter_module.lockout_windows(REPO_ROOT) + + assert not isinstance(resolved, filter_module.WindowError) + assert resolved == {filter_module.PYPI: timedelta(days=3), filter_module.NPM: timedelta(days=3)} + + def test_reads_a_custom_day_count(self, filter_module, tmp_path): + (tmp_path / "pyproject.toml").write_text('[tool.uv]\nexclude-newer = "7 days"\n') + (tmp_path / ".npmrc").write_text("ignore-scripts=true\nmin-release-age=5\n") + + assert filter_module.lockout_windows(tmp_path) == { + filter_module.PYPI: timedelta(days=7), + filter_module.NPM: timedelta(days=5), + } + + def test_no_window_configured_means_no_deferral(self, filter_module, tmp_path): + (tmp_path / "pyproject.toml").write_text("[tool.uv]\n") + + assert filter_module.lockout_windows(tmp_path) == { + filter_module.PYPI: timedelta(0), + filter_module.NPM: timedelta(0), + } + + def test_rejects_an_exclude_newer_form_it_cannot_interpret(self, filter_module, tmp_path): + (tmp_path / "pyproject.toml").write_text("[tool.uv]\nexclude-newer = 2026-01-01\n") + + assert isinstance(filter_module.lockout_windows(tmp_path), filter_module.WindowError) + + def test_errors_when_pyproject_is_missing(self, filter_module, tmp_path): + assert isinstance(filter_module.lockout_windows(tmp_path), filter_module.WindowError) + + +class TestMain: + def _write(self, tmp_path: Path, payload: dict) -> Path: + results = tmp_path / "osv-results.json" + results.write_text(json.dumps(payload), encoding="utf-8") + return results + + def test_exits_zero_when_the_only_finding_is_inside_the_window(self, filter_module, tmp_path, capsys): + results = self._write( + tmp_path, + _osv_payload( + path="uv.lock", + ecosystem="PyPI", + name="gitpython", + version="3.1.54", + vuln_id="GHSA-94p4-4cq8-9g67", + fixed=["3.1.55"], + ), + ) + recent = datetime.now(timezone.utc) - timedelta(hours=1) + + status = filter_module.main( + ["--results", str(results), "--repo-root", str(REPO_ROOT)], + publish_times=_times({"3.1.55": recent}), + ) + + assert status == 0 + assert "GHSA-94p4-4cq8-9g67" in capsys.readouterr().out + + def test_exits_one_when_the_fix_is_installable_today(self, filter_module, tmp_path): + results = self._write( + tmp_path, + _osv_payload( + path="uv.lock", + ecosystem="PyPI", + name="gitpython", + version="3.1.54", + vuln_id="GHSA-94p4-4cq8-9g67", + fixed=["3.1.55"], + ), + ) + old = datetime.now(timezone.utc) - timedelta(days=30) + + status = filter_module.main( + ["--results", str(results), "--repo-root", str(REPO_ROOT)], + publish_times=_times({"3.1.55": old}), + ) + + assert status == 1 + + def test_exits_zero_on_a_clean_scan(self, filter_module, tmp_path): + results = self._write(tmp_path, {"results": []}) + + assert filter_module.main(["--results", str(results), "--repo-root", str(REPO_ROOT)]) == 0 + + def test_exits_one_when_the_results_file_is_missing(self, filter_module, tmp_path): + assert filter_module.main(["--results", str(tmp_path / "nope.json"), "--repo-root", str(REPO_ROOT)]) == 1 + + def test_writes_a_step_summary_when_github_provides_one(self, filter_module, tmp_path, monkeypatch): + results = self._write( + tmp_path, + _osv_payload( + path="uv.lock", + ecosystem="PyPI", + name="gitpython", + version="3.1.54", + vuln_id="GHSA-94p4-4cq8-9g67", + fixed=["3.1.55"], + ), + ) + summary = tmp_path / "summary.md" + monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(summary)) + + filter_module.main( + ["--results", str(results), "--repo-root", str(REPO_ROOT)], + publish_times=_times({"3.1.55": datetime.now(timezone.utc)}), + ) + + assert "GHSA-94p4-4cq8-9g67" in summary.read_text(encoding="utf-8") + + +class TestRender: + def test_names_the_date_a_deferred_finding_starts_failing(self, filter_module, windows): + deferred = filter_module.Deferred( + finding=_finding(filter_module), + target="3.1.55", + published=datetime(2026, 7, 23, 2, 52, tzinfo=timezone.utc), + unlocks_at=datetime(2026, 7, 26, 2, 52, tzinfo=timezone.utc), + ) + + report = filter_module.render(windows, (deferred,), ()) + + assert "3.1.54 -> 3.1.55" in report + assert "2026-07-26 02:52 UTC" in report + assert "Blocking (0)" in report