| """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 |
| owner: str |
| repo: str |
| fixed_sha: str |
| published: str | None |
| 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: |
| |
| 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 |
| 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 |
| |
| return published[:10] >= since |
|
|
