diff --git a/.github/link_audit_state.json b/.github/link_audit_state.json new file mode 100644 index 00000000000..41f1767afb1 --- /dev/null +++ b/.github/link_audit_state.json @@ -0,0 +1,3 @@ +{ + "cursor": "zzzz966/CVE-2020-19360" +} diff --git a/.github/workflows/audit_poc_links.yml b/.github/workflows/audit_poc_links.yml new file mode 100644 index 00000000000..8d8f497a5d5 --- /dev/null +++ b/.github/workflows/audit_poc_links.yml @@ -0,0 +1,59 @@ +name: Audit PoC Links + +on: + schedule: + - cron: '40 3 * * 1' + workflow_dispatch: + +permissions: + contents: write + +concurrency: + group: repository-updates + cancel-in-progress: false + +jobs: + audit: + runs-on: ubuntu-latest + steps: + - name: Checkout repository + uses: actions/checkout@v7 + + - name: Set up Python + uses: actions/setup-python@v7 + with: + python-version: '3.13' + + # A quarter of the corpus a week keeps each run inside the token's + # hourly budget and sweeps everything within the month. + - name: Drop links to deleted repositories + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: python3 scripts/audit_poc_links.py --limit 12000 + + - name: Check for changes and commit + run: | + git config user.name 'CVE-BOT' + git config user.email 'cve@codepwn.win' + + git add --all -- \ + .github/link_audit_state.json \ + github.txt \ + ':(top,glob)[12][0-9][0-9][0-9]/CVE-*.md' + + if git diff --cached --quiet; then + echo "No changes to commit" + exit 0 + fi + + git commit -m "Drop PoC links to deleted repositories $(date -u +'%Y-%m-%d')" + + for attempt in 1 2 3; do + if git push origin HEAD:main; then + exit 0 + fi + echo "Push rejected; replaying onto the current main (attempt $attempt)" + git fetch --unshallow origin main 2>/dev/null || git fetch origin main + git rebase FETCH_HEAD + done + exit 1 diff --git a/.github/workflows/site.yml b/.github/workflows/site.yml index 05b5b0291b3..c8bf8212c12 100644 --- a/.github/workflows/site.yml +++ b/.github/workflows/site.yml @@ -8,6 +8,7 @@ on: workflows: - Sync CVE PoCs - Hot CVEs List + - Audit PoC Links types: - completed push: diff --git a/scripts/audit_poc_links.py b/scripts/audit_poc_links.py index db83845d104..a8430f96987 100644 --- a/scripts/audit_poc_links.py +++ b/scripts/audit_poc_links.py @@ -1,628 +1,166 @@ #!/usr/bin/env python3 -"""Locally validate and deduplicate CVE PoC links.""" +"""Remove PoC links whose GitHub repository no longer exists.""" from __future__ import annotations import argparse -import bisect -import hashlib import json -import re +import os import sys -import threading -import time from concurrent.futures import ThreadPoolExecutor, as_completed -from dataclasses import dataclass from pathlib import Path -from typing import Any, Callable, Iterable -from urllib import error, request -from urllib.parse import urlsplit, urlunsplit +from typing import Iterable sys.path.insert(0, str(Path(__file__).resolve().parent)) -import update_cves +from update_cves import ( + GITHUB_GRAPHQL_URL, + GitHubClient, + github_repo_from_url, + http_json, + replace_section, + section_links, +) ROOT = Path(__file__).resolve().parents[1] -STATE_FILE = ROOT / "data" / "link_audit_state.json" -INVENTORY_FILES = (ROOT / "github.txt", ROOT / "references.txt") -SECTION_EMPTY_TEXT = { - "#### Reference": "No PoCs from references.", - "#### Github": "No PoCs found on GitHub currently.", -} -GITHUB_RESERVED_OWNERS = { - "about", - "advisories", - "apps", - "collections", - "codespaces", - "events", - "explore", - "features", - "login", - "marketplace", - "notifications", - "orgs", - "pricing", - "search", - "security", - "settings", - "site", - "sponsors", - "topics", - "users", -} -USER_AGENT = "0xMarcio-cve-link-audit" +STATE_FILE = ROOT / ".github" / "link_audit_state.json" +GITHUB_LIST = ROOT / "github.txt" +GITHUB_SECTION = "#### Github" +GITHUB_EMPTY = "No PoCs found on GitHub currently." +BATCH = 20 -@dataclass(frozen=True) -class GitHubIdentity: - key: str - full_name: str - suffix: str - - -@dataclass -class ApplyStats: - markdown_files: int = 0 - inventory_files: int = 0 - removed: int = 0 - rewritten: int = 0 - deduplicated: int = 0 - - -class HostThrottle: - def __init__(self, interval: float) -> None: - self.interval = interval - self.guard = threading.Lock() - self.locks: dict[str, threading.Lock] = {} - self.last_start: dict[str, float] = {} - - def run(self, url: str, operation: Callable[[], str]) -> str: - host = (urlsplit(url).hostname or "").lower() - with self.guard: - lock = self.locks.setdefault(host, threading.Lock()) - with lock: - elapsed = time.monotonic() - self.last_start.get(host, 0.0) - if elapsed < self.interval: - time.sleep(self.interval - elapsed) - self.last_start[host] = time.monotonic() - return operation() - - -def raw_section_links(block: str) -> list[str]: - links: list[str] = [] - for raw_line in block.splitlines(): - line = raw_line.strip() - if line.startswith("- "): - line = line[2:].strip() - if line.startswith(("http://", "https://")): - links.append(line) - return links - - -def section_links(text: str, header: str) -> list[str]: - pattern = re.compile( - update_cves.SECTION_RE_TEMPLATE.format(header=re.escape(header)), - re.DOTALL, - ) - match = pattern.search(text) - return raw_section_links(match.group(2)) if match else [] - - -def github_identity(url: str) -> GitHubIdentity | None: - parsed = urlsplit(url) - if (parsed.hostname or "").lower() not in {"github.com", "www.github.com"}: - return None - parts = [part for part in parsed.path.split("/") if part] - if len(parts) < 2 or parts[0].lower() in GITHUB_RESERVED_OWNERS: - return None - owner = parts[0] - name = parts[1][:-4] if parts[1].lower().endswith(".git") else parts[1] - if not owner or not name: - return None - full_name = f"{owner}/{name}" - suffix = "/" + "/".join(parts[2:]) if len(parts) > 2 else "" - return GitHubIdentity(full_name.lower(), full_name, suffix) - - -def canonical_github_url(url: str, canonical_base: str) -> str: - identity = github_identity(url) - if not identity: - return url - parsed = urlsplit(url) - base = urlsplit(canonical_base.rstrip("/")) - path = base.path.rstrip("/") + identity.suffix - return urlunsplit(("https", "github.com", path, parsed.query, parsed.fragment)) - - -def is_profile_repository_root(url: str) -> bool: - identity = github_identity(url) - if not identity or identity.suffix: - return False - owner, name = identity.full_name.lower().split("/", 1) - return owner == name - - -def inventory_url(line: str) -> str: - value = line.strip() - if " - " in value: - value = value.split(" - ", 1)[1].strip() - return value if value.startswith(("http://", "https://")) else "" - - -def collect_links() -> set[str]: - links: set[str] = set() - for path in INVENTORY_FILES: - if not path.exists(): - continue - for line in path.read_text(encoding="utf-8").splitlines(): - if url := inventory_url(line): - links.add(url) - for path in ROOT.glob("[12][0-9][0-9][0-9]/CVE-*.md"): +def collect_repositories() -> dict[str, list[tuple[Path, str]]]: + """Map every referenced repository to the entries and links citing it.""" + references: dict[str, list[tuple[Path, str]]] = {} + for path in sorted(ROOT.glob("[12][0-9][0-9][0-9]/CVE-*.md")): text = path.read_text(encoding="utf-8", errors="replace") - for header in SECTION_EMPTY_TEXT: - links.update(section_links(text, header)) - return links + for url in section_links(text, GITHUB_SECTION): + full_name = github_repo_from_url(url) + if full_name and "/" in full_name: + references.setdefault(full_name, []).append((path, url)) + return references -def repository_names(links: Iterable[str]) -> dict[str, str]: - result: dict[str, str] = {} - for url in links: - identity = github_identity(url) - if identity: - result.setdefault(identity.key, identity.full_name) - return result +def missing_repositories(client: GitHubClient, names: list[str]) -> set[str]: + """Return the repositories GitHub reports as NOT_FOUND. + Anything else — a network error, a throttle, an unexpected response — leaves + the repository out of the result, so an audit failure never drops a link. + """ + batches = [names[index : index + BATCH] for index in range(0, len(names), BATCH)] + missing: set[str] = set() + checked = 0 -def rotation_key(value: str) -> str: - return f"{hashlib.sha256(value.encode('utf-8')).hexdigest()}:{value}" + def check(batch: list[str]) -> set[str]: + aliases = " ".join( + f"r{index}: repository(owner: {json.dumps(name.split('/', 1)[0])}, " + f"name: {json.dumps(name.split('/', 1)[1])}) {{ nameWithOwner }}" + for index, name in enumerate(batch) + ) + payload = http_json( + GITHUB_GRAPHQL_URL, + headers=client.headers, + data={"query": "query { " + aliases + " rateLimit { remaining resetAt } }"}, + ) + gone: set[str] = set() + for entry in payload.get("errors") or []: + path = entry.get("path") or [] + if str(entry.get("type") or "") == "NOT_FOUND" and path: + index = str(path[0])[1:] + if index.isdigit() and int(index) < len(batch): + gone.add(batch[int(index)]) + return gone - -def select_rotating( - values: Iterable[str], - after: str, - limit: int, - *, - key: Callable[[str], str] = lambda value: value, -) -> tuple[list[str], str]: - ordered = sorted(set(values), key=key) - if not ordered: - return [], "" - if limit <= 0 or limit >= len(ordered): - return ordered, "" - keys = [key(value) for value in ordered] - start = bisect.bisect_right(keys, after) - selected = [ordered[(start + offset) % len(ordered)] for offset in range(limit)] - return selected, key(selected[-1]) - - -def request_result(url: str, method: str, timeout: int) -> str: - headers = { - "Accept": "*/*", - "Accept-Encoding": "identity", - "User-Agent": USER_AGENT, - } - if method == "GET": - headers["Range"] = "bytes=0-0" - req = request.Request(url, headers=headers, method=method) - try: - with request.urlopen(req, timeout=timeout) as response: - if method == "GET": - response.read(1) - return "alive" if int(response.status) < 400 else "unknown" - except error.HTTPError as exc: - code = exc.code - exc.close() - if code in {404, 410}: - return "dead" - return "unknown" - except (error.URLError, TimeoutError, OSError): - return "unknown" - - -def check_http_url(url: str, timeout: int, throttle: HostThrottle) -> str: - def operation() -> str: - head = request_result(url, "HEAD", timeout) - if head == "alive": - return head - return request_result(url, "GET", timeout) - - return throttle.run(url, operation) - - -def audit_http( - urls: list[str], - *, - workers: int, - timeout: int, - host_interval: float, - label: str = "URLs", -) -> dict[str, str]: - results: dict[str, str] = {} - throttle = HostThrottle(host_interval) - with ThreadPoolExecutor(max_workers=workers) as executor: - futures = { - executor.submit(check_http_url, url, timeout, throttle): url - for url in urls - } - for count, future in enumerate(as_completed(futures), start=1): - url = futures[future] + with ThreadPoolExecutor(max_workers=4) as executor: + futures = {executor.submit(check, batch): batch for batch in batches} + for future in as_completed(futures): + batch = futures[future] try: - results[url] = future.result() - except Exception: - results[url] = "unknown" - if count == 1 or count % 250 == 0 or count == len(urls): - print(f"{label} checked: {count} of {len(urls)}") - return results + missing |= future.result() + except Exception as exc: + print(f"Kept {len(batch)} repositories after a failed check: {exc}", file=sys.stderr) + checked += len(batch) + if checked % 5000 < len(batch): + print(f"Checked {checked} of {len(names)} repositories") + return missing -def rewrite_url( - url: str, - canonical: dict[str, str], - missing: set[str], - dead_urls: set[str], -) -> str | None: - if url in dead_urls: - return None - identity = github_identity(url) - if not identity: - return url - if identity.key in missing: - return None - base = canonical.get(identity.key) - return canonical_github_url(url, base) if base else url - - -def rewrite_links( - links: Iterable[str], - canonical: dict[str, str], - missing: set[str], - dead_urls: set[str], - cache: dict[str, str | None] | None = None, - seen: set[str] | None = None, -) -> tuple[list[str], int, int, int]: - output: list[str] = [] - seen = set() if seen is None else seen - removed = 0 - rewritten = 0 - deduplicated = 0 - for url in links: - if cache is not None and url in cache: - updated = cache[url] - else: - updated = rewrite_url(url, canonical, missing, dead_urls) - if cache is not None: - cache[url] = updated - if updated is None: - removed += 1 - continue - if updated != url: - rewritten += 1 - key = update_cves.poc_link_key(updated) - if key in seen: - deduplicated += 1 - continue - seen.add(key) - output.append(updated) - return output, removed, rewritten, deduplicated - - -def update_markdown( - canonical: dict[str, str], - missing: set[str], - dead_urls: set[str], +def prune_entries( + references: dict[str, list[tuple[Path, str]]], + missing: Iterable[str], *, - apply: bool, - cache: dict[str, str | None] | None = None, -) -> ApplyStats: - stats = ApplyStats() - for path in ROOT.glob("[12][0-9][0-9][0-9]/CVE-*.md"): + dry_run: bool, +) -> tuple[int, int, set[str]]: + dead_by_path: dict[Path, set[str]] = {} + dead_urls: set[str] = set() + for full_name in missing: + for path, url in references.get(full_name, []): + dead_by_path.setdefault(path, set()).add(url) + dead_urls.add(url) + + for path, urls in sorted(dead_by_path.items()): with path.open("r", encoding="utf-8", newline="") as handle: - original = handle.read() - updated = original - changed = False - seen: set[str] = set() - for header, empty_text in SECTION_EMPTY_TEXT.items(): - links = section_links(updated, header) - accepted = [ - url - for url in links - if not update_cves.github_url_is_ambiguous_root(url, path.stem) - ] - stats.removed += len(links) - len(accepted) - replacement, removed, rewritten, deduplicated = rewrite_links( - accepted, - canonical, - missing, - dead_urls, - cache, - seen, - ) - stats.removed += removed - stats.rewritten += rewritten - stats.deduplicated += deduplicated - if replacement == links: - continue - updated, section_changed = update_cves.replace_section( - updated, - header, - replacement, - empty_text, - ) - changed = changed or section_changed - if not changed: - continue - stats.markdown_files += 1 - if apply: + text = handle.read() + kept = [url for url in section_links(text, GITHUB_SECTION) if url not in urls] + updated, changed = replace_section(text, GITHUB_SECTION, kept, GITHUB_EMPTY) + if changed and not dry_run: with path.open("w", encoding="utf-8", newline="") as handle: handle.write(updated) - return stats + removals = sum(len(urls) for urls in dead_by_path.values()) + return len(dead_by_path), removals, dead_urls -def update_inventory( - path: Path, - canonical: dict[str, str], - missing: set[str], - dead_urls: set[str], - *, - apply: bool, - cache: dict[str, str | None] | None = None, -) -> ApplyStats: - stats = ApplyStats() - original = path.read_text(encoding="utf-8") if path.exists() else "" - output: list[str] = [] - seen: set[str] = set() - for line in original.splitlines(): - url = inventory_url(line) - prefix = line.split(" - ", 1)[0] if " - " in line else "" - if not url: - updated_line = line - else: - if update_cves.github_url_is_ambiguous_root(url, prefix): - stats.removed += 1 - continue - if cache is not None and url in cache: - updated_url = cache[url] - else: - updated_url = rewrite_url(url, canonical, missing, dead_urls) - if cache is not None: - cache[url] = updated_url - if updated_url is None: - stats.removed += 1 - continue - if updated_url != url: - stats.rewritten += 1 - updated_line = f"{prefix} - {updated_url}" if prefix else updated_url - dedupe_key = updated_line - if url: - dedupe_key = update_cves.poc_link_key(updated_url) - if prefix: - dedupe_key = f"{prefix} - {dedupe_key}" - if dedupe_key in seen: - stats.deduplicated += 1 - continue - seen.add(dedupe_key) - output.append(updated_line) - updated = "\n".join(output) + ("\n" if output else "") - if updated != original: - stats.inventory_files = 1 - if apply: - path.write_text(updated, encoding="utf-8") - return stats +def prune_inventory(dead_urls: set[str], *, dry_run: bool) -> int: + if not GITHUB_LIST.exists() or not dead_urls: + return 0 + lines = GITHUB_LIST.read_text(encoding="utf-8").splitlines() + kept = [line for line in lines if line.rsplit(" - ", 1)[-1].strip() not in dead_urls] + if not dry_run and len(kept) != len(lines): + GITHUB_LIST.write_text("\n".join(kept) + "\n", encoding="utf-8") + return len(lines) - len(kept) -def add_stats(target: ApplyStats, source: ApplyStats) -> None: - target.markdown_files += source.markdown_files - target.inventory_files += source.inventory_files - target.removed += source.removed - target.rewritten += source.rewritten - target.deduplicated += source.deduplicated - - -def load_state(path: Path) -> dict[str, Any]: +def load_cursor() -> str: try: - state = json.loads(path.read_text(encoding="utf-8")) + return str(json.loads(STATE_FILE.read_text(encoding="utf-8")).get("cursor") or "") except (OSError, json.JSONDecodeError): - return {} - return state if isinstance(state, dict) else {} + return "" -def save_state(path: Path, state: dict[str, Any]) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n", encoding="utf-8") - - -def parse_args() -> argparse.Namespace: - parser = argparse.ArgumentParser(description="Locally audit CVE PoC links") - parser.add_argument("--apply", action="store_true", help="Write link and state changes") - parser.add_argument( - "--github-limit", - type=int, - default=0, - help="GitHub repository roots checked per run; zero checks all", - ) - parser.add_argument( - "--http-limit", - type=int, - default=0, - help="Exact URLs checked per run; zero checks all", - ) - parser.add_argument("--workers", type=int, default=12) - parser.add_argument("--timeout", type=int, default=15) - parser.add_argument("--host-interval", type=float, default=0.1) - parser.add_argument("--confirmations", type=int, default=2) - parser.add_argument("--skip-github", action="store_true") - parser.add_argument("--skip-http", action="store_true") - parser.add_argument("--state-file", type=Path, default=STATE_FILE) - args = parser.parse_args() - if args.github_limit < 0 or args.http_limit < 0: - parser.error("limits cannot be negative") - if args.workers < 1 or args.timeout < 1 or args.confirmations < 1: - parser.error("workers, timeout, and confirmations must be positive") - if args.host_interval < 0: - parser.error("--host-interval cannot be negative") - return args +def save_cursor(cursor: str, *, dry_run: bool) -> None: + if dry_run: + return + STATE_FILE.parent.mkdir(parents=True, exist_ok=True) + STATE_FILE.write_text(json.dumps({"cursor": cursor}, indent=2) + "\n", encoding="utf-8") def main() -> int: - args = parse_args() - state = load_state(args.state_file) - links = collect_links() - names_by_key = repository_names(links) - print(f"PoC links found: {len(links)} | GitHub repositories: {len(names_by_key)}") + parser = argparse.ArgumentParser(description="Drop PoC links to deleted repositories") + parser.add_argument("--limit", type=int, default=0, help="Repositories to check this run") + parser.add_argument("--dry-run", action="store_true", help="Report without writing files") + args = parser.parse_args() - github_after = str(state.get("github_after") or "") - missing_repo_candidates = { - str(key): int(count) - for key, count in (state.get("missing_repo_candidates") or {}).items() - if key in names_by_key and isinstance(count, int) and count > 0 - } - missing_repositories: set[str] = set() - if not args.skip_github: - selected_keys, github_after = select_rotating( - names_by_key, - github_after, - args.github_limit, - ) - roots_by_key = { - key: f"https://github.com/{names_by_key[key]}" - for key in selected_keys - } - results = audit_http( - list(roots_by_key.values()), - workers=args.workers, - timeout=args.timeout, - host_interval=args.host_interval, - label="GitHub repository roots", - ) - for key, url in roots_by_key.items(): - status = results[url] - if status == "alive": - missing_repo_candidates.pop(key, None) - elif status == "dead": - count = missing_repo_candidates.get(key, 0) + 1 - if count >= args.confirmations: - missing_repositories.add(key) - missing_repo_candidates.pop(key, None) - else: - missing_repo_candidates[key] = count - values = list(results.values()) - print( - f"GitHub repository result: {values.count('alive')} live | " - f"{values.count('dead')} dead | {values.count('unknown')} unknown | " - f"{len(missing_repositories)} confirmed removals" - ) + references = collect_repositories() + names = sorted(references) + print(f"{sum(len(v) for v in references.values())} links across {len(names)} repositories") - dead_candidates = { - str(url): int(count) - for url, count in (state.get("dead_candidates") or {}).items() - if url in links and isinstance(count, int) and count > 0 - } - dead_urls: set[str] = set() - http_after = str(state.get("http_after") or "") - if not args.skip_http: - http_candidates = [ - url - for url in links - if not ( - (identity := github_identity(url)) - and not identity.suffix - ) - and not ( - identity - and identity.key in missing_repositories - ) - ] - selected_urls, http_after = select_rotating( - http_candidates, - http_after, - args.http_limit, - key=rotation_key, - ) - results = audit_http( - selected_urls, - workers=args.workers, - timeout=args.timeout, - host_interval=args.host_interval, - label="Exact URLs", - ) - for url, status in results.items(): - if status == "alive": - dead_candidates.pop(url, None) - elif status == "dead": - count = dead_candidates.get(url, 0) + 1 - if count >= args.confirmations: - dead_urls.add(url) - dead_candidates.pop(url, None) - else: - dead_candidates[url] = count - values = list(results.values()) - counts = { - status: values.count(status) - for status in {"alive", "dead", "unknown"} - } - print( - f"Exact URL result: {counts['alive']} live | {counts['dead']} dead | " - f"{counts['unknown']} unknown | {len(dead_urls)} confirmed removals" - ) + cursor = load_cursor() + start = next((i for i, name in enumerate(names) if name > cursor), 0) if cursor else 0 + scope = names[start : start + args.limit] if args.limit else names + if args.limit and len(scope) < args.limit: + scope += names[: args.limit - len(scope)] # wrap around + print(f"Checking {len(scope)} repositories starting after {cursor or 'the beginning'}") - profile_roots = {url for url in links if is_profile_repository_root(url)} - dead_urls.update(profile_roots) - if profile_roots: - print(f"Profile repository roots to purge: {len(profile_roots)}") + token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") or "" + missing = missing_repositories(GitHubClient(token), scope) + print(f"{len(missing)} repositories no longer exist") - blacklist = update_cves.load_blacklist() - blocked_repositories = { - identity.key - for url in links - if (identity := github_identity(url)) - and update_cves.is_blacklisted_repo(identity.full_name, blacklist) - } - if blocked_repositories: - print(f"Blacklisted repositories to purge: {len(blocked_repositories)}") - removed_repositories = missing_repositories | blocked_repositories + entries, links, dead_urls = prune_entries(references, missing, dry_run=args.dry_run) + inventory = prune_inventory(dead_urls, dry_run=args.dry_run) + save_cursor(scope[-1] if scope else "", dry_run=args.dry_run) - rewrite_cache: dict[str, str | None] = {} - stats = update_markdown( - {}, - removed_repositories, - dead_urls, - apply=args.apply, - cache=rewrite_cache, - ) - for path in INVENTORY_FILES: - add_stats( - stats, - update_inventory( - path, - {}, - removed_repositories, - dead_urls, - apply=args.apply, - cache=rewrite_cache, - ), - ) - - if args.apply: - save_state( - args.state_file, - { - "dead_candidates": dead_candidates, - "github_after": github_after, - "http_after": http_after, - "missing_repo_candidates": missing_repo_candidates, - }, - ) - if stats.markdown_files: - update_cves.regenerate_json() - - print( - f"Changes: {stats.markdown_files} Markdown files | {stats.inventory_files} inventories | " - f"{stats.removed} links removed | {stats.rewritten} rewritten | " - f"{stats.deduplicated} duplicates removed" - ) - if not args.apply and (stats.markdown_files or stats.inventory_files): - print("Dry run only; pass --apply to write changes") + print(f"Removed {links} links from {entries} entries | {inventory} inventory lines") return 0