osint-investigation

1#!/usr/bin/env python32"""Search the Internet Archive Wayback Machine via the CDX server.3 4The CDX API indexes ~900B+ archived web pages. Anonymous read access,5no auth required. Useful for finding deleted / changed pages by URL,6domain, or substring match.7"""8from __future__ import annotations9 10import argparse11import csv12import sys13import urllib.parse14from pathlib import Path15 16sys.path.insert(0, str(Path(__file__).parent))17from _http import get_json  # noqa: E40218 19BASE = "https://web.archive.org/cdx/search/cdx"20 21COLUMNS = [22    "url",23    "timestamp",24    "wayback_url",25    "mimetype",26    "status",27    "digest",28    "length",29]30 31 32def fetch(33    url_or_host: str,34    match_type: str,35    from_date: str | None,36    to_date: str | None,37    status: str | None,38    mime: str | None,39    collapse: str | None,40    limit: int,41    out_path: str,42) -> int:43    params: dict[str, str] = {44        "url": url_or_host,45        "matchType": match_type,46        "output": "json",47        "limit": str(limit),48    }49    if from_date:50        params["from"] = from_date.replace("-", "")51    if to_date:52        params["to"] = to_date.replace("-", "")53    if status:54        params["filter"] = f"statuscode:{status}"55    if mime:56        params.setdefault("filter", "")57        # Multiple filters: CDX accepts repeated filter params via urlencode list58        params["filter"] = f"mimetype:{mime}"59    if collapse:60        params["collapse"] = collapse61 62    url = f"{BASE}?{urllib.parse.urlencode(params)}"63    try:64        payload = get_json(url)65    except Exception as e:  # noqa: BLE00166        print(f"Wayback CDX error: {e}", file=sys.stderr)67        payload = []68 69    rows: list[dict[str, str]] = []70    if isinstance(payload, list) and len(payload) > 1:71        header = payload[0]72        idx = {h: i for i, h in enumerate(header)}73        for entry in payload[1:]:74            ts = entry[idx["timestamp"]] if "timestamp" in idx else ""75            orig = entry[idx["original"]] if "original" in idx else ""76            rows.append(77                {78                    "url": orig,79                    "timestamp": ts,80                    "wayback_url": f"https://web.archive.org/web/{ts}/{orig}" if ts and orig else "",81                    "mimetype": entry[idx["mimetype"]] if "mimetype" in idx else "",82                    "status": entry[idx["statuscode"]] if "statuscode" in idx else "",83                    "digest": entry[idx["digest"]] if "digest" in idx else "",84                    "length": entry[idx["length"]] if "length" in idx else "",85                }86            )87 88    Path(out_path).parent.mkdir(parents=True, exist_ok=True)89    with open(out_path, "w", newline="", encoding="utf-8") as fh:90        w = csv.DictWriter(fh, fieldnames=COLUMNS)91        w.writeheader()92        w.writerows(rows)93    if not rows:94        print(95            f"Wayback Machine: 0 captures for {url_or_host!r} matchType={match_type}.",96            file=sys.stderr,97        )98    return len(rows)99 100 101def main() -> int:102    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)103    p.add_argument("--url", required=True, help="URL or host to look up in the archive")104    p.add_argument(105        "--match",106        default="exact",107        choices=["exact", "prefix", "host", "domain"],108        help=(109            "exact: this URL only. "110            "prefix: this URL's path-prefix. "111            "host: any URL on this host. "112            "domain: any URL on this domain or subdomains."113        ),114    )115    p.add_argument("--from-date", help="Earliest capture YYYY-MM-DD")116    p.add_argument("--to-date", help="Latest capture YYYY-MM-DD")117    p.add_argument("--status", help="HTTP status filter (e.g. 200)")118    p.add_argument("--mime", help="MIME type filter (e.g. text/html)")119    p.add_argument(120        "--collapse",121        help="Collapse adjacent identical entries (e.g. 'digest' for unique-content captures)",122    )123    p.add_argument("--limit", type=int, default=200)124    p.add_argument("--out", required=True)125    a = p.parse_args()126    n = fetch(127        url_or_host=a.url,128        match_type=a.match,129        from_date=a.from_date,130        to_date=a.to_date,131        status=a.status,132        mime=a.mime,133        collapse=a.collapse,134        limit=a.limit,135        out_path=a.out,136    )137    print(f"Wrote {n} Wayback capture rows to {a.out}")138    return 0139 140 141if __name__ == "__main__":142    raise SystemExit(main())143