Correct wire format to disassembly: raw S1||crypt(S2), radix 32, int snno/seconds, decryptData response parser, --probe
- envelope: 7 fixed keys + conditional dataArea, no data key; int
type/snno, seconds timestamp, data setter auto len/crc (CRC32 upper);
SequenceNumber 1..9999 (increaseSN)
- crypto: encrypt is plaintext-header-first a||transform(b), key radix 32
at both tryParse sites; decrypt_data ports decryptData (dual-0x58
markers, s+'}' equipNo probe, s+','+T return); decrypt_request kept
for the device-side known-header split
- socket: raw send (no length prefix), accumulate-and-try receive
- battery: queryAPowerData shape (data={'update':2}), radix default 32,
--probe prints exact TX/RX bytes per variant; device_local.send_probe
- docs: RE_DART_FINDINGS ss3/5/8 + open items, BATTERY_POC host recipe
- tests: 19 new protocol tests, suite 44/44 green
This commit is contained in:
kfj001 committed
1 parent
2438a04ed5
commit
9e69f07636
10 files changed
+744
-182
No files matched your search
@@ -0,0 +1,192 @@
|
||||
"""Offline tests for the corrected envelope/cipher/socket wire format.
|
||||
|
||||
No radio/hardware needed. Vectors mirror the blutter disassembly of
|
||||
``core/communication_object.dart``, ``core/encryption.dart`` and
|
||||
``socket/`` (see ``docs/RE_DART_FINDINGS.md`` §§3, 5, 8).
|
||||
"""
|
||||
|
||||
import json
|
||||
import unittest
|
||||
|
||||
from franklinwh_agate_client.protocol import (
|
||||
DEFAULT_KEY_RADIX,
|
||||
CommunicationCmd,
|
||||
CommunicationObject,
|
||||
SequenceNumber,
|
||||
SocketFrameDecoder,
|
||||
decrypt_data,
|
||||
decrypt_request,
|
||||
direct_connect_request,
|
||||
encrypt_object,
|
||||
frame_socket_payload,
|
||||
generate_secret_key,
|
||||
header_len_for,
|
||||
split_envelope,
|
||||
)
|
||||
from franklinwh_agate_client.protocol.crypto import MARKER
|
||||
|
||||
SN = "A03X2516013210060005"
|
||||
|
||||
|
||||
def _apower_json(snno=1, time_stamp=1792105196):
|
||||
env = direct_connect_request(
|
||||
CommunicationCmd.DEVICE_APOWER_INFO, SN,
|
||||
data={"update": 2}, snno=snno,
|
||||
)
|
||||
env.time_stamp = time_stamp
|
||||
env.set_data({"update": 2})
|
||||
return env.to_json_string()
|
||||
|
||||
|
||||
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_empty_sn_key(self):
|
||||
self.assertEqual(generate_secret_key(""), "0")
|
||||
|
||||
|
||||
class TestEnvelope(unittest.TestCase):
|
||||
def test_key_order_no_data_key(self):
|
||||
env = direct_connect_request(
|
||||
CommunicationCmd.DEVICE_APOWER_INFO, SN, data={"update": 2})
|
||||
d = env.to_dict()
|
||||
self.assertEqual(
|
||||
list(d),
|
||||
["cmdType", "equipNo", "type", "timeStamp", "snno", "len",
|
||||
"crc", "dataArea"],
|
||||
)
|
||||
self.assertNotIn("data", d)
|
||||
|
||||
def test_no_data_means_seven_keys(self):
|
||||
env = direct_connect_request(CommunicationCmd.BMS, SN, data=None)
|
||||
d = env.to_dict()
|
||||
self.assertEqual(
|
||||
list(d),
|
||||
["cmdType", "equipNo", "type", "timeStamp", "snno", "len", "crc"],
|
||||
)
|
||||
self.assertEqual(env.length, 0)
|
||||
self.assertEqual(env.crc, "")
|
||||
|
||||
def test_empty_map_means_no_data_area(self):
|
||||
env = direct_connect_request(CommunicationCmd.BMS, SN, data={})
|
||||
self.assertNotIn("dataArea", env.to_dict())
|
||||
self.assertEqual(env.length, 0)
|
||||
self.assertEqual(env.crc, "")
|
||||
|
||||
def test_len_crc_auto_uppercase(self):
|
||||
env = direct_connect_request(
|
||||
CommunicationCmd.DEVICE_APOWER_INFO, SN, data={"update": 2})
|
||||
self.assertEqual(env.length, len(b'{"update":2}'))
|
||||
import zlib
|
||||
self.assertEqual(
|
||||
env.crc, format(zlib.crc32(b'{"update":2}'), "X"))
|
||||
self.assertEqual(env.crc, env.crc.upper())
|
||||
|
||||
def test_int_fields(self):
|
||||
env = direct_connect_request(CommunicationCmd.BMS, SN, snno=7)
|
||||
d = env.to_dict()
|
||||
self.assertIsInstance(d["type"], int)
|
||||
self.assertIsInstance(d["timeStamp"], int)
|
||||
self.assertIsInstance(d["snno"], int)
|
||||
self.assertEqual(d["snno"], 7)
|
||||
self.assertEqual(d["type"], 0) # query
|
||||
# whole seconds, 10 digits
|
||||
self.assertEqual(len(str(d["timeStamp"])), 10)
|
||||
|
||||
def test_snno_sequence_wraps(self):
|
||||
seq = SequenceNumber()
|
||||
self.assertEqual([seq.next(), seq.next()], [1, 2])
|
||||
seq = SequenceNumber(start=9999)
|
||||
self.assertEqual(seq.next(), 1)
|
||||
|
||||
def test_empty_sn_placeholder(self):
|
||||
env = CommunicationObject(0, "")
|
||||
self.assertEqual(env.equip_no, "00000000")
|
||||
|
||||
def test_from_dict_roundtrip(self):
|
||||
env = direct_connect_request(
|
||||
CommunicationCmd.DEVICE_APOWER_INFO, SN, data={"update": 2},
|
||||
snno=3)
|
||||
clone = CommunicationObject.from_json_string(env.to_json_string())
|
||||
self.assertEqual(clone.to_dict(), env.to_dict())
|
||||
|
||||
|
||||
class TestCipher(unittest.TestCase):
|
||||
def test_header_plaintext_first(self):
|
||||
wire = encrypt_object(_apower_json(), generate_secret_key(SN))
|
||||
a, b = split_envelope(_apower_json())
|
||||
self.assertEqual(wire[:len(a)], a)
|
||||
self.assertTrue(a.endswith(b","))
|
||||
self.assertEqual(len(wire), len(a) + len(b))
|
||||
|
||||
def test_header_len_matches(self):
|
||||
a, _ = split_envelope(_apower_json())
|
||||
self.assertEqual(header_len_for(0x19, SN), len(a))
|
||||
|
||||
def test_stream_is_absolute_index_plus_n(self):
|
||||
a, b = split_envelope(_apower_json())
|
||||
n = int(generate_secret_key(SN), 32)
|
||||
wire = encrypt_object(_apower_json(), generate_secret_key(SN))
|
||||
tail = wire[len(a):]
|
||||
for i, byte in enumerate(b):
|
||||
self.assertEqual(tail[i], (byte + i + len(a) + n) & 0xFF)
|
||||
|
||||
def test_request_roundtrip(self):
|
||||
body = _apower_json()
|
||||
wire = encrypt_object(body, generate_secret_key(SN))
|
||||
self.assertEqual(
|
||||
decrypt_request(wire, generate_secret_key(SN),
|
||||
cmd_type=0x19, equip_no=SN),
|
||||
body,
|
||||
)
|
||||
|
||||
def test_decrypt_response_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)
|
||||
|
||||
def test_decrypt_data_failures_are_empty(self):
|
||||
self.assertEqual(decrypt_data(b""), "")
|
||||
self.assertEqual(decrypt_data(b"no markers here"), "")
|
||||
# only one marker
|
||||
self.assertEqual(decrypt_data(b"abc\x58def"), "")
|
||||
# probe not JSON
|
||||
self.assertEqual(decrypt_data(b"\x58garbage\x58\x00\x01"), "")
|
||||
|
||||
|
||||
class TestSocketFraming(unittest.TestCase):
|
||||
def test_send_is_raw(self):
|
||||
payload = b"\x01\x02\x03"
|
||||
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
|
||||
dec = SocketFrameDecoder()
|
||||
# first half: incomplete -> nothing yet
|
||||
self.assertEqual(dec.feed(resp[:len(resp) // 2]), [])
|
||||
# rest: one complete JSON string
|
||||
out = dec.feed(resp[len(resp) // 2:])
|
||||
self.assertEqual(len(out), 1)
|
||||
self.assertEqual(json.loads(out[0])["equipNo"], SN)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user