From 2cdc274fe37086cc15e9c2dcefcff37aac56faac Mon Sep 17 00:00:00 2001
From: kfj001 <kfj001@gmail.com>
Date: Thu, 15 Oct 2026 23:53:41 +0000
Subject: [PATCH] Add standalone host_probe2.py (stdlib-only transport probes)

---
 scripts/host_probe2.py | 133 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 133 insertions(+)
 create mode 100644 scripts/host_probe2.py

diff --git a/scripts/host_probe2.py b/scripts/host_probe2.py
new file mode 100644
index 0000000..92c48aa
--- /dev/null
+++ b/scripts/host_probe2.py
@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""Standalone host probe — no package install, no git needed (stdlib only).
+
+Drop this file on the machine joined to the aGate SoftAP and run:
+    python3 host_probe2.py --sn A03X2516013210060005
+
+Replicates ``battery --probe2``: the canonical queryAPowerData envelope
+(verified byte-for-byte against the Dart AOT disassembly) sent with
+transport-level variants. Prints exact TX/RX hex for each.
+"""
+
+import argparse
+import json
+import socket
+import time
+import zlib
+
+SNNO = [0]  # 1-based increaseSN counter (wraps at 9999)
+
+
+def secret_key(sn):
+    return format(sum(sn.encode()) & 0xFF, "x")
+
+
+def build_envelope(sn, cmd=25, data=None, snno=None, ts=None):
+    SNNO[0] = (SNNO[0] % 9999) + 1
+    env = {
+        "cmdType": cmd,
+        "equipNo": sn,
+        "type": 0,
+        "timeStamp": ts if ts is not None else int(time.time()),
+        "snno": snno if snno is not None else SNNO[0],
+        "len": 0,
+        "crc": "",
+    }
+    if data:
+        compact = json.dumps(data, separators=(",", ":"))
+        env["len"] = len(compact.encode())
+        env["crc"] = format(zlib.crc32(compact.encode()), "X")
+        env["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(sock, idle=4.0):
+    out = bytearray()
+    sock.settimeout(idle)
+    while True:
+        try:
+            chunk = sock.recv(65536)
+        except (socket.timeout, TimeoutError):
+            break
+        if not chunk:
+            break
+        out.extend(chunk)
+    return bytes(out)
+
+
+MQTT_CONNECT = bytes.fromhex("101200044d5154540402003c000670726f62653031")
+
+
+def run(label, host, port, tx, timeout, pre_read=0.0, post_idle=4.0,
+        split_send=False):
+    entry = (label, tx, b"", b"", None)
+    rx_pre, rx = b"", b""
+    err = None
+    try:
+        with socket.create_connection((host, port),
+                                      timeout=timeout) as sock:
+            if pre_read:
+                rx_pre = recv_all(sock, idle=pre_read)
+            if tx:
+                if split_send:
+                    half = len(tx) // 2
+                    sock.sendall(tx[:half])
+                    try:
+                        sock.settimeout(0.5)
+                        rx_pre += sock.recv(65536)
+                    except (socket.timeout, TimeoutError):
+                        pass
+                    sock.sendall(tx[half:])
+                else:
+                    sock.sendall(tx)
+                rx = recv_all(sock, idle=post_idle)
+    except (ConnectionError, socket.error, TimeoutError) as exc:
+        err = "%s: %s" % (type(exc).__name__, exc)
+    print("--- %s -> %s:%d" % (label, host, port))
+    print("  tx (%d B): %s" % (len(tx), tx.hex() or "(none)"))
+    if rx_pre:
+        print("  rx_pre (%d B): %s" % (len(rx_pre), rx_pre.hex()))
+    print("  rx (%d B): %s" % (len(rx), rx.hex() or "(empty)"))
+    if err:
+        print("  error: %s" % err)
+    return rx
+
+
+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))
+    tx = encrypt(build_envelope(args.sn, data={"update": 2}), key)
+    for port in [int(p) for p in args.ports.split(",") if p.strip()]:
+        run("greet-first/read-then-raw", args.host, port, tx,
+            args.timeout, pre_read=4.0)
+        run("raw/long-read", args.host, port, tx, args.timeout,
+            post_idle=8.0)
+        run("raw/split-send", args.host, port, tx, args.timeout,
+            split_send=True)
+        run("le16-prefix", args.host, port,
+            len(tx).to_bytes(2, "little") + tx, args.timeout)
+        run("be16-prefix", args.host, port,
+            len(tx).to_bytes(2, "big") + tx, args.timeout)
+        run("le32-prefix", args.host, port,
+            len(tx).to_bytes(4, "little") + tx, args.timeout)
+        run("mqtt-connect", args.host, port, MQTT_CONNECT, args.timeout)
+
+
+if __name__ == "__main__":
+    main()