cve/scripts/update_cves.py

#!/usr/bin/env python3
"""Discover CVE proof-of-concept links and update the repository."""

from __future__ import annotations

import argparse
import http.client
import json
import os
import re
import sys
import time
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Collection, Iterable, Iterator
from urllib import error, request
from urllib.parse import quote, urlparse

ROOT = Path(__file__).resolve().parents[1]
GITHUB_LIST = ROOT / "github.txt"
REFERENCE_LIST = ROOT / "references.txt"
BLACKLIST_FILE = ROOT / "blacklist.txt"
STATE_FILE = ROOT / ".github" / "cve_sync_state.json"

GITHUB_GRAPHQL_URL = "https://api.github.com/graphql"
CVE_API_URL = "https://cveawg.mitre.org/api/cve/{cve_id}"
CVELIST_DELTA_LOG_URL = (
    "https://raw.githubusercontent.com/CVEProject/cvelistV5/main/cves/deltaLog.json"
)

CVE_RE = re.compile(r"\bCVE[-_ ](?P<year>\d{4})[-_ ](?P<number>\d{4,})\b", re.IGNORECASE)
# A repository that calls itself an exploit in Chinese is still an exploit;
# an English-only vocabulary quietly reads every one of them as "not a PoC".
# CJK terms sit outside the \b group because CJK has no word boundaries.
POC_RE = re.compile(
    r"\b(?:poc|exploit(?:ation)?|proof[ _-]*of[ _-]*concept|"
    r"repro(?:ducer|duction|duce)?|payload|"
    r"prueba[ _-]*de[ _-]*concepto|explotaci(?:on|ón)|"
    r"explora(?:cao|ção)|preuve[ _-]*de[ _-]*concept)\b"
    r"|漏洞利用|漏洞复现|复现|利用代码|利用脚本|攻击脚本|驗證程式|検証コード"
    r"|эксплойт|익스플로잇",
    re.IGNORECASE,
)
CODE_FILE_RE = re.compile(
    r"\.(?:c|cc|cpp|cs|go|java|js|mjs|nse|php|pl|ps1|py|rb|rs|sh|ts)$",
    re.IGNORECASE,
)
NON_POC_RE = re.compile(
    r"(?:\b(?:advisory|aggregat(?:ion|or)|analysis|ansible|audits?|block(?:er|ing)?|blog|"
    r"bookmark|challenge|cheat[ _-]*sheets?|check(?:s|er|ing)?|collection|corpus|course|ctf|"
    r"dataset|defen(?:ce|se|sive)|detect(?:ion|or)?|dfir|epss|exam(?:en)?|feed|finder|framework|"
    r"github[ _-]*stars|hardening|hub|investigation|"
    r"honeypot|intelligence|ioc|lab|list|lookup|mitigat(?:e|ing|ion|or)|model(?:er|ing)|"
    r"monitor(?:ing)?|notes?|oscp|papers?|patch(?:es|ed|ing)?|pentest|portfolio|prediction|"
    r"pre[ _-]*checks?|prevent(?:ion|ive)?|profile|protect(?:ion|ive)|"
    r"remediat(?:e|ion)|reports?|"
    r"resources?|rules?|scan(?:s|ner|ning)?|shield|signature|suggester|templates?|toolkit|"
    r"training|tutorial|validat(?:e|ion|or)|verif(?:y|ier|ication)|walkthrough|wazuh|"
    r"workarounds?|fix(?:es|ed|ing)?|"
    r"write[ _-]*ups?)\b|searchpoc|watchdog)",
    re.IGNORECASE,
)
SECTION_RE_TEMPLATE = r"({header}\s*\n)(.*?)(?=\n#### |\n### |\Z)"
USER_AGENT = "0xMarcio-cve-poc-sync"

GITHUB_SEARCH_QUERY = """
query($query: String!, $cursor: String) {
  search(type: REPOSITORY, query: $query, first: 100, after: $cursor) {
    repositoryCount
    pageInfo { hasNextPage endCursor }
    nodes {
      ... on Repository {
        nameWithOwner
        url
        description
        isArchived
        isFork
        pushedAt
        repositoryTopics(first: 20) { nodes { topic { name } } }
      }
    }
  }
  rateLimit { cost remaining resetAt }
}
"""


@dataclass
class CVEDetails:
    description: str
    products: list[str]
    versions: list[str]
    vulnerabilities: list[str]


@dataclass
class SyncStats:
    created: list[str] = field(default_factory=list)
    updated: list[str] = field(default_factory=list)
    skipped: list[str] = field(default_factory=list)


def stable_unique(values: Iterable[str]) -> list[str]:
    seen: set[str] = set()
    result: list[str] = []
    for value in values:
        if value and value not in seen:
            seen.add(value)
            result.append(value)
    return result


def poc_link_key(url: str) -> str:
    parsed = urlparse(url)
    if (parsed.hostname or "").lower() not in {"github.com", "www.github.com"}:
        return url.rstrip("/")
    parts = [part for part in parsed.path.split("/") if part]
    if len(parts) < 2:
        return url
    repo = f"{parts[0]}/{parts[1]}".lower()
    suffix = "/" + "/".join(parts[2:]) if len(parts) > 2 else ""
    return f"github:{repo}{suffix}?{parsed.query}#{parsed.fragment}"


def stable_unique_poc_links(values: Iterable[str]) -> list[str]:
    seen: set[str] = set()
    result: list[str] = []
    for value in values:
        key = poc_link_key(value)
        if value and key not in seen:
            seen.add(key)
            result.append(value)
    return result


def is_valid_cve(cve_id: str) -> bool:
    match = re.fullmatch(r"CVE-(\d{4})-(\d{4,})", cve_id, re.IGNORECASE)
    if not match:
        return False
    return 1999 <= int(match.group(1)) <= 2100


def extract_cves(text: str | None, year: int | None = None) -> set[str]:
    result: set[str] = set()
    for match in CVE_RE.finditer(text or ""):
        match_year = int(match.group("year"))
        if year is not None and match_year != year:
            continue
        cve_id = f"CVE-{match_year}-{match.group('number')}"
        if is_valid_cve(cve_id):
            result.add(cve_id)
    return result


def load_blacklist(path: Path = BLACKLIST_FILE) -> set[str]:
    if not path.exists():
        return set()
    return {
        line.strip().lower()
        for line in path.read_text(encoding="utf-8").splitlines()
        if line.strip() and not line.lstrip().startswith("#")
    }


def is_blacklisted_repo(full_name: str, blacklist: Collection[str]) -> bool:
    return full_name.lower() in blacklist


def normalise_identity(text: str) -> str:
    return re.sub(r"[_-]+", " ", text)


def github_repo_from_url(url: str) -> str:
    parsed = urlparse(url)
    if (parsed.hostname or "").lower() != "github.com":
        return ""
    parts = [part for part in parsed.path.split("/") if part]
    return "/".join(parts[:2]) if len(parts) >= 2 else ""


def github_url_is_profile_root(url: str) -> bool:
    parsed = urlparse(url)
    if (parsed.hostname or "").lower() != "github.com":
        return False
    parts = [part for part in parsed.path.split("/") if part]
    return len(parts) == 2 and parts[0].lower() == parts[1].lower()


def github_url_is_ambiguous_root(url: str, cve_id: str) -> bool:
    parsed = urlparse(url)
    if (parsed.hostname or "").lower() != "github.com":
        return False
    parts = [part for part in parsed.path.split("/") if part]
    if len(parts) != 2:
        return False
    identity = normalise_identity(f"{parts[0]} {parts[1]}")
    if not NON_POC_RE.search(identity):
        return False
    return not (
        cve_id in extract_cves(identity)
        and POC_RE.search(identity)
    )


def readme_text(repo: dict[str, Any]) -> str:
    parts: list[str] = []
    for key in ("readmeMd", "readmeUpper", "readmeRst", "readmeBare"):
        value = repo.get(key)
        if isinstance(value, dict) and isinstance(value.get("text"), str):
            parts.append(value["text"])
    return "\n".join(stable_unique(parts))


def topic_names(repo: dict[str, Any]) -> list[str]:
    topics = repo.get("repositoryTopics") or {}
    names: list[str] = []
    for node in topics.get("nodes") or []:
        name = ((node or {}).get("topic") or {}).get("name")
        if name:
            names.append(str(name))
    return names


def root_entry_names(repo: dict[str, Any]) -> list[str]:
    root = repo.get("root") or {}
    return [
        str(entry.get("name") or "")
        for entry in root.get("entries") or []
        if entry.get("name")
    ]


def root_has_poc_artifact(repo: dict[str, Any], cve_id: str) -> bool:
    year, number = cve_id.split("-")[1:]
    cve_pattern = re.compile(
        rf"\bCVE[-_ ]{re.escape(year)}[-_ ]{re.escape(number)}\b",
        re.IGNORECASE,
    )
    for name in root_entry_names(repo):
        if name.lower().startswith(("readme", "license", "security")):
            continue
        if cve_pattern.search(name) or POC_RE.search(normalise_identity(name)):
            return True
    return False


def root_has_code(repo: dict[str, Any]) -> bool:
    return any(CODE_FILE_RE.search(name) for name in root_entry_names(repo))


def truncated_ids(cve_ids: Collection[str]) -> set[str]:
    """Identifiers that a longer identifier from the same year extends."""
    numbers: dict[str, set[str]] = defaultdict(set)
    for cve_id in cve_ids:
        _, year, number = cve_id.split("-")
        numbers[year].add(number)
    return {
        f"CVE-{year}-{number}"
        for year, values in numbers.items()
        for number in values
        if any(other.startswith(number) for other in values - {number})
    }


def readme_has_poc_context(readme: str, cve_id: str, full_name: str = "") -> bool:
    year, number = cve_id.split("-")[1:]
    pattern = re.compile(rf"\bCVE[-_ ]{re.escape(year)}[-_ ]{re.escape(number)}\b", re.IGNORECASE)
    for match in pattern.finditer(readme):
        line_start = readme.rfind("\n", 0, match.start()) + 1
        line_end = readme.find("\n", match.end())
        line = readme[line_start : line_end if line_end >= 0 else len(readme)]
        linked_repos = {
            github_repo_from_url(url).lower()
            for url in re.findall(r"https://github\.com/[^\s)>]+", line, re.IGNORECASE)
        }
        linked_repos.discard("")
        if linked_repos and full_name.lower() not in linked_repos:
            continue
        start = max(0, match.start() - 350)
        end = min(len(readme), match.end() + 350)
        if POC_RE.search(readme[start:end]):
            return True
    return False


def qualifying_repo_cves(
    repo: dict[str, Any],
    year: int,
    blacklist: Collection[str],
) -> set[str]:
    full_name = str(repo.get("nameWithOwner") or "")
    if not full_name or repo.get("isFork"):
        return set()
    owner, repo_name = full_name.lower().split("/", 1)
    if owner == repo_name or repo_name.endswith(".github.io"):
        return set()
    if is_blacklisted_repo(full_name, blacklist):
        return set()

    description = str(repo.get("description") or "")
    topics = " ".join(topic_names(repo))
    readme = readme_text(repo)
    # Underscores are word characters, so a raw name like CVE-A_CVE-B matches
    # neither identifier; read the separator-normalised form as well.
    name_cves = extract_cves(full_name, year) | extract_cves(normalise_identity(full_name), year)
    description_cves = extract_cves(description, year)
    topic_cves = extract_cves(topics, year)
    readme_cves = extract_cves(readme, year)
    # A caption or description that drops a digit — "CVE-2023-3883" written for
    # CVE-2023-38831 — still parses as a valid identifier and would otherwise
    # enrol the repository against whatever unrelated advisory owns it. Trust
    # the repository name, which may deliberately carry both identifiers.
    truncated = truncated_ids(name_cves | description_cves | topic_cves | readme_cves) - name_cves
    description_cves -= truncated
    topic_cves -= truncated
    readme_cves -= truncated
    identity_cves = name_cves | description_cves | topic_cves
    identity_text = normalise_identity(f"{full_name} {description}")
    identity_has_poc = bool(POC_RE.search(identity_text))
    metadata_has_poc = identity_has_poc or bool(POC_RE.search(topics))
    identity_is_non_poc = bool(NON_POC_RE.search(identity_text))
    has_code = root_has_code(repo)
    if identity_is_non_poc and not (name_cves and identity_has_poc and has_code):
        return set()
    if len(readme_cves) > 5 and not identity_cves:
        readme_cves = set()
    accepted: set[str] = set()
    for cve_id in name_cves:
        has_context = readme_has_poc_context(readme, cve_id, full_name)
        has_artifact = root_has_poc_artifact(repo, cve_id)
        has_content = bool(readme.strip()) or has_artifact or has_code
        if (
            (not identity_is_non_poc or has_artifact or has_code)
            and (metadata_has_poc or has_context or has_artifact)
            and has_content
        ):
            accepted.add(cve_id)
    for cve_id in description_cves | topic_cves:
        has_context = readme_has_poc_context(readme, cve_id, full_name)
        if (
            (not identity_is_non_poc or cve_id in name_cves)
            and (metadata_has_poc or has_context)
            and (has_context or root_has_poc_artifact(repo, cve_id) or has_code)
        ):
            accepted.add(cve_id)
    for cve_id in readme_cves:
        has_artifact = root_has_poc_artifact(repo, cve_id)
        if identity_cves and cve_id not in identity_cves and not has_artifact:
            continue
        if (
            readme_has_poc_context(readme, cve_id, full_name)
            and (has_artifact or has_code or identity_has_poc)
        ):
            accepted.add(cve_id)
    return accepted


def http_json(
    url: str,
    *,
    headers: dict[str, str] | None = None,
    data: dict[str, Any] | None = None,
    timeout: int = 45,
    allow_not_found: bool = False,
) -> Any:
    request_headers = {"Accept": "application/json", "User-Agent": USER_AGENT}
    request_headers.update(headers or {})
    body = None
    method = "GET"
    if data is not None:
        body = json.dumps(data).encode("utf-8")
        request_headers["Content-Type"] = "application/json"
        method = "POST"

    for attempt in range(3):
        req = request.Request(url, data=body, headers=request_headers, method=method)
        try:
            with request.urlopen(req, timeout=timeout) as response:
                return json.load(response)
        except error.HTTPError as exc:
            if allow_not_found and exc.code == 404:
                return None
            if exc.code not in {429, 500, 502, 503, 504} or attempt == 2:
                raise
        except (
            error.URLError,
            TimeoutError,
            json.JSONDecodeError,
            http.client.HTTPException,
        ):
            if attempt == 2:
                raise
        time.sleep(2 ** attempt)
    raise RuntimeError(f"Unable to fetch {url}")


class GitHubClient:
    def __init__(self, token: str) -> None:
        if not token:
            raise ValueError("GITHUB_TOKEN is required for GitHub PoC discovery")
        self.headers = {"Authorization": f"Bearer {token}"}

    def search_page(self, query_text: str, cursor: str | None = None) -> dict[str, Any]:
        payload = http_json(
            GITHUB_GRAPHQL_URL,
            headers=self.headers,
            data={"query": GITHUB_SEARCH_QUERY, "variables": {"query": query_text, "cursor": cursor}},
        )
        data = payload.get("data") or {}
        search = data.get("search")
        if not search:
            messages = "; ".join(str(item.get("message")) for item in payload.get("errors") or [])
            raise RuntimeError(f"GitHub search failed: {messages or 'missing response data'}")
        rate = data.get("rateLimit") or {}
        if int(rate.get("remaining") or 0) < 5:
            raise RuntimeError(f"GitHub GraphQL quota is nearly exhausted; resets at {rate.get('resetAt')}")
        return search

    def fetch_readmes(
        self,
        full_names: list[str],
    ) -> tuple[set[str], dict[str, str], dict[str, list[dict[str, str]]]]:
        fields = """
          readmeMd: object(expression: \"HEAD:README.md\") { ... on Blob { text } }
          readmeUpper: object(expression: \"HEAD:README.MD\") { ... on Blob { text } }
          readmeRst: object(expression: \"HEAD:README.rst\") { ... on Blob { text } }
          readmeBare: object(expression: \"HEAD:README\") { ... on Blob { text } }
          root: object(expression: \"HEAD:\") {
            ... on Tree { entries { name type } }
          }
        """
        aliases: list[str] = []
        names: list[str] = []
        for full_name in full_names:
            if "/" not in full_name:
                continue
            owner, name = full_name.split("/", 1)
            index = len(names)
            aliases.append(
                f"repo{index}: repository(owner: {json.dumps(owner)}, name: {json.dumps(name)}) "
                f"{{ {fields} }}"
            )
            names.append(full_name)
        if not aliases:
            return set(), {}, {}
        query_text = "query { " + " ".join(aliases) + " rateLimit { remaining resetAt } }"
        payload = http_json(
            GITHUB_GRAPHQL_URL,
            headers=self.headers,
            data={"query": query_text},
        )
        data = payload.get("data") or {}
        rate = data.get("rateLimit") or {}
        if int(rate.get("remaining") or 0) < 5:
            raise RuntimeError(f"GitHub GraphQL quota is nearly exhausted; resets at {rate.get('resetAt')}")
        result: dict[str, str] = {}
        roots: dict[str, list[dict[str, str]]] = {}
        existing: set[str] = set()
        for index, full_name in enumerate(names):
            repo = data.get(f"repo{index}")
            if repo is None:
                continue
            existing.add(full_name)
            text = readme_text(repo)
            if text:
                result[full_name] = text
            entries = ((repo.get("root") or {}).get("entries") or [])
            roots[full_name] = [
                {"name": str(entry.get("name") or ""), "type": str(entry.get("type") or "")}
                for entry in entries
                if entry.get("name")
            ]
        return existing, result, roots


def split_date_range(start: date, end: date) -> tuple[tuple[date, date], tuple[date, date]]:
    midpoint = start + timedelta(days=(end - start).days // 2)
    return (start, midpoint), (midpoint + timedelta(days=1), end)


def build_search_query(search_terms: str, qualifier: str, start: date, end: date) -> str:
    return (
        f"{search_terms} fork:false archived:false "
        f"{qualifier}:{start.isoformat()}..{end.isoformat()}"
    )


def search_range(
    client: GitHubClient,
    search_terms: str,
    qualifier: str,
    start: date,
    end: date,
) -> Iterator[dict[str, Any]]:
    query_text = build_search_query(search_terms, qualifier, start, end)
    first_page = client.search_page(query_text)
    count = int(first_page.get("repositoryCount") or 0)

    if count > 1000:
        if start == end:
            raise RuntimeError(f"GitHub search returned more than 1000 repositories for {query_text}")
        left, right = split_date_range(start, end)
        yield from search_range(client, search_terms, qualifier, *left)
        yield from search_range(client, search_terms, qualifier, *right)
        return

    print(f"GitHub {qualifier} {start}..{end}: {count} repositories")
    page = first_page
    while True:
        for repo in page.get("nodes") or []:
            if repo:
                yield repo
        page_info = page.get("pageInfo") or {}
        if not page_info.get("hasNextPage"):
            break
        page = client.search_page(query_text, str(page_info.get("endCursor")))


def search_without_range(client: GitHubClient, search_terms: str) -> Iterator[dict[str, Any]]:
    query_text = f"{search_terms} fork:false archived:false"
    cursor: str | None = None
    while True:
        page = client.search_page(query_text, cursor)
        if int(page.get("repositoryCount") or 0) > 1000:
            raise RuntimeError(f"GitHub search returned more than 1000 repositories for {query_text}")
        for repo in page.get("nodes") or []:
            if repo:
                yield repo
        page_info = page.get("pageInfo") or {}
        if not page_info.get("hasNextPage"):
            break
        cursor = str(page_info.get("endCursor"))


def discover_github_pocs(
    token: str,
    *,
    years: Iterable[int],
    lookback_days: int,
    backfill: bool,
    cve_filter: set[str],
) -> dict[str, list[str]]:
    client = GitHubClient(token)
    blacklist = load_blacklist()
    discovered: dict[str, list[str]] = defaultdict(list)
    repositories: dict[str, dict[str, Any]] = {}
    target_years = list(years)
    if cve_filter:
        target_years = sorted({int(cve_id.split("-")[1]) for cve_id in cve_filter})

    def remember(repo: dict[str, Any]) -> None:
        full_name = str(repo.get("nameWithOwner") or "")
        if full_name and not is_blacklisted_repo(full_name, blacklist):
            repositories[full_name] = repo

    def collect(repo: dict[str, Any], year: int) -> None:
        url = str(repo.get("url") or "").rstrip("/")
        if not url.startswith("https://github.com/"):
            return
        for cve_id in qualifying_repo_cves(repo, year, blacklist):
            if cve_filter and cve_id not in cve_filter:
                continue
            if url not in discovered[cve_id]:
                discovered[cve_id].append(url)

    def search_terms(identifier: str) -> list[str]:
        return [
            f'"{identifier}" in:name,description',
            f'"{identifier}" poc in:readme',
            f'"{identifier}" exploit in:readme',
            f'"{identifier}" "proof of concept" in:readme',
            f'"{identifier}" reproducer in:readme',
        ]

    if cve_filter:
        for cve_id in sorted(cve_filter):
            for terms in search_terms(cve_id):
                for repo in search_without_range(client, terms):
                    remember(repo)
    else:
        end = datetime.now(timezone.utc).date()
        for year in target_years:
            if backfill:
                qualifier = "created"
                start = date(2008, 1, 1)
            else:
                qualifier = "pushed"
                start = end - timedelta(days=max(1, lookback_days))
            for terms in search_terms(f"CVE-{year}"):
                for repo in search_range(client, terms, qualifier, start, end):
                    remember(repo)

    names = sorted(repositories)
    batches = [names[offset : offset + 10] for offset in range(0, len(names), 10)]
    completed = 0
    existing_names: set[str] = set()
    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = {executor.submit(client.fetch_readmes, batch): batch for batch in batches}
        for future in as_completed(futures):
            batch = futures[future]
            try:
                existing, readmes, roots = future.result()
            except Exception as exc:
                print(f"Skipped README batch starting with {batch[0]}: {exc}", file=sys.stderr)
                existing = set(batch)
                readmes = {}
                roots = {}
            existing_names.update(existing)
            for full_name, text in readmes.items():
                repositories[full_name]["readmeMd"] = {"text": text}
            for full_name, entries in roots.items():
                repositories[full_name]["root"] = {"entries": entries}
            completed += len(batch)
            if completed % 500 < len(batch):
                print(f"Fetched README data for {completed} of {len(names)} repositories")

    removed = set(repositories) - existing_names
    for full_name in removed:
        del repositories[full_name]
    if removed:
        print(f"Discarded {len(removed)} repositories removed during discovery")

    for repo in repositories.values():
        for year in target_years:
            collect(repo, year)
    return discovered


def cna_container(record: dict[str, Any]) -> dict[str, Any]:
    return (record.get("containers") or {}).get("cna") or {}


def record_is_published(record: dict[str, Any]) -> bool:
    return str((record.get("cveMetadata") or {}).get("state") or "").upper() == "PUBLISHED"


def record_cve_id(record: dict[str, Any]) -> str:
    return str((record.get("cveMetadata") or {}).get("cveId") or "").upper()


def english_description(container: dict[str, Any]) -> str:
    descriptions = container.get("descriptions") or []
    for item in descriptions:
        if str(item.get("lang") or "").lower().startswith("en") and item.get("value"):
            return str(item["value"]).strip()
    for item in descriptions:
        if item.get("value"):
            return str(item["value"]).strip()
    return ""


def details_from_record(record: dict[str, Any]) -> CVEDetails | None:
    if not record_is_published(record):
        return None
    cna = cna_container(record)
    description = english_description(cna)
    if not description:
        return None

    products: list[str] = []
    versions: list[str] = []
    for affected in cna.get("affected") or []:
        product = str(affected.get("product") or "").strip()
        if product:
            products.append(product)
        for version in affected.get("versions") or []:
            status = str(version.get("status") or "affected").lower()
            value = str(version.get("version") or "").strip()
            if status == "affected" and value:
                versions.append(value)

    vulnerabilities: list[str] = []
    for problem in cna.get("problemTypes") or []:
        for item in problem.get("descriptions") or []:
            value = item.get("description") or item.get("cweId")
            if value:
                vulnerabilities.append(str(value).strip())

    return CVEDetails(
        description=description,
        products=stable_unique(products),
        versions=stable_unique(versions),
        vulnerabilities=stable_unique(vulnerabilities),
    )


def reference_is_poc(
    reference: dict[str, Any],
    blacklist: Collection[str],
    cve_id: str = "",
) -> bool:
    url = str(reference.get("url") or "").strip()
    if not url.startswith(("http://", "https://")):
        return False
    repo = github_repo_from_url(url)
    if repo and (
        is_blacklisted_repo(repo, blacklist)
        or github_url_is_profile_root(url)
        or (cve_id and github_url_is_ambiguous_root(url, cve_id))
    ):
        return False
    tags = [str(tag).lower() for tag in reference.get("tags") or []]
    evidence = " ".join([url, str(reference.get("name") or ""), *tags])
    if any("exploit" in tag for tag in tags):
        return True
    return bool(POC_RE.search(evidence))


def poc_references(
    record: dict[str, Any],
    blacklist: Collection[str] | None = None,
) -> list[str]:
    cve_id = record_cve_id(record)
    if not cve_id or not record_is_published(record):
        return []
    blacklist = load_blacklist() if blacklist is None else blacklist
    candidates = [cna_container(record)]
    references: list[str] = []
    for container in candidates:
        for item in container.get("references") or []:
            url = str(item.get("url") or "").strip()
            if reference_is_poc(item, blacklist, cve_id):
                references.append(url)
    return stable_unique_poc_links(references)


def reference_url_is_live(url: str, timeout: int = 15) -> bool:
    headers = {
        "Accept": "*/*",
        "Accept-Encoding": "identity",
        "User-Agent": USER_AGENT,
    }
    for method in ("HEAD", "GET"):
        request_headers = dict(headers)
        if method == "GET":
            request_headers["Range"] = "bytes=0-0"
        req = request.Request(url, headers=request_headers, method=method)
        try:
            with request.urlopen(req, timeout=timeout) as response:
                if method == "GET":
                    response.read(1)
                if int(response.status) < 400:
                    return True
        except error.HTTPError as exc:
            code = exc.code
            exc.close()
            if method == "GET" and code in {404, 410}:
                return False
        except (error.URLError, TimeoutError, OSError):
            pass
    return True


def filter_live_references(
    references: dict[str, list[str]],
    checker: Callable[[str], bool] | None = None,
) -> dict[str, list[str]]:
    checker = reference_url_is_live if checker is None else checker
    urls = stable_unique(url for values in references.values() for url in values)
    if not urls:
        return references
    dead: set[str] = set()
    with ThreadPoolExecutor(max_workers=min(8, len(urls))) as executor:
        futures = {executor.submit(checker, url): url for url in urls}
        for future in as_completed(futures):
            url = futures[future]
            try:
                if not future.result():
                    dead.add(url)
            except Exception:
                continue
    if dead:
        print(f"Discarded {len(dead)} references returning 404 or 410")
    return {
        cve_id: [url for url in urls_for_cve if url not in dead]
        for cve_id, urls_for_cve in references.items()
        if any(url not in dead for url in urls_for_cve)
    }


def read_record(path: Path) -> dict[str, Any] | None:
    try:
        with path.open("r", encoding="utf-8") as handle:
            return json.load(handle)
    except (OSError, json.JSONDecodeError):
        return None


def load_backfill_records(
    cve_dir: Path,
    year: int,
    cve_filter: set[str],
) -> tuple[dict[str, dict[str, Any]], dict[str, list[str]], dict[str, Path]]:
    records: dict[str, dict[str, Any]] = {}
    references: dict[str, list[str]] = {}
    paths: dict[str, Path] = {}
    blacklist = load_blacklist()

    for path in sorted(cve_dir.rglob("CVE-*.json")):
        cve_id = path.stem.upper()
        if not is_valid_cve(cve_id) or int(cve_id.split("-")[1]) != year:
            continue
        if cve_filter and cve_id not in cve_filter:
            continue
        paths[cve_id] = path
        record = read_record(path)
        if not record or not record_is_published(record):
            continue
        selected = poc_references(record, blacklist)
        if selected:
            records[cve_id] = record
            references[cve_id] = selected
    return records, references, paths


def load_state(path: Path = STATE_FILE) -> str | None:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return None
    value = data.get("cvelist_fetch_time")
    return str(value) if value else None


def delta_changes(last_fetch_time: str | None) -> tuple[dict[str, str], str]:
    log = http_json(CVELIST_DELTA_LOG_URL)
    if not isinstance(log, list) or not log:
        raise RuntimeError("CVE List V5 delta log is empty")
    newest = str(log[0].get("fetchTime") or "")
    selected: list[dict[str, Any]] = []
    for batch in log:
        fetch_time = str(batch.get("fetchTime") or "")
        if last_fetch_time and fetch_time <= last_fetch_time:
            break
        selected.append(batch)
        if not last_fetch_time:
            break

    changes: dict[str, str] = {}
    for batch in reversed(selected):
        for group in ("new", "updated"):
            for item in batch.get(group) or []:
                cve_id = str(item.get("cveId") or "").upper()
                url = str(item.get("githubLink") or "")
                if is_valid_cve(cve_id) and url:
                    changes[cve_id] = url
    print(f"CVE List V5: {len(selected)} delta batches, {len(changes)} changed CVEs")
    return changes, newest


def fetch_records(urls: dict[str, str], *, allow_not_found: bool) -> dict[str, dict[str, Any]]:
    records: dict[str, dict[str, Any]] = {}
    failures: list[str] = []

    def fetch(cve_id: str, url: str) -> tuple[str, dict[str, Any] | None]:
        result = http_json(url, allow_not_found=allow_not_found)
        return cve_id, result if isinstance(result, dict) else None

    with ThreadPoolExecutor(max_workers=12) as executor:
        futures = {executor.submit(fetch, cve_id, url): cve_id for cve_id, url in urls.items()}
        for future in as_completed(futures):
            cve_id = futures[future]
            try:
                fetched_cve, record = future.result()
            except Exception as exc:
                failures.append(f"{cve_id}: {exc}")
                continue
            if record:
                records[fetched_cve] = record

    if failures and not allow_not_found:
        raise RuntimeError("Failed to fetch CVE records: " + "; ".join(failures[:10]))
    if failures:
        print(f"Skipped {len(failures)} CVE API failures", file=sys.stderr)
    return records


def load_incremental_records(
    cve_filter: set[str],
) -> tuple[dict[str, dict[str, Any]], dict[str, list[str]], str]:
    changes, newest = delta_changes(load_state())
    if cve_filter:
        changes = {cve_id: url for cve_id, url in changes.items() if cve_id in cve_filter}
    records = fetch_records(changes, allow_not_found=False) if changes else {}
    blacklist = load_blacklist()
    references = {
        cve_id: selected
        for cve_id, record in records.items()
        if (selected := poc_references(record, blacklist))
    }
    return records, references, newest


def fetch_missing_records(
    cve_ids: Iterable[str],
    records: dict[str, dict[str, Any]],
) -> None:
    urls = {
        cve_id: CVE_API_URL.format(cve_id=cve_id)
        for cve_id in cve_ids
        if cve_id not in records
    }
    if urls:
        records.update(fetch_records(urls, allow_not_found=True))


def badge(label: str, message: str, color: str) -> str:
    return (
        "![](https://img.shields.io/static/v1?"
        f"label={quote(label, safe='')}&message={quote(message, safe='')}&color={color})"
    )


def build_markdown(
    cve_id: str,
    details: CVEDetails,
    github_links: list[str],
    references: list[str],
) -> str:
    description = "\n".join(
        line.expandtabs(4).rstrip() for line in details.description.strip().splitlines()
    )
    lines = [f"### [{cve_id}](https://cve.mitre.org/cgi-bin/cvename.cgi?name={cve_id})"]
    products = details.products or ["n/a"]
    versions = details.versions or ["n/a"]
    vulnerabilities = details.vulnerabilities or ["n/a"]
    version_color = "brightgreen" if details.versions else "blue"
    vuln_color = "brightgreen" if details.vulnerabilities else "blue"
    lines.extend(badge("Product", value, "blue") for value in products)
    lines.extend(badge("Version", value, version_color) for value in versions)
    lines.extend(badge("Vulnerability", value, vuln_color) for value in vulnerabilities)
    lines.extend(["", "### Description", "", description, "", "### POC", "", "#### Reference"])
    lines.extend((f"- {url}" for url in references) if references else ["No PoCs from references."])
    lines.extend(["", "#### Github"])
    lines.extend((f"- {url}" for url in github_links) if github_links else ["No PoCs found on GitHub currently."])
    lines.append("")
    return "\n".join(lines)


def parse_section_links(block: str) -> list[str]:
    links: list[str] = []
    for raw_line in block.splitlines():
        line = raw_line.strip()
        if not line or line.startswith("No PoCs"):
            continue
        if line.startswith("- "):
            line = line[2:].strip()
        if line.startswith(("http://", "https://")):
            links.append(line)
    return stable_unique(links)


def section_links(text: str, header: str) -> list[str]:
    pattern = re.compile(SECTION_RE_TEMPLATE.format(header=re.escape(header)), re.DOTALL)
    match = pattern.search(text)
    return parse_section_links(match.group(2)) if match else []


def merge_section(text: str, header: str, links: Iterable[str], empty_text: str) -> tuple[str, bool]:
    incoming = stable_unique_poc_links(links)
    if not incoming:
        return text, False
    pattern = re.compile(SECTION_RE_TEMPLATE.format(header=re.escape(header)), re.DOTALL)
    match = pattern.search(text)
    if not match:
        return text, False
    existing = parse_section_links(match.group(2))
    combined = stable_unique_poc_links([*existing, *incoming])
    if combined == existing:
        return text, False
    return replace_section(text, header, combined, empty_text)


def replace_section(
    text: str,
    header: str,
    links: Iterable[str],
    empty_text: str,
) -> tuple[str, bool]:
    values = stable_unique_poc_links(links)
    pattern = re.compile(SECTION_RE_TEMPLATE.format(header=re.escape(header)), re.DOTALL)
    match = pattern.search(text)
    if not match:
        return text, False
    newline = "\r\n" if "\r\n" in match.group(1) else "\n"
    replacement = newline.join(f"- {url}" for url in values) if values else empty_text
    body = match.group(2)
    trailing = body[len(body.rstrip()) :] or newline
    updated = text[: match.start(2)] + replacement + trailing + text[match.end(2) :]
    return updated, updated != text


def update_existing_markdown(
    path: Path,
    github_links: list[str],
    references: list[str],
    *,
    dry_run: bool,
) -> bool:
    with path.open("r", encoding="utf-8", newline="") as handle:
        original = handle.read()
    updated, reference_changed = merge_section(
        original,
        "#### Reference",
        references,
        "No PoCs from references.",
    )
    reference_keys = {
        poc_link_key(url)
        for url in section_links(updated, "#### Reference")
    }
    existing_github = section_links(updated, "#### Github")
    github_values = [
        url
        for url in stable_unique_poc_links([*existing_github, *github_links])
        if poc_link_key(url) not in reference_keys
    ]
    updated, github_changed = replace_section(
        updated,
        "#### Github",
        github_values,
        "No PoCs found on GitHub currently.",
    )
    if not (reference_changed or github_changed):
        return False
    if not updated.endswith(("\n", "\r")):
        updated += "\r\n" if "\r\n" in original else "\n"
    if not dry_run:
        with path.open("w", encoding="utf-8", newline="") as handle:
            handle.write(updated)
    return True


def sync_markdown(
    records: dict[str, dict[str, Any]],
    github: dict[str, list[str]],
    references: dict[str, list[str]],
    *,
    dry_run: bool,
) -> tuple[SyncStats, dict[str, list[str]], dict[str, list[str]]]:
    stats = SyncStats()
    accepted_github: dict[str, list[str]] = {}
    accepted_references: dict[str, list[str]] = {}
    cve_ids = sorted(set(github) | set(references))

    for cve_id in cve_ids:
        year = cve_id.split("-")[1]
        path = ROOT / year / f"{cve_id}.md"
        github_links = stable_unique(github.get(cve_id, []))
        reference_links = stable_unique(references.get(cve_id, []))
        record = records.get(cve_id)

        if record and not record_is_published(record):
            stats.skipped.append(f"{cve_id}: CVE is not published")
            continue

        if path.exists():
            if update_existing_markdown(
                path,
                github_links,
                reference_links,
                dry_run=dry_run,
            ):
                stats.updated.append(cve_id)
        else:
            details = details_from_record(record or {})
            if not details:
                stats.skipped.append(f"{cve_id}: published CVE metadata is unavailable")
                continue
            if not dry_run:
                path.parent.mkdir(parents=True, exist_ok=True)
                path.write_text(
                    build_markdown(cve_id, details, github_links, reference_links),
                    encoding="utf-8",
                )
            stats.created.append(cve_id)

        accepted_github[cve_id] = github_links
        accepted_references[cve_id] = reference_links

    return stats, accepted_github, accepted_references


def append_inventory(path: Path, mappings: dict[str, list[str]], *, dry_run: bool) -> int:
    existing = set(path.read_text(encoding="utf-8").splitlines()) if path.exists() else set()
    additions = sorted(
        f"{cve_id} - {url}"
        for cve_id, urls in mappings.items()
        for url in urls
        if url and f"{cve_id} - {url}" not in existing
    )
    if not additions or dry_run:
        return len(additions)
    prefix = ""
    if path.exists() and path.stat().st_size and not path.read_bytes().endswith(b"\n"):
        prefix = "\n"
    with path.open("a", encoding="utf-8") as handle:
        handle.write(prefix + "\n".join(additions) + "\n")
    return len(additions)


def write_state(fetch_time: str, *, dry_run: bool) -> None:
    if not fetch_time or dry_run:
        return
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(
        json.dumps({"cvelist_fetch_time": fetch_time}, indent=2) + "\n",
        encoding="utf-8",
    )


def newest_delta_time() -> str:
    log = http_json(CVELIST_DELTA_LOG_URL)
    if not isinstance(log, list) or not log:
        raise RuntimeError("CVE List V5 delta log is empty")
    return str(log[0].get("fetchTime") or "")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Discover and ingest CVE PoC links")
    parser.add_argument("--backfill-dir", type=Path, help="Local CVE List V5 year directory")
    parser.add_argument("--year", type=int, help="Year to backfill")
    parser.add_argument("--lookback-days", type=int, default=3, help="GitHub pushed-date overlap")
    parser.add_argument("--years", type=int, default=2, help="Number of recent years to scan")
    parser.add_argument("--cve", action="append", default=[], help="Process one CVE ID")
    parser.add_argument("--skip-github", action="store_true", help="Skip GitHub repository discovery")
    parser.add_argument("--skip-cvelist", action="store_true", help="Skip CVE List V5 references")
    parser.add_argument("--dry-run", action="store_true", help="Report changes without writing files")
    args = parser.parse_args()
    if bool(args.backfill_dir) != bool(args.year):
        parser.error("--backfill-dir and --year must be used together")
    if args.lookback_days < 1 or args.years < 1:
        parser.error("--lookback-days and --years must be positive")
    return args


def main() -> int:
    args = parse_args()
    cve_filter = {cve.upper() for cve in args.cve if is_valid_cve(cve.upper())}
    if args.cve and len(cve_filter) != len(args.cve):
        raise SystemExit("Every --cve value must be a valid CVE ID")

    backfill = bool(args.backfill_dir)
    records: dict[str, dict[str, Any]] = {}
    references: dict[str, list[str]] = {}
    record_paths: dict[str, Path] = {}
    checkpoint = ""

    if not args.skip_cvelist:
        if backfill:
            records, references, record_paths = load_backfill_records(
                args.backfill_dir,
                args.year,
                cve_filter,
            )
            checkpoint = newest_delta_time()
            print(f"CVE List V5 backfill: {len(references)} CVEs with PoC references")
        else:
            records, references, checkpoint = load_incremental_records(cve_filter)
            references = filter_live_references(references)

    github: dict[str, list[str]] = {}
    if not args.skip_github:
        token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") or ""
        current_year = datetime.now(timezone.utc).year
        years = [args.year] if backfill else list(range(current_year, current_year - args.years, -1))
        github = discover_github_pocs(
            token,
            years=years,
            lookback_days=args.lookback_days,
            backfill=backfill,
            cve_filter=cve_filter,
        )
        print(f"GitHub discovery: {sum(len(urls) for urls in github.values())} links for {len(github)} CVEs")

    if backfill:
        for cve_id in set(github) - set(records):
            path = record_paths.get(cve_id)
            if path and (record := read_record(path)):
                records[cve_id] = record
    fetch_missing_records(set(github) | set(references), records)

    stats, accepted_github, accepted_references = sync_markdown(
        records,
        github,
        references,
        dry_run=args.dry_run,
    )
    github_additions = append_inventory(GITHUB_LIST, accepted_github, dry_run=args.dry_run)
    reference_additions = append_inventory(REFERENCE_LIST, accepted_references, dry_run=args.dry_run)

    write_state(checkpoint, dry_run=args.dry_run)

    print(
        f"Created: {len(stats.created)} | Updated: {len(stats.updated)} | "
        f"Skipped: {len(stats.skipped)}"
    )
    print(
        f"Inventory additions: {github_additions} GitHub | "
        f"{reference_additions} references"
    )
    if stats.skipped:
        for entry in stats.skipped[:20]:
            print(f"Skipped {entry}")
        if len(stats.skipped) > 20:
            print(f"Skipped {len(stats.skipped) - 20} additional CVEs")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())