mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-14 23:21:35 +00:00
ci: hold osv-scan findings whose fix is inside the dependency lockout window
uv is pinned to `exclude-newer = "3 days"` and .npmrc to `min-release-age=3`, so a release published in the last three days cannot be resolved even if we want it. When osv-scanner points at such a release, the job failed with nothing anyone could do about it, and a check that fails for three days at a time stops being read. The scan now emits JSON and a filter step decides the verdict. For each finding it takes the lowest fix above the installed version, reads that version's publish date from PyPI or the npm registry, and compares its age against the window read from pyproject.toml and .npmrc. Findings still inside the window are printed with the timestamp they become installable and do not fail the job; everything else fails as before. Registry lookup failures, unparseable versions and unknown ecosystems all fail closed.
This commit is contained in:
Yuneng Jiang
parent
4edf8f1551
commit
cafc2cf86a
3 changed files with 812 additions and 0 deletions
379
.github/scripts/osv_lockout_filter.py
vendored
Normal file
379
.github/scripts/osv_lockout_filter.py
vendored
Normal file
|
|
@ -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())
|
||||
21
.github/workflows/osv-scan.yml
vendored
21
.github/workflows/osv-scan.yml
vendored
|
|
@ -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"
|
||||
|
|
|
|||
412
tests/test_litellm/test_github_osv_lockout_filter.py
Normal file
412
tests/test_litellm/test_github_osv_lockout_filter.py
Normal file
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue