Add host_probe5 (session sequencing + close-vs-timeout)

scripts/host_probe5.py

@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""Host probe v5 — session behaviour on 9000 (stdlib only).
+
+v4 (correct ids + opt payloads) is still met with silence, so this probe
+tests *session* hypotheses instead of bytes:
+* persistent connection: discovery → directed queries on ONE socket, with
+  app-like sequential snno (the app reuses its SocketManager connection;
+  the field captures show a discovery frame minutes before directed ones);
+* close-vs-timeout distinction for every read (immediate close = active
+  reject; 8 s idle + open conn = device thinking/forwarding upstream);
+* high-snno opens (in case the device tracks sequence per sender).
+"""
+
+import argparse
+import json
+import socket
+import time
+import zlib
+
+APOWER, QUERY_INFO, BMS = 0x51B, 0x44D, 0x6A9
+SNNO = [100]
+
+
+def secret_key(sn):
+    return format(sum(sn.encode()) & 0xFF, "x")
+
+
+def build_envelope(sn, cmd, args=None, snno=None, ts=None, equip=None):
+    if snno is None:
+        SNNO[0] = (SNNO[0] % 9999) + 1
+        snno = SNNO[0]
+    data = {"opt": 0}
+    if args:
+        data.update(args)
+    compact = json.dumps(data, separators=(",", ":"))
+    env = {"cmdType": cmd, "equipNo": equip if equip is not None else sn,
+           "type": 0,
+           "timeStamp": ts if ts is not None else int(time.time()),
+           "snno": snno,
+           "len": len(compact.encode()),
+           "crc": format(zlib.crc32(compact.encode()), "X"),
+           "dataArea": data}
+    return json.dumps(env, separators=(",", ":"))
+
+
+def encrypt(json_str, key):
+    parts = json_str.split(",")
+    a = (parts[0] + "," + parts[1] + ",").encode()
+    b = ",".join(parts[2:]).encode()
+    n = int(key, 32)
+    return a + bytes((b[i] + i + len(a) + n) & 0xFF
+                     for i in range(len(b)))
+
+
+def recv_all_verbose(sock, idle=8.0):
+    """Return (bytes, how) where how is closed|idle-timeout."""
+    out = bytearray()
+    sock.settimeout(idle)
+    while True:
+        try:
+            chunk = sock.recv(65536)
+        except (socket.timeout, TimeoutError):
+            return bytes(out), "idle-timeout(%.1fs)" % idle
+        if not chunk:
+            return bytes(out), "closed-by-peer"
+        out.extend(chunk)
+
+
+def one_shot(label, host, port, tx, timeout, snno):
+    try:
+        with socket.create_connection((host, port),
+                                      timeout=timeout) as sock:
+            t0 = time.time()
+            sock.sendall(tx)
+            rx, how = recv_all_verbose(sock)
+            dt = time.time() - t0
+    except (ConnectionError, socket.error, TimeoutError) as exc:
+        print("--- %s snno=%d -> %s:%d" % (label, snno, host, port))
+        print("  tx (%d B): %s" % (len(tx), tx.hex()))
+        print("  error: %s: %s" % (type(exc).__name__, exc))
+        return b""
+    print("--- %s snno=%d -> %s:%d (%.1fs, %s)" % (
+        label, snno, host, port, dt, how))
+    print("  tx (%d B): %s" % (len(tx), tx.hex()))
+    print("  rx (%d B): %s" % (len(rx), rx.hex() or "(empty)"))
+    return rx
+
+
+def session(label, host, port, frames, timeout):
+    print("=== %s -> %s:%d (one connection)" % (label, host, port))
+    try:
+        with socket.create_connection((host, port),
+                                      timeout=timeout) as sock:
+            for name, tx in frames:
+                t0 = time.time()
+                sock.sendall(tx)
+                rx, how = recv_all_verbose(sock)
+                dt = time.time() - t0
+                print("--- %s tx=%dB rx=%dB (%.1fs, %s)" % (
+                    name, len(tx), len(rx), dt, how))
+                if rx:
+                    print("  rx: %s" % rx.hex()[:3000])
+                if how == "closed-by-peer" and not rx:
+                    print("  (peer closed; stopping session)")
+                    break
+    except (ConnectionError, socket.error, TimeoutError) as exc:
+        print("  error: %s: %s" % (type(exc).__name__, exc))
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--sn", required=True)
+    ap.add_argument("--host", default="10.100.1.1")
+    ap.add_argument("--ports", default="9000")
+    ap.add_argument("--timeout", type=float, default=5.0)
+    args = ap.parse_args()
+    key = secret_key(args.sn)
+    print("secret key for %s: %s (radix 32)" % (args.sn, key))
+
+    def tx(cmd, data, equip=None, snno=None):
+        return encrypt(build_envelope(
+            args.sn, cmd, data, snno=snno,
+            equip=args.sn if equip is None else equip or "00000000"), key)
+
+    ports = [int(p) for p in args.ports.split(",") if p.strip()]
+    for port in ports:
+        # A. persistent session: discovery first, then directed (app order)
+        session("discovery-then-directed", args.host, port, [
+            ("disc/1101+empty",
+             tx(QUERY_INFO, {"minProtocolVer": "V1.00.00"}, equip="")),
+            ("info/1101+SN",
+             tx(QUERY_INFO, {"minProtocolVer": "V1.00.00"})),
+            ("apower/1307",
+             tx(APOWER, {"update": 2})),
+            ("bms/1705", tx(BMS, None)),
+        ], args.timeout)
+        # B. high-snno fresh opens
+        one_shot("apower-high-snno", args.host, port,
+                 tx(APOWER, {"update": 2}, snno=500), args.timeout, 500)
+        one_shot("info-high-snno", args.host, port,
+                 tx(QUERY_INFO, {"minProtocolVer": "V1.00.00"}, snno=501),
+                 args.timeout, 501)
+
+
+if __name__ == "__main__":
+    main()