asdf

4 files changed, 1075 insertions, 0 deletions

diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..c61ec2e
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,3 @@
+*.json
+*.tsv
+*.csv
diff --git a/csv/critical.sh b/csv/critical.sh
new file mode 100755
index 0000000..ab20b34
--- /dev/null
+++ b/csv/critical.sh
@@ -0,0 +1,89 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [ "$#" -lt 1 ]; then
+    echo "Usage: $0 doca-*-vuln.csv [more.csv ...]" >&2
+    exit 1
+fi
+
+python3 - "$@" <<'PY'
+import csv
+import os
+import re
+import sys
+
+FILENAME_RE = re.compile(
+    r"^doca-(?P<container>.+)-(?P<version>\d+\.\d+\.\d+)-(?P<host>host|dpu)-vuln\.csv$"
+)
+
+META_COLUMNS = ["doca_version", "doca_host", "doca_container"]
+
+
+def parse_metadata(path):
+    base = os.path.basename(path)
+    match = FILENAME_RE.match(base)
+
+    if not match:
+        raise ValueError(
+            f"Filename does not match expected pattern: {base}\n"
+            f"Expected: doca-(container)-(version)-(host)-vuln.csv"
+        )
+
+    return {
+        "doca_version": match.group("version"),
+        "doca_host": match.group("host"),
+        "doca_container": match.group("container"),
+    }
+
+
+def is_critical(row):
+    fields_to_check = [
+        "nvd_severity",
+        "severity",
+        "osv_severity",
+    ]
+
+    for field in fields_to_check:
+        value = row.get(field, "")
+        if value and "CRITICAL" in value.upper():
+            return True
+
+    return False
+
+
+def main(paths):
+    writer = None
+    output_header = None
+
+    for path in paths:
+        metadata = parse_metadata(path)
+
+        with open(path, "r", encoding="utf-8", newline="") as f:
+            reader = csv.DictReader(f)
+
+            if reader.fieldnames is None:
+                continue
+
+            if writer is None:
+                original_header = reader.fieldnames
+                output_header = META_COLUMNS + original_header
+                writer = csv.DictWriter(
+                    sys.stdout,
+                    fieldnames=output_header,
+                    extrasaction="ignore",
+                )
+                writer.writeheader()
+
+            for row in reader:
+                if not is_critical(row):
+                    continue
+
+                output_row = {}
+                output_row.update(metadata)
+                output_row.update(row)
+                writer.writerow(output_row)
+
+
+if __name__ == "__main__":
+    main(sys.argv[1:])
+PY
diff --git a/csv/sort.sh b/csv/sort.sh
new file mode 100755
index 0000000..1a025a4
--- /dev/null
+++ b/csv/sort.sh
@@ -0,0 +1,135 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [ "$#" -lt 2 ]; then
+    echo "Usage: $0 <name> doca-*-vuln.csv [more.csv ...]" >&2
+    echo "Example: $0 doca doca-*-vuln.csv" >&2
+    exit 1
+fi
+
+NAME="$1"
+shift
+
+python3 - "$NAME" "$@" <<'PY'
+import csv
+import os
+import re
+import sys
+
+FILENAME_RE = re.compile(
+    r"^doca-(?P<container>.+)-(?P<version>\d+\.\d+\.\d+)-(?P<host>host|dpu)-vuln\.csv$"
+)
+
+META_COLUMNS = ["doca_version", "doca_host", "doca_container"]
+
+OUTPUT_BUCKETS = [
+    ("network", "nopriv"),
+    ("network", "priv"),
+    ("local", "nopriv"),
+    ("local", "priv"),
+]
+
+
+def parse_metadata(path):
+    base = os.path.basename(path)
+    match = FILENAME_RE.match(base)
+
+    if not match:
+        raise ValueError(
+            f"Filename does not match expected pattern: {base}\n"
+            f"Expected: doca-(container)-(version)-(host)-vuln.csv"
+        )
+
+    return {
+        "doca_version": match.group("version"),
+        "doca_host": match.group("host"),
+        "doca_container": match.group("container"),
+    }
+
+
+def get_av(row):
+    return (
+        row.get("nvd_AV")
+        or row.get("AV")
+        or row.get("cvss_AV")
+        or ""
+    ).strip().upper()
+
+
+def get_pr(row):
+    return (
+        row.get("nvd_PR")
+        or row.get("PR")
+        or row.get("cvss_PR")
+        or ""
+    ).strip().upper()
+
+
+def av_bucket(av):
+    # User requested N and A in the same file.
+    if av in {"N", "A"}:
+        return "network"
+
+    return "local"
+
+
+def pr_bucket(pr):
+    if pr == "N":
+        return "nopriv"
+
+    # L, H, empty, unknown all go to privileged bucket.
+    return "priv"
+
+
+def main(name, paths):
+    output_files = {
+        bucket: open(f"{name}-{bucket[0]}-{bucket[1]}.csv", "w", encoding="utf-8", newline="")
+        for bucket in OUTPUT_BUCKETS
+    }
+
+    writers = {}
+    output_header = None
+
+    try:
+        for path in paths:
+            metadata = parse_metadata(path)
+
+            with open(path, "r", encoding="utf-8", newline="") as f:
+                reader = csv.DictReader(f)
+
+                if reader.fieldnames is None:
+                    continue
+
+                if output_header is None:
+                    original_header = reader.fieldnames
+                    output_header = META_COLUMNS + original_header
+
+                    for bucket, fh in output_files.items():
+                        writer = csv.DictWriter(
+                            fh,
+                            fieldnames=output_header,
+                            extrasaction="ignore",
+                        )
+                        writer.writeheader()
+                        writers[bucket] = writer
+
+                for row in reader:
+                    av = get_av(row)
+                    pr = get_pr(row)
+
+                    bucket = (av_bucket(av), pr_bucket(pr))
+
+                    output_row = {}
+                    output_row.update(metadata)
+                    output_row.update(row)
+
+                    writers[bucket].writerow(output_row)
+
+    finally:
+        for fh in output_files.values():
+            fh.close()
+
+
+if __name__ == "__main__":
+    main(sys.argv[1], sys.argv[2:])
+PY
diff --git a/query.py b/query.py
new file mode 100755
index 0000000..6680815
--- /dev/null
+++ b/query.py
@@ -0,0 +1,848 @@
+#!/usr/bin/env python3
+
+import argparse
+import csv
+import json
+import os
+import re
+import time
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Set
+
+import requests
+
+
+OSV_BATCH_URL = "https://api.osv.dev/v1/querybatch"
+NVD_CVE_API = "https://services.nvd.nist.gov/rest/json/cves/2.0"
+
+UBUNTU_CVE_RE = re.compile(r"^UBUNTU-(CVE-\d{4}-\d{4,})$")
+CVE_RE = re.compile(r"^(CVE-\d{4}-\d{4,})$")
+
+
+# ----------------------------
+# Package parsing
+# ----------------------------
+
+def strip_debian_arch_suffix(pkg_name: str) -> str:
+    """
+    Convert names like:
+      libyaml-0-2:arm64
+
+    into:
+      libyaml-0-2
+    """
+    if ":" not in pkg_name:
+        return pkg_name
+
+    name, suffix = pkg_name.rsplit(":", 1)
+
+    known_arches = {
+        "amd64",
+        "arm64",
+        "armhf",
+        "i386",
+        "ppc64el",
+        "s390x",
+        "riscv64",
+        "all",
+    }
+
+    if suffix in known_arches:
+        return name
+
+    return pkg_name
+
+
+def parse_package_file(path: Path) -> List[Dict[str, str]]:
+    """
+    Parses package files with rows like:
+
+      package<TAB>version<TAB>arch
+
+    Example:
+
+      libyaml-0-2:arm64    0.2.2-1build2    arm64
+      login                1:4.8.1-2ubuntu2.2 arm64
+    """
+    packages = []
+
+    with path.open("r", encoding="utf-8") as f:
+        for line_no, line in enumerate(f, start=1):
+            line = line.strip()
+
+            if not line or line.startswith("#"):
+                continue
+
+            parts = line.split()
+
+            if len(parts) < 2:
+                print(f"[WARN] Skipping malformed line {line_no}: {line}")
+                continue
+
+            raw_name = parts[0]
+            version = parts[1]
+            arch = parts[2] if len(parts) >= 3 else ""
+
+            packages.append({
+                "package_raw": raw_name,
+                "package": strip_debian_arch_suffix(raw_name),
+                "version": version,
+                "arch": arch,
+            })
+
+    return packages
+
+
+def chunked(items: List[Any], size: int):
+    for i in range(0, len(items), size):
+        yield i, items[i:i + size]
+
+
+# ----------------------------
+# OSV logic
+# ----------------------------
+
+def extract_fixed_versions(affected: List[Dict[str, Any]]) -> List[str]:
+    fixed = set()
+
+    for affected_entry in affected:
+        for range_entry in affected_entry.get("ranges", []):
+            for event in range_entry.get("events", []):
+                if "fixed" in event:
+                    fixed.add(event["fixed"])
+
+    return sorted(fixed)
+
+
+def derive_cve_from_osv(osv_id: str, aliases: List[str]) -> str:
+    """
+    Prefer actual alias CVE.
+
+    If OSV ID looks like:
+      UBUNTU-CVE-2025-1149
+
+    derive:
+      CVE-2025-1149
+    """
+    for alias in aliases:
+        if CVE_RE.match(alias):
+            return alias
+
+    match = UBUNTU_CVE_RE.match(osv_id)
+    if match:
+        return match.group(1)
+
+    return ""
+
+
+def ubuntu_cve_url(cve_id: str) -> str:
+    if cve_id.startswith("CVE-"):
+        return f"https://ubuntu.com/security/{cve_id}"
+    return ""
+
+
+def query_osv(
+    packages: List[Dict[str, str]],
+    ecosystem: str,
+    batch_size: int,
+    sleep_seconds: float,
+) -> List[Dict[str, Any]]:
+    rows = []
+
+    for start_idx, batch in chunked(packages, batch_size):
+        print(
+            f"[INFO] Querying OSV packages "
+            f"{start_idx + 1}-{start_idx + len(batch)} of {len(packages)}"
+        )
+
+        queries = [
+            {
+                "package": {
+                    "name": pkg["package"],
+                    "ecosystem": ecosystem,
+                },
+                "version": pkg["version"],
+            }
+            for pkg in batch
+        ]
+
+        response = requests.post(
+            OSV_BATCH_URL,
+            json={"queries": queries},
+            timeout=90,
+        )
+        response.raise_for_status()
+
+        results = response.json().get("results", [])
+
+        if len(results) != len(batch):
+            print(
+                f"[WARN] OSV returned {len(results)} results "
+                f"for {len(batch)} queries"
+            )
+
+        for pkg, result in zip(batch, results):
+            for vuln in result.get("vulns", []):
+                osv_id = vuln.get("id", "")
+                aliases = vuln.get("aliases", [])
+                cves = [a for a in aliases if a.startswith("CVE-")]
+
+                derived_cve = derive_cve_from_osv(osv_id, aliases)
+
+                if not cves and derived_cve:
+                    cves = [derived_cve]
+
+                severity_entries = vuln.get("severity", [])
+                osv_severity = "; ".join(
+                    f"{s.get('type', '')}:{s.get('score', '')}"
+                    for s in severity_entries
+                )
+
+                fixed_versions = extract_fixed_versions(vuln.get("affected", []))
+
+                rows.append({
+                    "package_raw": pkg["package_raw"],
+                    "package": pkg["package"],
+                    "version": pkg["version"],
+                    "arch": pkg["arch"],
+                    "ecosystem": ecosystem,
+
+                    "osv_id": osv_id,
+                    "cves": ",".join(cves),
+                    "derived_cve": derived_cve,
+                    "osv_summary": vuln.get("summary", ""),
+                    "osv_details": vuln.get("details", ""),
+                    "osv_severity": osv_severity,
+                    "osv_published": vuln.get("published", ""),
+                    "osv_modified": vuln.get("modified", ""),
+                    "osv_fixed_versions": ",".join(fixed_versions),
+                    "ubuntu_cve_url": ubuntu_cve_url(derived_cve),
+                    "osv_references": ";".join(
+                        ref.get("url", "")
+                        for ref in vuln.get("references", [])
+                        if ref.get("url")
+                    ),
+                })
+
+        time.sleep(sleep_seconds)
+
+    return rows
+
+
+# ----------------------------
+# NVD parsing
+# ----------------------------
+
+def parse_cvss_vector(vector: str) -> Dict[str, str]:
+    """
+    Parse vector strings like:
+
+      CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H
+
+    into:
+
+      {
+        "AV": "N",
+        "AC": "L",
+        "PR": "N",
+        ...
+      }
+    """
+    fields = {}
+
+    if not vector:
+        return fields
+
+    for part in vector.split("/"):
+        if ":" not in part:
+            continue
+
+        key, value = part.split(":", 1)
+
+        if key == "CVSS":
+            continue
+
+        fields[key] = value
+
+    return fields
+
+
+def choose_best_cvss(metrics: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Prefer newer CVSS versions.
+
+    NVD may contain:
+      metrics.cvssMetricV40
+      metrics.cvssMetricV31
+      metrics.cvssMetricV30
+      metrics.cvssMetricV2
+    """
+    priority = [
+        ("cvssMetricV40", "4.0"),
+        ("cvssMetricV31", "3.1"),
+        ("cvssMetricV30", "3.0"),
+        ("cvssMetricV2", "2.0"),
+    ]
+
+    for metric_key, version in priority:
+        entries = metrics.get(metric_key, [])
+
+        if not entries:
+            continue
+
+        selected = None
+
+        for entry in entries:
+            if entry.get("source") == "nvd@nist.gov":
+                selected = entry
+                break
+
+        if selected is None:
+            selected = entries[0]
+
+        cvss_data = selected.get("cvssData", {})
+
+        return {
+            "nvd_cvss_version": version,
+            "nvd_vector": cvss_data.get("vectorString", ""),
+            "nvd_base_score": cvss_data.get("baseScore", ""),
+            "nvd_severity": (
+                cvss_data.get("baseSeverity")
+                or selected.get("baseSeverity")
+                or ""
+            ),
+            "nvd_metric_source": selected.get("source", ""),
+            "nvd_metric_type": selected.get("type", ""),
+        }
+
+    return empty_nvd_cvss_fields()
+
+
+def empty_nvd_cvss_fields() -> Dict[str, Any]:
+    return {
+        "nvd_cvss_version": "",
+        "nvd_vector": "",
+        "nvd_base_score": "",
+        "nvd_severity": "",
+        "nvd_metric_source": "",
+        "nvd_metric_type": "",
+    }
+
+
+def extract_english_description(cve_obj: Dict[str, Any]) -> str:
+    for desc in cve_obj.get("descriptions", []):
+        if desc.get("lang") == "en":
+            return desc.get("value", "")
+
+    return ""
+
+
+def parse_nvd_vulnerability(vuln: Dict[str, Any]) -> Dict[str, Any]:
+    cve_obj = vuln.get("cve", {})
+    cve_id = cve_obj.get("id", "")
+
+    cvss = choose_best_cvss(cve_obj.get("metrics", {}))
+    vector_fields = parse_cvss_vector(cvss.get("nvd_vector", ""))
+
+    return {
+        "cve_id": cve_id,
+        "nvd_found": "yes",
+        "nvd_error": "",
+        "nvd_description": extract_english_description(cve_obj),
+        "nvd_published": cve_obj.get("published", ""),
+        "nvd_last_modified": cve_obj.get("lastModified", ""),
+        "nvd_vuln_status": cve_obj.get("vulnStatus", ""),
+        **cvss,
+        "nvd_AV": vector_fields.get("AV", ""),
+        "nvd_PR": vector_fields.get("PR", ""),
+    }
+
+
+def empty_nvd_record(error: str = "") -> Dict[str, Any]:
+    return {
+        "nvd_found": "no",
+        "nvd_error": error,
+        "nvd_description": "",
+        "nvd_published": "",
+        "nvd_last_modified": "",
+        "nvd_vuln_status": "",
+        "nvd_cvss_version": "",
+        "nvd_vector": "",
+        "nvd_base_score": "",
+        "nvd_severity": "",
+        "nvd_metric_source": "",
+        "nvd_metric_type": "",
+        "nvd_AV": "",
+        "nvd_PR": "",
+    }
+
+
+# ----------------------------
+# NVD API
+# ----------------------------
+
+def cve_year(cve_id: str) -> str:
+    """
+    CVE-2025-1149 -> 2025
+    """
+    parts = cve_id.split("-")
+    if len(parts) >= 3:
+        return parts[1]
+    return ""
+
+
+def nvd_headers(api_key: Optional[str]) -> Dict[str, str]:
+    if api_key:
+        return {"apiKey": api_key}
+    return {}
+
+
+def fetch_nvd_cve_one_by_one(
+    cve_id: str,
+    api_key: Optional[str],
+    sleep_seconds: float,
+) -> Dict[str, Any]:
+    response = requests.get(
+        NVD_CVE_API,
+        params={"cveId": cve_id},
+        headers=nvd_headers(api_key),
+        timeout=90,
+    )
+
+    if response.status_code == 429:
+        raise RuntimeError(
+            "NVD rate limit hit. Use NVD_API_KEY or increase --nvd-sleep."
+        )
+
+    response.raise_for_status()
+    time.sleep(sleep_seconds)
+
+    data = response.json()
+    vulns = data.get("vulnerabilities", [])
+
+    if not vulns:
+        return empty_nvd_record("CVE not found in NVD")
+
+    return parse_nvd_vulnerability(vulns[0])
+
+
+def fetch_nvd_year_page(
+    year: str,
+    start_index: int,
+    results_per_page: int,
+    api_key: Optional[str],
+    sleep_seconds: float,
+) -> Dict[str, Any]:
+    params = {
+        "pubStartDate": f"{year}-01-01T00:00:00.000",
+        "pubEndDate": f"{year}-12-31T23:59:59.999",
+        "startIndex": start_index,
+        "resultsPerPage": results_per_page,
+    }
+
+    response = requests.get(
+        NVD_CVE_API,
+        params=params,
+        headers=nvd_headers(api_key),
+        timeout=120,
+    )
+
+    if response.status_code == 429:
+        raise RuntimeError(
+            "NVD rate limit hit. Use NVD_API_KEY or increase --nvd-sleep."
+        )
+
+    response.raise_for_status()
+    time.sleep(sleep_seconds)
+
+    return response.json()
+
+
+def fetch_nvd_by_years(
+    needed_cves: List[str],
+    api_key: Optional[str],
+    sleep_seconds: float,
+    results_per_page: int,
+) -> Dict[str, Dict[str, Any]]:
+    """
+    Batch method.
+
+    It fetches all NVD CVEs published in the years represented by the CVE IDs,
+    then keeps only the CVEs we need.
+
+    Example:
+      needed CVEs: CVE-2025-1149, CVE-2025-1150, CVE-2022-27943
+
+    Fetch:
+      all NVD CVEs published in 2025
+      all NVD CVEs published in 2022
+
+    This is much faster than one API call per CVE.
+    """
+    needed: Set[str] = set(needed_cves)
+    years = sorted({cve_year(cve) for cve in needed if cve_year(cve)})
+
+    nvd_map: Dict[str, Dict[str, Any]] = {}
+
+    print(f"[INFO] Fetching NVD in year batches: {', '.join(years)}")
+
+    for year in years:
+        start_index = 0
+
+        while True:
+            print(
+                f"[INFO] NVD batch year={year}, "
+                f"startIndex={start_index}, resultsPerPage={results_per_page}"
+            )
+
+            data = fetch_nvd_year_page(
+                year=year,
+                start_index=start_index,
+                results_per_page=results_per_page,
+                api_key=api_key,
+                sleep_seconds=sleep_seconds,
+            )
+
+            total_results = data.get("totalResults", 0)
+            vulns = data.get("vulnerabilities", [])
+
+            for vuln in vulns:
+                parsed = parse_nvd_vulnerability(vuln)
+                cve_id = parsed.get("cve_id", "")
+
+                if cve_id in needed:
+                    nvd_map[cve_id] = parsed
+
+            start_index += results_per_page
+
+            if start_index >= total_results:
+                break
+
+    return nvd_map
+
+
+# ----------------------------
+# Cache
+# ----------------------------
+
+def load_cache(path: Path) -> Dict[str, Dict[str, Any]]:
+    if path.exists():
+        with path.open("r", encoding="utf-8") as f:
+            return json.load(f)
+
+    return {}
+
+
+def save_cache(path: Path, cache: Dict[str, Dict[str, Any]]) -> None:
+    with path.open("w", encoding="utf-8") as f:
+        json.dump(cache, f, indent=2, ensure_ascii=False)
+
+
+# ----------------------------
+# Enrichment
+# ----------------------------
+
+def enrich_with_nvd(
+    rows: List[Dict[str, Any]],
+    cache_path: Path,
+    nvd_sleep_seconds: float,
+    results_per_page: int,
+    use_batch: bool,
+    fallback_one_by_one: bool,
+) -> List[Dict[str, Any]]:
+    api_key = "d753aa27-a3ae-4566-b52b-b3128208db21"
+    cache = load_cache(cache_path)
+
+    unique_cves = sorted({
+        row.get("derived_cve", "")
+        for row in rows
+        if row.get("derived_cve", "")
+    })
+
+    missing_cves = [
+        cve for cve in unique_cves
+        if cve not in cache
+    ]
+
+    print(f"[INFO] Unique CVEs to enrich with NVD: {len(unique_cves)}")
+    print(f"[INFO] Missing from local NVD cache: {len(missing_cves)}")
+    print(f"[INFO] NVD API key present: {'yes' if api_key else 'no'}")
+    print(f"[INFO] NVD mode: {'batch by year' if use_batch else 'one by one'}")
+
+    if missing_cves:
+        if use_batch:
+            try:
+                batch_results = fetch_nvd_by_years(
+                    needed_cves=missing_cves,
+                    api_key=api_key,
+                    sleep_seconds=nvd_sleep_seconds,
+                    results_per_page=results_per_page,
+                )
+
+                cache.update(batch_results)
+                save_cache(cache_path, cache)
+
+            except Exception as e:
+                print(f"[WARN] NVD batch fetch failed: {e}")
+
+        still_missing = [
+            cve for cve in missing_cves
+            if cve not in cache
+        ]
+
+        if still_missing and fallback_one_by_one:
+            print(
+                f"[INFO] Falling back to one-by-one NVD lookup for "
+                f"{len(still_missing)} CVEs"
+            )
+
+            for idx, cve_id in enumerate(still_missing, start=1):
+                print(
+                    f"[INFO] NVD fallback {idx}/{len(still_missing)}: {cve_id}"
+                )
+
+                try:
+                    cache[cve_id] = fetch_nvd_cve_one_by_one(
+                        cve_id=cve_id,
+                        api_key=api_key,
+                        sleep_seconds=nvd_sleep_seconds,
+                    )
+                except Exception as e:
+                    cache[cve_id] = empty_nvd_record(str(e))
+
+                save_cache(cache_path, cache)
+
+        elif still_missing:
+            for cve_id in still_missing:
+                cache[cve_id] = empty_nvd_record(
+                    "Not found by batch method; fallback disabled"
+                )
+
+            save_cache(cache_path, cache)
+
+    enriched_rows = []
+
+    for row in rows:
+        cve_id = row.get("derived_cve", "")
+        nvd_data = cache.get(cve_id, {}) if cve_id else {}
+
+        new_row = dict(row)
+
+        for key in [
+            "nvd_found",
+            "nvd_error",
+            "nvd_severity",
+            "nvd_base_score",
+            "nvd_cvss_version",
+            "nvd_vector",
+            "nvd_AV",
+            "nvd_PR",
+            "nvd_metric_source",
+            "nvd_metric_type",
+            "nvd_vuln_status",
+            "nvd_published",
+            "nvd_last_modified",
+            "nvd_description",
+        ]:
+            new_row[key] = nvd_data.get(key, "")
+
+        enriched_rows.append(new_row)
+
+    return enriched_rows
+
+
+# ----------------------------
+# Output
+# ----------------------------
+
+def write_csv(rows: List[Dict[str, Any]], path: Path) -> None:
+    fieldnames = [
+        "package_raw",
+        "package",
+        "version",
+        "arch",
+        "ecosystem",
+
+        "osv_id",
+        "cves",
+        "derived_cve",
+        "osv_summary",
+        "osv_severity",
+        "osv_published",
+        "osv_modified",
+        "osv_fixed_versions",
+        "ubuntu_cve_url",
+
+        "nvd_found",
+        "nvd_error",
+        "nvd_severity",
+        "nvd_base_score",
+        "nvd_cvss_version",
+        "nvd_vector",
+        "nvd_AV",
+        "nvd_PR",
+        "nvd_metric_source",
+        "nvd_metric_type",
+        "nvd_vuln_status",
+        "nvd_published",
+        "nvd_last_modified",
+        "nvd_description",
+
+        "osv_references",
+        "osv_details",
+    ]
+
+    with path.open("w", encoding="utf-8", newline="") as f:
+        writer = csv.DictWriter(
+            f,
+            fieldnames=fieldnames,
+            extrasaction="ignore",
+        )
+        writer.writeheader()
+
+        for row in rows:
+            writer.writerow(row)
+
+
+def write_json(rows: List[Dict[str, Any]], path: Path) -> None:
+    with path.open("w", encoding="utf-8") as f:
+        json.dump(rows, f, indent=2, ensure_ascii=False)
+
+
+# ----------------------------
+# Main
+# ----------------------------
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Query OSV for vulnerable Ubuntu packages, then enrich CVEs "
+            "with NVD severity, CVSS vector, AV, and PR."
+        )
+    )
+
+    parser.add_argument(
+        "input",
+        type=Path,
+        help="Input dpkg package list TSV",
+    )
+
+    parser.add_argument(
+        "--ecosystem",
+        default="Ubuntu:22.04:LTS",
+        help="OSV ecosystem, e.g. Ubuntu:22.04:LTS",
+    )
+
+    parser.add_argument(
+        "--csv",
+        type=Path,
+        default=Path("osv-nvd-results.csv"),
+        help="Output CSV path",
+    )
+
+    parser.add_argument(
+        "--json",
+        type=Path,
+        default=Path("osv-nvd-results.json"),
+        help="Output JSON path",
+    )
+
+    parser.add_argument(
+        "--nvd-cache",
+        type=Path,
+        default=Path("nvd-cache.json"),
+        help="Local NVD cache file path",
+    )
+
+    parser.add_argument(
+        "--osv-batch-size",
+        type=int,
+        default=1000,
+        help="Number of packages per OSV batch query",
+    )
+
+    parser.add_argument(
+        "--osv-sleep",
+        type=float,
+        default=0.2,
+        help="Seconds to sleep between OSV batch requests",
+    )
+
+    parser.add_argument(
+        "--nvd-sleep",
+        type=float,
+        default=None,
+        help=(
+            "Seconds to sleep between NVD requests. "
+            "Default: 6.0 without NVD_API_KEY, 0.7 with NVD_API_KEY."
+        ),
+    )
+
+    parser.add_argument(
+        "--nvd-results-per-page",
+        type=int,
+        default=2000,
+        help="NVD resultsPerPage for year-based batch fetching",
+    )
+
+    parser.add_argument(
+        "--skip-nvd",
+        action="store_true",
+        help="Only query OSV; do not enrich with NVD",
+    )
+
+    parser.add_argument(
+        "--no-nvd-batch",
+        action="store_true",
+        help="Disable year-based NVD batch fetching",
+    )
+
+    parser.add_argument(
+        "--no-nvd-fallback",
+        action="store_true",
+        help="Disable one-by-one fallback for CVEs missed by NVD batch fetch",
+    )
+
+    args = parser.parse_args()
+
+    if args.nvd_sleep is None:
+        nvd_sleep_seconds = 0.7 if os.environ.get("NVD_API_KEY") else 6.0
+    else:
+        nvd_sleep_seconds = args.nvd_sleep
+
+    packages = parse_package_file(args.input)
+
+    print(f"[INFO] Loaded packages: {len(packages)}")
+    print(f"[INFO] OSV ecosystem: {args.ecosystem}")
+
+    rows = query_osv(
+        packages=packages,
+        ecosystem=args.ecosystem,
+        batch_size=args.osv_batch_size,
+        sleep_seconds=args.osv_sleep,
+    )
+
+    print(f"[INFO] OSV vulnerability rows: {len(rows)}")
+
+    if not args.skip_nvd:
+        rows = enrich_with_nvd(
+            rows=rows,
+            cache_path=args.nvd_cache,
+            nvd_sleep_seconds=nvd_sleep_seconds,
+            results_per_page=args.nvd_results_per_page,
+            use_batch=not args.no_nvd_batch,
+            fallback_one_by_one=not args.no_nvd_fallback,
+        )
+
+    write_csv(rows, args.csv)
+    write_json(rows, args.json)
+
+    print(f"[INFO] Wrote CSV:  {args.csv}")
+    print(f"[INFO] Wrote JSON: {args.json}")
+
+    if not args.skip_nvd:
+        print(f"[INFO] Wrote NVD cache: {args.nvd_cache}")
+
+
+if __name__ == "__main__":
+    main()