From 2438a04ed505032e6bb14edada5159b13e996872 Mon Sep 17 00:00:00 2001 From: kfj001 Date: Thu, 15 Oct 2026 01:26:28 +0000 Subject: [PATCH] Add SoftAP battery PoC (apower/BMS/EMS socket read) with docs and tests Probes 10.100.1.1:9000,18000, sends 0x19/0x1D/0x1E envelopes, extracts currentBatterySoc/soc/avg(socArray). Includes offline extract_soc tests, BATTERY_POC docs, README usage, and empirical port-9000 RE notes. --- README.md | 14 ++ docs/BATTERY_POC.md | 58 ++++++++ docs/RE_DART_FINDINGS.md | 31 ++++- franklinwh_agate_client/__init__.py | 13 ++ franklinwh_agate_client/battery.py | 207 ++++++++++++++++++++++++++++ tests/test_battery.py | 39 ++++++ 6 files changed, 359 insertions(+), 3 deletions(-) create mode 100644 docs/BATTERY_POC.md create mode 100644 franklinwh_agate_client/battery.py create mode 100644 tests/test_battery.py diff --git a/README.md b/README.md index 8a7c97e..9d1efa8 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,8 @@ Python client library for interacting with FranklinWH AGATE energy storage devic - **[Dart AOT Reverse-Engineering Findings](docs/RE_DART_FINDINGS.md)** — detailed protocol analysis recovered from the app's `libapp.so` with [blutter](https://github.com/worawit/blutter) (transports, wire envelope, command/topic enums, crypto/CRC, BLE framing, hotspot credentials). +- **[Battery PoC](docs/BATTERY_POC.md)** — read charge % over the SoftAP socket + (`python -m franklinwh_agate_client.battery --sn `). ## Reverse Engineering Summary @@ -51,6 +53,18 @@ client = AgateClient() device_info = client.query_agate_info(device_id) ``` +## Battery PoC (SoftAP socket) + +```bash +python -m franklinwh_agate_client.battery --sn <20-char-SN> --show-hotspot +python -m franklinwh_agate_client.battery --sn <20-char-SN> # e.g. 78% +python -m franklinwh_agate_client.battery --sn <20-char-SN> --raw +``` + +Probes `10.100.1.1:9000,18000`, sends `0x19/0x1D/0x1E`, prints +`currentBatterySoc → soc → avg(socArray)`. Must run where the Wi-Fi adapter +is on the device AP — see [Battery PoC](docs/BATTERY_POC.md). + ## Local connect (device hotspot) The aGate runs its own SoftAP (gateway `10.100.1.1`); joining it is what diff --git a/docs/BATTERY_POC.md b/docs/BATTERY_POC.md new file mode 100644 index 0000000..9717636 --- /dev/null +++ b/docs/BATTERY_POC.md @@ -0,0 +1,58 @@ +# Battery PoC — direct-connect charge state + +Reads the aggregate battery percentage straight from the aGate over its own +SoftAP. No cloud, no MQTT, no Bluetooth. Recovered from the Dart AOT layer; +see `RE_DART_FINDINGS.md` §§3–5, 8, 10. + +## Requirements + +- Host whose Wi-Fi adapter is associated to the device AP + (`AP_` / ``, `wifi.derive_hotspot_credentials`). +- L2 reachability to `10.100.1.1`. In a container this means a bridged + `macvlan`/`ipvlan` address on `10.100.1.0/24` or `network_mode: host` + (see repo-root `.devcontainer/docker-compose.yml`). A default Docker + bridge (`172.17.0.0/16`) is NATed to a non-`10.100.1.x` source and gets + `ECONNREFUSED`. +- Python `requests` for the HTTP helpers in `device_local.py`. + +## What it does + +1. Probes TCP ports `9000, 18000` in order (`9000` was open on the tested + unit; `0x4650 = 18000` is the app constant but was closed there). +2. Sends `DEVICE_APOWER_INFO (0x19)`, then `BMS (0x1D)`, `EMS (0x1E)`: + compact-JSON `CommunicationObject` → `encrypt_object` (key + `hex(sum(sn) & 0xFF)`) → int16LE length framing. +3. Decrypts the reply and extracts `currentBatterySoc` → `soc` → + `avg(socArray)`. + +## Usage + +```bash +# show derived hotspot creds only +python -m franklinwh_agate_client.battery --sn <20-char-SN> --show-hotspot + +# print just the percentage +python -m franklinwh_agate_client.battery --sn <20-char-SN> + +# full reply envelope +python -m franklinwh_agate_client.battery --sn <20-char-SN> --raw +``` + +Options: `--host` (default `10.100.1.1`), `--ports 9000,18000`, +`--key-radix {16,32}` (default 16; 32 is the alternate candidate in the +disassembly), `--timeout`. + +Programmatic: + +```python +from franklinwh_agate_client import get_battery +out = get_battery("...", host="10.100.1.1") +print(out["value"], out["source"]) # e.g. 78.0 currentBatterySoc +``` + +## Limits + +- No live round-trip has succeeded yet from an isolated container netns + (all probes `ECONNREFUSED`); first success needs a `10.100.1.x` source. +- `key_radix`, BLE field order, and `X-Device-*` binding remain open + (`RE_DART_FINDINGS.md` Open items). diff --git a/docs/RE_DART_FINDINGS.md b/docs/RE_DART_FINDINGS.md index f4f49d2..00c8a25 100644 --- a/docs/RE_DART_FINDINGS.md +++ b/docs/RE_DART_FINDINGS.md @@ -193,6 +193,25 @@ authS2C=2 "fwh/s2c/auth/" mngC2S=5 "fwh/c2s/mng/" deviceS2C "fwh/s2c/ - a `Duration` and retry loop (up to ~2 retries) are involved. - `BytesToInt16/Int16ToBytes` LE framing; `Socket Manager send object` / `receive object`. +### Empirical — live aGate on its SoftAP (2026-10-13, from a client on `10.100.1.0/24`) + +Full TCP sweep of `10.100.1.1` (ports 1–65535): exactly **two** ports open — **53** and **9000**. No 80/443, no web UI, and **18000 (`0x4650`) is closed** on this unit. + +- **Port 53** — authoritative, *closed* DNS: well-formed `NXDOMAIN` for everything + (even `example.com`), no PTR for `1.1.0.100.in-addr.arpa`, no forwarding (the SoftAP + is an island). Serves some local name pattern we haven't identified yet. +- **Port 9000** — a **length-prefixed framing service** (int16LE length), consistent + with the app's socket layer: + - every TLS `ClientHello` gets `UNEXPECTED_EOF` mid-handshake — first two bytes + `16 03` = int16LE `790`, so the decoder waits for a 790-byte frame that never + completes, then closes on idle; + - cleartext HTTP `GET` (`47 45` = `17735`) hangs forever (read timeout). + - A *complete* 147-byte frame (int16LE length + encrypted `QUERY_DEVICE_INFO` + envelope, `equipNo = X25160132`, key from `sum(utf8(sn))&0xFF`) is accepted + but met with **zero bytes** — every command tried (`0x00`/`0x01`/`0x02`/`0x06`) + and both key radices (16/32). Content-level rejection; the leading hypothesis is + `equipNo` validation against the real serial. **Pending: retry with the true SN.** + ## 9. Bluetooth (BLE) transport — `communication_center/bluetooth` - `BluetoothManager` (singleton), `BluetoothStatus`: `closed / connected / connecting / interrupted`. @@ -323,9 +342,15 @@ Still open: 1. Exact BLE frame field order & the 2-byte magic semantics (confirm with `btsnoop`). 2. Exact `encryptObject` radix + trailing-header handling. -3. Local HTTP port + TLS behavior for `https://IP:PORT` (device self-signed cert?). -4. Local socket port confirmation (`18000` candidate) and whether local RPC uses - socket vs `/terminal` HTTP vs local MQTT. +3. ~~Local HTTP port + TLS behavior for `https://IP:PORT`~~ — **revised (2026-10-13, + §8 empirical):** no HTTP/TLS port exists on the device at all (only 53 + 9000). + Either local mode is socket-only on this firmware, or the HTTP service is + state-dependent and wasn't running. Revisit if a `/terminal` response is ever + captured from the SoftAP. +4. ~~Local socket port confirmation~~ — **resolved on one unit (2026-10-13): `9000`** + (§8 empirical). The `0x4650` constant is not listening; the port may be + firmware-/device-dependent. Remaining: first confirmed RPC round-trip + (blocked on the real SN — see §8 empirical). 5. Which class in `classes5.dex` sets the `X-Device-*` headers and their exact values (string-pool location confirmed; class binding in progress — §13). 6. (Tooling, optional) Enable REA's DEX tools: obtain a **jadx-headless-mcp 0.7.1** JAR diff --git a/franklinwh_agate_client/__init__.py b/franklinwh_agate_client/__init__.py index 639dace..bc655d3 100644 --- a/franklinwh_agate_client/__init__.py +++ b/franklinwh_agate_client/__init__.py @@ -33,7 +33,20 @@ __all__ = [ "HotspotProfile", "HotspotJoinError", "derive_hotspot_credentials", + "extract_soc", + "get_battery", + "SOCKET_PORT_CANDIDATES", "DEVICE_GATEWAY_IP", "FALLBACK_PASSWORD", "VALID_SN_LENGTH", ] + + +def __getattr__(name: str): + # Lazy so `python -m franklinwh_agate_client.battery` does not import + # the submodule twice (avoids a runpy RuntimeWarning). + if name in {"extract_soc", "get_battery", "SOCKET_PORT_CANDIDATES"}: + from . import battery as _battery + + return getattr(_battery, name) + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/franklinwh_agate_client/battery.py b/franklinwh_agate_client/battery.py new file mode 100644 index 0000000..1f51bad --- /dev/null +++ b/franklinwh_agate_client/battery.py @@ -0,0 +1,207 @@ +"""Battery charge-state proof of concept. + +Direct-connect (SoftAP) read path recovered from the app's Dart AOT layer +(see ``docs/RE_DART_FINDINGS.md``): join the device hotspot, open the local +TCP socket, send the encrypted ``CommunicationObject`` envelope, decrypt the +reply, and pull the aggregate state-of-charge out of it. + +No cloud, no MQTT, no Bluetooth needed. Requires L2 reachability to the +aGate gateway (``10.100.1.1``) — i.e. run on the host whose Wi-Fi adapter is +associated to the device AP, or in a container bridged onto ``10.100.1.0/24``. +""" + +from __future__ import annotations + +import argparse +import json +import logging +import socket +from typing import Any, Dict, List, Optional, Tuple + +from .device_local import LocalDeviceClient +from .protocol import CommunicationCmd +from .wifi import derive_hotspot_credentials + +log = logging.getLogger(__name__) + +#: Ports probed in order. ``9000`` was observed open on one live unit +#: (``docs/RE_DART_FINDINGS.md`` §8); ``18000`` (``0x4650``) is the constant +#: in the app's connect path but was closed on that unit. +SOCKET_PORT_CANDIDATES: Tuple[int, ...] = (9000, 18000) + +#: Reply keys searched for the aggregate percentage, in priority order. +SOC_KEYS: Tuple[str, ...] = ("currentBatterySoc", "soc", "batterySoc", "vppSoc") + + +def _as_dict(data: Any) -> Optional[Dict[str, Any]]: + if isinstance(data, dict): + return data + if isinstance(data, str): + try: + parsed = json.loads(data) + except (ValueError, TypeError): + return None + return parsed if isinstance(parsed, dict) else None + return None + + +def _walk(obj: Any, key: str) -> List[Any]: + """Collect every value stored under ``key`` anywhere in nested dicts/lists.""" + found: List[Any] = [] + if isinstance(obj, dict): + for k, v in obj.items(): + if k == key: + found.append(v) + found.extend(_walk(v, key)) + elif isinstance(obj, list): + for item in obj: + found.extend(_walk(item, key)) + return found + + +def _to_float(value: Any) -> Optional[float]: + try: + if isinstance(value, bool): + return None + num = float(value) + except (TypeError, ValueError): + return None + return num + + +def extract_soc(reply: Dict[str, Any]) -> Dict[str, Any]: + """Pull the aggregate charge percentage out of a parsed reply envelope. + + Returns ``{"value": float|None, "source": str|None, "detail": {...}}``. + ``source`` names the key the value came from; ``detail`` carries the + supporting fields (``socArray``, ``batteryCount``, charge/discharge + power) when present. + """ + data = _as_dict(reply.get("data", reply)) or {} + + for key in SOC_KEYS: + for container in (data, reply): + for hit in _walk(container, key): + num = _to_float(hit) + if num is not None: + return { + "value": num, + "source": key, + "detail": _soc_detail(data), + } + soc_arrays = _walk(data, "socArray") + for arr in soc_arrays: + if isinstance(arr, list) and arr: + nums = [_to_float(v) for v in arr] + nums = [n for n in nums if n is not None] + if nums: + return { + "value": sum(nums) / len(nums), + "source": "socArray(avg)", + "detail": _soc_detail(data), + } + return {"value": None, "source": None, "detail": _soc_detail(data)} + + +def _soc_detail(data: Dict[str, Any]) -> Dict[str, Any]: + detail: Dict[str, Any] = {} + for key in ( + "socArray", + "batteryCount", + "batteryList", + "chargePower", + "dischargePower", + "currentBatterySoc", + "soc", + ): + hits = _walk(data, key) + if hits: + detail[key] = hits[0] if len(hits) == 1 else hits + # Always record where we looked so failures are debuggable. + return detail + + +def get_battery( + sn: str, + host: str = "10.100.1.1", + ports: Tuple[int, ...] = SOCKET_PORT_CANDIDATES, + key_radix: int = 16, + timeout: float = 5.0, +) -> Dict[str, Any]: + """Query the aggregate charge percentage over the local TCP socket. + + Tries ``DEVICE_APOWER_INFO`` (0x19) first, then ``BMS`` (0x1D) and + ``EMS`` (0x1E) 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. + """ + last_err: Optional[Exception] = None + for port in ports: + client = LocalDeviceClient( + sn=sn, host=host, socket_port=port, + key_radix=key_radix, timeout=timeout, + ) + for cmd in ( + CommunicationCmd.DEVICE_APOWER_INFO, + CommunicationCmd.BMS, + CommunicationCmd.EMS, + ): + try: + reply = client.send_rpc(cmd) + except (ConnectionError, socket.error, TimeoutError) as exc: + last_err = exc + log.debug("port %d cmd %s failed: %s", port, cmd, exc) + break # port unreachable — try next port + except Exception as exc: # envelope-level failure; try next cmd + last_err = exc + log.debug("port %d cmd %s error: %s", port, cmd, exc) + continue + result = extract_soc(reply) + result.update({"cmd": str(cmd), "port": port, "reply": reply}) + if result["value"] is not None: + return result + last_err = ValueError(f"reply on port {port} cmd {cmd} has no SoC field") + raise ConnectionError(f"no battery reply via {host}:{list(ports)}") from last_err + + +def main(argv: Optional[List[str]] = None) -> int: + parser = argparse.ArgumentParser( + description="Read aGate battery percentage over the device SoftAP socket." + ) + parser.add_argument("--sn", required=True, help="20-char device serial") + parser.add_argument("--host", default="10.100.1.1", help="aGate gateway IP") + parser.add_argument( + "--ports", default=",".join(map(str, SOCKET_PORT_CANDIDATES)), + help="comma-separated socket ports to probe", + ) + 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") + parser.add_argument("--raw", action="store_true", help="print full reply JSON") + args = parser.parse_args(argv) + + if args.show_hotspot: + profile = derive_hotspot_credentials(args.sn) + print(json.dumps( + {"ssid": profile.ssid, "password": profile.password, + "gateway_ip": profile.gateway_ip}, indent=2)) + return 0 + + ports = tuple(int(p) for p in args.ports.split(",") if p.strip()) + try: + result = get_battery(args.sn, host=args.host, ports=ports, + key_radix=args.key_radix, timeout=args.timeout) + except Exception as exc: + print(f"ERROR: {exc}") + return 1 + if args.raw: + print(json.dumps(result["reply"], indent=2, default=str)) + else: + value = result["value"] + print(f"{value:g}%" if value is not None else "unknown") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_battery.py b/tests/test_battery.py new file mode 100644 index 0000000..5002861 --- /dev/null +++ b/tests/test_battery.py @@ -0,0 +1,39 @@ +"""Offline tests for battery SoC extraction (no radio/hardware needed).""" + +import unittest + +from franklinwh_agate_client.battery import extract_soc + + +class TestExtractSoc(unittest.TestCase): + def test_current_battery_soc_wins(self): + reply = {"cmdType": 25, "data": {"currentBatterySoc": 78, "soc": 10}} + out = extract_soc(reply) + self.assertEqual(out["value"], 78) + self.assertEqual(out["source"], "currentBatterySoc") + + def test_soc_fallback(self): + out = extract_soc({"data": {"soc": "55"}}) + self.assertEqual(out["value"], 55) + self.assertEqual(out["source"], "soc") + + def test_soc_array_averaged(self): + out = extract_soc({"data": {"socArray": [80, 90]}}) + self.assertEqual(out["value"], 85) + self.assertEqual(out["source"], "socArray(avg)") + + def test_nested_data_string(self): + import json + + reply = {"data": json.dumps({"aPower": {"soc": 62}})} + out = extract_soc(reply) + self.assertEqual(out["value"], 62) + + def test_missing_soc(self): + out = extract_soc({"data": {"foo": 1}}) + self.assertIsNone(out["value"]) + self.assertIsNone(out["source"]) + + +if __name__ == "__main__": + unittest.main()