captchaforge 0.2.40

#!/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())