chuanqi/cq2_sdk/scripts/parse_saz.py

"""Parse a Fiddler .saz capture and extract request/response pairs.

A .saz file is just a ZIP archive. Inside, each captured session has:
    raw/<n>_c_request.txt    — client request (request line + headers + body)
    raw/<n>_s_response.txt   — server response (status line + headers + body)
    raw/<n>_c_mclient.txt    — metadata (client IP, timing)

Usage:
    python scripts/parse_saz.py <file.saz>                     # list all sessions (method URL status)
    python scripts/parse_saz.py <file.saz> --kw sms            # only sessions whose req/resp mentions 'sms'
    python scripts/parse_saz.py <file.saz> --kw phone --dump   # also print full req/resp bodies
    python scripts/parse_saz.py <file.saz> --index 42 --dump   # dump one session by number
"""
from __future__ import annotations
import argparse
import io
import re
import sys
import zipfile
from pathlib import Path

# Windows GBK console cannot print \ufffd etc. — force UTF-8 with replacement.
if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
    sys.stderr.reconfigure(encoding="utf-8", errors="replace")


def parse_request(text: bytes) -> dict:
    """Split a raw HTTP request into method/url/headers/body."""
    # Split headers from body at first blank line
    parts = text.split(b"\r\n\r\n", 1)
    head = parts[0].decode("utf-8", errors="replace")
    body = parts[1] if len(parts) > 1 else b""
    lines = head.split("\r\n")
    request_line = lines[0] if lines else ""
    m = re.match(r"([A-Z]+)\s+(\S+)\s+HTTP", request_line)
    method, url = (m.group(1), m.group(2)) if m else ("?", request_line)
    headers: dict[str, str] = {}
    for ln in lines[1:]:
        if ":" in ln:
            k, v = ln.split(":", 1)
            headers[k.strip().lower()] = v.strip()
    return {"method": method, "url": url, "headers": headers, "body": body}


def parse_response(text: bytes) -> dict:
    """Split a raw HTTP response into status/headers/body."""
    parts = text.split(b"\r\n\r\n", 1)
    head = parts[0].decode("utf-8", errors="replace")
    body = parts[1] if len(parts) > 1 else b""
    lines = head.split("\r\n")
    status_line = lines[0] if lines else ""
    m = re.match(r"HTTP/[\d.]+\s+(\d+)", status_line)
    status = int(m.group(1)) if m else 0
    headers: dict[str, str] = {}
    for ln in lines[1:]:
        if ":" in ln:
            k, v = ln.split(":", 1)
            headers[k.strip().lower()] = v.strip()
    # Handle chunked encoding crudely: strip chunk size lines
    if headers.get("transfer-encoding", "").lower() == "chunked":
        body = dechunk(body)
    return {"status": status, "headers": headers, "body": body}


def dechunk(body: bytes) -> bytes:
    """Remove HTTP chunked-encoding frame markers."""
    out = io.BytesIO()
    pos = 0
    while pos < len(body):
        nl = body.find(b"\r\n", pos)
        if nl < 0:
            out.write(body[pos:])
            break
        try:
            size = int(body[pos:nl], 16)
        except ValueError:
            out.write(body[pos:])
            break
        if size == 0:
            break
        out.write(body[nl + 2: nl + 2 + size])
        pos = nl + 2 + size + 2  # skip chunk + trailing CRLF
    return out.getvalue()


def maybe_decompress(resp: dict) -> bytes:
    body = resp["body"]
    enc = resp["headers"].get("content-encoding", "")
    if "gzip" in enc:
        import gzip
        try:
            return gzip.decompress(body)
        except Exception:
            return body
    if "br" in enc:
        try:
            import brotli
            return brotli.decompress(body)
        except ImportError:
            return body  # brotli lib missing; raw bytes shown as-is
        except Exception:
            return body
    if "deflate" in enc:
        import zlib
        try:
            return zlib.decompress(body)
        except Exception:
            return zlib.decompress(body, -zlib.MAX_WBITS)
    return body


def load_sessions(saz_path: Path) -> list[dict]:
    """Load all sessions from a .saz zip.

    Fiddler uses two naming conventions depending on version:
      raw/<n>_c_request.txt + raw/<n>_s_response.txt   (newer)
      raw/<n>_c.txt       + raw/<n>_s.txt              (older / this capture)
    Detect which one exists and use it.
    """
    sessions: list[dict] = []
    with zipfile.ZipFile(saz_path) as zf:
        names = set(zf.namelist())
        # Detect naming convention
        req_files = sorted(
            (n for n in names if re.search(r"raw/(\d+)_c(_request)?\.txt$", n)),
            key=lambda n: int(re.search(r"(\d+)", n).group(1)),
        )
        for req_name in req_files:
            num = int(re.search(r"(\d+)", req_name).group(1))
            # Find the matching response file (either convention)
            resp_name = next(
                (n for n in names
                 if re.search(rf"raw/{num:03d}_s(_response)?\.txt$", n)
                 or re.search(rf"raw/{num}_s(_response)?\.txt$", n)),
                None,
            )
            if resp_name is None:
                continue
            try:
                req = parse_request(zf.read(req_name))
                resp = parse_response(zf.read(resp_name))
            except Exception as e:
                print(f"  [warn] session {num}: parse error {e}", file=sys.stderr)
                continue
            resp_body = maybe_decompress(resp)
            sessions.append({
                "num": num,
                "method": req["method"],
                "url": req["url"],
                "req_headers": req["headers"],
                "req_body": req["body"],
                "status": resp["status"],
                "resp_headers": resp["headers"],
                "resp_body": resp_body,
            })
    return sessions


def body_text(b: bytes, limit: int = 2000) -> str:
    try:
        return b.decode("utf-8", errors="replace")[:limit]
    except Exception:
        return repr(b[:limit])


def main() -> int:
    parser = argparse.ArgumentParser(description="Parse Fiddler .saz capture")
    parser.add_argument("saz", help="Path to .saz file")
    parser.add_argument("--kw", default=None,
                        help="Only show sessions where url or bodies contain this keyword")
    parser.add_argument("--dump", action="store_true",
                        help="Print full request/response bodies for matched sessions")
    parser.add_argument("--index", type=int, default=None,
                        help="Dump a single session by number (ignores --kw)")
    parser.add_argument("--limit", type=int, default=30,
                        help="Max sessions to list (default 30)")
    args = parser.parse_args()

    saz = Path(args.saz)
    if not saz.exists():
        print(f"FAIL: {saz} not found")
        return 1

    sessions = load_sessions(saz)
    print(f"loaded {len(sessions)} sessions from {saz.name}\n")

    def matches(s: dict) -> bool:
        if not args.kw:
            return True
        kw = args.kw.lower()
        hay = (s["url"] + " " +
               body_text(s["req_body"], 4000) + " " +
               body_text(s["resp_body"], 4000)).lower()
        return kw in hay

    shown = 0
    for s in sessions:
        if args.index is not None and s["num"] != args.index:
            continue
        if not matches(s):
            continue
        shown += 1
        if shown > args.limit and not args.dump:
            print(f"  ... (more hidden, use --limit)")
            break
        print(f"[{s['num']:>4}] {s['method']:<6} {s['status']:>3}  {s['url'][:110]}")
        if args.dump or args.index is not None:
            print(f"      --- request headers ---")
            for k, v in s["req_headers"].items():
                print(f"      {k}: {v[:120]}")
            if s["req_body"]:
                print(f"      --- request body ({len(s['req_body'])} bytes) ---")
                print(f"      {body_text(s['req_body'], 1500)}")
            print(f"      --- response body ({len(s['resp_body'])} bytes) ---")
            print(f"      {body_text(s['resp_body'], 1500)}")
            print()

    if args.kw and shown == 0:
        print(f"[!] no sessions matched keyword {args.kw!r}")
    return 0


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