From 62e0fd649b3fcb1769c9b5a32799e4cc4ca7b2cd Mon Sep 17 00:00:00 2001 From: kfj001 Date: Fri, 16 Oct 2026 02:50:23 +0000 Subject: [PATCH] Radix 16 (live-verified), universal reply splitter, golden capture vectors, host_probe6 --- docs/RE_DART_FINDINGS.md | 45 +++-- franklinwh_agate_client/battery.py | 24 ++- franklinwh_agate_client/device_local.py | 31 +-- franklinwh_agate_client/protocol/__init__.py | 4 + franklinwh_agate_client/protocol/crypto.py | 95 ++++++--- franklinwh_agate_client/protocol/framing.py | 7 +- scripts/host_probe6.py | 194 +++++++++++++++++++ tests/fixtures/device_info_reply.bin | Bin 0 -> 1154 bytes tests/fixtures/discovery_req.bin | 2 + tests/test_protocol.py | 98 +++++++--- 10 files changed, 418 insertions(+), 82 deletions(-) create mode 100644 scripts/host_probe6.py create mode 100644 tests/fixtures/device_info_reply.bin create mode 100644 tests/fixtures/discovery_req.bin diff --git a/docs/RE_DART_FINDINGS.md b/docs/RE_DART_FINDINGS.md index 1a0edb0..bed80ec 100644 --- a/docs/RE_DART_FINDINGS.md +++ b/docs/RE_DART_FINDINGS.md @@ -161,14 +161,20 @@ unpadded. E.g. SN `A03X2516013210060005` → byte sum `1052 & 0xFF = 28` → `"1 2. `S1 = parts[0] + "," + parts[1] + ","` (header, *with* trailing comma); `a = UTF-8(S1)`. 3. `S2 = join(",", parts[2..])` (the rest); `b = UTF-8(S2)`. -4. `keyInt = int.tryParse(key, radix: 32)` (literal `0x20` at the call site; - null → 0). +4. `keyInt = int.tryParse(key)` (default radix **16** — two live frames + decrypt cleanly only with radix 16 and then verify byte-exact on + `len`/CRC; an earlier radix-32 reading of a nearby `0x20` literal is + superseded). 5. `out[i] = (b[i] + i + len(a) + keyInt) & 0xFF` for all i (so each tail byte is shifted by its *absolute* JSON index plus the key). 6. Result = `a ++ out` — the plaintext header goes on the wire **first**, transformed tail after (`addAll` onto the header copy). **`decryptData(bytes)`** (single arg, the accumulated receive buffer): +as recovered from the disassembly (two-`0x58`-marker scan, `s + "}"` +equipNo probe, `s + "," + T` return). NOTE (2026-10-16): live socket +replies decrypt with the *request-side* split instead (see below); the +marker model is kept for the unverified Bluetooth path. 1. `i = indexWhere(byte == 0x58)`, `j = indexWhere(byte == 0x58, start: i+1)`; either missing → return `""`. 2. `s = utf8(wire[0..j])`; `probe = jsonDecode(s + "}")`, null → `""`. @@ -273,17 +279,32 @@ Full TCP sweep of `10.100.1.1` (ports 1–65535): exactly **two** ports open — for *every* probe (TLS ClientHello, HTTP GET, MQTT CONNECT, exact-spec envelopes raw/split/length-prefixed, greeting-first reads). -### Update (2026-10-15) — socket cmd space + live capture +### Update (2026-10-16) — binary capture fully decoded -All pre-2026-10-15 probes used `off_8` ids (`25/0/29/30`) and were dropped -silently — the socket transport speaks the `off_14` id space (§4), so the -device saw unknown commands. A PCAPdroid capture of the working app shows -real socket frames: `{"cmdType":1101,…}` (queryDeviceInfo, empty SN) and -`{"cmdType":1102,"equipNo":"1006005A03X25160132",…}` (922 B). Open: -dst IP:port + direction per frame (need capture metadata / full `.pcap`), -the meaning of `1102` (no pool cmd maps to `0x44E`), and the 17-char -`equipNo` (sub-device? battery unit? reply convention?). Current probes -use socket-space ids (`1101/1307/1705/1707`) via `scripts/host_probe3.py`. +Raw `.bin` payloads (no text mangling) decrypt completely offline with +radix 16 (`tests/fixtures/*.bin` are golden vectors): + +- Discovery request (153 B): `{"cmdType":1101,"equipNo":"00000000",…, + "dataArea":{"opt":0,"minProtocolVer":"V1.00.00"}}`, key `"80"`/n=128. +- Device-info reply (1154 B): `cmdType` **1102**, `equipNo` + `10060005A03X25160132` (the unit's IBG SN — label SN + `A03X2516013210060005` shares its trailing 12 chars), `len` 1024 exact, + CRC exact, `dataArea.result == 0` with the full version inventory + (`FHP_NUM: 2`, two `FHP_SN`, two `BMS_SN`, `protocolVer V1.12.36`). + Encrypted with the **gateway** key (`"1c"`/n=28) while addressed to the + IBG SN — the gateway is the crypto endpoint; sub-unit addressing rides + inside. + +Reply rules learned: same `S1 || transform(S2)` framing both directions; +reply id observed as request + 1 (`1101 → 1102`); universal split at the +2nd comma needs no ids up front (`decrypt_wire`). The remaining unknown +is the live endpoint (dst IP:port + direction per capture). + +Superseded notes: pre-2026-10-16 probes additionally used the wrong key +radix (32 instead of 16) and mostly omitted the `opt`-seeded payload, so +every silent run is fully explained by bytes, with no need for exotic +causes. Current probes use socket-space ids + radix 16 + `opt` payloads +via `scripts/host_probe6.py`. ## 14. Direct-connect flow (manual AP join → return to app) diff --git a/franklinwh_agate_client/battery.py b/franklinwh_agate_client/battery.py index 6598723..68d7d98 100644 --- a/franklinwh_agate_client/battery.py +++ b/franklinwh_agate_client/battery.py @@ -125,16 +125,18 @@ def get_battery( sn: str, host: str = "10.100.1.1", ports: Tuple[int, ...] = SOCKET_PORT_CANDIDATES, - key_radix: int = 32, + key_radix: int = 16, timeout: float = 5.0, ) -> Dict[str, Any]: """Query the aggregate charge percentage over the local TCP socket. Mirrors the app's ``queryAPowerData`` first (``DEVICE_APOWER_INFO`` with - ``data={"update": 2}``), then ``BMS`` and ``EMS`` as fallbacks, probing - each candidate port. Returns the ``extract_soc`` dict plus - ``cmd``/``port``/``reply`` context. Raises the last socket error if no - port answers. + ``data={"opt": 0, "update": 2}``), then ``BMS`` and ``EMS`` as fallbacks, + probing each candidate port. Like the app's direct-connect flow, a + discovery query (``QUERY_DEVICE_INFO``, empty serial) opens each port's + sequence — its reply also confirms two-way crypto before the battery + read. Returns the ``extract_soc`` dict plus ``cmd``/``port``/``reply`` + context. Raises the last socket error if no port answers. """ last_err: Optional[Exception] = None for port in ports: @@ -142,6 +144,12 @@ def get_battery( sn=sn, host=host, socket_port=port, key_radix=key_radix, timeout=timeout, ) + try: + client.send_rpc(CommunicationCmd.QUERY_DEVICE_INFO, + equip_no="", + data={"minProtocolVer": "V1.00.00"}) + except Exception as exc: # discovery is best-effort; keep going + log.debug("port %d discovery failed: %s", port, exc) for cmd, data in ( (CommunicationCmd.DEVICE_APOWER_INFO, {"update": 2}), (CommunicationCmd.BMS, None), @@ -182,7 +190,7 @@ def probe_device( sn: str, host: str = "10.100.1.1", ports: Tuple[int, ...] = SOCKET_PORT_CANDIDATES, - key_radix: int = 32, + key_radix: int = 16, timeout: float = 5.0, ) -> List[Dict[str, Any]]: """Send every :data:`PROBE_VARIANTS` envelope; return exact TX/RX bytes.""" @@ -251,7 +259,7 @@ def probe_transport( sn: str, host: str = "10.100.1.1", ports: Tuple[int, ...] = (9000,), - key_radix: int = 32, + key_radix: int = 16, timeout: float = 5.0, ) -> List[Dict[str, Any]]: """Transport-level probes against one service port. @@ -323,7 +331,7 @@ def main(argv: Optional[List[str]] = None) -> int: "--ports", default=",".join(map(str, SOCKET_PORT_CANDIDATES)), help="comma-separated socket ports to probe", ) - parser.add_argument("--key-radix", type=int, default=32, choices=(16, 32)) + parser.add_argument("--key-radix", type=int, default=16, choices=(16, 32)) parser.add_argument("--timeout", type=float, default=5.0) parser.add_argument("--show-hotspot", action="store_true", help="print derived AP SSID/password and exit") diff --git a/franklinwh_agate_client/device_local.py b/franklinwh_agate_client/device_local.py index 0054eb7..d1fb23a 100644 --- a/franklinwh_agate_client/device_local.py +++ b/franklinwh_agate_client/device_local.py @@ -29,6 +29,7 @@ from .protocol import ( SocketFrameDecoder, decrypt_data, decrypt_request, + decrypt_wire, direct_connect_request, encrypt_object, frame_socket_payload, @@ -64,8 +65,8 @@ class LocalDeviceClient: :param socket_port: local TCP socket port (candidate 18000). :param verify: TLS verification for the device API (device likely serves a self-signed cert, so ``False`` is the practical default). - :param key_radix: key parse radix for the envelope cipher (32, per the - ``int.tryParse(key, radix: 32)`` call in both directions). + :param key_radix: key parse radix for the envelope cipher (16 — + default ``int.tryParse`` radix, verified against live captures). :param timeout: per-request timeout in seconds. """ @@ -77,7 +78,7 @@ class LocalDeviceClient: scheme: str = "https", verify: Optional[bool] = False, socket_port: Optional[int] = None, - key_radix: int = 32, + key_radix: int = 16, timeout: float = 10.0, extra_headers: Optional[Dict[str, str]] = None, session: Optional[requests.Session] = None, @@ -185,7 +186,7 @@ class LocalDeviceClient: """ if envelope is None: envelope = direct_connect_request( - cmd, equip_no or self.sn, data=data, + cmd, self.sn if equip_no is None else equip_no, data=data, snno=self._sn_seq.next(), ) return self.send_envelope(envelope) @@ -194,8 +195,9 @@ class LocalDeviceClient: wire = encrypt_object(envelope.to_json_string(), self.key, radix=self.key_radix) raw = self._socket_roundtrip(frame_socket_payload(wire)) - reply = decrypt_data(raw, radix=self.key_radix) - if not reply: + try: + reply = decrypt_wire(raw, radix=self.key_radix) + except ValueError: raise ConnectionError( f"undecodable reply ({len(raw)} bytes) from " f"{self.socket_host}:{self.socket_port}" @@ -224,7 +226,10 @@ class LocalDeviceClient: ) reply: Optional[Dict[str, Any]] = None if reply_raw: - decoded = decrypt_data(reply_raw, radix=self.key_radix) + try: + decoded = decrypt_wire(reply_raw, radix=self.key_radix) + except ValueError: + decoded = "" if decoded: try: reply = json.loads(decoded) @@ -268,11 +273,13 @@ class LocalDeviceClient: break if collect_only: return bytes(received) - if decrypt_data(bytes(received), radix=self.key_radix): - return bytes(received) - raise ConnectionError( - f"no reply frame from {self.socket_host}:{self.socket_port}" - ) + try: + decrypt_wire(bytes(received), radix=self.key_radix) + except ValueError: + raise ConnectionError( + f"no reply frame from {self.socket_host}:{self.socket_port}" + ) + return bytes(received) # ------------------------------------------------------- convenience RPCs def query_device_info(self) -> Dict[str, Any]: diff --git a/franklinwh_agate_client/protocol/__init__.py b/franklinwh_agate_client/protocol/__init__.py index a10eefc..48bddf5 100644 --- a/franklinwh_agate_client/protocol/__init__.py +++ b/franklinwh_agate_client/protocol/__init__.py @@ -26,10 +26,12 @@ from .crypto import ( decrypt_data, decrypt_object, decrypt_request, + decrypt_wire, encrypt_object, generate_secret_key, header_len_for, split_envelope, + split_wire, ) from .envelope import ( KEY_ORDER, @@ -81,6 +83,8 @@ __all__ = [ "decrypt_object", "decrypt_request", "decrypt_data", + "decrypt_wire", + "split_wire", "header_len_for", "KEY_ORDER", "SNNO_MIN", diff --git a/franklinwh_agate_client/protocol/crypto.py b/franklinwh_agate_client/protocol/crypto.py index 6c40936..03c7043 100644 --- a/franklinwh_agate_client/protocol/crypto.py +++ b/franklinwh_agate_client/protocol/crypto.py @@ -5,42 +5,42 @@ Reconstructed from the blutter disassembly; see Request wire form of an envelope:: - s = compact JSON, e.g. {"cmdType":25,"equipNo":"SN",...} + s = compact JSON, e.g. {"cmdType":1101,"equipNo":"SN",...} S1 = parts[0] + "," + parts[1] + "," # quartet header, trailing comma S2 = ",".join(parts[2:]) # the rest of the JSON a = UTF-8(S1) # sent PLAINTEXT first - n = int.tryParse(key, radix: 32) or 0 + n = int.tryParse(key) or 0 # default radix 16 wire = a || bytes((S2[i] + i + len(a) + n) & 0xFF) i.e. the two leading JSON parts go on the wire **unchanged first**, and the tail is transformed byte-wise with a position-dependent additive stream (index ``i`` is relative to the tail; ``i + len(a)`` is the absolute JSON -index). The receiver that knows the serial can compute ``len(a)`` up front -(``header_len_for``) and split the wire deterministically. +index). Both directions use this same framing (verified live: a 1154-byte +device reply decrypts with the identical split — see ``decrypt_wire``). -Responses use a different framing (``decrypt_data``): the wire holds the -plaintext JSON prefix ``s`` (which itself contains the serial's ``X``), -a literal ``0x58`` separator, then the encoded tail:: +.. note:: An earlier revision claimed radix 32 from a ``0x20`` literal + next to the ``tryParse`` call sites. Two independent live frames (an + app discovery request and a device info reply) decrypt cleanly only + with radix 16 (``int("80", 16) == 128``, ``int("1c", 16) == 28``) and + then verify byte-exact (``len``/uppercase CRC-32). Ground truth wins: + the ``0x20`` belongs to a neighbouring call. - wire = UTF-8(s) || 0x58 || encode(tail) - -``decrypt_data`` locates the *second* ``0x58`` byte ``j``, takes -``s = UTF-8(wire[:j])``, probes ``s + "}"`` as JSON to recover ``equipNo`` -(and hence the key), decodes the tail with the same stream in reverse, and -returns ``s + "," + tail_plain``. Anything unexpected yields ``""`` (the -app's ``decryptData`` returns ``""`` on marker/parse failure and the socket -layer keeps accumulating bytes). - -Confirmed from the disassembly: +Confirmed from the disassembly *and live traffic*: * ``generateSecretKey`` = ``sum(utf8(sn)) & 0xFF`` rendered as lowercase hex - with **no** zero padding (Dart ``_toPow2String(16)``). -* The key is parsed with ``int.tryParse(key, radix: 32)`` — the literal - ``0x20`` (32) is materialised right before the call in both the encrypt - and decrypt paths. + with **no** zero padding (Dart ``_toPow2String(16)``). E.g. ``"1c"`` for + the gateway SN, ``"80"`` for the empty/placeholder SN. +* The key is parsed with ``int.tryParse(key)`` — default radix **16**, + verified against two live frames (``tests/fixtures/*.bin``). * ``encryptObject(object, key)`` takes the object in the first slot and the key in the second; its result is ``a ++ transform(b)`` (plaintext header first — ``addAll(transformed)`` onto the header copy). +* Replies use the same ``S1 || transform(S2)`` framing (the device info + reply decrypts identically). The reply ``cmdType`` observed live is the + request id + 1 (``1101`` → ``1102``); replies are therefore parsed with + :func:`decrypt_wire`, which locates the header without knowing the ids + up front. ``decrypt_data`` (``0x58``-marker model) is kept for the + Bluetooth path, which is unverified live. """ from __future__ import annotations @@ -48,8 +48,10 @@ from __future__ import annotations import json from typing import Optional, Tuple -#: Key radix used by both ``encryptObject`` and ``decryptData`` (literal 0x20). -DEFAULT_KEY_RADIX = 32 +#: Key radix used by the envelope cipher (default ``tryParse`` radix). +#: Verified live: ``int("80", 16) == 128`` and ``int("1c", 16) == 28`` +#: decode two independent captured frames byte-exact. +DEFAULT_KEY_RADIX = 16 #: Response separator byte (``0x58``, ASCII ``X``). MARKER = 0x58 @@ -143,6 +145,53 @@ def decrypt_request( decrypt_object = decrypt_request +def split_wire(wire: bytes) -> Tuple[bytes, bytes]: + """Split any envelope wire into ``(a, b)`` at the 2nd comma. + + The header ``S1 = parts[0] + "," + parts[1] + ","`` is always ASCII + with no commas inside, so the first two commas on the wire are its + structural boundaries — no ids or serials needed up front. Works for + requests and replies alike. + """ + raw = bytes(wire) + first = raw.find(b",") + if first < 0: + raise ValueError("no header comma in wire buffer") + second = raw.find(b",", first + 1) + if second < 0: + raise ValueError("truncated header in wire buffer") + return raw[: second + 1], raw[second + 1:] + + +def decrypt_wire(wire: bytes, radix: int = DEFAULT_KEY_RADIX) -> str: + """Decode one complete envelope wire (request or reply) to JSON. + + Splits via :func:`split_wire`, recovers ``equipNo`` from the + ``a + "}"`` probe (exactly how the receiver learns which key to use), + and reverses the additive stream. Raises ``ValueError`` on any + failure (use :func:`decrypt_data`-style ``""`` handling when + accumulating). + """ + a, b = split_wire(wire) + try: + # ``a`` ends with the header's trailing comma — drop it so the + # probe is valid JSON (mirrors the ``s + "}"`` equipNo probe). + probe = json.loads(a[:-1].decode("utf-8") + "}") + except (UnicodeDecodeError, ValueError) as exc: + raise ValueError(f"header probe is not JSON: {exc}") from exc + if not isinstance(probe, dict): + raise ValueError("header probe is not an object") + key = generate_secret_key(str(probe.get("equipNo", ""))) + key_int = _parse_key(key, radix) + plain_b = bytes( + (byte - i - len(a) - key_int) & 0xFF for i, byte in enumerate(b) + ) + try: + return (a + plain_b).decode("utf-8") + except UnicodeDecodeError as exc: + raise ValueError(f"tail is not UTF-8: {exc}") from exc + + def decrypt_data(wire: bytes, radix: int = DEFAULT_KEY_RADIX) -> str: """``decryptData`` — decode one accumulated *response* buffer. diff --git a/franklinwh_agate_client/protocol/framing.py b/franklinwh_agate_client/protocol/framing.py index 9c488d0..f4a0910 100644 --- a/franklinwh_agate_client/protocol/framing.py +++ b/franklinwh_agate_client/protocol/framing.py @@ -224,11 +224,12 @@ class SocketFrameDecoder: def feed(self, data: bytes) -> List[str]: """Append *data*; return a list of complete decoded JSON strings.""" - from .crypto import decrypt_data # local import: cheap, no cycles + from .crypto import decrypt_wire # local import: cheap, no cycles self._buf.extend(data) - decoded = decrypt_data(bytes(self._buf)) - if not decoded: + try: + decoded = decrypt_wire(bytes(self._buf)) + except ValueError: return [] try: json.loads(decoded) diff --git a/scripts/host_probe6.py b/scripts/host_probe6.py new file mode 100644 index 0000000..2f5ff26 --- /dev/null +++ b/scripts/host_probe6.py @@ -0,0 +1,194 @@ +#!/usr/bin/env python3 +"""Host probe v6 — verified-against-capture wire format (stdlib only). + +Locked by two live frames (tests/fixtures/*.bin in the repo): +* key radix 16 (``int("80",16)=128``, ``int("1c",16)=28``); +* ``S1 || transform(S2)`` both directions; reply id = request id + 1 + observed (1101 -> 1102); +* socket payloads seeded with ``"opt": 0``; +* replies parsed by splitting at the 2nd comma (no ids needed up front). + +Run on the SoftAP-joined host: + python3 host_probe6.py --sn A03X2516013210060005 --ports 9000 +""" + +import argparse +import json +import socket +import time +import zlib + +APOWER, QUERY_INFO, BMS, EMS = 0x51B, 0x44D, 0x6A9, 0x6AB +SNNO = [0] +SOC_KEYS = ("currentBatterySoc", "soc", "batterySoc", "vppSoc") + + +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, 16) + return a + bytes((b[i] + i + len(a) + n) & 0xFF + for i in range(len(b))) + + +def decrypt_wire(wire): + first = wire.find(b",") + second = wire.find(b",", first + 1) + if first < 0 or second < 0: + return "" + a, b = wire[:second + 1], wire[second + 1:] + try: + probe = json.loads(a[:-1].decode() + "}") + except Exception: + return "" + if not isinstance(probe, dict): + return "" + n = int(secret_key(str(probe.get("equipNo", ""))), 16) + try: + return (a + bytes((t - i - len(a) - n) & 0xFF + for i, t in enumerate(b))).decode() + except Exception: + return "" + + +def recv_all(sock, idle=8.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) + + +def walk(obj, key, found): + if isinstance(obj, dict): + for k, v in obj.items(): + if k == key: + found.append(v) + walk(v, key, found) + elif isinstance(obj, list): + for item in obj: + walk(item, key, found) + return found + + +def extract_soc(reply): + data = reply.get("dataArea", reply) + if isinstance(data, str): + try: + data = json.loads(data) + except ValueError: + data = {} + for key in SOC_KEYS: + hits = walk(reply, key, []) + for hit in hits: + if isinstance(hit, bool): + continue + try: + return float(hit), key + except (TypeError, ValueError): + continue + arrs = walk(data if isinstance(data, dict) else {}, "socArray", []) + for arr in arrs: + if isinstance(arr, list) and arr: + try: + nums = [float(v) for v in arr + if not isinstance(v, bool)] + if nums: + return sum(nums) / len(nums), "socArray(avg)" + except (TypeError, ValueError): + continue + return None, None + + +def run(label, host, port, tx, timeout, pre_read=0.0): + rx_pre, rx, err, reply = b"", b"", None, 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: + sock.sendall(tx) + rx = recv_all(sock) + if rx: + dec = decrypt_wire(rx) + reply = json.loads(dec) if dec else {"_raw": rx.hex()} + 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()[:400] or "(empty)")) + if reply is not None: + print(" reply: " + json.dumps(reply, default=str)[:3000]) + if isinstance(reply, dict): + val, src = extract_soc(reply) + print(" soc: %s (source=%s)" % (val, src)) + 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 16)" % (args.sn, key)) + variants = [ + ("discovery/1101+empty", QUERY_INFO, + {"minProtocolVer": "V1.00.00"}, ""), + ("queryInfo/1101+SN", QUERY_INFO, + {"minProtocolVer": "V1.00.00"}, None), + ("apower/1307+SN", APOWER, {"update": 2}, None), + ("apower/1307+IBG", APOWER, {"update": 2}, + "10060005A03X25160132"), + ("bms/1705+SN", BMS, None, None), + ("ems/1707+SN", EMS, None, None), + ] + for port in [int(p) for p in args.ports.split(",") if p.strip()]: + for label, cmd, data, equip in variants: + tx = encrypt(build_envelope( + args.sn, cmd, data, + equip=args.sn if equip is None else equip or "00000000", + ), key) + if run(label, args.host, port, tx, args.timeout, + pre_read=4.0 if label.startswith("discovery") else 0.0): + pass + + +if __name__ == "__main__": + main() diff --git a/tests/fixtures/device_info_reply.bin b/tests/fixtures/device_info_reply.bin new file mode 100644 index 0000000000000000000000000000000000000000..f23a0fa5f4149717b29ff5cdfdf433fd610079a7 GIT binary patch literal 1154 zcmV-|1bzE^B4cf2RC#b^B04cKFfuG6WpQ{=bW{z zt+2DKqviGIr@*PcyS>5B(A3egy|nlB3HR6v{` zNlJQec5!}nZE1pKb#;Didt6(KUt>{RQ-5@QgO!bai(GYHlW%X5a&m=ub&7X*sEePg z!n(G@#=v!ecbBfKxUitEf0%)`l9IKRm8hAXt(u$5z`(tdos_i1#nsKd%bc~I)32}6 zva-dsy~?+^tG2D*+3n@k@-{kY);nd&O z2m%%k{tO!a3=I?F`{Wt*^%(g03;O*K{r>U%^B5Z>G$bq~`xFfb|0E|cKs7N2A3+Hj z85$rQ9UmkyAtFmADkmx^EH5n{DIhQ~F*h?bHakc-IbuCQK1Dx7MMO1mK{#h!cx`Du zS9w8EQBqk{RaRVSSzUu(VqaxoVr6B7W>{ldYHDk8ZEkOMf^v48cYAn#d3%0-oq%zz zeRP?moR_w)o~(VAxPY~ik+hSQr6shrrTpTfe##NpY= z*QmkctE#QIudcD}y|mZY+1d8)-t79_?e4+a{KWaz)%e%i>*Cz--QLmM(+K6{2j=Pe z@$3KW?cf~j;|2x@2qhg28z~JN93t@wEc7V`1tBPvz%f`sd%iPVr`pUuL?)3ut>iEgx1k3^6-2mU>^Wy9H<>uMs+Ys&S z5AN{-@%jh!_2?w@>kJnwBNHL=3oZ332L>q!3mgj!Ar21$4FowB6*w0fD;gUx9Uc)) z9TZ1AR7^Zd8#GoP9UvQ*?lblz4-efsvY$f{t%EQye)8X04*Qmj&tG2D* U2G;ZD_5=p~1O&O#%F50e8pk3#cmMzZ literal 0 HcmV?d00001 diff --git a/tests/fixtures/discovery_req.bin b/tests/fixtures/discovery_req.bin new file mode 100644 index 0000000..a5bd6b3 --- /dev/null +++ b/tests/fixtures/discovery_req.bin @@ -0,0 +1,2 @@ +{"cmdType":1101,"equipNo":"00000000",Ç ÌåÜÙÐ#(#'Úóëòõïïð÷úò÷ðç9568ìýùð;5?ô  ù;K=ýÿ%+ +MK_M.`TQ,ndfk3*'jgmPsqwshus^n|-F/d@>ABADE8”• \ No newline at end of file diff --git a/tests/test_protocol.py b/tests/test_protocol.py index cfb109c..7549a0b 100644 --- a/tests/test_protocol.py +++ b/tests/test_protocol.py @@ -6,6 +6,7 @@ No radio/hardware needed. Vectors mirror the blutter disassembly of """ import json +import os import unittest from franklinwh_agate_client.protocol import ( @@ -17,12 +18,14 @@ from franklinwh_agate_client.protocol import ( SocketFrameDecoder, decrypt_data, decrypt_request, + decrypt_wire, direct_connect_request, encrypt_object, frame_socket_payload, generate_secret_key, header_len_for, split_envelope, + split_wire, ) from franklinwh_agate_client.protocol.crypto import MARKER @@ -62,9 +65,9 @@ class TestSecretKey(unittest.TestCase): def test_real_sn_key(self): self.assertEqual(generate_secret_key(SN), "1c") - def test_key_radix_is_32(self): - self.assertEqual(DEFAULT_KEY_RADIX, 32) - self.assertEqual(int(generate_secret_key(SN), 32), 44) + def test_key_radix_is_16(self): + self.assertEqual(DEFAULT_KEY_RADIX, 16) + self.assertEqual(int(generate_secret_key(SN), 16), 28) def test_empty_sn_key(self): self.assertEqual(generate_secret_key(""), "0") @@ -148,7 +151,7 @@ class TestCipher(unittest.TestCase): def test_stream_is_absolute_index_plus_n(self): a, b = split_envelope(_apower_json()) - n = int(generate_secret_key(SN), 32) + n = int(generate_secret_key(SN), 16) wire = encrypt_object(_apower_json(), generate_secret_key(SN)) tail = wire[len(a):] for i, byte in enumerate(b): @@ -163,16 +166,10 @@ class TestCipher(unittest.TestCase): body, ) - def test_decrypt_response_roundtrip(self): + def test_wire_roundtrip(self): body = _apower_json() - cut = body.index('"type"') - s, tail = body[:cut - 1], body[cut:] - n = int(generate_secret_key(SN), 32) - prefix_len = len((s + ",").encode("utf-8")) - enc = bytes((byte + k + prefix_len + n) & 0xFF - for k, byte in enumerate(tail.encode("utf-8"))) - resp = s.encode("utf-8") + bytes((MARKER,)) + enc - self.assertEqual(decrypt_data(resp), body) + wire = encrypt_object(body, generate_secret_key(SN)) + self.assertEqual(decrypt_wire(wire), body) def test_decrypt_data_failures_are_empty(self): self.assertEqual(decrypt_data(b""), "") @@ -189,22 +186,75 @@ class TestSocketFraming(unittest.TestCase): self.assertEqual(frame_socket_payload(payload), payload) def test_decoder_accumulates_until_json(self): - body = _apower_json() - cut = body.index('"type"') - s, tail = body[:cut - 1], body[cut:] - n = int(generate_secret_key(SN), 32) - prefix_len = len((s + ",").encode("utf-8")) - enc = bytes((byte + k + prefix_len + n) & 0xFF - for k, byte in enumerate(tail.encode("utf-8"))) - resp = s.encode("utf-8") + bytes((MARKER,)) + enc + wire = open(os.path.join(FIXTURES, "discovery_req.bin"), + "rb").read() dec = SocketFrameDecoder() # first half: incomplete -> nothing yet - self.assertEqual(dec.feed(resp[:len(resp) // 2]), []) + self.assertEqual(dec.feed(wire[:len(wire) // 2]), []) # rest: one complete JSON string - out = dec.feed(resp[len(resp) // 2:]) + out = dec.feed(wire[len(wire) // 2:]) self.assertEqual(len(out), 1) - self.assertEqual(json.loads(out[0])["equipNo"], SN) + self.assertEqual(json.loads(out[0])["equipNo"], "00000000") if __name__ == "__main__": unittest.main() + + +FIXTURES = os.path.join(os.path.dirname(__file__), "fixtures") + + +class TestLiveCaptures(unittest.TestCase): + """Golden vectors: two real frames (app request + device reply).""" + + def test_key_radix_is_16(self): + self.assertEqual(DEFAULT_KEY_RADIX, 16) + self.assertEqual(generate_secret_key("00000000"), "80") + self.assertEqual(int("80", 16), 128) + self.assertEqual(int(generate_secret_key(SN), 16), 28) + + def test_discovery_request_vector(self): + wire = open(os.path.join(FIXTURES, "discovery_req.bin"), "rb").read() + self.assertEqual(len(wire), 153) + a, b = split_wire(wire) + self.assertEqual(a, b'{"cmdType":1101,"equipNo":"00000000",') + self.assertEqual( + decrypt_wire(wire), + '{"cmdType":1101,"equipNo":"00000000","type":0,' + '"timeStamp":1792117904,"snno":1,"len":37,"crc":"A474C16F",' + '"dataArea":{"opt":0,"minProtocolVer":"V1.00.00"}}', + ) + + def test_device_info_reply_vector(self): + wire = open(os.path.join(FIXTURES, "device_info_reply.bin"), + "rb").read() + self.assertEqual(len(wire), 1154) + a, b = split_wire(wire) + self.assertEqual( + a, b'{"cmdType":1102,"equipNo":"10060005A03X25160132",') + reply = json.loads(decrypt_wire(wire)) + self.assertEqual(reply["cmdType"], 1102) + self.assertEqual(reply["equipNo"], "10060005A03X25160132") + self.assertEqual(reply["len"], 1024) + data = reply["dataArea"] + self.assertEqual(data["opt"], 0) + self.assertEqual(data["result"], 0) + self.assertEqual(data["IBG_SN"], "10060005A03X25160132") + self.assertEqual(data["FHP_NUM"], 2) + self.assertEqual(len(data["FHP_SN"]), 2) + self.assertEqual(len(data["BMS_SN"]), 2) + # len/crc integrity, as the device checks them + import zlib + compact = json.dumps(data, separators=(",", ":"), + ensure_ascii=False) + self.assertEqual(reply["len"], len(compact.encode("utf-8"))) + self.assertEqual( + reply["crc"], format(zlib.crc32(compact.encode()), "X")) + + def test_split_wire_errors(self): + with self.assertRaises(ValueError): + split_wire(b"no commas here") + with self.assertRaises(ValueError): + split_wire(b'{"cmdType":1101') + with self.assertRaises(ValueError): + decrypt_wire(b"not json at all")