From 0fdc3cd5752e107aec5ce5e7d2cb84ee73a65240 Mon Sep 17 00:00:00 2001 From: 0xMarcio Date: Thu, 22 Oct 2026 13:26:22 +0200 Subject: [PATCH] Remove confirmed dead PoC links during metadata refresh --- .github/getTrending.py | 5 + .github/workflows/audit_poc_links.yml | 17 +- .github/workflows/hot_cves.yml | 3 +- scripts/audit_poc_links.py | 195 ++++++++++++++--------- scripts/build_site.py | 7 +- scripts/link_health.py | 125 +++++++++++++++ scripts/sync_collections.py | 2 + scripts/sync_exploits.py | 2 + scripts/sync_nuclei.py | 3 + scripts/sync_repo_metadata.py | 43 +++++- scripts/update_cves.py | 36 ++--- tests/test_audit_poc_links.py | 35 +++-- tests/test_link_health.py | 213 ++++++++++++++++++++++++++ tests/test_repo_metadata.py | 8 +- 14 files changed, 567 insertions(+), 127 deletions(-) create mode 100644 scripts/link_health.py create mode 100644 tests/test_link_health.py diff --git a/.github/getTrending.py b/.github/getTrending.py index 82079ea059c..2e11bd8cead 100644 --- a/.github/getTrending.py +++ b/.github/getTrending.py @@ -312,6 +312,7 @@ def just_landed(token: str, ledger: dict | None = None) -> list[dict]: def ledger_landed(ledger: dict, candidates: list[dict]) -> list[dict]: + import link_health from build_site import build_cve_list, build_landed, dedupe_source_links, source_key, CURATED, REPO_META entries, _ = build_cve_list(load_blacklist()) published = {entry["cve"]: entry for entry in entries} @@ -326,6 +327,8 @@ def ledger_landed(ledger: dict, candidates: list[dict]) -> list[dict]: for repo in candidates: cve = cve_of(repo) url = repo.get("_artifact_url") or repo["html_url"] + if link_health.unavailable(url): + continue entry = published.get(cve, {}) curated = [link for _, field in CURATED for link in entry.get(field, []) if source_key(link) == source_key(url)] @@ -865,6 +868,7 @@ repository URL. def main(*, history_limit: int = HISTORY_LIMIT, retry_history: bool = False) -> int: + import link_health global HISTORY_REMAINING, HISTORY_RETRY_HOURS HISTORY_REMAINING = history_limit HISTORY_RETRY_HOURS = 0 if retry_history else 4 @@ -884,6 +888,7 @@ def main(*, history_limit: int = HISTORY_LIMIT, retry_history: bool = False) -> searched = len(repositories) repositories, paths = qualifying_repositories(repositories, token) repositories = published_repositories(repositories) + repositories = [repo for repo in repositories if not link_health.unavailable(artifact_link(repo)[0])] observations.extend(repositories) release_dates(repositories, token, paths, ledger) shipped = code_pushed( diff --git a/.github/workflows/audit_poc_links.yml b/.github/workflows/audit_poc_links.yml index edf442044ee..4be818d8b77 100644 --- a/.github/workflows/audit_poc_links.yml +++ b/.github/workflows/audit_poc_links.yml @@ -29,13 +29,15 @@ jobs: with: python-version: '3.13' - # 4,000 repositories and 3,000 reference links a day sweeps the whole - # index about weekly, in around half an hour. A sample found one link in - # eight pointing at a deleted repository under the old fortnightly sweep. + # The hourly sweep checks front-page repositories plus 2,000 older repos + # and 500 artifact URLs. This deeper scan also includes reference links, + # curated file paths and recovery checks for previously unavailable URLs. - name: Drop dead links and refresh repository metadata + id: link_audit + continue-on-error: true env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - run: python3 scripts/audit_poc_links.py --limit 4000 --references 3000 + run: python3 scripts/audit_poc_links.py --limit 4000 --references 10000 --workers 32 --timeout 5 - name: Verify the generated index run: python3 scripts/build_site.py --offline @@ -56,10 +58,13 @@ jobs: git add --all -- \ .github/link_audit_state.json \ + index/link_health.json \ index/github.txt \ index/repo_meta.json \ index/kev.json \ index/references.txt \ + index/reference_pocs.txt \ + index/trending.json \ README.md \ docs/hero.svg \ docs/social-card.svg \ @@ -93,3 +98,7 @@ jobs: env: GH_TOKEN: ${{ github.token }} run: gh workflow run site.yml --ref main + + - name: Report incomplete repository checks + if: always() && steps.link_audit.outcome == 'failure' + run: exit 1 diff --git a/.github/workflows/hot_cves.yml b/.github/workflows/hot_cves.yml index 8ab0d00f368..cad05646e7a 100644 --- a/.github/workflows/hot_cves.yml +++ b/.github/workflows/hot_cves.yml @@ -45,7 +45,7 @@ jobs: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} run: python3 scripts/update_cves.py --lookback-days 1 --years 2 --skip-cvelist - - name: Refresh repository stars and push dates + - name: Refresh repository metadata and remove confirmed dead links id: repository_metadata continue-on-error: true env: @@ -88,6 +88,7 @@ jobs: git add --all -- \ README.md \ .github/repo_metadata_state.json \ + index/link_health.json \ index/repo_meta.json \ index/metadata/ \ index/trending.json \ diff --git a/scripts/audit_poc_links.py b/scripts/audit_poc_links.py index 1738db3d180..67b0d639375 100644 --- a/scripts/audit_poc_links.py +++ b/scripts/audit_poc_links.py @@ -6,13 +6,13 @@ from __future__ import annotations import argparse import json import os +import re import sys import tempfile from concurrent.futures import ThreadPoolExecutor, as_completed -from datetime import date +from datetime import date, datetime, timezone from pathlib import Path from typing import Iterable -from urllib import error, request from urllib.parse import urlsplit sys.path.insert(0, str(Path(__file__).resolve().parent)) @@ -20,11 +20,10 @@ sys.path.insert(0, str(Path(__file__).resolve().parent)) from update_cves import ( readme_text, repo_is_bare, + repository_is_nonpublic, GITHUB_GRAPHQL_URL, kev_badge, - USER_AGENT, GitHubClient, - github_repo_from_url, http_json, is_blacklisted_repo, is_valid_cve, @@ -32,6 +31,7 @@ from update_cves import ( replace_section, section_links, ) +import link_health ROOT = Path(__file__).resolve().parents[1] CVES = ROOT / "cves" @@ -50,6 +50,11 @@ REFERENCE_SECTION = "#### Reference" SECTION_EMPTY = { GITHUB_SECTION: "No PoCs found on GitHub currently.", REFERENCE_SECTION: "No PoCs from references.", + "#### Nuclei": "No PoCs from Nuclei.", + "#### ExploitDB": "No PoCs from ExploitDB.", + "#### Metasploit": "No PoCs from Metasploit.", + "#### Vulhub": "No PoCs from Vulhub.", + "#### Collections": "No PoCs from collections.", } GITHUB_EMPTY = "No PoCs found on GitHub currently." BATCH = 20 @@ -58,8 +63,8 @@ BATCH = 20 def collect_repositories() -> tuple[dict[str, list[tuple[Path, str]]], set[str]]: """Map prunable links to their entries, and name every repository the site shows. - Only the GitHub section is prunable; a reference comes from the CVE record - itself, so it stays even when it rots. Both sections need star counts. + The GitHub section also has a content-quality check. Confirmed public + availability failures are pruned from every section by prune_unavailable. """ prunable: dict[str, list[tuple[Path, str]]] = {} referenced: set[str] = set() @@ -82,7 +87,7 @@ def collect_repositories() -> tuple[dict[str, list[tuple[Path, str]]], set[str]] REPO_FIELDS = ( - "{ nameWithOwner stargazerCount pushedAt " + "{ nameWithOwner isPrivate visibility stargazerCount pushedAt " '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 } } ' @@ -91,7 +96,7 @@ REPO_FIELDS = ( ) -def survey(client: GitHubClient, names: list[str]) -> tuple[set[str], dict[str, list]]: +def survey(client: GitHubClient, names: list[str]) -> tuple[set[str], dict[str, list], set[str]]: """Return the repositories to drop, and stars/last push for the rest. Dropped means GitHub reports NOT_FOUND, or the repository is bare: no code, @@ -102,10 +107,11 @@ def survey(client: GitHubClient, names: list[str]) -> tuple[set[str], dict[str, """ batches = [names[index : index + BATCH] for index in range(0, len(names), BATCH)] missing: set[str] = set() + unavailable: set[str] = set() metadata: dict[str, list] = {} checked = 0 - def check(batch: list[str]) -> tuple[set[str], dict[str, list]]: + def check(batch: list[str]) -> tuple[set[str], dict[str, list], set[str]]: aliases = " ".join( f"r{index}: repository(owner: {json.dumps(name.split('/', 1)[0])}, " f"name: {json.dumps(name.split('/', 1)[1])}) " @@ -118,56 +124,60 @@ def survey(client: GitHubClient, names: list[str]) -> tuple[set[str], dict[str, data={"query": "query { " + aliases + " rateLimit { remaining resetAt } }"}, ) gone: set[str] = set() - failed_aliases: set[str] = set() - for entry in payload.get("errors") or []: - path = entry.get("path") or [] - if not path: - return set(), {} - alias = str(path[0]) - failed_aliases.add(alias) - index = alias[1:] - if ( - str(entry.get("type") or "") == "NOT_FOUND" - and len(path) == 1 - and alias.startswith("r") - and index.isdigit() - and int(index) < len(batch) - ): - gone.add(batch[int(index)]) + absent: set[str] = set() + errors = payload.get("errors") or [] + if any(not entry.get("path") for entry in errors): + return set(), {}, set() found: dict[str, list] = {} data = payload.get("data") or {} for index, name in enumerate(batch): alias = f"r{index}" # A failed README or root lookup is not evidence of a bare repo. - if alias in failed_aliases: + problems = [entry for entry in errors if entry["path"][0] == alias] + if problems: + if (all(entry.get("type") == "NOT_FOUND" and len(entry["path"]) == 1 + for entry in problems) + and link_health.check_url(f"https://github.com/{name}") is False): + gone.add(name) + absent.add(name) continue repo = data.get(alias) if not repo: continue + if repository_is_nonpublic(repo): + if link_health.check_url(f"https://github.com/{name}") is False: + gone.add(name) + absent.add(name) + continue + if repo.get("isPrivate") is not False or repo.get("visibility") != "PUBLIC": + continue if repo_is_bare(repo, readme_text(repo)): gone.add(name) continue + if type(repo.get("stargazerCount")) is not int or repo["stargazerCount"] < 0 or "pushedAt" not in repo: + continue pushed = str(repo.get("pushedAt") or "")[:10] found[name.lower()] = [ int(repo.get("stargazerCount") or 0), pushed, ] - return gone, found + return gone, found, absent 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: - gone, found = future.result() + gone, found, absent = future.result() missing |= gone + unavailable |= absent metadata.update(found) 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, metadata + return missing, metadata, unavailable def save_metadata(metadata: dict[str, list], alive: set[str], *, dry_run: bool) -> int: @@ -222,22 +232,77 @@ def prune_inventory(dead_urls: set[str], *, dry_run: bool) -> int: def collect_references() -> dict[str, list[tuple[Path, str]]]: - """Every link that is not a GitHub repository, and where it appears.""" + """Exact artifact/reference URLs, including files inside healthy repositories.""" references: dict[str, list[tuple[Path, str]]] = {} for path in sorted(CVES.glob("[12][0-9][0-9][0-9]/CVE-*.md")): text = path.read_text(encoding="utf-8", errors="replace") - for header in (REFERENCE_SECTION, GITHUB_SECTION): + for header in SECTION_EMPTY: for url in section_links(text, header): - if not github_repo_from_url(url): + if not is_repository_url(url): references.setdefault(url, []).append((path, header)) return references +def is_repository_url(url: str) -> bool: + try: + parsed = urlsplit(url) + except ValueError: + return False + return (parsed.hostname in {"github.com", "www.github.com"} + and bool(link_health.repository(url)) + and len(parsed.path.strip("/").split("/")) == 2) + + +def prune_unavailable(*, dry_run: bool = False) -> tuple[int, int]: + """Apply confirmed findings to every source, inventory and front-page cache.""" + if not any(link_health.load().values()): + return 0, 0 + changed = removed = 0 + for path in sorted(CVES.glob("[12][0-9][0-9][0-9]/CVE-*.md")): + with path.open(encoding="utf-8", newline="") as handle: + original = text = handle.read() + for header, empty in SECTION_EMPTY.items(): + links = section_links(text, header) + kept = [url for url in links if not link_health.unavailable(url)] + if len(kept) != len(links): + removed += len(links) - len(kept) + text, _ = replace_section(text, header, kept, empty) + if text != original: + changed += 1 + if not dry_run: + with path.open("w", encoding="utf-8", newline="") as handle: + handle.write(text) + for path in (GITHUB_LIST, REFERENCE_LIST, INDEX / "reference_pocs.txt"): + if path.exists(): + lines = path.read_text(encoding="utf-8").splitlines() + kept = [line for line in lines if not link_health.unavailable(line.rsplit(" - ", 1)[-1].strip())] + if len(kept) != len(lines) and not dry_run: + path.write_text("\n".join(kept) + "\n", encoding="utf-8") + trending = INDEX / "trending.json" + if trending.exists(): + rows = json.loads(trending.read_text(encoding="utf-8")) + kept = [row for row in rows if not link_health.unavailable(row.get("url") or "")] + if len(kept) != len(rows) and not dry_run: + trending.write_text(json.dumps(kept, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + readme = ROOT / "README.md" + if readme.exists(): + lines = readme.read_text(encoding="utf-8").splitlines(keepends=True) + kept = [line for line in lines if not (line.startswith("|") and any( + link_health.unavailable(url) for url in re.findall(r"\]\((https?://[^\s)]+)\)", line)))] + if kept != lines and not dry_run: + readme.write_text("".join(kept), encoding="utf-8") + print(f"Confirmed unavailable links: {removed} removed from {changed} CVE entries") + return changed, removed + + def interleave_by_host(urls: list[str]) -> list[str]: """Round-robin the hosts so a slice does not hammer one server in a burst.""" hosts: dict[str, list[str]] = {} for url in urls: - host = urlsplit(url).netloc.lower() + try: + host = urlsplit(url).netloc.lower() + except ValueError: + host = "" hosts.setdefault(host, []).append(url) queues = list(hosts.values()) ordered: list[str] = [] @@ -251,38 +316,7 @@ def interleave_by_host(urls: list[str]) -> list[str]: def dead_references(urls: list[str], *, workers: int, timeout: int) -> set[str]: """URLs the web answers 404 or 410 for. Everything else keeps its link.""" - dead: set[str] = set() - checked = 0 - - def check(url: str) -> tuple[str, bool]: - headers = {"User-Agent": USER_AGENT, "Accept": "*/*"} - for method in ("HEAD", "GET"): - try: - req = request.Request(url, method=method, headers=headers) - with request.urlopen(req, timeout=timeout) as response: - if method == "GET": - response.read(1) - return url, False - except error.HTTPError as exc: - code = exc.code - exc.close() - # a server that dislikes HEAD is not a server without the page - if method == "GET" or code not in {403, 404, 405, 410, 501}: - return url, code in {404, 410} - except Exception: - # a timeout or a refused connection keeps the link, and asking - # the same unreachable host a second time only doubles the wait - return url, False - return url, False - - with ThreadPoolExecutor(max_workers=workers) as executor: - for url, gone in executor.map(check, urls): - if gone: - dead.add(url) - checked += 1 - if checked % 2000 == 0: - print(f"Checked {checked} of {len(urls)} references") - return dead + return link_health.survey_urls(urls, workers=workers, timeout=timeout)[0] def prune_references( @@ -461,7 +495,8 @@ def main() -> int: ) state = load_state() - scope = slice_after(names, str(state.get("cursor") or ""), args.limit) + rotating = slice_after(names, str(state.get("cursor") or ""), args.limit) + scope = sorted(set(rotating) | set(link_health.load()["repositories"])) print(f"Checking {len(scope)} repositories starting after {state.get('cursor') or 'the beginning'}") token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") or "" @@ -469,20 +504,25 @@ def main() -> int: name for name in references if is_blacklisted_repo(name, load_blacklist()) } - missing, metadata = survey( + missing, metadata, unavailable = survey( GitHubClient(token), [name for name in scope if name not in blocked], ) print(f"{len(missing)} repositories no longer exist or hold nothing runnable") + if not args.dry_run: + link_health.record(missing=unavailable, healthy=metadata) if blocked: print(f"Pruning {len(blocked)} explicitly blacklisted repositories") missing |= blocked + deferred = set(scope) - missing - metadata.keys() + if deferred: + print(f"::warning::{len(deferred)} repository checks were inconclusive; existing links retained") entries, links, dead_urls = prune_entries(references, missing, dry_run=args.dry_run) inventory = prune_inventory(dead_urls, dry_run=args.dry_run) alive = {name.lower() for name in names if name not in missing} tracked = save_metadata(metadata, alive, dry_run=args.dry_run) - state["cursor"] = scope[-1] if scope else "" + state["cursor"] = rotating[-1] if rotating else "" print(f"Removed {links} links from {entries} entries | {inventory} inventory lines") print(f"Refreshed stars and last-push for {len(metadata)} repositories | {tracked} tracked") @@ -492,20 +532,29 @@ def main() -> int: urls = sorted(references) window = slice_after(urls, str(state.get("reference_cursor") or ""), 0 if args.references < 0 else args.references) - print(f"Checking {len(window)} of {len(urls)} reference links") - gone = dead_references( - interleave_by_host(window), workers=args.workers, timeout=args.timeout - ) + # Recovery gets a separate bounded rotation, so historical failures + # cannot consume the entire audit and starve the still-published URLs. + recovery = slice_after(sorted(link_health.load()["urls"]), + str(state.get("recovery_cursor") or ""), 1000) + selected = sorted(set(window) | set(recovery)) + print(f"Checking {len(selected)} of {len(urls)} artifact/reference links") + gone, live = link_health.survey_urls( + interleave_by_host(selected), workers=args.workers, timeout=args.timeout) + if not args.dry_run: + link_health.record(dead_urls=gone, live_urls=live) ref_links, ref_lines = prune_references(references, gone, dry_run=args.dry_run) state["reference_cursor"] = window[-1] if window else "" + state["recovery_cursor"] = recovery[-1] if recovery else "" print(f"Reference links returning 404 or 410: {len(gone)} | " f"removed {ref_links} links, {ref_lines} inventory lines") + prune_unavailable(dry_run=args.dry_run) + state["checked"] = datetime.now(timezone.utc).isoformat() save_state(state, dry_run=args.dry_run) print(f"KEV catalogue: {refresh_kev(dry_run=args.dry_run)} known-exploited CVEs") badged, unbadged = sync_kev_badges(dry_run=args.dry_run) print(f"KEV badges: {badged} written, {unbadged} withdrawn") - return 0 + return int(bool(deferred)) if __name__ == "__main__": diff --git a/scripts/build_site.py b/scripts/build_site.py index f060e8296d1..d83b5ff0dde 100644 --- a/scripts/build_site.py +++ b/scripts/build_site.py @@ -15,6 +15,7 @@ from typing import Collection, Dict, Iterable, List, Optional from urllib.parse import unquote, urlparse from update_cves import markdown_record_status +import link_health ROOT = Path(__file__).resolve().parents[1] CVES = ROOT / "cves" @@ -387,7 +388,7 @@ def collect_links(block: str, *, blacklist: Optional[Collection[str]] = None) -> continue except ValueError: continue - if is_blacklisted(entry, blacklist): + if is_blacklisted(entry, blacklist) or link_health.unavailable(entry): continue if entry not in links: links.append(entry) @@ -538,7 +539,8 @@ def build_metadata(cve_entries: List[Dict[str, object]]) -> Dict[str, object]: entry: Dict[str, object] = {} if value.get("cvss"): entry["cvss"] = value["cvss"] - advisories = dedupe_advisories(value.get("advisories", []), cve) + advisories = [row for row in dedupe_advisories(value.get("advisories", []), cve) + if not link_health.unavailable(row[0])] if advisories: entry["advisories"] = advisories if entry: @@ -593,6 +595,7 @@ def build_trending(blacklist: Collection[str]) -> List[Dict[str, object]]: return [ row for row in rows if row.get("url") and not is_blacklisted(str(row["url"]), blacklist) + and not link_health.unavailable(str(row["url"])) and (row.get("cve"), repo_from_url(str(row["url"]))) not in load_source_exclusions() ] diff --git a/scripts/link_health.py b/scripts/link_health.py new file mode 100644 index 00000000000..8b56146d788 --- /dev/null +++ b/scripts/link_health.py @@ -0,0 +1,125 @@ +"""Confirmed public-link failures shared by ingestion, audits and publication.""" + +from __future__ import annotations + +import json +import tempfile +from concurrent.futures import ThreadPoolExecutor +from datetime import datetime, timezone +from functools import lru_cache +from http.client import HTTPException +from pathlib import Path +from urllib import error, request +from urllib.parse import urlsplit, urlunsplit + +STATE = Path(__file__).resolve().parents[1] / "index" / "link_health.json" + + +def repository(url: str) -> str: + from build_site import GITHUB_NON_REPOS + try: + parsed = urlsplit(url) + except ValueError: + return "" + if parsed.hostname not in {"github.com", "www.github.com", "raw.githubusercontent.com"}: + return "" + parts = parsed.path.strip("/").split("/") + if len(parts) < 2 or parts[0].lower() in GITHUB_NON_REPOS: + return "" + return "/".join(parts[:2]).lower().removesuffix(".git") + + +def url_key(url: str) -> str: + try: + parsed = urlsplit(url) + except ValueError: + return url + host = parsed.netloc.lower() + if host == "www.github.com": + host = "github.com" + path = parsed.path + if host in {"github.com", "raw.githubusercontent.com"}: + parts = path.split("/") + parts[1:3] = [part.lower() for part in parts[1:3]] + path = "/".join(parts) + # A fragment never reaches the server; path case and query strings do. + return urlunsplit((parsed.scheme.lower(), host, path, parsed.query, "")) + + +@lru_cache(maxsize=1) +def load() -> dict: + try: + state = json.loads(STATE.read_text(encoding="utf-8")) + except FileNotFoundError: + return {"repositories": {}, "urls": {}} + if not isinstance(state, dict) or any( + not isinstance(state.get(field), dict) + or any(not isinstance(key, str) or not isinstance(stamp, str) + for key, stamp in state[field].items()) + for field in ("repositories", "urls") + ): + raise ValueError("Invalid link health state; refusing to forget confirmed dead links") + return state + + +def unavailable(url: str) -> bool: + state = load() + return repository(url) in state["repositories"] or url_key(url) in state["urls"] + + +def record(*, missing=(), healthy=(), dead_urls=(), live_urls=(), now: str | None = None) -> None: + state = {field: dict(values) for field, values in load().items()} + stamp = now or datetime.now(timezone.utc).isoformat() + for name in healthy: + state["repositories"].pop(name.lower(), None) + for name in missing: + state["repositories"][name.lower()] = stamp + for url in live_urls: + state["urls"].pop(url_key(url), None) + for url in dead_urls: + state["urls"][url_key(url)] = stamp + if state == load(): + return + STATE.parent.mkdir(parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=STATE.parent, delete=False) as handle: + json.dump(state, handle, sort_keys=True, indent=2) + handle.write("\n") + temporary = Path(handle.name) + try: + temporary.replace(STATE) + finally: + temporary.unlink(missing_ok=True) + load.cache_clear() + + +def check_url(url: str, timeout: int = 8) -> bool | None: + """Anonymous GET: True live, False confirmed 404/410, None inconclusive. + + Follow redirects and read only one byte. HEAD is not authoritative: several + hosts serve different routes or statuses for HEAD and GET. + """ + try: + parsed = urlsplit(url) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + return None + req = request.Request(url, headers={"User-Agent": "pocindex-link-audit", "Accept": "*/*"}) + with request.urlopen(req, timeout=timeout) as response: + response.read(1) + return True if 200 <= response.status < 300 else None + except error.HTTPError as exc: + code = exc.code + exc.close() + return False if code in {404, 410} else None + except (OSError, error.URLError, ValueError, HTTPException): + return None + + +def survey_urls(urls: list[str], *, workers: int = 8, timeout: int = 8) -> tuple[set[str], set[str]]: + dead, live = set(), set() + with ThreadPoolExecutor(max_workers=workers) as pool: + for url, result in zip(urls, pool.map(lambda url: check_url(url, timeout), urls)): + if result is False: + dead.add(url) + elif result is True: + live.add(url) + return dead, live diff --git a/scripts/sync_collections.py b/scripts/sync_collections.py index 2883fa2a1a5..a1113ea0dc0 100644 --- a/scripts/sync_collections.py +++ b/scripts/sync_collections.py @@ -293,6 +293,8 @@ def section_bounds(text: str) -> tuple[int, int, int] | None: def apply_links(cve: str, urls: list[str], *, dry_run: bool) -> str: + import link_health + urls = [url for url in urls if not link_health.unavailable(url)] path = CVES / cve.split("-")[1] / f"{cve}.md" if not path.exists(): return "absent" diff --git a/scripts/sync_exploits.py b/scripts/sync_exploits.py index 37dc0fed1b8..aeb105ba292 100644 --- a/scripts/sync_exploits.py +++ b/scripts/sync_exploits.py @@ -161,6 +161,8 @@ def section_bounds(text: str, header: str) -> tuple[int, int] | None: def apply_to_markdown(cve: str, header: str, urls: list[str], dry_run: bool) -> str: + import link_health + urls = [url for url in urls if not link_health.unavailable(url)] path = CVES / cve.split("-")[1] / f"{cve}.md" if not path.exists(): return "absent" diff --git a/scripts/sync_nuclei.py b/scripts/sync_nuclei.py index 7859d7a0f57..bd87c63bdff 100644 --- a/scripts/sync_nuclei.py +++ b/scripts/sync_nuclei.py @@ -167,6 +167,9 @@ def section_bounds(text: str, header: str) -> tuple[int, int] | None: def apply_to_markdown(cve: str, url: str, dry_run: bool) -> str: """Add the template link to one CVE entry. Returns what happened.""" + import link_health + if link_health.unavailable(url): + return "unchanged" path = CVES / cve.split("-")[1] / f"{cve}.md" if not path.exists(): return "absent" diff --git a/scripts/sync_repo_metadata.py b/scripts/sync_repo_metadata.py index 0b776c5714a..88e8800fd53 100644 --- a/scripts/sync_repo_metadata.py +++ b/scripts/sync_repo_metadata.py @@ -12,6 +12,7 @@ from datetime import datetime, timedelta, timezone from pathlib import Path from update_cves import GITHUB_GRAPHQL_URL, GitHubClient, http_json, repository_is_nonpublic +import link_health ROOT = Path(__file__).resolve().parents[1] METADATA = ROOT / "index" / "repo_meta.json" @@ -124,26 +125,44 @@ def scope(entries: list[dict], stored: dict, frontpage: list[dict], cursor: str, names = sorted(linked) rotating = [name for name in names if name > cursor] + [name for name in names if name <= cursor] rotating = rotating[:limit] if limit else rotating - selected = ((linked - stored.keys()) | (priority & linked) | set(rotating)) - (deferred or set()) + selected = ((linked - stored.keys()) | priority | set(rotating) + | set(link_health.load()["repositories"])) - (deferred or set()) + # A transient failure must not hide a front-page check for up to 24 hours. + selected |= priority return sorted(selected), rotating[-1] if rotating else cursor def main() -> int: from build_site import build_cve_list, build_landed, build_trending, load_blacklist + from audit_poc_links import is_repository_url, prune_unavailable, slice_after, interleave_by_host + from build_site import CURATED parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--limit", type=int, default=2000, help="Older repositories per run; 0 refreshes all") + parser.add_argument("--references", type=int, default=500, help="Older artifact URLs per run; 0 skips them") + parser.add_argument("--retry-deferred", action="store_true", help="Also recheck repositories in retry cooldown") args = parser.parse_args() - if args.limit < 0: - parser.error("--limit must not be negative") + if args.limit < 0 or args.references < 0: + parser.error("Limits must not be negative") entries, _ = build_cve_list(load_blacklist()) stored, state = load(METADATA), load(STATE) frontpage = build_trending(load_blacklist()) + build_landed(entries, stored, {}) now = datetime.now(timezone.utc) retries = state.get("retries", {}) deferred = {name for name, row in retries.items() if row.get("after", "") > now.isoformat()} + if args.retry_deferred: + deferred = set() names, cursor = scope(entries, stored, frontpage, state.get("cursor", ""), args.limit, deferred) client = GitHubClient(os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN") or "") + print(f"Checking public availability and metadata for {len(names)} repositories", flush=True) readings, unavailable, failed = survey(names, client) + # An API NOT_FOUND can reflect token permissions. Require the same public + # GET failure a visitor sees before withdrawing any existing public link. + public_urls = [f"https://github.com/{name}" for name in sorted(unavailable)] + confirmed, _ = link_health.survey_urls(public_urls) + missing = {link_health.repository(url) for url in confirmed} + failed |= unavailable - missing + unavailable = missing + link_health.record(missing=unavailable, healthy=readings) stored.update(readings) for name in unavailable: stored.pop(name, None) @@ -154,8 +173,24 @@ def main() -> int: hours = 6 if name in unavailable else min(24, 2 ** min(attempts, 5)) retries[name] = {"attempts": attempts, "after": (now + timedelta(hours=hours)).isoformat()} save(METADATA, stored) + reference_cursor = state.get("reference_cursor", "") + if args.references: + urls = sorted({url for entry in entries for field in ["poc", *(f for _, f in CURATED)] + for url in entry.get(field, []) if not is_repository_url(url) + and not link_health.unavailable(url)}) + rotating = slice_after(urls, reference_cursor, args.references) + priority = {row["url"] for row in frontpage if row.get("url") + and not is_repository_url(row["url"]) and not link_health.unavailable(row["url"])} + window = interleave_by_host(sorted(set(rotating) | priority)) + dead, live = link_health.survey_urls(window, workers=16, timeout=5) + link_health.record(dead_urls=dead, live_urls=live) + reference_cursor = rotating[-1] if rotating else reference_cursor + print(f"Artifact URLs: {len(window)} checked, {len(dead)} confirmed dead, " + f"{len(window) - len(dead) - len(live)} inconclusive") + prune_unavailable() # Healthy work keeps moving even when one repository persistently fails. - save(STATE, {"cursor": cursor, "checked": now.isoformat(), "retries": retries}) + save(STATE, {"cursor": cursor, "reference_cursor": reference_cursor, + "checked": now.isoformat(), "retries": retries}) print(f"Repository metadata: {len(names)} checked, {len(readings)} refreshed, " f"{len(unavailable)} unavailable, {len(failed)} deferred") return int(bool(failed)) diff --git a/scripts/update_cves.py b/scripts/update_cves.py index b8e2495158d..011a4f88150 100755 --- a/scripts/update_cves.py +++ b/scripts/update_cves.py @@ -148,11 +148,12 @@ def poc_link_key(url: str) -> str: def stable_unique_poc_links(values: Iterable[str]) -> list[str]: + import link_health seen: set[str] = set() result: list[str] = [] for value in values: key = poc_link_key(value) - if value and key not in seen: + if value and key not in seen and not link_health.unavailable(value): seen.add(key) result.append(value) return result @@ -1080,30 +1081,8 @@ def poc_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 + import link_health + return link_health.check_url(url, timeout) is not False def filter_live_references( @@ -1725,6 +1704,8 @@ def build_markdown( references: list[str], kev: dict[str, list] | None = None, ) -> str: + github_links = stable_unique_poc_links(github_links) + references = stable_unique_poc_links(references) description = "\n".join( line.expandtabs(4).rstrip() for line in details.description.strip().splitlines() ) @@ -1888,12 +1869,13 @@ def sync_markdown( def append_inventory(path: Path, mappings: dict[str, list[str]], *, dry_run: bool) -> int: + import link_health 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 url and f"{cve_id} - {url}" not in existing and not link_health.unavailable(url) ) if not additions or dry_run: return len(additions) @@ -1913,6 +1895,7 @@ def reconcile_inventory( dry_run: bool, ) -> tuple[int, int]: """Make an inventory an exact view of its markdown section.""" + import link_health expected = { f"{md_path.stem} - {url}" for md_path in CVES.glob("[12][0-9][0-9][0-9]/CVE-*.md") @@ -1924,6 +1907,7 @@ def reconcile_inventory( for url in urls if url ) + expected = {line for line in expected if not link_health.unavailable(line.split(" - ", 1)[1])} current_lines = path.read_text(encoding="utf-8").splitlines() if path.exists() else [] current = set(current_lines) additions = len(expected - current) diff --git a/tests/test_audit_poc_links.py b/tests/test_audit_poc_links.py index 0cca3c99b46..56fb5be252d 100644 --- a/tests/test_audit_poc_links.py +++ b/tests/test_audit_poc_links.py @@ -1,17 +1,17 @@ from __future__ import annotations -import io import json import sys import tempfile import unittest from pathlib import Path -from unittest.mock import patch +from unittest.mock import patch, MagicMock from urllib.error import HTTPError sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) import audit_poc_links +import link_health from update_cves import GitHubClient @@ -33,6 +33,8 @@ class RepositorySurveyTests(unittest.TestCase): def repo(self, filename: str = "README.md") -> dict: return { "nameWithOwner": "owner/poc", + "isPrivate": False, + "visibility": "PUBLIC", "stargazerCount": 10, "pushedAt": "2026-10-06T12:00:00Z", "root": {"entries": [{"name": filename, "type": "blob"}]}, @@ -40,8 +42,10 @@ class RepositorySurveyTests(unittest.TestCase): } def survey(self, payload: dict, names: list[str] | None = None) -> tuple: - with patch.object(audit_poc_links, "http_json", return_value=payload): - return audit_poc_links.survey(GitHubClient("test-token"), names or ["owner/poc"]) + with patch.object(audit_poc_links, "http_json", return_value=payload), \ + patch.object(link_health, "check_url", return_value=False): + missing, metadata, _ = audit_poc_links.survey(GitHubClient("test-token"), names or ["owner/poc"]) + return missing, metadata def test_preserves_repository_when_readme_lookup_fails(self) -> None: self.assertEqual(self.survey({ @@ -78,27 +82,26 @@ class RepositorySurveyTests(unittest.TestCase): class ReferenceAuditTests(unittest.TestCase): - def test_head_not_found_keeps_reference_when_get_succeeds(self) -> None: + def test_checks_get_instead_of_trusting_head_routes(self) -> None: url = "https://example.com/poc" - for code in (404, 410): - with self.subTest(code=code), patch.object( - audit_poc_links.request, - "urlopen", - side_effect=[HTTPError(url, code, "No HEAD route", {}, None), io.BytesIO(b"PoC")], - ) as fetch: - self.assertEqual(audit_poc_links.dead_references([url], workers=1, timeout=1), set()) - self.assertEqual([call.args[0].method for call in fetch.call_args_list], ["HEAD", "GET"]) + response = MagicMock() + response.__enter__.return_value.status = 206 + with patch.object(link_health.request, "urlopen", return_value=response) as fetch: + self.assertEqual(audit_poc_links.dead_references([url], workers=1, timeout=1), set()) + self.assertEqual(fetch.call_args.args[0].get_method(), "GET") + self.assertNotIn("Authorization", fetch.call_args.args[0].headers) + response.__enter__.return_value.read.assert_called_once_with(1) def test_get_confirms_reference_is_missing(self) -> None: url = "https://example.com/poc" for code in (404, 410): with self.subTest(code=code), patch.object( - audit_poc_links.request, + link_health.request, "urlopen", - side_effect=[HTTPError(url, code, "Missing", {}, None), HTTPError(url, code, "Missing", {}, None)], + side_effect=HTTPError(url, code, "Missing", {}, None), ) as fetch: self.assertEqual(audit_poc_links.dead_references([url], workers=1, timeout=1), {url}) - self.assertEqual([call.args[0].method for call in fetch.call_args_list], ["HEAD", "GET"]) + self.assertEqual([call.args[0].get_method() for call in fetch.call_args_list], ["GET"]) class KevRefreshTests(unittest.TestCase): diff --git a/tests/test_link_health.py b/tests/test_link_health.py new file mode 100644 index 00000000000..7ceed18a40e --- /dev/null +++ b/tests/test_link_health.py @@ -0,0 +1,213 @@ +from __future__ import annotations + +import json +import sys +import tempfile +import unittest +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from threading import Thread +from unittest.mock import patch +from urllib.error import HTTPError, URLError + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) +import audit_poc_links as audit +import build_site +import link_health as health +import sync_repo_metadata as metadata +import update_cves + + +class LinkHealthTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.root = Path(self.directory.name) + self.index = self.root / 'index' + self.index.mkdir() + self.cves = self.root / 'cves' + (self.cves / '2026').mkdir(parents=True) + self.state = self.index / 'link_health.json' + self.patch = patch.object(health, 'STATE', self.state) + self.patch.start() + self.addCleanup(self.patch.stop) + health.load.cache_clear() + self.addCleanup(health.load.cache_clear) + + def test_repository_failures_cover_aliases_children_and_raw_files(self): + health.record(missing={'owner/gone'}) + for url in ('https://github.com/Owner/Gone', 'https://www.github.com/Owner/Gone.git/', + 'https://github.com/Owner/Gone/blob/main/poc.py', + 'https://raw.githubusercontent.com/Owner/Gone/main/poc.py', + 'https://github.com/Owner/Gone/security/advisories/GHSA-abcd'): + self.assertTrue(health.unavailable(url), url) + self.assertFalse(health.unavailable('https://github.com/owner/working')) + self.assertFalse(health.unavailable('https://github.com/owner/gone-v2')) + self.assertFalse(health.unavailable('https://notgithub.com/owner/gone')) + + def test_artifact_failure_never_hides_siblings_or_discards_case_and_query(self): + url = 'https://github.com/Owner/Repo/blob/main/PoC.py?raw=1' + health.record(dead_urls={url}) + self.assertTrue(health.unavailable(url + '#L10')) + self.assertFalse(health.unavailable(url.replace('PoC.py', 'poc.py'))) + self.assertFalse(health.unavailable(url.replace('?raw=1', '?raw=2'))) + self.assertFalse(health.unavailable('https://github.com/Owner/Repo')) + health.record(missing={'owner/repo'}) + health.record(healthy={'owner/repo'}) + self.assertTrue(health.unavailable(url)) + health.record(live_urls={url}) + self.assertFalse(health.unavailable(url)) + + def test_transient_http_and_network_errors_are_not_evidence_of_deletion(self): + url = 'https://example.com/poc' + failures = [HTTPError(url, code, 'Unavailable', {}, None) + for code in (401, 403, 405, 416, 429, 500, 502, 503)] + failures += [URLError('temporary DNS failure'), TimeoutError('timeout')] + for failure in failures: + with self.subTest(failure=str(failure)), patch.object(health.request, 'urlopen', side_effect=failure): + self.assertEqual(health.survey_urls([url]), (set(), set())) + + def test_api_not_found_without_public_404_does_not_prune(self): + payload = {'data': {'r0': None}, 'errors': [{'type': 'NOT_FOUND', 'path': ['r0']}]} + for public_result in (True, None): + with patch.object(audit, 'http_json', return_value=payload), \ + patch.object(health, 'check_url', return_value=public_result): + self.assertEqual(audit.survey(update_cves.GitHubClient('test'), ['owner/poc']), (set(), {}, set())) + + def test_private_repository_visible_to_token_cannot_clear_suppression(self): + health.record(missing={'owner/private'}) + payload = {'data': {'r0': {'isPrivate': True, 'visibility': 'PRIVATE', + 'stargazerCount': 100, 'pushedAt': '2026-10-22T00:00:00Z'}}} + with patch.object(audit, 'http_json', return_value=payload), \ + patch.object(health, 'check_url', return_value=False): + missing, readings, unavailable = audit.survey(update_cves.GitHubClient('test'), ['owner/private']) + self.assertEqual((missing, readings, unavailable), ({'owner/private'}, {}, {'owner/private'})) + health.record(missing=unavailable, healthy=readings) + self.assertTrue(health.unavailable('https://github.com/owner/private')) + + def test_invalid_urls_are_inconclusive_without_a_network_request(self): + with patch.object(health.request, 'urlopen') as fetch: + for url in ('https://[broken', 'file:///tmp/poc', '', 'https://'): + self.assertIsNone(health.check_url(url)) + fetch.assert_not_called() + + def test_redirects_follow_the_destination_without_losing_the_original_identity(self): + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + if self.path == '/renamed': + self.send_response(302) + self.send_header('Location', '/live') + else: + self.send_response(200 if self.path == '/live' else 404) + self.end_headers() + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + base = f'http://127.0.0.1:{server.server_port}' + self.assertEqual(health.survey_urls([base+'/renamed', base+'/gone']), + ({base+'/gone'}, {base+'/renamed'})) + finally: + server.shutdown() + server.server_close() + thread.join() + + def test_publication_hides_dead_links_in_all_fields_but_preserves_cve_enrichment(self): + dead = 'https://github.com/owner/gone' + live = 'https://github.com/owner/live' + health.record(missing={'owner/gone'}) + text = '### Description\nExample vulnerability\n\n### POC\n' + for header in audit.SECTION_EMPTY: + text += f'\n{header}\n- {dead}\n- {live}-{header.split()[-1]}/blob/main/poc.py\n' + cve = 'CVE-2026-1234' + (self.cves / f'2026/{cve}.md').write_text(text) + stored = {cve: {'cvss': {'v3': [9.8, 'CRITICAL']}, 'advisories': [ + [dead, ['Exploit']], ['https://vendor.example/advisory', ['Vendor Advisory']]]}} + with patch.multiple(build_site, CVES=self.cves, load_dates=lambda: {}, + load_metadata=lambda: stored, load_verified_references=lambda: set()): + entries, total = build_site.build_cve_list(set()) + enriched = build_site.build_metadata(entries) + self.assertEqual((len(entries), total), (1, 1)) + for field in ['poc', *(f for _, f in build_site.CURATED)]: + self.assertTrue(entries[0][field]) + self.assertTrue(all(url.startswith(live) for url in entries[0][field])) + self.assertEqual(enriched[cve]['cvss'], stored[cve]['cvss']) + self.assertEqual(enriched[cve]['advisories'], [['https://vendor.example/advisory', ['Vendor Advisory']]]) + + def test_all_sources_prune_and_stale_imports_cannot_republish(self): + dead_repo = 'https://github.com/owner/gone' + dead_file = 'https://github.com/owner/collection/blob/main/removed.py' + live_file = 'https://github.com/owner/collection/blob/main/live.py' + health.record(missing={'owner/gone'}, dead_urls={dead_file}) + text = '### Description\nExample\n\n### POC\n' + for header in audit.SECTION_EMPTY: + text += f'\n{header}\n- {dead_repo}\n- {dead_file}\n- {live_file}\n' + path = self.cves / '2026/CVE-2026-1234.md' + path.write_text(text) + for name in ('github.txt', 'references.txt', 'reference_pocs.txt'): + (self.index / name).write_text(f'CVE-2026-1234 - {dead_repo}\nCVE-2026-1234 - {live_file}\n') + (self.index / 'trending.json').write_text(json.dumps([{'url': dead_repo}, {'url': live_file}])) + readme = self.root / 'README.md' + readme.write_text(f'| [dead]({dead_repo}) |\n| [live]({live_file}) |\n') + with patch.multiple(audit, ROOT=self.root, CVES=self.cves, INDEX=self.index, + GITHUB_LIST=self.index/'github.txt', REFERENCE_LIST=self.index/'references.txt'): + self.assertEqual(audit.prune_unavailable(dry_run=True), (1, 14)) + self.assertEqual(path.read_text(), text) + self.assertEqual(audit.prune_unavailable(), (1, 14)) + for header in audit.SECTION_EMPTY: + self.assertEqual(update_cves.section_links(path.read_text(), header), [live_file]) + restored, _ = update_cves.replace_section(path.read_text(), header, [dead_file, live_file], 'No PoCs') + self.assertNotIn(dead_file, restored) + for name in ('github.txt', 'references.txt', 'reference_pocs.txt', 'trending.json'): + self.assertNotIn(dead_repo, (self.index / name).read_text()) + self.assertNotIn(dead_repo, readme.read_text()) + self.assertEqual(build_site.collect_links(text), [live_file]) + with patch.object(build_site, 'TRENDING_INPUT', self.index/'trending.json'): + self.assertEqual([r['url'] for r in build_site.build_trending(set())], [live_file]) + + def test_exact_url_scope_includes_curated_and_github_files(self): + path = self.cves / '2026/CVE-2026-1234.md' + root = 'https://github.com/owner/collection' + text = f'#### Github\n- {root}\n- {root}/blob/main/poc.py\n' + text += f'#### Collections\n- {root}/blob/main/another.py\n' + path.write_text(text) + with patch.object(audit, 'CVES', self.cves): + self.assertEqual(set(audit.collect_references()), + {root+'/blob/main/poc.py', root+'/blob/main/another.py'}) + + def test_hourly_refresh_withdraws_confirmed_dead_repo_and_schedules_recovery(self): + url = 'https://github.com/owner/gone' + entries = [{'poc': [url, 'https://github.com/owner/flaky']}] + cache, state = self.index/'repo_meta.json', self.root/'metadata_state.json' + cache.write_text('{"owner/gone":[10,"2026-10-21"],"owner/flaky":[0,"2026-10-21"]}') + with patch.multiple(metadata, METADATA=cache, STATE=state), \ + patch.multiple(build_site, build_cve_list=lambda _: (entries, 1), + build_trending=lambda _: [], build_landed=lambda *a: []), \ + patch.object(metadata, 'survey', return_value=({}, {'owner/gone'}, {'owner/flaky'})), \ + patch.object(health, 'check_url', return_value=False), \ + patch.dict(metadata.os.environ, {'GITHUB_TOKEN': 'test'}), \ + patch.object(audit, 'prune_unavailable') as prune, \ + patch.object(sys, 'argv', ['sync_repo_metadata', '--references', '0']): + self.assertEqual(metadata.main(), 1) + prune.assert_called_once() + self.assertTrue(health.unavailable(url)) + self.assertEqual(json.loads(cache.read_text()), {'owner/flaky': [0, '2026-10-21']}) + self.assertIn('owner/gone', json.loads(state.read_text())['retries']) + # Still eligible after removal from the live index, unless cooling down. + self.assertEqual(metadata.scope([], {}, [], '', 10)[0], ['owner/gone']) + self.assertEqual(metadata.scope([], {}, [], '', 10, {'owner/gone'})[0], []) + self.assertEqual(metadata.scope([], {}, [{'url': 'https://github.com/owner/flaky'}], '', 10, + {'owner/gone', 'owner/flaky'})[0], ['owner/flaky']) + + def test_corrupt_state_fails_instead_of_republishing_everything(self): + self.state.write_text('{"repositories":[]}') + with self.assertRaises(ValueError): + health.unavailable('https://github.com/owner/poc') + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_repo_metadata.py b/tests/test_repo_metadata.py index 1ee1a01534b..7c466ffe2f6 100644 --- a/tests/test_repo_metadata.py +++ b/tests/test_repo_metadata.py @@ -14,6 +14,11 @@ from update_cves import GitHubClient class RepositoryMetadataTests(unittest.TestCase): + def setUp(self): + health = patch('link_health.load', return_value={'repositories': {}, 'urls': {}}) + health.start() + self.addCleanup(health.stop) + def test_priority_covers_missing_curated_and_frontpage_repositories(self): entries = [{"poc": ["https://github.com/owner/older", "https://github.com/owner/front"], "collections": ["https://github.com/collection/new/blob/main/poc.py"], @@ -82,7 +87,8 @@ class RepositoryMetadataTests(unittest.TestCase): build_trending=lambda blocked: [], build_landed=lambda *args: []), \ patch.object(metadata, 'survey', return_value=({}, set(), {'owner/a'})), \ patch.dict(metadata.os.environ, {'GITHUB_TOKEN': 'test'}), \ - patch.object(sys, 'argv', ['sync_repo_metadata', '--limit', '1']): + patch.object(sys, 'argv', ['sync_repo_metadata', '--limit', '1', '--references', '0']), \ + patch('audit_poc_links.prune_unavailable'): self.assertEqual(metadata.main(), 1) saved = json.loads(state.read_text()) self.assertEqual(saved['cursor'], 'owner/a')