Clean and verify CVE PoC sources

scripts/audit_poc_links.py

@@ -0,0 +1,630 @@
+#!/usr/bin/env python3
+"""Locally validate and deduplicate CVE PoC links."""
+
+from __future__ import annotations
+
+import argparse
+import bisect
+import hashlib
+import json
+import re
+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
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+import update_cves
+
+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"
+
+
+@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"):
+        text = path.read_text(encoding="utf-8", errors="replace")
+        for header in SECTION_EMPTY_TEXT:
+            links.update(section_links(text, header))
+    return links
+
+
+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 rotation_key(value: str) -> str:
+    return f"{hashlib.sha256(value.encode('utf-8')).hexdigest()}:{value}"
+
+
+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]
+            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
+
+
+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],
+    *,
+    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"):
+        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:
+            with path.open("w", encoding="utf-8", newline="") as handle:
+                handle.write(updated)
+    return stats
+
+
+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 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]:
+    try:
+        state = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    return state if isinstance(state, dict) else {}
+
+
+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 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)}")
+
+    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"
+        )
+
+    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"
+        )
+
+    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)}")
+
+    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
+
+    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")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/build_site.py

@@ -5,7 +5,7 @@ import json
 import re
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Collection, Dict, List, Optional
 from urllib.parse import urlparse
 
 from jinja2 import Environment, FileSystemLoader, select_autoescape
@@ -15,7 +15,6 @@ from utils import DOCS_DIR, TEMPLATES_DIR, ensure_dirs, load_blacklist, parse_tr
 ROOT = DOCS_DIR.parent
 README_PATH = ROOT / "README.md"
 CVE_OUTPUT = DOCS_DIR / "CVE_list.json"
-REMOVED_OUTPUT = DOCS_DIR / "CVE_blacklist_removed.json"
 TRENDING_OUTPUT = DOCS_DIR / "trending_poc.json"
 
 
@@ -72,31 +71,32 @@ def repo_from_url(url: str) -> str:
         path = url
     parts = path.strip("/").split("/")
     if len(parts) >= 2:
-        return parts[1].lower()
+        return "/".join(parts[:2]).lower()
     return (parts[-1] if parts else "").lower()
 
 
-def is_blacklisted(url: str, blacklist: List[str]) -> bool:
-    repo = repo_from_url(url)
-    if not repo:
+def is_blacklisted(url: str, blacklist: Collection[str]) -> bool:
+    full_name = repo_from_url(url)
+    if not full_name:
         return False
-    for entry in blacklist:
-        slug = entry.lower()
-        if not slug:
-            continue
-        if slug.endswith("*"):
-            if repo.startswith(slug[:-1]):
-                return True
-        elif repo == slug:
-            return True
-    return False
+    return full_name in blacklist
 
 
-def collect_links(block: str, *, blacklist: Optional[List[str]] = None, removed: Optional[List[str]] = None) -> List[str]:
+def link_key(url: str) -> str:
+    parsed = urlparse(url)
+    if (parsed.hostname or "").lower() not in {"github.com", "www.github.com"}:
+        return url
+    parts = [part for part in parsed.path.split("/") if part]
+    if len(parts) < 2:
+        return url
+    repo = f"{parts[0]}/{parts[1]}".lower()
+    suffix = "/" + "/".join(parts[2:]) if len(parts) > 2 else ""
+    return f"github:{repo}{suffix}?{parsed.query}#{parsed.fragment}"
+
+
+def collect_links(block: str, *, blacklist: Optional[Collection[str]] = None) -> List[str]:
     links: List[str] = []
-    blacklist = blacklist or []
-    if removed is None:
-        removed = []
+    blacklist = blacklist or set()
     for raw in block.splitlines():
         entry = raw.strip()
         if not entry or "No PoCs" in entry:
@@ -106,38 +106,31 @@ def collect_links(block: str, *, blacklist: Optional[List[str]] = None, removed:
         if not entry:
             continue
         if is_blacklisted(entry, blacklist):
-            removed.append(entry)
             continue
         if entry not in links:
             links.append(entry)
     return links
 
 
-def build_cve_list(blacklist: List[str]) -> Dict[str, object]:
+def build_cve_list(blacklist: Collection[str]) -> List[Dict[str, object]]:
     cve_entries = []
-    removed_by_cve: Dict[str, List[str]] = {}
-    removed_seen: set[str] = set()
 
     for md_path in sorted(ROOT.glob("[12][0-9][0-9][0-9]/CVE-*.md")):
         content = md_path.read_text(encoding="utf-8")
         sections = parse_sections(content)
         description = normalise_block(sections.get("### Description", ""))
-        removed_links: List[str] = []
-        references = collect_links(sections.get("#### Reference", ""), blacklist=blacklist, removed=removed_links)
-        github_links = collect_links(sections.get("#### Github", ""), blacklist=blacklist, removed=removed_links)
+        references = collect_links(sections.get("#### Reference", ""), blacklist=blacklist)
+        github_links = collect_links(sections.get("#### Github", ""), blacklist=blacklist)
 
         poc_entries: List[str] = []
         seen = set()
         for link in references + github_links:
-            if link not in seen:
+            key = link_key(link)
+            if key not in seen:
                 poc_entries.append(link)
-                seen.add(link)
+                seen.add(key)
 
         cve_id = md_path.stem
-        if removed_links:
-            removed_by_cve[cve_id] = sorted(set(removed_links))
-            removed_seen.update(removed_links)
-
         if not poc_entries:
             continue
 
@@ -147,16 +140,10 @@ def build_cve_list(blacklist: List[str]) -> Dict[str, object]:
             "poc": poc_entries,
         })
 
-    return {
-        "entries": cve_entries,
-        "removed": {
-            "removed": sorted(removed_seen),
-            "by_cve": removed_by_cve,
-        },
-    }
+    return cve_entries
 
 
-def build_trending(blacklist: List[str]) -> List[Dict[str, object]]:
+def build_trending(blacklist: Collection[str]) -> List[Dict[str, object]]:
     rows = parse_trending_from_readme(README_PATH)
     if not rows:
         return []
@@ -210,8 +197,7 @@ def main() -> int:
     blacklist = load_blacklist()
 
     cve_payload = build_cve_list(blacklist)
-    write_json(CVE_OUTPUT, cve_payload["entries"])
-    write_json(REMOVED_OUTPUT, cve_payload["removed"], indent=2)
+    write_json(CVE_OUTPUT, cve_payload)
 
     trending_items = build_trending(blacklist)
     write_json(

scripts/update_cves.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import argparse
+import http.client
 import json
 import os
 import re
@@ -15,7 +16,7 @@ from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass, field
 from datetime import date, datetime, timedelta, timezone
 from pathlib import Path
-from typing import Any, Iterable, Iterator
+from typing import Any, Callable, Collection, Iterable, Iterator
 from urllib import error, request
 from urllib.parse import quote, urlparse
 
@@ -39,20 +40,22 @@ POC_RE = re.compile(
     r"repro(?:ducer|duction|duce)?|payload)\b",
     re.IGNORECASE,
 )
-UNSAFE_RE = re.compile(
-    r"\b(?:fully[ _-]*weaponized|weaponiz(?:e|es|ed|ing|ation)|"
-    r"mass(?:ive)?[ _-]*(?:attack|exploit(?:ation)?|scanners?|scanning)|"
-    r"bulk[ _-]*(?:attack|exploit(?:ation)?|scanners?|scanning))\b",
+CODE_FILE_RE = re.compile(
+    r"\.(?:c|cc|cpp|cs|go|java|js|mjs|nse|php|pl|ps1|py|rb|rs|sh|ts)$",
     re.IGNORECASE,
 )
 NON_POC_RE = re.compile(
     r"(?:\b(?:advisory|aggregat(?:ion|or)|analysis|ansible|audits?|block(?:er|ing)?|blog|"
-    r"bookmark|challenge|check(?:er|ing)|collection|corpus|course|ctf|dataset|defen(?:ce|se|sive)|"
-    r"detect(?:ion|or)?|dfir|exam|feed|finder|github[ _-]*stars|hardening|investigation|"
+    r"bookmark|challenge|cheat[ _-]*sheets?|check(?:s|er|ing)?|collection|corpus|course|ctf|"
+    r"dataset|defen(?:ce|se|sive)|detect(?:ion|or)?|dfir|epss|exam(?:en)?|feed|finder|framework|"
+    r"github[ _-]*stars|hardening|hub|investigation|"
     r"honeypot|intelligence|ioc|lab|list|lookup|mitigat(?:e|ing|ion|or)|model(?:er|ing)|"
-    r"monitor(?:ing)?|notes?|papers?|patch(?:es)?|pentest|portfolio|prediction|"
-    r"prevent(?:ion|ive)?|profile|protect(?:ion|ive)|remediat(?:e|ion)|reports?|"
-    r"resources?|rules?|scanners?|shield|signature|suggester|toolkit|walkthrough|wazuh|"
+    r"monitor(?:ing)?|notes?|oscp|papers?|patch(?:es|ed|ing)?|pentest|portfolio|prediction|"
+    r"pre[ _-]*checks?|prevent(?:ion|ive)?|profile|protect(?:ion|ive)|prueba|"
+    r"remediat(?:e|ion)|reports?|"
+    r"resources?|rules?|scan(?:s|ner|ning)?|shield|signature|suggester|templates?|toolkit|"
+    r"training|tutorial|validat(?:e|ion|or)|verif(?:y|ier|ication)|walkthrough|wazuh|"
+    r"workarounds?|fix(?:es|ed|ing)?|"
     r"write[ _-]*ups?)\b|searchpoc|watchdog)",
     re.IGNORECASE,
 )
@@ -110,6 +113,29 @@ def stable_unique(values: Iterable[str]) -> list[str]:
     return result
 
 
+def poc_link_key(url: str) -> str:
+    parsed = urlparse(url)
+    if (parsed.hostname or "").lower() not in {"github.com", "www.github.com"}:
+        return url
+    parts = [part for part in parsed.path.split("/") if part]
+    if len(parts) < 2:
+        return url
+    repo = f"{parts[0]}/{parts[1]}".lower()
+    suffix = "/" + "/".join(parts[2:]) if len(parts) > 2 else ""
+    return f"github:{repo}{suffix}?{parsed.query}#{parsed.fragment}"
+
+
+def stable_unique_poc_links(values: Iterable[str]) -> list[str]:
+    seen: set[str] = set()
+    result: list[str] = []
+    for value in values:
+        key = poc_link_key(value)
+        if value and key not in seen:
+            seen.add(key)
+            result.append(value)
+    return result
+
+
 def is_valid_cve(cve_id: str) -> bool:
     match = re.fullmatch(r"CVE-(\d{4})-(\d{4,})", cve_id, re.IGNORECASE)
     if not match:
@@ -129,51 +155,24 @@ def extract_cves(text: str | None, year: int | None = None) -> set[str]:
     return result
 
 
-def load_blacklist(path: Path = BLACKLIST_FILE) -> list[str]:
+def load_blacklist(path: Path = BLACKLIST_FILE) -> set[str]:
     if not path.exists():
-        return []
-    return [
+        return set()
+    return {
         line.strip().lower()
         for line in path.read_text(encoding="utf-8").splitlines()
         if line.strip() and not line.lstrip().startswith("#")
-    ]
+    }
 
 
-def is_blacklisted_repo(full_name: str, blacklist: list[str]) -> bool:
-    repo = full_name.rsplit("/", 1)[-1].lower()
-    full_name = full_name.lower()
-    for entry in blacklist:
-        target = full_name if "/" in entry else repo
-        if entry.endswith("*") and target.startswith(entry[:-1]):
-            return True
-        if target == entry:
-            return True
-    return False
+def is_blacklisted_repo(full_name: str, blacklist: Collection[str]) -> bool:
+    return full_name.lower() in blacklist
 
 
 def normalise_identity(text: str) -> str:
     return re.sub(r"[_-]+", " ", text)
 
 
-def has_unsafe_claim(text: str) -> bool:
-    text = normalise_identity(text)
-    for match in UNSAFE_RE.finditer(text):
-        before = text[max(0, match.start() - 80) : match.start()]
-        after = text[match.end() : match.end() + 80]
-        if re.search(
-            r"\b(?:avoid|cannot|no|non|not|without)\b.{0,50}$",
-            before,
-            re.IGNORECASE | re.DOTALL,
-        ):
-            continue
-        if re.search(r"\b(?:can|could|may)\s+be\s*$", before, re.IGNORECASE):
-            continue
-        if re.match(r".{0,30}\b(?:is|are)\s+withheld\b", after, re.IGNORECASE):
-            continue
-        return True
-    return False
-
-
 def github_repo_from_url(url: str) -> str:
     parsed = urlparse(url)
     if (parsed.hostname or "").lower() != "github.com":
@@ -182,6 +181,30 @@ def github_repo_from_url(url: str) -> str:
     return "/".join(parts[:2]) if len(parts) >= 2 else ""
 
 
+def github_url_is_profile_root(url: str) -> bool:
+    parsed = urlparse(url)
+    if (parsed.hostname or "").lower() != "github.com":
+        return False
+    parts = [part for part in parsed.path.split("/") if part]
+    return len(parts) == 2 and parts[0].lower() == parts[1].lower()
+
+
+def github_url_is_ambiguous_root(url: str, cve_id: str) -> bool:
+    parsed = urlparse(url)
+    if (parsed.hostname or "").lower() != "github.com":
+        return False
+    parts = [part for part in parsed.path.split("/") if part]
+    if len(parts) != 2:
+        return False
+    identity = normalise_identity(f"{parts[0]} {parts[1]}")
+    if not NON_POC_RE.search(identity):
+        return False
+    return not (
+        cve_id in extract_cves(identity)
+        and POC_RE.search(identity)
+    )
+
+
 def readme_text(repo: dict[str, Any]) -> str:
     parts: list[str] = []
     for key in ("readmeMd", "readmeUpper", "readmeRst", "readmeBare"):
@@ -201,6 +224,33 @@ def topic_names(repo: dict[str, Any]) -> list[str]:
     return names
 
 
+def root_entry_names(repo: dict[str, Any]) -> list[str]:
+    root = repo.get("root") or {}
+    return [
+        str(entry.get("name") or "")
+        for entry in root.get("entries") or []
+        if entry.get("name")
+    ]
+
+
+def root_has_poc_artifact(repo: dict[str, Any], cve_id: str) -> bool:
+    year, number = cve_id.split("-")[1:]
+    cve_pattern = re.compile(
+        rf"\bCVE[-_ ]{re.escape(year)}[-_ ]{re.escape(number)}\b",
+        re.IGNORECASE,
+    )
+    for name in root_entry_names(repo):
+        if name.lower().startswith(("readme", "license", "security")):
+            continue
+        if cve_pattern.search(name) or POC_RE.search(normalise_identity(name)):
+            return True
+    return False
+
+
+def root_has_code(repo: dict[str, Any]) -> bool:
+    return any(CODE_FILE_RE.search(name) for name in root_entry_names(repo))
+
+
 def readme_has_poc_context(readme: str, cve_id: str, full_name: str = "") -> bool:
     year, number = cve_id.split("-")[1:]
     pattern = re.compile(rf"\bCVE[-_ ]{re.escape(year)}[-_ ]{re.escape(number)}\b", re.IGNORECASE)
@@ -225,10 +275,10 @@ def readme_has_poc_context(readme: str, cve_id: str, full_name: str = "") -> boo
 def qualifying_repo_cves(
     repo: dict[str, Any],
     year: int,
-    blacklist: list[str],
+    blacklist: Collection[str],
 ) -> set[str]:
     full_name = str(repo.get("nameWithOwner") or "")
-    if not full_name or repo.get("isFork") or repo.get("isArchived"):
+    if not full_name or repo.get("isFork"):
         return set()
     owner, repo_name = full_name.lower().split("/", 1)
     if owner == repo_name or repo_name.endswith(".github.io"):
@@ -239,34 +289,47 @@ def qualifying_repo_cves(
     description = str(repo.get("description") or "")
     topics = " ".join(topic_names(repo))
     readme = readme_text(repo)
-    if has_unsafe_claim(f"{full_name} {description} {topics} {readme}"):
-        return set()
     name_cves = extract_cves(full_name, year)
     description_cves = extract_cves(description, year)
     topic_cves = extract_cves(topics, year)
     readme_cves = extract_cves(readme, year)
+    identity_cves = name_cves | description_cves | topic_cves
     identity_text = normalise_identity(f"{full_name} {description}")
     identity_has_poc = bool(POC_RE.search(identity_text))
     metadata_has_poc = identity_has_poc or bool(POC_RE.search(topics))
     identity_is_non_poc = bool(NON_POC_RE.search(identity_text))
-    if len(extract_cves(readme)) > 25:
+    has_code = root_has_code(repo)
+    name_has_poc = bool(POC_RE.search(normalise_identity(full_name)))
+    if identity_is_non_poc and not (name_cves and name_has_poc and has_code):
+        return set()
+    if len(readme_cves) > 5 and not identity_cves:
         readme_cves = set()
-
     accepted: set[str] = set()
     for cve_id in name_cves:
-        name_has_poc = bool(POC_RE.search(normalise_identity(full_name)))
-        if not identity_is_non_poc or name_has_poc:
+        has_context = readme_has_poc_context(readme, cve_id, full_name)
+        has_artifact = root_has_poc_artifact(repo, cve_id)
+        has_content = bool(readme.strip()) or has_artifact or has_code
+        if (
+            (not identity_is_non_poc or has_artifact or has_code)
+            and (metadata_has_poc or has_context or has_artifact)
+            and has_content
+        ):
             accepted.add(cve_id)
     for cve_id in description_cves | topic_cves:
-        if not identity_is_non_poc and (
-            metadata_has_poc or readme_has_poc_context(readme, cve_id, full_name)
+        has_context = readme_has_poc_context(readme, cve_id, full_name)
+        if (
+            (not identity_is_non_poc or cve_id in name_cves)
+            and (metadata_has_poc or has_context)
+            and (has_context or root_has_poc_artifact(repo, cve_id) or has_code)
         ):
             accepted.add(cve_id)
     for cve_id in readme_cves:
+        has_artifact = root_has_poc_artifact(repo, cve_id)
+        if identity_cves and cve_id not in identity_cves and not has_artifact:
+            continue
         if (
-            identity_has_poc
-            and not identity_is_non_poc
-            and readme_has_poc_context(readme, cve_id, full_name)
+            readme_has_poc_context(readme, cve_id, full_name)
+            and (has_artifact or has_code or identity_has_poc)
         ):
             accepted.add(cve_id)
     return accepted
@@ -299,7 +362,12 @@ def http_json(
                 return None
             if exc.code not in {429, 500, 502, 503, 504} or attempt == 2:
                 raise
-        except (error.URLError, TimeoutError, json.JSONDecodeError):
+        except (
+            error.URLError,
+            TimeoutError,
+            json.JSONDecodeError,
+            http.client.HTTPException,
+        ):
             if attempt == 2:
                 raise
         time.sleep(2 ** attempt)
@@ -328,12 +396,18 @@ class GitHubClient:
             raise RuntimeError(f"GitHub GraphQL quota is nearly exhausted; resets at {rate.get('resetAt')}")
         return search
 
-    def fetch_readmes(self, full_names: list[str]) -> tuple[set[str], dict[str, str]]:
+    def fetch_readmes(
+        self,
+        full_names: list[str],
+    ) -> tuple[set[str], dict[str, str], dict[str, list[dict[str, str]]]]:
         fields = """
           readmeMd: object(expression: \"HEAD:README.md\") { ... on Blob { text } }
           readmeUpper: object(expression: \"HEAD:README.MD\") { ... on Blob { text } }
           readmeRst: object(expression: \"HEAD:README.rst\") { ... on Blob { text } }
           readmeBare: object(expression: \"HEAD:README\") { ... on Blob { text } }
+          root: object(expression: \"HEAD:\") {
+            ... on Tree { entries { name type } }
+          }
         """
         aliases: list[str] = []
         names: list[str] = []
@@ -348,7 +422,7 @@ class GitHubClient:
             )
             names.append(full_name)
         if not aliases:
-            return set(), {}
+            return set(), {}, {}
         query_text = "query { " + " ".join(aliases) + " rateLimit { remaining resetAt } }"
         payload = http_json(
             GITHUB_GRAPHQL_URL,
@@ -360,6 +434,7 @@ class GitHubClient:
         if int(rate.get("remaining") or 0) < 5:
             raise RuntimeError(f"GitHub GraphQL quota is nearly exhausted; resets at {rate.get('resetAt')}")
         result: dict[str, str] = {}
+        roots: dict[str, list[dict[str, str]]] = {}
         existing: set[str] = set()
         for index, full_name in enumerate(names):
             repo = data.get(f"repo{index}")
@@ -369,7 +444,13 @@ class GitHubClient:
             text = readme_text(repo)
             if text:
                 result[full_name] = text
-        return existing, result
+            entries = ((repo.get("root") or {}).get("entries") or [])
+            roots[full_name] = [
+                {"name": str(entry.get("name") or ""), "type": str(entry.get("type") or "")}
+                for entry in entries
+                if entry.get("name")
+            ]
+        return existing, result, roots
 
 
 def split_date_range(start: date, end: date) -> tuple[tuple[date, date], tuple[date, date]]:
@@ -498,14 +579,17 @@ def discover_github_pocs(
         for future in as_completed(futures):
             batch = futures[future]
             try:
-                existing, readmes = future.result()
+                existing, readmes, roots = future.result()
             except Exception as exc:
                 print(f"Skipped README batch starting with {batch[0]}: {exc}", file=sys.stderr)
                 existing = set(batch)
                 readmes = {}
+                roots = {}
             existing_names.update(existing)
             for full_name, text in readmes.items():
                 repositories[full_name]["readmeMd"] = {"text": text}
+            for full_name, entries in roots.items():
+                repositories[full_name]["root"] = {"entries": entries}
             completed += len(batch)
             if completed % 500 < len(batch):
                 print(f"Fetched README data for {completed} of {len(names)} repositories")
@@ -580,12 +664,20 @@ def details_from_record(record: dict[str, Any]) -> CVEDetails | None:
     )
 
 
-def reference_is_poc(reference: dict[str, Any], blacklist: list[str]) -> bool:
+def reference_is_poc(
+    reference: dict[str, Any],
+    blacklist: Collection[str],
+    cve_id: str = "",
+) -> bool:
     url = str(reference.get("url") or "").strip()
     if not url.startswith(("http://", "https://")):
         return False
     repo = github_repo_from_url(url)
-    if repo and is_blacklisted_repo(repo, blacklist):
+    if repo and (
+        is_blacklisted_repo(repo, blacklist)
+        or github_url_is_profile_root(url)
+        or (cve_id and github_url_is_ambiguous_root(url, cve_id))
+    ):
         return False
     tags = [str(tag).lower() for tag in reference.get("tags") or []]
     evidence = " ".join([url, str(reference.get("name") or ""), *tags])
@@ -596,7 +688,7 @@ def reference_is_poc(reference: dict[str, Any], blacklist: list[str]) -> bool:
 
 def poc_references(
     record: dict[str, Any],
-    blacklist: list[str] | None = None,
+    blacklist: Collection[str] | None = None,
 ) -> list[str]:
     cve_id = record_cve_id(record)
     if not cve_id or not record_is_published(record):
@@ -607,9 +699,63 @@ def poc_references(
     for container in candidates:
         for item in container.get("references") or []:
             url = str(item.get("url") or "").strip()
-            if reference_is_poc(item, blacklist):
+            if reference_is_poc(item, blacklist, cve_id):
                 references.append(url)
-    return stable_unique(references)
+    return stable_unique_poc_links(references)
+
+
+def reference_url_is_live(url: str, timeout: int = 15) -> bool:
+    headers = {
+        "Accept": "*/*",
+        "Accept-Encoding": "identity",
+        "User-Agent": USER_AGENT,
+    }
+    for method in ("HEAD", "GET"):
+        request_headers = dict(headers)
+        if method == "GET":
+            request_headers["Range"] = "bytes=0-0"
+        req = request.Request(url, headers=request_headers, method=method)
+        try:
+            with request.urlopen(req, timeout=timeout) as response:
+                if method == "GET":
+                    response.read(1)
+                if int(response.status) < 400:
+                    return True
+        except error.HTTPError as exc:
+            code = exc.code
+            exc.close()
+            if method == "GET" and code in {404, 410}:
+                return False
+        except (error.URLError, TimeoutError, OSError):
+            pass
+    return True
+
+
+def filter_live_references(
+    references: dict[str, list[str]],
+    checker: Callable[[str], bool] | None = None,
+) -> dict[str, list[str]]:
+    checker = reference_url_is_live if checker is None else checker
+    urls = stable_unique(url for values in references.values() for url in values)
+    if not urls:
+        return references
+    dead: set[str] = set()
+    with ThreadPoolExecutor(max_workers=min(8, len(urls))) as executor:
+        futures = {executor.submit(checker, url): url for url in urls}
+        for future in as_completed(futures):
+            url = futures[future]
+            try:
+                if not future.result():
+                    dead.add(url)
+            except Exception:
+                continue
+    if dead:
+        print(f"Discarded {len(dead)} references returning 404 or 410")
+    return {
+        cve_id: [url for url in urls_for_cve if url not in dead]
+        for cve_id, urls_for_cve in references.items()
+        if any(url not in dead for url in urls_for_cve)
+    }
 
 
 def read_record(path: Path) -> dict[str, Any] | None:
@@ -783,8 +929,14 @@ def parse_section_links(block: str) -> list[str]:
     return stable_unique(links)
 
 
+def section_links(text: str, header: str) -> list[str]:
+    pattern = re.compile(SECTION_RE_TEMPLATE.format(header=re.escape(header)), re.DOTALL)
+    match = pattern.search(text)
+    return parse_section_links(match.group(2)) if match else []
+
+
 def merge_section(text: str, header: str, links: Iterable[str], empty_text: str) -> tuple[str, bool]:
-    incoming = stable_unique(links)
+    incoming = stable_unique_poc_links(links)
     if not incoming:
         return text, False
     pattern = re.compile(SECTION_RE_TEMPLATE.format(header=re.escape(header)), re.DOTALL)
@@ -792,14 +944,29 @@ def merge_section(text: str, header: str, links: Iterable[str], empty_text: str)
     if not match:
         return text, False
     existing = parse_section_links(match.group(2))
-    combined = stable_unique([*existing, *incoming])
+    combined = stable_unique_poc_links([*existing, *incoming])
     if combined == existing:
         return text, False
-    replacement = "\n".join(f"- {url}" for url in combined) if combined else empty_text
+    return replace_section(text, header, combined, empty_text)
+
+
+def replace_section(
+    text: str,
+    header: str,
+    links: Iterable[str],
+    empty_text: str,
+) -> tuple[str, bool]:
+    values = stable_unique_poc_links(links)
+    pattern = re.compile(SECTION_RE_TEMPLATE.format(header=re.escape(header)), re.DOTALL)
+    match = pattern.search(text)
+    if not match:
+        return text, False
+    newline = "\r\n" if "\r\n" in match.group(1) else "\n"
+    replacement = newline.join(f"- {url}" for url in values) if values else empty_text
     body = match.group(2)
-    trailing = body[len(body.rstrip()) :] or "\n"
+    trailing = body[len(body.rstrip()) :] or newline
     updated = text[: match.start(2)] + replacement + trailing + text[match.end(2) :]
-    return updated, True
+    return updated, updated != text
 
 
 def update_existing_markdown(
@@ -809,25 +976,37 @@ def update_existing_markdown(
     *,
     dry_run: bool,
 ) -> bool:
-    original = path.read_text(encoding="utf-8")
+    with path.open("r", encoding="utf-8", newline="") as handle:
+        original = handle.read()
     updated, reference_changed = merge_section(
         original,
         "#### Reference",
         references,
         "No PoCs from references.",
     )
-    updated, github_changed = merge_section(
+    reference_keys = {
+        poc_link_key(url)
+        for url in section_links(updated, "#### Reference")
+    }
+    existing_github = section_links(updated, "#### Github")
+    github_values = [
+        url
+        for url in stable_unique_poc_links([*existing_github, *github_links])
+        if poc_link_key(url) not in reference_keys
+    ]
+    updated, github_changed = replace_section(
         updated,
         "#### Github",
-        github_links,
+        github_values,
         "No PoCs found on GitHub currently.",
     )
     if not (reference_changed or github_changed):
         return False
-    if not updated.endswith("\n"):
-        updated += "\n"
+    if not updated.endswith(("\n", "\r")):
+        updated += "\r\n" if "\r\n" in original else "\n"
     if not dry_run:
-        path.write_text(updated, encoding="utf-8")
+        with path.open("w", encoding="utf-8", newline="") as handle:
+            handle.write(updated)
     return True
 
 
@@ -925,7 +1104,7 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--backfill-dir", type=Path, help="Local CVE List V5 year directory")
     parser.add_argument("--year", type=int, help="Year to backfill")
     parser.add_argument("--lookback-days", type=int, default=3, help="GitHub pushed-date overlap")
-    parser.add_argument("--years", type=int, default=6, help="Number of recent years to scan")
+    parser.add_argument("--years", type=int, default=2, help="Number of recent years to scan")
     parser.add_argument("--cve", action="append", default=[], help="Process one CVE ID")
     parser.add_argument("--skip-github", action="store_true", help="Skip GitHub repository discovery")
     parser.add_argument("--skip-cvelist", action="store_true", help="Skip CVE List V5 references")
@@ -962,6 +1141,7 @@ def main() -> int:
             print(f"CVE List V5 backfill: {len(references)} CVEs with PoC references")
         else:
             records, references, checkpoint = load_incremental_records(cve_filter)
+            references = filter_live_references(references)
 
     github: dict[str, list[str]] = {}
     if not args.skip_github:

scripts/utils.py

@@ -4,7 +4,7 @@ import json
 import re
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Dict, Iterable, List, Optional, Tuple
+from typing import Collection, Dict, Iterable, List, Optional, Tuple
 
 import requests
 
@@ -261,20 +261,20 @@ def hash_key(text: str) -> str:
     return hashlib.sha256(text.encode("utf-8")).hexdigest()
 
 
-def load_blacklist(path: Path | None = None) -> List[str]:
+def load_blacklist(path: Path | None = None) -> set[str]:
     target = path or ROOT / "blacklist.txt"
     if not target.exists():
-        return []
-    entries: List[str] = []
+        return set()
+    entries: set[str] = set()
     for raw in target.read_text(encoding="utf-8").splitlines():
         line = raw.strip()
         if line and not line.startswith("#"):
-            entries.append(line)
+            entries.add(line.lower())
     return entries
 
 
 def extract_repo_from_url(url: str) -> str:
-    """Return repository name segment from a URL (best effort)."""
+    """Return the owner and repository segments from a URL."""
     try:
         from urllib.parse import urlparse
 
@@ -287,30 +287,20 @@ def extract_repo_from_url(url: str) -> str:
         path = url
     parts = path.strip("/").split("/")
     if len(parts) >= 2:
-        return parts[1].lower()
+        return "/".join(parts[:2]).lower()
     if parts:
         return parts[-1].lower()
     return ""
 
 
-def is_blacklisted_repo(url: str, blacklist: List[str]) -> bool:
-    repo = extract_repo_from_url(url)
-    if not repo:
+def is_blacklisted_repo(url: str, blacklist: Collection[str]) -> bool:
+    full_name = extract_repo_from_url(url)
+    if not full_name:
         return False
-    for entry in blacklist:
-        slug = entry.strip().lower()
-        if not slug:
-            continue
-        if slug.endswith("*"):
-            prefix = slug[:-1]
-            if prefix and repo.startswith(prefix):
-                return True
-        elif repo == slug:
-            return True
-    return False
+    return full_name in blacklist
 
 
-def filter_links_by_blacklist(links: List[str], blacklist: List[str]) -> List[str]:
+def filter_links_by_blacklist(links: List[str], blacklist: Collection[str]) -> List[str]:
     if not blacklist:
         return links
     filtered: List[str] = []