ksjsb/tools/analyze_region_ticket.py

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import re
from pathlib import Path
from typing import Any, Iterator
from urllib.parse import urlsplit, urlunsplit


_REGION_COOKIE_RE = re.compile(r"(?:^|;\s*)region_ticket=([^;]*)", re.IGNORECASE)


def _endpoint(url: str) -> str:
    parts = urlsplit(url)
    return urlunsplit((parts.scheme, parts.netloc, parts.path, "", ""))


def _headers(message: dict[str, Any]) -> Iterator[tuple[str, str]]:
    for header in message.get("headers") or []:
        if not isinstance(header, dict):
            continue
        name = header.get("name")
        value = header.get("value")
        if isinstance(name, str) and isinstance(value, str):
            yield name.lower(), value


def _ticket_summary(ticket: str) -> dict[str, Any]:
    return {
        "prefix": ticket[:3],
        "length": len(ticket),
        "sha256_12": hashlib.sha256(ticket.encode()).hexdigest()[:12],
    }


def _cookie_ticket(message: dict[str, Any]) -> str | None:
    for name, value in _headers(message):
        if name != "cookie":
            continue
        match = _REGION_COOKIE_RE.search(value)
        if match and match.group(1):
            return match.group(1)
    for cookie in message.get("cookies") or []:
        if not isinstance(cookie, dict):
            continue
        if str(cookie.get("name", "")).lower() == "region_ticket":
            value = cookie.get("value")
            if isinstance(value, str) and value:
                return value
    return None


def _set_cookie_tickets(message: dict[str, Any]) -> Iterator[str]:
    for name, value in _headers(message):
        if name != "set-cookie":
            continue
        match = _REGION_COOKIE_RE.search(value)
        if match and match.group(1):
            yield match.group(1)


def _response_json(response: dict[str, Any]) -> Any | None:
    content = response.get("content") or {}
    text = content.get("text")
    if not isinstance(text, str) or not text:
        return None
    if str(content.get("encoding", "")).lower() == "base64":
        try:
            text = base64.b64decode(text).decode("utf-8", errors="replace")
        except (ValueError, UnicodeError):
            return None
    try:
        return json.loads(text.lstrip("\ufeff"))
    except (TypeError, ValueError):
        return None


def _region_nodes(value: Any, path: str = "$") -> Iterator[tuple[str, dict[str, Any]]]:
    if isinstance(value, dict):
        for key, child in value.items():
            child_path = f"{path}.{key}"
            if key == "region" and isinstance(child, dict):
                ticket = child.get("ticket")
                if isinstance(ticket, str) and ticket:
                    yield child_path, child
            yield from _region_nodes(child, child_path)
    elif isinstance(value, list):
        for index, child in enumerate(value):
            yield from _region_nodes(child, f"{path}[{index}]")


def _event_base(index: int, entry: dict[str, Any]) -> dict[str, Any]:
    request = entry.get("request") or {}
    response = entry.get("response") or {}
    return {
        "entry_index": index,
        "started": entry.get("startedDateTime", ""),
        "method": request.get("method", ""),
        "url": _endpoint(str(request.get("url", ""))),
        "status": response.get("status", 0),
    }


def analyze_har(path: Path) -> dict[str, Any]:
    with path.open("r", encoding="utf-8", errors="replace") as stream:
        har = json.load(stream)

    entries = har.get("log", {}).get("entries", [])
    response_regions: list[dict[str, Any]] = []
    response_set_cookies: list[dict[str, Any]] = []
    request_cookies: list[dict[str, Any]] = []

    for index, entry in enumerate(entries):
        if not isinstance(entry, dict):
            continue
        request = entry.get("request") or {}
        response = entry.get("response") or {}
        event = _event_base(index, entry)

        request_ticket = _cookie_ticket(request)
        if request_ticket:
            request_cookies.append({**event, "ticket": _ticket_summary(request_ticket)})

        for ticket in _set_cookie_tickets(response):
            response_set_cookies.append({**event, "ticket": _ticket_summary(ticket)})

        body = _response_json(response)
        for json_path, region in _region_nodes(body):
            ticket = str(region["ticket"])
            response_regions.append(
                {
                    **event,
                    "json_path": json_path,
                    "uid": str(region.get("uid", "")),
                    "name": str(region.get("name", "")),
                    "ticket": _ticket_summary(ticket),
                }
            )

    return {
        "source": str(path),
        "entry_count": len(entries),
        "response_regions": response_regions,
        "response_set_cookies": response_set_cookies,
        "request_cookies": request_cookies,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="审计 HAR 中 region_ticket 的下发和使用时间线")
    parser.add_argument("har", nargs="+", type=Path)
    parser.add_argument("--out", type=Path)
    args = parser.parse_args()

    reports = [analyze_har(path) for path in args.har]
    output = json.dumps(reports, ensure_ascii=False, indent=2)
    if args.out:
        args.out.parent.mkdir(parents=True, exist_ok=True)
        args.out.write_text(output + "\n", encoding="utf-8")
    else:
        print(output)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())