261 lines
9.5 KiB
Python
261 lines
9.5 KiB
Python
"""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 os
|
|
import unittest
|
|
|
|
from franklinwh_agate_client.protocol import (
|
|
DEFAULT_KEY_RADIX,
|
|
CommunicationCmd,
|
|
CommunicationObject,
|
|
SequenceNumber,
|
|
socket_cmd_type,
|
|
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
|
|
|
|
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 TestSocketCmdSpace(unittest.TestCase):
|
|
def test_socket_ids(self):
|
|
self.assertEqual(socket_cmd_type(CommunicationCmd.QUERY_DEVICE_INFO), 1101)
|
|
self.assertEqual(socket_cmd_type(CommunicationCmd.DEVICE_APOWER_INFO), 1307)
|
|
self.assertEqual(socket_cmd_type(CommunicationCmd.BMS), 1705)
|
|
self.assertEqual(socket_cmd_type(CommunicationCmd.EMS), 1707)
|
|
|
|
def test_direct_request_uses_socket_id(self):
|
|
env = direct_connect_request(CommunicationCmd.DEVICE_APOWER_INFO, SN)
|
|
self.assertEqual(env.cmd_type, 1307)
|
|
self.assertTrue(env.to_json_string().startswith('{"cmdType":1307,'))
|
|
|
|
def test_discovery_shape_like_capture(self):
|
|
env = direct_connect_request(CommunicationCmd.QUERY_DEVICE_INFO, "")
|
|
body = env.to_json_string()
|
|
self.assertTrue(body.startswith(
|
|
'{"cmdType":1101,"equipNo":"00000000",'))
|
|
|
|
|
|
class TestSecretKey(unittest.TestCase):
|
|
def test_real_sn_key(self):
|
|
self.assertEqual(generate_secret_key(SN), "1c")
|
|
|
|
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")
|
|
|
|
|
|
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)
|
|
# socket payload is always seeded with the operation type first
|
|
self.assertEqual(d["dataArea"], {"opt": 0, "update": 2})
|
|
|
|
def test_no_data_still_carries_opt(self):
|
|
env = direct_connect_request(CommunicationCmd.BMS, SN, data=None)
|
|
d = env.to_dict()
|
|
self.assertIn("dataArea", d)
|
|
self.assertEqual(d["dataArea"], {"opt": 0})
|
|
self.assertEqual(env.length, len(b'{"opt":0}'))
|
|
self.assertTrue(env.crc)
|
|
|
|
def test_empty_map_still_carries_opt(self):
|
|
env = direct_connect_request(CommunicationCmd.BMS, SN, data={})
|
|
self.assertEqual(env.to_dict()["dataArea"], {"opt": 0})
|
|
|
|
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'{"opt":0,"update":2}'))
|
|
import zlib
|
|
self.assertEqual(
|
|
env.crc, format(zlib.crc32(b'{"opt":0,"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(1307, SN), len(a))
|
|
|
|
def test_stream_is_absolute_index_plus_n(self):
|
|
a, b = split_envelope(_apower_json())
|
|
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):
|
|
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=1307, equip_no=SN),
|
|
body,
|
|
)
|
|
|
|
def test_wire_roundtrip(self):
|
|
body = _apower_json()
|
|
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""), "")
|
|
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):
|
|
wire = open(os.path.join(FIXTURES, "discovery_req.bin"),
|
|
"rb").read()
|
|
dec = SocketFrameDecoder()
|
|
# first half: incomplete -> nothing yet
|
|
self.assertEqual(dec.feed(wire[:len(wire) // 2]), [])
|
|
# rest: one complete JSON string
|
|
out = dec.feed(wire[len(wire) // 2:])
|
|
self.assertEqual(len(out), 1)
|
|
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")
|