File size: 4,251 Bytes
5bfc214 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | """OSV.dev bulk-export parsing (spec amendment: primary OOD source, see
DECISIONS.md "OOD source switched to OSV.dev bulk export").
OSV's `GIT`-ecosystem export (`https://storage.googleapis.com/osv-vulnerabilities/GIT/all.zip`,
~200MB, ~111k individual advisory JSON files bundled in one zip) carries
structured `affected[].ranges[]` entries with `type: "GIT"` giving a repo URL
and `events` with `fixed` (and usually `introduced`) commit SHAs directly —
exactly the field NVD's free-text reference list doesn't reliably have. No
need to regex-scrape a reference URL list; the fix commit is a first-class
field.
"""
from __future__ import annotations
import json
import re
import zipfile
from dataclasses import dataclass
_GITHUB_REPO_RE = re.compile(r"github\.com/(?P<owner>[\w.-]+)/(?P<repo>[\w.-]+?)(?:\.git)?/?$")
@dataclass
class OSVCandidate:
osv_id: str
cve_id: str | None # from `aliases`, first CVE-looking alias if present
owner: str
repo: str
fixed_sha: str
published: str | None # ISO8601 string, as given by OSV
cwe: str | None
def _first_cve_alias(aliases: list) -> str | None:
for a in aliases or []:
if isinstance(a, str) and a.upper().startswith("CVE-"):
return a
return None
def _extract_cwe(entry: dict) -> str | None:
# OSV's CWE placement varies by importer; check the common spots.
db_specific = entry.get("database_specific") or {}
cwe_ids = db_specific.get("cwe_ids") or db_specific.get("cwe") or []
if isinstance(cwe_ids, str):
return cwe_ids
if isinstance(cwe_ids, list) and cwe_ids:
return cwe_ids[0]
for sev in entry.get("severity") or []:
if sev.get("type") == "CWE":
return sev.get("score")
return None
def parse_osv_entry(entry: dict) -> list[OSVCandidate]:
"""One OSV advisory can list multiple affected packages/repos and
multiple GIT ranges (e.g. multiple repos, or multiple fix commits across
branches) — returns one OSVCandidate per (repo, fixed_sha) pair found."""
osv_id = entry.get("id", "")
cve_id = _first_cve_alias(entry.get("aliases"))
published = entry.get("published")
cwe = _extract_cwe(entry)
out: list[OSVCandidate] = []
seen: set[tuple[str, str]] = set()
for affected in entry.get("affected", []):
for rng in affected.get("ranges", []):
if rng.get("type") != "GIT":
continue
repo_url = rng.get("repo", "")
m = _GITHUB_REPO_RE.search(repo_url)
if not m:
continue # non-GitHub repo (e.g. a GitLab/self-hosted URL) — skip, we only fetch via GitHub API
owner, repo = m.group("owner"), m.group("repo")
for event in rng.get("events", []):
fixed_sha = event.get("fixed")
if not fixed_sha:
continue
key = (repo_url, fixed_sha)
if key in seen:
continue
seen.add(key)
out.append(OSVCandidate(
osv_id=osv_id, cve_id=cve_id, owner=owner, repo=repo,
fixed_sha=fixed_sha, published=published, cwe=cwe,
))
return out
def iter_osv_candidates_from_zip(zip_path: str, since: str | None = None):
"""Yields OSVCandidate for every GIT-type fix-commit event across every
advisory in the zip, optionally filtered to `published >= since`
(ISO date string, e.g. "2025-01-01")."""
with zipfile.ZipFile(zip_path) as zf:
for name in zf.namelist():
if not name.endswith(".json"):
continue
with zf.open(name) as f:
try:
entry = json.load(f)
except (json.JSONDecodeError, UnicodeDecodeError):
continue
if since and not _published_on_or_after(entry.get("published"), since):
continue
yield from parse_osv_entry(entry)
def _published_on_or_after(published: str | None, since: str) -> bool:
if not published:
return False
# Plain string comparison works for ISO8601-prefixed dates ("2025-01-03..." >= "2025-01-01").
return published[:10] >= since
|