websafe-kb/scripts/intel/sources/osv_api.py

from __future__ import annotations

import re
from typing import Any, Dict, List

import requests

from intel.http_client import build_session, request
from intel.models import Candidate
from intel.utils import unique


QUERY_BATCH_URL = "https://api.osv.dev/v1/querybatch"
DETAIL_URL = "https://api.osv.dev/v1/vulns/{vuln_id}"
CVSS_SCORE_RE = re.compile(r"/CVSS:3\.[01]/AV:[A-Z]/AC:[A-Z]/PR:[A-Z]/UI:[A-Z]/S:[A-Z]/C:[A-Z]/I:[A-Z]/A:[A-Z]")
NUMERIC_SCORE_RE = re.compile(r"([0-9]+(?:\.[0-9]+)?)")


def _fetch_detail(session: requests.Session, vuln_id: str) -> Dict[str, Any]:
    response = request(
        "GET",
        DETAIL_URL.format(vuln_id=vuln_id),
        session=session,
        headers={"User-Agent": "websafe-intel"},
    )
    response.raise_for_status()
    return response.json()


def _fixed_versions(vuln: Dict[str, Any]) -> List[str]:
    fixed = []
    for affected in vuln.get("affected", []):
        for rng in affected.get("ranges", []):
            for event in rng.get("events", []):
                if event.get("fixed"):
                    fixed.append(event["fixed"])
    return unique(fixed)


def _affected_versions(vuln: Dict[str, Any]) -> List[str]:
    versions = []
    ranges = []
    for affected in vuln.get("affected", []):
        versions.extend(affected.get("versions", [])[:20])
        for rng in affected.get("ranges", []):
            introduced = None
            fixed = None
            last_affected = None
            limit = None
            for event in rng.get("events", []):
                introduced = introduced or event.get("introduced")
                fixed = fixed or event.get("fixed")
                last_affected = last_affected or event.get("last_affected")
                limit = limit or event.get("limit")
            if introduced or fixed or last_affected or limit:
                parts = []
                if introduced:
                    parts.append(f"introduced={introduced}")
                if last_affected:
                    parts.append(f"last_affected={last_affected}")
                if fixed:
                    parts.append(f"fixed<{fixed}")
                if limit:
                    parts.append(f"limit<{limit}")
                ranges.append(", ".join(parts))
    return unique(versions + ranges)


def _severity(vuln: Dict[str, Any]) -> tuple[str, float | None]:
    best_score = None
    for sev in vuln.get("severity", []):
        score = sev.get("score", "")
        match = NUMERIC_SCORE_RE.search(score)
        if match:
            try:
                best_score = float(match.group(1))
                break
            except ValueError:
                continue
    if best_score is None:
        return "unknown", None
    if best_score >= 9.0:
        return "critical", best_score
    if best_score >= 7.0:
        return "high", best_score
    if best_score >= 4.0:
        return "medium", best_score
    return "low", best_score


def fetch(system: Dict[str, Any], source: Dict[str, Any]) -> List[Candidate]:
    packages = system.get("package_names", [])
    if not packages:
        return []

    queries = [{"package": {"name": pkg["name"], "ecosystem": pkg["ecosystem"]}} for pkg in packages]
    session = build_session(source)
    response = request(
        "POST",
        QUERY_BATCH_URL,
        source=source,
        session=session,
        json={"queries": queries},
        headers={"User-Agent": "websafe-intel"},
    )
    response.raise_for_status()
    payload = response.json()

    detail_cache: Dict[str, Dict[str, Any]] = {}
    candidates: List[Candidate] = []
    for package, result in zip(packages, payload.get("results", [])):
        for summary in result.get("vulns", []):
            vuln_id = summary.get("id")
            if not vuln_id:
                continue
            if vuln_id not in detail_cache:
                detail_cache[vuln_id] = _fetch_detail(session, vuln_id)
            vuln = detail_cache[vuln_id]

            aliases = unique(vuln.get("aliases", []) + [vuln.get("id")])
            refs = [ref.get("url") for ref in vuln.get("references", []) if ref.get("url")]
            severity, cvss_score = _severity(vuln)
            package_name = package["name"]
            if not package_name:
                for affected in vuln.get("affected", []):
                    pkg = affected.get("package") or {}
                    if pkg.get("name"):
                        package_name = pkg["name"]
                        break

            candidates.append(
                Candidate(
                    system_id=system["system_id"],
                    display_name=system["display_name"],
                    category=system["category"],
                    advisory_mode=source.get("advisory_mode", "core"),
                    source_kind=source["kind"],
                    source_name=source["name"],
                    source_confidence=source["confidence"],
                    source_url=refs[0] if refs else DETAIL_URL.format(vuln_id=vuln_id),
                    title=vuln.get("summary") or vuln.get("id") or f"OSV advisory for {package['name']}",
                    published_at=vuln.get("published"),
                    updated_at=vuln.get("modified"),
                    summary=vuln.get("details") or "",
                    severity=severity,
                    cvss_score=cvss_score,
                    aliases=aliases,
                    cve_ids=[item for item in aliases if item and item.startswith("CVE-")],
                    ghsa_ids=[item for item in aliases if item and item.startswith("GHSA-")],
                    osv_ids=[vuln.get("id")] if vuln.get("id") else [],
                    affected_versions=_affected_versions(vuln),
                    fixed_versions=_fixed_versions(vuln),
                    package_name=package_name,
                    references=refs,
                    raw=vuln,
                )
            )
    return candidates