mirror of
https://github.com/0xMarcio/cve.git
synced 2026-10-23 16:35:19 +00:00
562 lines
23 KiB
Python
562 lines
23 KiB
Python
#!/usr/bin/env python3
|
|
"""Prune PoC links to deleted repositories and refresh the metadata the site shows."""
|
|
|
|
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, datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
from urllib.parse import urlsplit
|
|
|
|
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,
|
|
GitHubClient,
|
|
http_json,
|
|
is_blacklisted_repo,
|
|
is_valid_cve,
|
|
load_blacklist,
|
|
replace_section,
|
|
section_links,
|
|
)
|
|
import link_health
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
CVES = ROOT / "cves"
|
|
INDEX = ROOT / "index"
|
|
STATE_FILE = ROOT / ".github" / "link_audit_state.json"
|
|
GITHUB_LIST = INDEX / "github.txt"
|
|
REFERENCE_LIST = INDEX / "references.txt"
|
|
REPO_META = INDEX / "repo_meta.json"
|
|
KEV_FILE = INDEX / "kev.json"
|
|
KEV_URL = (
|
|
"https://www.cisa.gov/sites/default/files/feeds/"
|
|
"known_exploited_vulnerabilities.json"
|
|
)
|
|
GITHUB_SECTION = "#### Github"
|
|
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
|
|
|
|
|
|
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.
|
|
|
|
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()
|
|
from build_site import CURATED, repo_from_url
|
|
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 url in section_links(text, GITHUB_SECTION):
|
|
full_name = repo_from_url(url)
|
|
if full_name and "/" in full_name:
|
|
prunable.setdefault(full_name, []).append((path, url))
|
|
referenced.add(full_name)
|
|
for url in section_links(text, REFERENCE_SECTION):
|
|
full_name = repo_from_url(url)
|
|
if full_name and "/" in full_name:
|
|
referenced.add(full_name)
|
|
for section, _ in CURATED:
|
|
referenced.update(name for url in section_links(text, section)
|
|
if (name := repo_from_url(url)))
|
|
return prunable, referenced
|
|
|
|
|
|
REPO_FIELDS = (
|
|
"{ 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 } } '
|
|
'readmeBare: object(expression: "HEAD:README") { ... on Blob { text } } '
|
|
'root: object(expression: "HEAD:") { ... on Tree { entries { name type } } } }'
|
|
)
|
|
|
|
|
|
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,
|
|
no directory, no artifact and a README with nothing runnable in it, which
|
|
the discovery gate would reject today. Anything else, a network error or a
|
|
throttle or an unexpected response, leaves the repository out of both
|
|
results, 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()
|
|
unavailable: set[str] = set()
|
|
metadata: dict[str, list] = {}
|
|
checked = 0
|
|
|
|
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])}) "
|
|
f"{REPO_FIELDS}"
|
|
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()
|
|
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.
|
|
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, 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, 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, unavailable
|
|
|
|
|
|
def save_metadata(metadata: dict[str, list], alive: set[str], *, dry_run: bool) -> int:
|
|
"""Merge this run's readings over the stored ones and forget deleted repositories."""
|
|
try:
|
|
stored = json.loads(REPO_META.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
stored = {}
|
|
stored.update(metadata)
|
|
stored = {name: value for name, value in stored.items() if name.lower() in alive}
|
|
if not dry_run:
|
|
REPO_META.write_text(
|
|
json.dumps(stored, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
return len(stored)
|
|
|
|
|
|
def prune_entries(
|
|
references: dict[str, list[tuple[Path, str]]],
|
|
missing: Iterable[str],
|
|
*,
|
|
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:
|
|
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)
|
|
removals = sum(len(urls) for urls in dead_by_path.values())
|
|
return len(dead_by_path), removals, dead_urls
|
|
|
|
|
|
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 collect_references() -> dict[str, list[tuple[Path, str]]]:
|
|
"""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 SECTION_EMPTY:
|
|
for url in section_links(text, header):
|
|
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:
|
|
try:
|
|
host = urlsplit(url).netloc.lower()
|
|
except ValueError:
|
|
host = ""
|
|
hosts.setdefault(host, []).append(url)
|
|
queues = list(hosts.values())
|
|
ordered: list[str] = []
|
|
while queues:
|
|
for queue in list(queues):
|
|
ordered.append(queue.pop(0))
|
|
if not queue:
|
|
queues.remove(queue)
|
|
return ordered
|
|
|
|
|
|
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."""
|
|
return link_health.survey_urls(urls, workers=workers, timeout=timeout)[0]
|
|
|
|
|
|
def prune_references(
|
|
references: dict[str, list[tuple[Path, str]]],
|
|
dead: set[str],
|
|
*,
|
|
dry_run: bool,
|
|
) -> tuple[int, int]:
|
|
by_path: dict[Path, dict[str, set[str]]] = {}
|
|
for url in dead:
|
|
for path, header in references.get(url, []):
|
|
by_path.setdefault(path, {}).setdefault(header, set()).add(url)
|
|
|
|
removed = 0
|
|
for path, headers in sorted(by_path.items()):
|
|
with path.open("r", encoding="utf-8", newline="") as handle:
|
|
text = path_text = handle.read()
|
|
for header, urls in headers.items():
|
|
empty = SECTION_EMPTY[header]
|
|
kept = [url for url in section_links(text, header) if url not in urls]
|
|
removed += len(section_links(text, header)) - len(kept)
|
|
text, _ = replace_section(text, header, kept, empty)
|
|
if text != path_text and not dry_run:
|
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
|
handle.write(text)
|
|
|
|
lines = REFERENCE_LIST.read_text(encoding="utf-8").splitlines() if REFERENCE_LIST.exists() else []
|
|
kept_lines = [line for line in lines if line.rsplit(" - ", 1)[-1].strip() not in dead]
|
|
if not dry_run and len(kept_lines) != len(lines):
|
|
REFERENCE_LIST.write_text("\n".join(kept_lines) + "\n", encoding="utf-8")
|
|
return removed, len(lines) - len(kept_lines)
|
|
|
|
|
|
def refresh_kev(*, dry_run: bool) -> int:
|
|
"""Store CISA's known-exploited catalogue: the CVEs attackers actually use.
|
|
|
|
Keeps only what the page shows, so the whole catalogue costs a few tens of
|
|
kilobytes. A failure leaves the previous file in place rather than
|
|
dropping the badge off every entry.
|
|
"""
|
|
try:
|
|
payload = http_json(KEV_URL)
|
|
if not isinstance(payload, dict) or not isinstance(payload.get("vulnerabilities"), list):
|
|
raise ValueError("Missing KEV vulnerability list")
|
|
vulnerabilities = payload["vulnerabilities"]
|
|
if type(payload.get("count")) is not int or payload["count"] != len(vulnerabilities):
|
|
raise ValueError("KEV count does not match its vulnerability list")
|
|
if not vulnerabilities:
|
|
raise ValueError("Empty KEV catalogue")
|
|
entries = {}
|
|
for item in vulnerabilities:
|
|
if not isinstance(item, dict):
|
|
raise ValueError("Invalid KEV entry")
|
|
cve_id = str(item.get("cveID") or "").upper()
|
|
if not is_valid_cve(cve_id) or cve_id in entries:
|
|
raise ValueError("Invalid or duplicate KEV CVE ID")
|
|
added = str(item.get("dateAdded") or "")
|
|
if date.fromisoformat(added).isoformat() != added:
|
|
raise ValueError("Invalid KEV dateAdded")
|
|
ransomware = str(item.get("knownRansomwareCampaignUse") or "Unknown").lower()
|
|
if ransomware not in {"known", "unknown"}:
|
|
raise ValueError("Invalid KEV ransomware status")
|
|
entries[cve_id] = [added, 1 if ransomware == "known" else 0]
|
|
except Exception as exc:
|
|
print(f"Kept the stored KEV catalogue after a failed fetch: {exc}", file=sys.stderr)
|
|
return 0
|
|
if not dry_run:
|
|
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=KEV_FILE.parent, delete=False) as handle:
|
|
handle.write(json.dumps(entries, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n")
|
|
temporary = Path(handle.name)
|
|
try:
|
|
temporary.replace(KEV_FILE)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
return len(entries)
|
|
|
|
|
|
KEV_MARKER = "label=CISA%20KEV"
|
|
|
|
|
|
def sync_kev_badges(*, dry_run: bool) -> tuple[int, int]:
|
|
"""Put the known-exploited badge on the entries CISA lists, and only those.
|
|
|
|
The site shows KEV in a search result, but the repository is what most
|
|
people read, and nothing in the markdown said a vulnerability was under
|
|
active attack.
|
|
"""
|
|
try:
|
|
catalogue = json.loads(KEV_FILE.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return 0, 0
|
|
|
|
added = removed = 0
|
|
for path in sorted(CVES.glob("[12][0-9][0-9][0-9]/CVE-*.md")):
|
|
cve_id = path.stem.upper()
|
|
wanted = kev_badge(cve_id, catalogue)
|
|
with path.open("r", encoding="utf-8", newline="") as handle:
|
|
text = handle.read()
|
|
has = KEV_MARKER in text
|
|
if not wanted and not has:
|
|
continue
|
|
|
|
newline = "\r\n" if "\r\n" in text else "\n"
|
|
lines = text.split(newline)
|
|
lines = [line for line in lines if KEV_MARKER not in line]
|
|
if wanted:
|
|
anchor = next((i for i, line in enumerate(lines) if line.strip() == "### Description"), None)
|
|
if anchor is None:
|
|
continue
|
|
insert = anchor
|
|
while insert > 0 and not lines[insert - 1].strip():
|
|
insert -= 1
|
|
lines.insert(insert, wanted)
|
|
|
|
updated = newline.join(lines)
|
|
if updated == text:
|
|
continue
|
|
added += 1 if wanted else 0
|
|
removed += 1 if has and not wanted else 0
|
|
if not dry_run:
|
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
|
handle.write(updated)
|
|
return added, removed
|
|
|
|
|
|
def load_state() -> dict:
|
|
try:
|
|
return json.loads(STATE_FILE.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def save_state(state: dict, *, dry_run: bool) -> None:
|
|
if dry_run:
|
|
return
|
|
STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
STATE_FILE.write_text(json.dumps(state, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
|
|
|
|
def slice_after(items: list[str], cursor: str, limit: int) -> list[str]:
|
|
"""The next `limit` items after the cursor, wrapping once at the end.
|
|
|
|
A limit of zero, or one larger than the list, means everything -- capped so
|
|
that wrapping can never hand back the same item twice.
|
|
"""
|
|
limit = len(items) if not limit else min(limit, len(items))
|
|
start = next((i for i, item in enumerate(items) if item > cursor), 0) if cursor else 0
|
|
window = items[start : start + limit]
|
|
if len(window) < limit:
|
|
window += items[: limit - len(window)]
|
|
return window
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="Drop PoC links to deleted repositories and refresh repo_meta.json"
|
|
)
|
|
parser.add_argument("--limit", type=int, default=0, help="Repositories to check this run")
|
|
parser.add_argument("--references", type=int, default=0,
|
|
help="Reference links to check this run (0 skips them, -1 checks all)")
|
|
parser.add_argument("--workers", type=int, default=16, help="Concurrent reference requests")
|
|
parser.add_argument("--timeout", type=int, default=8, help="Seconds per reference request")
|
|
parser.add_argument("--kev-only", action="store_true", help="Refresh CISA KEV without auditing links")
|
|
parser.add_argument("--dry-run", action="store_true", help="Report without writing files")
|
|
args = parser.parse_args()
|
|
|
|
if args.kev_only:
|
|
print(f"KEV catalogue: {refresh_kev(dry_run=args.dry_run)} known-exploited CVEs")
|
|
return 0
|
|
|
|
references, referenced = collect_repositories()
|
|
names = sorted(referenced)
|
|
print(
|
|
f"{sum(len(v) for v in references.values())} prunable links | "
|
|
f"{len(names)} repositories to survey"
|
|
)
|
|
|
|
state = load_state()
|
|
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 ""
|
|
blocked = {
|
|
name for name in references
|
|
if is_blacklisted_repo(name, load_blacklist())
|
|
}
|
|
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"] = 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")
|
|
|
|
if args.references:
|
|
references = collect_references()
|
|
urls = sorted(references)
|
|
window = slice_after(urls, str(state.get("reference_cursor") or ""),
|
|
0 if args.references < 0 else args.references)
|
|
# 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 int(bool(deferred))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|