captchaforge 0.2.34

#!/usr/bin/env python3
"""pcap2fixture: convert a captured HTTP exchange into a captchaforge
bench fixture OR adversarial_replay::AdversarialCapture JSON.

Usage:
  python3 tools/pcap2fixture/convert.py [--out-mode bench|adversarial]

Reads every file in `incoming/`, dispatches by extension:
  .har  → HAR v1.2 parser
  .txt  → curl --trace-ascii parser
  .pcap → scapy parser (requires `pip install scapy` + TLS keys)

Emits files into `bench/src/fixtures.rs` (bench mode) or
`adversarial-corpus/<vendor>/<id>.json` (adversarial mode).

Stub for now — protocol locked so the implementation can land
incrementally without disrupting consumers.
"""

import argparse
import hashlib
import json
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent
INCOMING = ROOT / "incoming"
ROOT_REPO = ROOT.parent.parent  # captchaforge/


def parse_har(path: Path) -> dict | None:
    """Extract the first interesting captcha response from a HAR file."""
    har = json.loads(path.read_text())
    for entry in har.get("log", {}).get("entries", []):
        url = entry.get("request", {}).get("url", "")
        if any(s in url for s in ("turnstile", "recaptcha", "hcaptcha", "/cdn-cgi/challenge-platform/")):
            resp = entry.get("response", {})
            content = resp.get("content", {})
            body = content.get("text", "")
            return {
                "url": url,
                "status": resp.get("status", 0),
                "headers": {h["name"].lower(): h["value"] for h in resp.get("headers", [])},
                "body": body,
            }
    return None


def parse_curl_trace(path: Path) -> dict | None:
    """Minimal curl --trace-ascii parser. Returns the first response."""
    text = path.read_text()
    # Curl trace marks response bodies with `<= Recv data`.
    if "<= Recv data" not in text:
        return None
    body_chunks = []
    for line in text.splitlines():
        if line.startswith("0000:"):
            # `0000: 3c 21 44 4f 43 54 ...   <!DOCTY...` ascii right of last spaces
            ascii_part = line.split("  ", 1)
            if len(ascii_part) > 1:
                body_chunks.append(ascii_part[1])
    body = "".join(body_chunks)
    if not body:
        return None
    return {
        "url": "unknown",
        "status": 0,
        "headers": {},
        "body": body,
    }


def parse_pcap(_path: Path) -> dict | None:
    print("[pcap2fixture] PCAP parsing not yet implemented; "
          "use HAR or curl --trace-ascii.", file=sys.stderr)
    return None


def to_adversarial(captured: dict, vendor: str, out_root: Path) -> Path:
    body = captured["body"] or ""
    cid = hashlib.sha256(body.encode("utf-8", "replace")).hexdigest()[:16]
    payload = {
        "id": cid,
        "vendor": vendor,
        "status": captured.get("status", 0),
        "headers": captured.get("headers", {}),
        "body": body,
        "cookie_names": [],
        "expected": "recognised",
        "notes": f"pcap2fixture import from {captured.get('url', 'unknown')}",
        "captured_at_unix": int(time.time()),
    }
    dest_dir = out_root / vendor
    dest_dir.mkdir(parents=True, exist_ok=True)
    dest = dest_dir / f"{cid}.json"
    dest.write_text(json.dumps(payload, indent=2))
    return dest


def infer_vendor(captured: dict) -> str:
    url = captured.get("url", "").lower()
    body = (captured.get("body") or "").lower()
    if "turnstile" in url or "cf-turnstile" in body:
        return "cloudflare-turnstile"
    if "/cdn-cgi/challenge-platform/" in url:
        return "cloudflare-interstitial"
    if "hcaptcha" in url or "h-captcha" in body:
        return "hcaptcha"
    if "recaptcha" in url or "g-recaptcha" in body:
        return "recaptcha-v2"
    return "unknown"


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--out-mode",
        choices=["bench", "adversarial"],
        default="adversarial",
    )
    args = parser.parse_args()

    if not INCOMING.is_dir():
        INCOMING.mkdir(parents=True, exist_ok=True)
        print(f"[pcap2fixture] created {INCOMING} — drop captures in and rerun")
        return 0

    files = sorted(INCOMING.iterdir())
    if not files:
        print(f"[pcap2fixture] {INCOMING} is empty")
        return 0

    out_root = ROOT_REPO / "adversarial-corpus"
    count = 0
    for path in files:
        if path.suffix == ".har":
            captured = parse_har(path)
        elif path.suffix == ".txt":
            captured = parse_curl_trace(path)
        elif path.suffix == ".pcap":
            captured = parse_pcap(path)
        else:
            print(f"[pcap2fixture] skipping {path.name} (unknown extension)")
            continue
        if not captured:
            print(f"[pcap2fixture] no challenge response found in {path.name}")
            continue
        vendor = infer_vendor(captured)
        if args.out_mode == "adversarial":
            dest = to_adversarial(captured, vendor, out_root)
            print(f"[pcap2fixture] {path.name} → {dest.relative_to(ROOT_REPO)}")
            count += 1
        else:
            print(f"[pcap2fixture] bench-mode emission not yet implemented "
                  "(use adversarial mode)")
    print(f"[pcap2fixture] {count} capture(s) imported")
    return 0


if __name__ == "__main__":
    sys.exit(main())