mirror of
https://github.com/0xMarcio/cve.git
synced 2026-10-27 16:51:03 +00:00
0xMarcio
2b6547ef29
KEV reached the site three days ago but nothing in the markdown said a vulnerability was under active attack, and the repository is what most people actually read. 1,595 entries now carry a red CISA KEV shield beside their product and vulnerability badges, marked for ransomware use where CISA records it. update_cves writes the badge on entries it creates, and the weekly audit reconciles the whole set against the refreshed catalogue, so an entry CISA adds gains the badge and one it withdraws loses it. Re-running the sync is a no-op.
1238 lines
45 KiB
Python
Executable File
1238 lines
45 KiB
Python
Executable File
#!/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"
|
|
KEV_FILE = ROOT / "kev.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 load_kev(path: Path = KEV_FILE) -> dict[str, list]:
|
|
"""CISA's known-exploited catalogue, as refreshed by the weekly audit."""
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def kev_badge(cve_id: str, kev: dict[str, list]) -> str:
|
|
"""The one badge that says an attacker is using this today."""
|
|
entry = kev.get(cve_id.upper())
|
|
if not entry:
|
|
return ""
|
|
ransomware = len(entry) > 1 and entry[1]
|
|
return badge("CISA KEV", "Known Exploited · Ransomware" if ransomware else "Known Exploited", "red")
|
|
|
|
|
|
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],
|
|
kev: dict[str, list] | None = None,
|
|
) -> 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)
|
|
known_exploited = kev_badge(cve_id, kev or {})
|
|
if known_exploited:
|
|
lines.append(known_exploited)
|
|
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()
|
|
kev = load_kev()
|
|
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, kev),
|
|
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())
|