mirror of
https://github.com/0xMarcio/cve.git
synced 2026-10-23 16:35:19 +00:00
807 lines
43 KiB
Python
807 lines
43 KiB
Python
#!/usr/bin/env python3
|
|
"""Durable release evidence for published CVE/artifact pairs.
|
|
|
|
Observing a link is not evidence of its publication date. Imports remain
|
|
undated until source history supplies evidence; removed links stay recorded.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from collections import defaultdict
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from urllib.parse import quote, unquote, urlencode, urlsplit
|
|
from urllib.request import Request, urlopen
|
|
|
|
import source_artifacts
|
|
from update_cves import CODE_FILE_RE, INDEX, POC_RE, poc_link_key
|
|
|
|
DIRECTORY = INDEX / "releases"
|
|
VERSION = 1
|
|
DATE_ORDER = {"commit": 1, "source": 2, "created": 3, "public": 4, "merged": 5, "seen": 6}
|
|
EXCLUDED_DIRS = {".git", ".github", "docs", "doc", "images", "img", "assets", "screenshots", "media",
|
|
"tests", "test", "node_modules", "vendor", "third_party", "__pycache__"}
|
|
BUILD_FILES = {"setup.py", "conftest.py", "__init__.py", "requirements.txt", "package.json", "package-lock.json",
|
|
"go.mod", "go.sum", "cargo.toml", "cargo.lock", "makefile", "cmakelists.txt", "dockerfile",
|
|
"docker-compose.yml", "pyproject.toml"}
|
|
DATA_SUFFIXES = {".json", ".yml", ".yaml", ".toml", ".lock", ".txt", ".csv", ".xml", ".ini", ".cfg", ".md",
|
|
".rst", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".pdf", ".gz", ".zip", ".xz", ".bz2",
|
|
".tar", ".7z", ".o", ".so", ".a", ".dll", ".exe"}
|
|
PAPERWORK = re.compile(r"^(?:readme|license|licence|security|contributing|changelog|code.of.conduct)(?:\.|$)", re.I)
|
|
SUPPORT_FILE = re.compile(r"^(?:setup|install|example|demo|test|conftest|helper|util|common)(?:[_.-]|$)", re.I)
|
|
HISTORY_VERSION = 3
|
|
HISTORY_FIELDS = {"paths", "rev", "blobs", "commit", "sha", "commit_verified", "history_paths",
|
|
"merged", "merged_verified", "history_version", "bulk"}
|
|
COMMIT_FIELDS = {"commit", "sha", "commit_verified", "history_version"}
|
|
|
|
|
|
class Ledger(dict):
|
|
"""Keep identity comparisons local to one CVE during large backfills."""
|
|
def __init__(self, values=()):
|
|
super().__init__()
|
|
self.by_cve = defaultdict(set)
|
|
for identity, row in dict(values).items():
|
|
self[identity] = row
|
|
|
|
def __setitem__(self, identity, row):
|
|
super().__setitem__(identity, row)
|
|
self.by_cve[row["cve"]].add(identity)
|
|
|
|
def __deepcopy__(self, memo):
|
|
return Ledger((identity, copy.deepcopy(row, memo)) for identity, row in self.items())
|
|
|
|
|
|
def utcnow() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def timestamp(value: object) -> datetime | None:
|
|
if not isinstance(value, str):
|
|
return None
|
|
try:
|
|
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
return parsed.replace(tzinfo=timezone.utc) if parsed.tzinfo is None else parsed.astimezone(timezone.utc)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def key(cve: str, url: str) -> str:
|
|
if not re.fullmatch(r"CVE-\d{4}-\d{4,}", cve):
|
|
raise ValueError(f"Invalid CVE: {cve}")
|
|
parsed = urlsplit(url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.hostname or parsed.username or parsed.password:
|
|
raise ValueError("Invalid artifact URL")
|
|
return f"{cve} {poc_link_key(url)}"
|
|
|
|
|
|
def artifact_paths(entries: list[dict], cve: str, *, readme_path: str | None = None,
|
|
readme_qualified: bool = False, limit: int = 12) -> list[str]:
|
|
"""Select paths in an already qualified repository, before applying a cap.
|
|
|
|
A selected directory still needs its qualifying files checked for dating.
|
|
README fallback is explicit: an advisory/claim alone is not an artifact.
|
|
"""
|
|
ranked = []
|
|
for entry in entries:
|
|
path = str(entry.get("path") or entry.get("name") or "")
|
|
parts = path.split("/")
|
|
name = parts[-1].lower()
|
|
if (not path or any(part.lower() in EXCLUDED_DIRS for part in parts)
|
|
or name in BUILD_FILES or PAPERWORK.match(name) or SUPPORT_FILE.match(name)):
|
|
continue
|
|
kind = str(entry.get("type") or "blob").lower()
|
|
code = bool(CODE_FILE_RE.search(name)) or name.endswith((".html", ".asm", ".lua"))
|
|
if not code and Path(name).suffix in DATA_SUFFIXES:
|
|
continue
|
|
rank = (0 if cve.lower() in name else 1 if code else 2 if POC_RE.search(name)
|
|
else 3 if kind == "tree" else 4)
|
|
ranked.append((rank, path))
|
|
selected = [path for _, path in sorted(set(ranked))[:limit]]
|
|
return selected or ([readme_path] if readme_qualified and readme_path else [])
|
|
|
|
|
|
def artifact_identity(entries: list[dict], paths: list[str]) -> dict:
|
|
"""Fingerprint selected content, independently of path moves or paperwork."""
|
|
selected = {entry.get("path") or entry.get("name"): entry for entry in entries}
|
|
objects, blobs = [], []
|
|
for path in paths:
|
|
entry = selected.get(path) or {}
|
|
oid = entry.get("sha256") or entry.get("oid") or entry.get("sha")
|
|
if not isinstance(oid, str) or not re.fullmatch(r"[0-9a-f]{40}|[0-9a-f]{64}", oid):
|
|
continue
|
|
kind = str(entry.get("type") or "blob").lower()
|
|
objects.append((kind, oid))
|
|
size = entry.get("size", entry.get("byteSize", 0))
|
|
if kind == "blob" and isinstance(size, int) and size >= 256:
|
|
blobs.append(oid)
|
|
result = {"paths": sorted(paths)}
|
|
if objects:
|
|
result["rev"] = hashlib.sha256(json.dumps(sorted(set(objects)), separators=(",", ":")).encode()).hexdigest()
|
|
if blobs:
|
|
result["blobs"] = sorted(set(blobs))
|
|
return result
|
|
|
|
|
|
def date_release(evidence: dict, *, now: str | None = None) -> tuple[str | None, str]:
|
|
"""Latest evidenced lower bound; client commit clocks have strict ceilings."""
|
|
if evidence.get("pending"):
|
|
return None, "pending"
|
|
current = timestamp(now or utcnow())
|
|
ceiling = min(current, timestamp(evidence.get("pushed_at")) or current)
|
|
observed_ceiling = min(ceiling, timestamp(evidence.get("seen")) or ceiling)
|
|
bounds = []
|
|
for name in ("created", "public", "commit", "merged", "source"):
|
|
value = evidence.get(name)
|
|
moment = timestamp(value)
|
|
if moment is None or moment > (ceiling if name in {"commit", "merged"} else observed_ceiling):
|
|
continue
|
|
if name == "commit" and (observed := timestamp(evidence.get("history_observed"))) and moment > observed:
|
|
continue
|
|
if name in {"commit", "merged"} and evidence.get(f"{name}_verified") is not True:
|
|
continue
|
|
bounds.append((moment, DATE_ORDER[name], value, name))
|
|
chosen = max(bounds) if bounds else None
|
|
absent, seen = timestamp(evidence.get("absent")), timestamp(evidence.get("seen"))
|
|
# Parser/gate changes and import baselines never establish absence.
|
|
if (not evidence.get("imported") and evidence.get("absence_verified") is True
|
|
and evidence.get("previous_head") and absent and seen and absent < seen <= current
|
|
and (chosen is None or chosen[0] <= absent)):
|
|
return evidence["seen"], "seen"
|
|
return (chosen[2], chosen[3]) if chosen else (None, "unknown")
|
|
|
|
|
|
def load_ledger(directory: Path | None = None) -> dict[str, dict]:
|
|
result = Ledger()
|
|
for path in sorted((directory or DIRECTORY).glob("[12][0-9][0-9][0-9].json")):
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
if payload.get("version") != VERSION or not isinstance(payload.get("entries"), dict):
|
|
raise ValueError(f"Invalid release shard: {path}")
|
|
for identity, row in payload["entries"].items():
|
|
if (not isinstance(row, dict) or identity != key(row.get("cve", ""), row.get("url", ""))
|
|
or row["cve"].split("-")[1] != path.stem or identity in result):
|
|
raise ValueError(f"Invalid release entry: {path}")
|
|
result[identity] = row
|
|
return result
|
|
|
|
|
|
def save_ledger(ledger: dict[str, dict], directory: Path | None = None) -> None:
|
|
directory = directory or DIRECTORY
|
|
previous = load_ledger(directory)
|
|
if not previous.keys() <= ledger.keys():
|
|
raise ValueError("Release history cannot shrink; retain removed entries as tombstones")
|
|
for identity, old in previous.items():
|
|
if old.get("seen") != ledger[identity].get("seen") or old.get("imported") != ledger[identity].get("imported"):
|
|
raise ValueError("First release observation and import provenance are immutable")
|
|
shards = defaultdict(dict)
|
|
for identity, row in sorted(ledger.items()):
|
|
if identity != key(row["cve"], row["url"]):
|
|
raise ValueError("Release key differs from its artifact")
|
|
shards[row["cve"].split("-")[1]][identity] = row
|
|
directory.mkdir(parents=True, exist_ok=True)
|
|
for year, entries in shards.items():
|
|
# One sorted entry per line keeps backfills and hourly diffs reviewable.
|
|
body = '{"version":1,"entries":{\n' + ",\n".join(
|
|
json.dumps(identity) + ":" + json.dumps(row, sort_keys=True, separators=(",", ":"))
|
|
for identity, row in entries.items()) + "\n}}\n"
|
|
path = directory / f"{year}.json"
|
|
if path.exists() and path.read_text(encoding="utf-8") == body:
|
|
continue
|
|
temporary = path.with_suffix(".tmp")
|
|
temporary.write_text(body, encoding="utf-8")
|
|
temporary.replace(path)
|
|
|
|
|
|
def _repository_artifact(url: str) -> tuple[str, str] | None:
|
|
parsed = urlsplit(url)
|
|
parts = unquote(parsed.path).strip("/").split("/")
|
|
if parsed.hostname not in {"github.com", "www.github.com"} or len(parts) < 2:
|
|
return None
|
|
if len(parts) == 2:
|
|
return "/".join(parts[:2]).lower(), ""
|
|
if len(parts) >= 5 and parts[2] in {"blob", "tree"}:
|
|
return "/".join(parts[:2]).lower(), "/".join(parts[4:])
|
|
return None
|
|
|
|
|
|
def _same_artifact(row: dict, candidate: dict) -> bool:
|
|
left, right = _repository_artifact(row["url"]), _repository_artifact(candidate["url"])
|
|
if (left and right and row.get("rev") and row.get("rev") == candidate.get("rev")
|
|
and (left[0] == right[0] or row.get("repo") and row.get("repo") == candidate.get("repo"))):
|
|
return True
|
|
# Repository IDs identify transfers, but never collapse different variants.
|
|
return bool(left and right and row.get("repo") and row.get("repo") == candidate.get("repo")
|
|
and left[1] == right[1] and left[0] != right[0])
|
|
|
|
|
|
def _repository_rejected_shas(row: dict, candidates) -> set[str]:
|
|
rejected = set(row.get("future_shas", []))
|
|
location = _repository_artifact(row["url"])
|
|
if location:
|
|
for other in candidates:
|
|
if other["cve"] != row["cve"] or not other.get("future_shas"):
|
|
continue
|
|
peer = _repository_artifact(other["url"])
|
|
same = (row["repo"] == other["repo"] if row.get("repo") and other.get("repo")
|
|
else peer and location[0] == peer[0])
|
|
if peer and same:
|
|
rejected.update(other["future_shas"])
|
|
return rejected
|
|
|
|
|
|
def _discard_rejected_commit(row: dict, rejected: set[str]) -> bool:
|
|
if rejected:
|
|
row["future_shas"] = sorted(rejected)
|
|
if row.get("sha") not in rejected:
|
|
return False
|
|
for name in COMMIT_FIELDS:
|
|
row.pop(name, None)
|
|
if row.get("basis") == "commit":
|
|
row.update(released=None, basis="unknown")
|
|
return True
|
|
|
|
|
|
def record(ledger: dict[str, dict], cve: str, url: str, evidence: dict | None = None, *,
|
|
observed_at: str | None = None, import_mode: bool = False) -> dict:
|
|
observed_at = observed_at or utcnow()
|
|
identity = key(cve, url)
|
|
old = ledger.get(identity)
|
|
row = copy.deepcopy(old) if old else {"cve": cve, "url": url, "basis": "unknown", "released": None}
|
|
row.pop("gone", None)
|
|
row.pop("gone_reason", None)
|
|
if not old:
|
|
if import_mode:
|
|
row["imported"] = True
|
|
else:
|
|
row["seen"] = observed_at
|
|
evidence = evidence or {}
|
|
if evidence.get("error") and row.get("released"):
|
|
# A failed refresh may advance retry bookkeeping, never valid evidence.
|
|
evidence = {name: value for name, value in evidence.items() if name in {"error", "checked", "head"}}
|
|
identities = ledger.by_cve.get(cve, ()) if isinstance(ledger, Ledger) else ledger.keys()
|
|
future_shas = _repository_rejected_shas(
|
|
{**row, "repo": evidence.get("repo") or row.get("repo")}, (ledger[item] for item in identities))
|
|
observed = min(timestamp(value) for value in (
|
|
observed_at, evidence.get("history_observed") or evidence.get("checked")
|
|
) if timestamp(value))
|
|
# A stale pushed_at is retryable; a SHA already seen before its clock is not.
|
|
for proof, checked in ((evidence, observed),
|
|
(row, timestamp(row.get("checked")) if not row.get("history_version") else None)):
|
|
clock = timestamp(proof.get("commit"))
|
|
if proof.get("sha") and proof.get("commit_verified") is True and clock and checked and clock > checked:
|
|
future_shas.add(proof["sha"])
|
|
_discard_rejected_commit(row, future_shas)
|
|
if evidence.get("sha") in future_shas:
|
|
evidence = {name: value for name, value in evidence.items() if name not in COMMIT_FIELDS}
|
|
method = evidence.get("history_version", 0)
|
|
method = method if type(method) is int and method > 0 else 0
|
|
old_method = row.get("history_version", 0)
|
|
old_method = old_method if type(old_method) is int and old_method > 0 else 0
|
|
moment = timestamp(evidence.get("commit"))
|
|
ceiling = min(timestamp(value) for value in (
|
|
observed_at, evidence.get("pushed_at", row.get("pushed_at"))
|
|
) if timestamp(value))
|
|
verified_history = bool(not evidence.get("error") and evidence.get("commit_verified") is True
|
|
and moment and moment <= ceiling)
|
|
correction = bool(old and verified_history and method > old_method)
|
|
if old_method and evidence.get("commit_verified") and method < old_method:
|
|
evidence = {name: value for name, value in evidence.items() if name not in HISTORY_FIELDS}
|
|
if correction:
|
|
for name in HISTORY_FIELDS | {"copy", "absence_verified"}:
|
|
row.pop(name, None)
|
|
row.update(released=None, basis="unknown")
|
|
for name in ("created", "public", "repo", "paths", "rev", "blobs", "source", "absent", "previous_head",
|
|
"absence_verified", "commit", "sha", "commit_verified", "merged", "merged_verified", "pushed_at", "bulk", "checked", "head", "history_paths"):
|
|
if name not in evidence:
|
|
continue
|
|
value = evidence[name]
|
|
if name in {"paths", "rev", "blobs", "absent", "previous_head", "absence_verified"} and name in row:
|
|
continue
|
|
if name in {"commit", "merged", "source"} and timestamp(row.get(name)):
|
|
if not timestamp(value) or timestamp(value) >= timestamp(row[name]):
|
|
continue
|
|
if name == "sha" and row.get("commit") != evidence.get("commit"):
|
|
continue
|
|
row[name] = value
|
|
if verified_history and method >= old_method and method:
|
|
row["history_version"] = method
|
|
if evidence.get("error"):
|
|
if not row.get("released"):
|
|
row.update(basis="pending", pending=True)
|
|
else:
|
|
if evidence.get("commit_verified") or evidence.get("source") or evidence.get("absence_verified"):
|
|
row.pop("pending", None)
|
|
released, basis = date_release(row, now=observed_at)
|
|
if released:
|
|
# Re-adds and subsequent observations are not another release.
|
|
if row.get("basis") != "seen" or not row.get("released"):
|
|
row.update(released=released, basis=basis)
|
|
if (not old or correction or not any(old.get(field) for field in ("repo", "rev", "blobs"))) and any(row.get(field) for field in ("repo", "rev", "blobs")):
|
|
identities = ledger.by_cve.get(cve, ()) if isinstance(ledger, Ledger) else ledger.keys()
|
|
candidates = [(other_key, ledger[other_key]) for other_key in identities
|
|
if other_key != identity and ledger[other_key]["cve"] == cve]
|
|
matches = [(other_key, other) for other_key, other in candidates if _same_artifact(row, other)]
|
|
copies = [(other_key, other) for other_key, other in candidates
|
|
if row.get("blobs") and set(row["blobs"]) == set(other.get("blobs", []))]
|
|
if matches or copies:
|
|
other_key, other = min(matches or copies, key=lambda item: (item[1].get("released") or "9999", item[0]))
|
|
row["copy"] = other.get("copy", other_key)
|
|
row["released"], row["basis"] = other.get("released"), other.get("basis", "unknown")
|
|
elif row.get("copy") in ledger:
|
|
original = ledger[row["copy"]]
|
|
row["released"], row["basis"] = original.get("released"), original.get("basis", "unknown")
|
|
ledger[identity] = row
|
|
return row
|
|
|
|
|
|
def mark_bulk(ledger: dict[str, dict]) -> None:
|
|
groups = defaultdict(list)
|
|
for identity, row in ledger.items():
|
|
location = (_repository_artifact(row["url"]) or (row["url"],))[0]
|
|
source = row.get("repo") or location
|
|
if row.get("sha") and row.get("commit_verified"):
|
|
groups[(str(source), row["sha"], "commit")].append(identity)
|
|
if row.get("seen") and (row.get("basis") == "seen" or row.get("pending")):
|
|
groups[(location, row["seen"], "seen")].append(identity)
|
|
for (*_, basis), identities in groups.items():
|
|
count = len({ledger[identity]["cve"] for identity in identities}) if basis == "commit" else len(identities)
|
|
if count > (10 if basis == "commit" else 25):
|
|
for identity in identities:
|
|
if basis == "commit" or ledger[identity].get("basis") == "seen":
|
|
ledger[identity]["bulk"] = True
|
|
|
|
|
|
def resolve_copies(ledger: dict[str, dict], *, now: str | None = None) -> None:
|
|
"""Resolve identities after a batch, independent of enrichment order.
|
|
|
|
Entire selected nontrivial blob sets must match. One shared helper cannot
|
|
collapse distinct variants. Tombstones still supply original provenance.
|
|
"""
|
|
rejected = defaultdict(list)
|
|
for row in ledger.values():
|
|
if row.get("future_shas"):
|
|
rejected[row["cve"]].append(dict(row))
|
|
# Existing payloads can predate the peer that establishes clock rejection.
|
|
for row in ledger.values():
|
|
if row["cve"] in rejected:
|
|
shas = _repository_rejected_shas(row, rejected[row["cve"]])
|
|
if _discard_rejected_commit(row, shas):
|
|
row["released"], row["basis"] = date_release(row, now=now)
|
|
groups, parents = defaultdict(list), {}
|
|
for identity, row in ledger.items():
|
|
location = _repository_artifact(row["url"])
|
|
if blobs := row.get("blobs"):
|
|
groups[(row["cve"], "content", tuple(sorted(set(blobs))))].append(identity)
|
|
if row.get("rev") and location:
|
|
groups[(row["cve"], "revision", str(row.get("repo") or location[0]), row["rev"])].append(identity)
|
|
if row.get("repo") and location:
|
|
groups[(row["cve"], "transfer", str(row["repo"]), location[1])].append(identity)
|
|
|
|
def root(identity):
|
|
parents.setdefault(identity, identity)
|
|
if parents[identity] != identity:
|
|
parents[identity] = root(parents[identity])
|
|
return parents[identity]
|
|
|
|
for signature, identities in groups.items():
|
|
if len(identities) < 2:
|
|
continue
|
|
if signature[1] == "transfer" and len({_repository_artifact(ledger[item]["url"])[0] for item in identities}) < 2:
|
|
continue
|
|
first = min(identities)
|
|
for identity in identities:
|
|
parents[root(identity)] = root(first)
|
|
components = defaultdict(list)
|
|
for identity in parents:
|
|
components[root(identity)].append(identity)
|
|
for identities in components.values():
|
|
own_dates = {identity: date_release(ledger[identity], now=now) for identity in identities}
|
|
original = min(identities, key=lambda identity: (
|
|
timestamp(own_dates[identity][0]) or datetime.max.replace(tzinfo=timezone.utc), identity))
|
|
released, basis = own_dates[original]
|
|
for identity in identities:
|
|
row = ledger[identity]
|
|
row["released"], row["basis"] = released, basis
|
|
if identity == original:
|
|
row.pop("copy", None)
|
|
else:
|
|
row["copy"] = original
|
|
for identity, row in ledger.items():
|
|
if row.get("copy") and identity not in parents:
|
|
row.pop("copy", None)
|
|
row["released"], row["basis"] = date_release(row, now=now)
|
|
|
|
|
|
def reconcile(ledger: dict[str, dict], pairs: list[tuple[str, str]], *, evidence: dict[str, dict] | None = None,
|
|
observed_at: str | None = None, import_mode: bool = False, complete: bool = True) -> dict[str, dict]:
|
|
observed_at = observed_at or utcnow()
|
|
result = copy.deepcopy(ledger) if isinstance(ledger, Ledger) else Ledger(copy.deepcopy(ledger))
|
|
present = set()
|
|
for cve, url in sorted(set(pairs)):
|
|
identity = key(cve, url)
|
|
present.add(identity)
|
|
record(result, cve, url, (evidence or {}).get(identity), observed_at=observed_at, import_mode=import_mode)
|
|
if complete:
|
|
for identity, row in result.items():
|
|
if identity not in present:
|
|
if not row.get('gone'):
|
|
import link_health
|
|
row['gone_reason'] = 'unavailable' if link_health.unavailable(row['url']) else 'withdrawn'
|
|
row['gone'] = observed_at
|
|
resolve_copies(result, now=observed_at)
|
|
mark_bulk(result)
|
|
return result
|
|
|
|
|
|
def landed(ledger: dict[str, dict], *, now: str | None = None, days: int = 10) -> list[dict]:
|
|
moment = timestamp(now or utcnow())
|
|
found = [(identity, row) for identity, row in ledger.items()
|
|
if not any(row.get(flag) for flag in ("gone", "copy", "bulk"))
|
|
and row.get("basis") not in {"unknown", "pending"}
|
|
and timestamp(row.get("released")) is not None
|
|
and moment - timedelta(days=days) <= timestamp(row["released"]) <= moment]
|
|
found.sort(key=lambda item: item[0])
|
|
found.sort(key=lambda item: timestamp(item[1]["released"]), reverse=True)
|
|
return [dict(row) for _, row in found]
|
|
|
|
|
|
def published_pairs(entries: list[dict] | None = None) -> list[tuple[str, str]]:
|
|
import archive_pocs
|
|
managed = {row['snapshot']['url'] for row in archive_pocs.load_records().values() if archive_pocs.valid_snapshot(row)}
|
|
if entries is None:
|
|
# Lazy import also lets the site builder consume this ledger.
|
|
from build_site import build_cve_list, load_blacklist
|
|
entries, _ = build_cve_list(load_blacklist())
|
|
return sorted({(row["cve"], url) for row in entries
|
|
for field in ("poc", "nuclei", "edb", "msf", "vulhub", "collections")
|
|
for url in row.get(field, []) if url not in managed})
|
|
|
|
|
|
def qualifying_content(repository: str, cve: str, path: str, source: bytes, reviews: list[dict], *,
|
|
related_cves=()) -> bool:
|
|
"""Confirm that a historical file contains reproduction material, not a stub.
|
|
|
|
This only dates already accepted links. It is not a discovery/approval gate.
|
|
Ambiguous historical content stays undated until reviewed.
|
|
"""
|
|
if cve in source_artifacts.approved_cves_for_artifact(repository, path, source, reviews=reviews):
|
|
return True
|
|
if source_artifacts.identity_cves(path) - {cve, *related_cves}:
|
|
return False
|
|
if len(source.strip()) < 40:
|
|
return False
|
|
text = source.decode("utf-8", "replace")
|
|
name = Path(path).name.lower()
|
|
if repository.lower() == "github/securitylab" and name in {"bug.pdf", "fuzzer-poc.djvu", "poc.bin"}:
|
|
return bool(source)
|
|
if repository.lower() == "github/securitylab" and name.endswith(".cue"):
|
|
return bool(re.search(r"(?m)^\s*(?:FILE|TRACK|INDEX)\b", text))
|
|
if name.endswith((".md", ".rst")) or name == "readme":
|
|
context = ""
|
|
for chunk in re.split(r"(```[^\n]*\n.*?```)", text, flags=re.S):
|
|
if not chunk.startswith("```"):
|
|
context = chunk[-1000:]
|
|
continue
|
|
if not POC_RE.search(context) or re.search(r"(?im)^#+\s*(?:install|setup|requirements|mitigation)", context):
|
|
continue
|
|
if re.search(r"(?m)^\s*(?:curl\s+(?!.*(?:install|update)\.).*(?:https?://|--url)|(?:GET|POST|PUT)\s+/|.*(?:requests\.(?:get|post)|\.sendall)\s*\()", chunk):
|
|
return True
|
|
return False
|
|
if repository in {"tenable/poc", "pedrib/PoC"}:
|
|
from sync_collections import author_artifact_ids
|
|
if cve not in author_artifact_ids(repository, path, text):
|
|
return False
|
|
if name in {"diff.txt", "stroke_patch.txt"} and repository.lower() == "github/securitylab":
|
|
return bool(re.search(r"(?m)^@@ .*@@", text) and re.search(r"(?m)^\+[^+]", text))
|
|
if not (CODE_FILE_RE.search(name) or name.endswith((".html", ".asm", ".lua"))):
|
|
return False
|
|
# Imports/comments alone and placeholder text do not establish introduction.
|
|
executable = "\n".join(line for line in text.splitlines()
|
|
if line.strip() and not re.match(r"\s*(?:#|//|/\*|\*|import\b|from\s+\S+\s+import\b)", line))
|
|
# Function/class declarations, pass/return stubs and arbitrary method calls
|
|
# are not historical exploit evidence. Require an actual action/payload.
|
|
return bool(re.search(r"(?:\b(?:socket|send|sendto|sendmsg|sendmmsg|write|ioctl|syscall|mmap|mprotect|ptrace|"
|
|
r"prctl|setsockopt|msgsnd|msgrcv|setxattr|execve|system|memcpy|strcpy|gets|"
|
|
r"remote|process|fetch|eval|unserialize)\s*\(|"
|
|
r"\b(?:requests|urllib|http|axios|Net::HTTP)\W.{0,40}\b(?:get|post|request|open|start)\s*\(|"
|
|
r"\.(?:send|sendall|send_request_cgi|write|Write|writeObject|Dial|NewRequest|"
|
|
r"connect|Connect|exec|execute|communicate)\s*\(|"
|
|
r"<(?:form|script)\b|\bcurl\s+(?:.*(?:https?://|--url)))", executable))
|
|
|
|
|
|
class GitHubHistory:
|
|
"""Pinned, bounded history reads. Failed requests never become absence."""
|
|
def __init__(self, headers: dict, *, fetch=None):
|
|
self.headers, self.fetch, self.cache = headers, fetch, {}
|
|
self.reviews = source_artifacts.load_reviews()
|
|
self.budget = threading.local()
|
|
|
|
def begin(self, *, budget_seconds: float = 45, max_requests: int = 24) -> None:
|
|
self.budget.deadline = time.monotonic() + budget_seconds
|
|
self.budget.remaining = max_requests
|
|
|
|
def get(self, path: str):
|
|
if path not in self.cache:
|
|
if not hasattr(self.budget, "deadline"):
|
|
self.begin()
|
|
remaining = self.budget.deadline - time.monotonic()
|
|
if remaining <= 0 or self.budget.remaining <= 0:
|
|
raise TimeoutError("Artifact history inspection budget exhausted")
|
|
self.budget.remaining -= 1
|
|
url = "https://api.github.com" + path
|
|
if self.fetch is not None:
|
|
payload = self.fetch(url, headers=self.headers, timeout=min(15, remaining))
|
|
else:
|
|
request = Request(url, headers={"User-Agent": "pocindex-release-evidence", **self.headers})
|
|
with urlopen(request, timeout=min(15, remaining)) as response:
|
|
payload = json.load(response)
|
|
self.cache[path] = payload
|
|
return self.cache[path]
|
|
|
|
def tree(self, repository: str, revision: str, scope: str = "") -> list[dict]:
|
|
expression = revision + (":" + scope if scope else "")
|
|
payload = self.get(f"/repos/{repository}/git/trees/{quote(expression, safe='')}?recursive=1")
|
|
if not isinstance(payload, dict) or payload.get("truncated") is not False or not isinstance(payload.get("tree"), list):
|
|
raise ValueError("Incomplete artifact tree")
|
|
result = []
|
|
for item in payload["tree"]:
|
|
if not isinstance(item, dict) or not source_artifacts.valid_path(str(item.get("path") or "")):
|
|
raise ValueError("Invalid artifact tree entry")
|
|
if item.get("type") == "blob" and item.get("mode") in {"100644", "100755"}:
|
|
result.append({**item, "path": (scope + "/" if scope else "") + item["path"]})
|
|
return result
|
|
|
|
def blob(self, repository: str, entry: dict) -> bytes:
|
|
if not isinstance(entry.get("size"), int) or not 0 < entry["size"] <= 2_000_000:
|
|
raise ValueError("Artifact blob is empty or exceeds the history inspection bound")
|
|
payload = self.get(f"/repos/{repository}/git/blobs/{entry['sha']}")
|
|
if payload.get("encoding") != "base64" or not isinstance(payload.get("content"), str):
|
|
raise ValueError("Missing artifact blob")
|
|
source = base64.b64decode("".join(payload["content"].split()), validate=True)
|
|
digest = hashlib.sha1(f"blob {len(source)}\0".encode() + source).hexdigest()
|
|
if len(source) != entry["size"] or digest != entry["sha"]:
|
|
raise ValueError("Artifact blob differs from its pinned tree")
|
|
return source
|
|
|
|
def qualifying(self, repository: str, revision: str, paths: list[str], cve: str) -> list[dict]:
|
|
# Read each parent tree once; a requested directory is expanded into its
|
|
# real files so its initial README/config commit cannot date the PoC.
|
|
found = {}
|
|
for path in paths:
|
|
parent = path.rsplit("/", 1)[0] if "/" in path else ""
|
|
rows = self.tree(repository, revision, parent)
|
|
candidates = [r for r in rows if r["path"] == path or r["path"].startswith(path.rstrip("/") + "/")]
|
|
explicit = [r for r in candidates if r["path"] == path]
|
|
selected = {r["path"] for r in explicit} or set(artifact_paths(candidates, cve))
|
|
if repository.lower() == "github/securitylab":
|
|
selected.update(r["path"] for r in candidates if Path(r["path"]).name.lower() in {
|
|
"bug.pdf", "fuzzer-poc.djvu", "poc.bin", "diff.txt", "stroke_patch.txt"} or r["path"].endswith(".cue"))
|
|
for entry in candidates:
|
|
size = entry.get("size")
|
|
if not isinstance(size, int) or not 0 < size <= 2_000_000:
|
|
continue
|
|
if entry["path"] in selected and qualifying_content(repository, cve, entry["path"], self.blob(repository, entry), self.reviews):
|
|
found[entry["path"]] = entry
|
|
return list(found.values())
|
|
|
|
def history(self, repository: str, revision: str, path: str) -> list[dict]:
|
|
result = []
|
|
for page in range(1, 21):
|
|
query = urlencode({"sha": revision, "path": path, "per_page": 100, "page": page})
|
|
rows = self.get(f"/repos/{repository}/commits?{query}")
|
|
if not isinstance(rows, list) or any(not isinstance(row, dict) or not re.fullmatch(r"[a-f0-9]{40}", str(row.get("sha") or "")) for row in rows):
|
|
raise ValueError("Incomplete artifact history")
|
|
result.extend(rows)
|
|
if len(rows) < 100:
|
|
return list(reversed(result))
|
|
raise ValueError("Artifact history exceeds the bounded scan; use the local backfill")
|
|
|
|
def introduction(self, repository: str, revision: str, path: str, cve: str, *, depth: int = 0) -> dict:
|
|
if depth > 8:
|
|
raise ValueError("Artifact rename history exceeds inspection bound")
|
|
for commit in self.history(repository, revision, path):
|
|
# First path existence alone is insufficient: inspect those bytes.
|
|
entries = self.qualifying(repository, commit["sha"], [path], cve)
|
|
if not entries:
|
|
continue
|
|
detail = self.get(f"/repos/{repository}/commits/{commit['sha']}")
|
|
files = detail.get("files")
|
|
if not isinstance(files, list) or not any(item.get("filename") == path for item in files):
|
|
raise ValueError("Missing introduction diff")
|
|
renamed = next((item.get("previous_filename") for item in files
|
|
if item.get("filename") == path and item.get("status") == "renamed"), None)
|
|
if renamed and detail.get("parents"):
|
|
parent = detail["parents"][0]["sha"]
|
|
if self.qualifying(repository, parent, [renamed], cve):
|
|
return self.introduction(repository, parent, renamed, cve, depth=depth + 1)
|
|
result = {"commit": commit["commit"]["committer"]["date"], "sha": commit["sha"],
|
|
"commit_verified": True, "history_paths": [entry["path"] for entry in entries]}
|
|
introduced = {cve for item in files if item.get("status") == "added"
|
|
for cve in source_artifacts.identity_cves(item.get("filename", ""))}
|
|
if len(introduced) > 10:
|
|
result["bulk"] = True
|
|
return result
|
|
raise ValueError("No verified qualifying artifact introduction in source history")
|
|
|
|
def absent(self, repository: str, revision: str, paths: list[str]) -> bool:
|
|
# A complete known commit tree is stronger than an ambiguous 404 from
|
|
# a contents endpoint (which may mean the commit/repository is missing).
|
|
entries = self.tree(repository, revision)
|
|
present = {entry["path"] for entry in entries}
|
|
return all(path not in present for path in paths)
|
|
|
|
|
|
def current_repository_evidence(repo: dict, paths: list[str], headers: dict, *, cve: str | None = None,
|
|
cache_dir: Path | None = None, observed_at: str | None = None,
|
|
history: GitHubHistory | None = None, budget_seconds: float = 45,
|
|
max_requests: int = 24) -> dict:
|
|
"""Verify first qualifying file history for an already accepted repository.
|
|
|
|
The supplied HEAD must be captured by the caller. Complete results are
|
|
cached by HEAD, paths and review rules; errors return pending evidence.
|
|
"""
|
|
name = str(repo.get("full_name") or repo.get("nameWithOwner") or "")
|
|
revision = str(repo.get("revision") or repo.get("head") or
|
|
((repo.get("defaultBranchRef") or {}).get("target") or {}).get("oid") or "")
|
|
cve = cve or repo.get("cve")
|
|
if not cve:
|
|
cves = source_artifacts.identity_cves(name)
|
|
cve = next(iter(cves)) if len(cves) == 1 else None
|
|
evidence = {"repo": repo.get("id", repo.get("databaseId")),
|
|
"created": repo.get("created_at", repo.get("createdAt")),
|
|
"pushed_at": repo.get("pushed_at", repo.get("pushedAt")),
|
|
"checked": observed_at or utcnow(), "head": revision}
|
|
evidence = {k: v for k, v in evidence.items() if v is not None}
|
|
if not cve or not re.fullmatch(r"[0-9a-f]{40}", revision) or not paths:
|
|
return {**evidence, "error": "Missing CVE, pinned HEAD or qualifying artifact paths"}
|
|
history = history or GitHubHistory(headers)
|
|
history.begin(budget_seconds=budget_seconds, max_requests=max_requests)
|
|
rules = [(r["cve"], r["path"], r["sha256"]) for r in history.reviews if r["repository"].lower() == name.lower()]
|
|
cache_key = hashlib.sha256(json.dumps([HISTORY_VERSION, name.lower(), revision, cve, sorted(paths), sorted(rules)]).encode()).hexdigest()
|
|
cache_dir = cache_dir or Path(tempfile.gettempdir()) / "pocindex-release-cache"
|
|
cache_path = cache_dir / f"{cache_key}.json"
|
|
try:
|
|
if cache_path.exists():
|
|
cached = json.loads(cache_path.read_text())
|
|
if cached.get("history_version") == HISTORY_VERSION and cached.get("commit_verified") is True:
|
|
cached.setdefault("history_observed", cached.get("checked"))
|
|
return {**cached, **evidence}
|
|
entries = history.qualifying(name, revision, paths, cve)
|
|
if not entries:
|
|
raise ValueError("Selected paths contain no verified reproduction material")
|
|
entrypoints = [entry for entry in entries if source_artifacts.PAYLOAD_RE.search(Path(entry["path"]).name)
|
|
or cve.lower() in Path(entry["path"]).name.lower()]
|
|
if entrypoints:
|
|
entries = entrypoints
|
|
identity = artifact_identity(entries, [entry["path"] for entry in entries])
|
|
introductions = [history.introduction(name, revision, entry["path"], cve) for entry in entries]
|
|
first = min(introductions, key=lambda row: timestamp(row["commit"]))
|
|
result = {**evidence, **identity, **first, "history_version": HISTORY_VERSION,
|
|
"history_observed": evidence["checked"]}
|
|
cache_dir.mkdir(parents=True, exist_ok=True)
|
|
temporary = cache_path.with_suffix(".tmp")
|
|
temporary.write_text(json.dumps(result, sort_keys=True) + "\n")
|
|
temporary.replace(cache_path)
|
|
return result
|
|
except Exception as problem:
|
|
return {**evidence, "error": str(problem)}
|
|
|
|
|
|
def record_collection_snapshot(previous: dict, current: dict, pairs: list[tuple[str, str]], *,
|
|
directory: Path | None = None, observed_at: str | None = None,
|
|
history: GitHubHistory | None = None, limit: int = 8) -> None:
|
|
"""Record only applied source links, after every collection completed.
|
|
|
|
Existing undated imports belong to the historical backfill. New/pending
|
|
pairs get bounded evidence reads; skipped work remains pending for retry.
|
|
"""
|
|
directory = directory or DIRECTORY
|
|
if not directory.exists():
|
|
return # Establish the explicit historical baseline before enabling hooks.
|
|
observed_at = observed_at or utcnow()
|
|
ledger = load_ledger(directory)
|
|
old_sources, sources = previous.get("sources", {}), current.get("sources", {})
|
|
token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")
|
|
history = history or GitHubHistory({"Authorization": f"Bearer {token}"} if token else {})
|
|
present, attempted = set(), 0
|
|
ordered = sorted(set(pairs), key=lambda pair: (
|
|
timestamp(ledger.get(key(*pair), {}).get("checked")) or datetime.min.replace(tzinfo=timezone.utc), pair))
|
|
for cve, url in ordered:
|
|
identity = key(cve, url)
|
|
present.add(identity)
|
|
location = _repository_artifact(url)
|
|
if not location:
|
|
continue
|
|
repository = next((name for name in sources if name.lower() == location[0]), None)
|
|
if repository is None:
|
|
continue
|
|
snapshot, prior = sources[repository], old_sources.get(repository, {})
|
|
old = ledger.get(identity) or {}
|
|
# A newly indexed historical source or newly published CVE record is
|
|
# not evidence of an artifact appearing at this observation.
|
|
already_in_source = url in (prior.get("links", {}).get(cve) or [])
|
|
import_mode = bool(old.get("imported")) if old else not prior.get("observed_at") or already_in_source
|
|
evidence = None
|
|
if not old or old.get("basis") == "pending" or old.get("basis") == "unknown" and not old.get("imported"):
|
|
previous_head = old.get("previous_head") or prior.get("revision")
|
|
absent_at = old.get("absent") or prior.get("observed_at")
|
|
evidence = {"error": "Deferred by collection history budget"}
|
|
if previous_head and absent_at:
|
|
evidence.update(previous_head=previous_head, absent=absent_at)
|
|
if attempted < limit:
|
|
attempted += 1
|
|
evidence.update(checked=observed_at, head=snapshot["revision"])
|
|
try:
|
|
history.begin(budget_seconds=15, max_requests=3)
|
|
repo = history.get(f"/repos/{repository}")
|
|
# The parser and these reads use exactly the same captured HEAD.
|
|
path = unquote(url.split(f"/{snapshot['branch']}/", 1)[1])
|
|
verified = current_repository_evidence({**repo, "revision": snapshot["revision"]}, [path], {},
|
|
cve=cve, history=history, observed_at=observed_at, budget_seconds=30, max_requests=20)
|
|
evidence.update(verified)
|
|
if "error" not in verified:
|
|
evidence.pop("error", None)
|
|
if previous_head and absent_at and not import_mode:
|
|
evidence["absence_verified"] = history.absent(repository, previous_head, verified["paths"])
|
|
except Exception as problem:
|
|
evidence["error"] = str(problem)
|
|
record(ledger, cve, url, evidence, observed_at=observed_at, import_mode=import_mode)
|
|
previously_owned = {key(cve, url) for snapshot in old_sources.values()
|
|
for cve, urls in snapshot.get("links", {}).items() for url in urls}
|
|
removed = previously_owned.intersection(ledger) - present
|
|
if removed:
|
|
published = {key(*pair) for pair in published_pairs()}
|
|
for identity in removed:
|
|
if identity in published:
|
|
ledger[identity].pop("gone", None)
|
|
ledger[identity].pop("gone_reason", None)
|
|
else:
|
|
if not ledger[identity].get('gone'):
|
|
import link_health
|
|
ledger[identity]['gone_reason'] = 'unavailable' if link_health.unavailable(ledger[identity]['url']) else 'withdrawn'
|
|
ledger[identity]['gone'] = observed_at
|
|
resolve_copies(ledger, now=observed_at)
|
|
mark_bulk(ledger)
|
|
save_ledger(ledger, directory)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
mode = parser.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument("--seed", action="store_true", help="historical import; no observation or release dates")
|
|
mode.add_argument("--record", action="store_true", help="reconcile the complete published artifact set")
|
|
parser.add_argument("--directory", type=Path, default=DIRECTORY)
|
|
parser.add_argument("--input", type=Path, help="published CVE_list.json; otherwise parse the current corpus")
|
|
parser.add_argument("--evidence", type=Path, help="JSON mapping of artifact keys to verified evidence")
|
|
parser.add_argument("--dry-run", action="store_true")
|
|
args = parser.parse_args()
|
|
entries = json.loads(args.input.read_text()) if args.input else None
|
|
if entries is None:
|
|
from build_site import build_cve_list, load_blacklist
|
|
entries, _ = build_cve_list(load_blacklist())
|
|
evidence = json.loads(args.evidence.read_text()) if args.evidence else {}
|
|
ledger = reconcile(load_ledger(args.directory), published_pairs(entries), evidence=evidence, import_mode=args.seed)
|
|
for entry in entries:
|
|
if int(entry['cve'].split('-')[1]) < 2026:
|
|
continue
|
|
for field in ('poc', 'nuclei', 'edb', 'msf', 'vulhub', 'collections'):
|
|
for url in entry.get(field, []):
|
|
if urlsplit(url).hostname != 'web.archive.org':
|
|
ledger[key(entry['cve'], url)]['field'] = field
|
|
if not args.dry_run:
|
|
save_ledger(ledger, args.directory)
|
|
print(f"releases: {len(ledger):,} pairs, {sum(bool(row.get('released')) for row in ledger.values()):,} dated")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|