chuanqi/cq2_sdk/scripts/analyze_seq.py

"""Analyze the seq/flags of captured game-gateway packets.

Answers (from a capture jsonl with the seq/flags-enhanced capture_cq2.py):
  1. Are the 3x 6B zero packets the same seq (retransmission) or distinct?
  2. Does the server FIN or RST after them?
"""
import json
import sys
from pathlib import Path

path = Path(sys.argv[1] if len(sys.argv) > 1 else "wire_test2.jsonl")
if not path.exists():
    print(f"{path} not found", file=sys.stderr)
    sys.exit(1)

FLAG = {0x01: "FIN", 0x02: "SYN", 0x04: "RST", 0x08: "PSH", 0x10: "ACK"}


def fstr(f):
    return "+".join(n for b, n in FLAG.items() if f & b) or "."


recs = []
with path.open(encoding="utf-8") as f:
    for line in f:
        line = line.strip()
        if not line:
            continue
        try:
            recs.append(json.loads(line))
        except json.JSONDecodeError:
            continue

print(f"=== {len(recs)} records from {path} ===\n")
t0 = recs[0]["ts"] if recs else 0
for r in recs:
    t = r.get("ts", 0) - t0
    if r.get("ctrl"):
        print(f"  t+{t:6.3f}s  {r['src']:>22} -> {r['dst']:<22} [CTRL {fstr(r.get('flags',0))}] seq={r.get('seq')}")
    else:
        seq = r.get("seq", "?")
        fl = fstr(r.get("flags", 0))
        print(f"  t+{t:6.3f}s  {r['src']:>22} -> {r['dst']:<22} DATA[{r['len']:>3}B] seq={seq} {fl} hex={r.get('hex','')[:24]}")

# Analysis: the 6B zeros
zeros = [r for r in recs if r.get("len") == 6 and r.get("hex") == "0" * 12]
print(f"\n--- 6B-zero packets: {len(zeros)} ---")
if zeros:
    seqs = [r.get("seq") for r in zeros]
    print(f"seqs: {seqs}")
    if len(set(seqs)) == 1:
        print("*** ALL SAME SEQ -> TCP RETRANSMISSIONS of ONE 6-byte frame ***")
        print("    (server sent ONE 6B zero frame; our side didn't ACK fast enough)")
    else:
        print("*** DISTINCT SEQs -> server deliberately sent multiple 6B frames ***")
        diffs = [round(zeros[i + 1]["ts"] - zeros[i]["ts"], 3) for i in range(len(zeros) - 1)]
        print(f"    inter-arrival: {diffs}s")

ctrls = [r for r in recs if r.get("ctrl")]
fins = [r for r in ctrls if r.get("flags", 0) & 0x01]
rsts = [r for r in ctrls if r.get("flags", 0) & 0x04]
print(f"\n--- control packets: FIN={len(fins)} RST={len(rsts)} ---")
if fins:
    print("    server FINs -> graceful close (we can detect EOF properly)")
if rsts:
    print("    RST present -> hard abort")