239 lines
9.0 KiB
Python
239 lines
9.0 KiB
Python
"""Envelope cipher — ``communication_center/core/encryption.dart``.
|
|
|
|
Reconstructed from the blutter disassembly; see
|
|
``docs/RE_DART_FINDINGS.md`` §5.
|
|
|
|
Request wire form of an envelope::
|
|
|
|
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) 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). Both directions use this same framing (verified live: a 1154-byte
|
|
device reply decrypts with the identical split — see ``decrypt_wire``).
|
|
|
|
.. 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.
|
|
|
|
Confirmed from the disassembly *and live traffic*:
|
|
|
|
* ``generateSecretKey`` = ``sum(utf8(sn)) & 0xFF`` rendered as lowercase hex
|
|
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
|
|
|
|
import json
|
|
from typing import Optional, Tuple
|
|
|
|
#: 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
|
|
|
|
|
|
def generate_secret_key(equip_no: str) -> str:
|
|
"""``generateSecretKey(equipNo)``: sum of UTF-8 bytes, ``& 0xFF``, hex.
|
|
|
|
Example: ``generate_secret_key("A03X2516013210060005") == "1c"``.
|
|
"""
|
|
total = sum(equip_no.encode("utf-8")) & 0xFF
|
|
return format(total, "x")
|
|
|
|
|
|
def _parse_key(key: str, radix: int) -> int:
|
|
"""Mirror of Dart ``int.tryParse(key, radix: r)`` -> null -> 0."""
|
|
try:
|
|
return int(key, radix)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def split_envelope(json_str: str) -> Tuple[bytes, bytes]:
|
|
"""Split the JSON into ``(a, b)`` as the app does (UTF-8 encoded).
|
|
|
|
``a`` is the plaintext header ``parts[0] + "," + parts[1] + ","``
|
|
(trailing comma included); ``b`` is the remaining tail.
|
|
"""
|
|
parts = json_str.split(",")
|
|
a = (parts[0] + "," + parts[1] + ",").encode("utf-8")
|
|
b = ",".join(parts[2:]).encode("utf-8")
|
|
return a, b
|
|
|
|
|
|
def encrypt_object(
|
|
json_str: str, key: str, radix: int = DEFAULT_KEY_RADIX
|
|
) -> bytes:
|
|
"""Encrypt an envelope JSON string into wire bytes (``a || crypt(b)``)."""
|
|
a, b = split_envelope(json_str)
|
|
key_int = _parse_key(key, radix)
|
|
transformed = bytes(
|
|
(byte + i + len(a) + key_int) & 0xFF for i, byte in enumerate(b)
|
|
)
|
|
return bytes(a) + transformed
|
|
|
|
|
|
def header_len_for(cmd_type: int, equip_no: str) -> int:
|
|
"""``len(a)`` for an envelope with the given ``cmdType`` / ``equipNo``.
|
|
|
|
Derived from the *known* key order (``cmdType``, ``equipNo`` first) and
|
|
compact JSON — this is what a receiver that knows the serial uses to
|
|
locate the end of the unencrypted header of an incoming *request* frame.
|
|
"""
|
|
part0 = '"cmdType":%d' % int(cmd_type)
|
|
part1 = '"equipNo":"%s"' % equip_no
|
|
# parts[0] + "," + parts[1] + "," — plus the leading "{" of the JSON.
|
|
return len(("{" + part0).encode("utf-8")) + 1 + len(part1.encode("utf-8")) + 1
|
|
|
|
|
|
def decrypt_request(
|
|
wire: bytes,
|
|
key: str,
|
|
header_len: Optional[int] = None,
|
|
radix: int = DEFAULT_KEY_RADIX,
|
|
cmd_type: Optional[int] = None,
|
|
equip_no: Optional[str] = None,
|
|
) -> str:
|
|
"""Inverse of :func:`encrypt_object` -> original JSON string.
|
|
|
|
Splits the wire at the known plaintext-header length (device side —
|
|
the device knows its own serial). ``header_len`` is ``len(a)``; when
|
|
omitted it is derived from ``cmd_type`` + ``equip_no``.
|
|
"""
|
|
if header_len is None:
|
|
if cmd_type is None or equip_no is None:
|
|
raise ValueError(
|
|
"header_len (or cmd_type + equip_no) is required to split "
|
|
"the wire buffer"
|
|
)
|
|
header_len = header_len_for(cmd_type, equip_no)
|
|
a = wire[:header_len]
|
|
b = wire[header_len:]
|
|
key_int = _parse_key(key, radix)
|
|
plain_b = bytes(
|
|
(byte - i - len(a) - key_int) & 0xFF for i, byte in enumerate(b)
|
|
)
|
|
return (a + plain_b).decode("utf-8")
|
|
|
|
|
|
# Backwards-compatible alias (device-side request split).
|
|
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.
|
|
|
|
Faithful port of the app's routine: find the first two ``0x58`` bytes,
|
|
take ``s`` as everything before the second one, probe ``s + "}"`` for
|
|
``equipNo`` to derive the key, decode the tail after the marker, and
|
|
return ``s + "," + tail``. Returns ``""`` on any failure (markers
|
|
missing, probe not JSON, decode error) — the caller keeps accumulating.
|
|
"""
|
|
try:
|
|
raw = bytes(wire)
|
|
i = raw.find(bytes((MARKER,)))
|
|
if i < 0:
|
|
return ""
|
|
j = raw.find(bytes((MARKER,)), i + 1)
|
|
if j < 0:
|
|
return ""
|
|
try:
|
|
s = raw[:j].decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return ""
|
|
try:
|
|
probe = json.loads(s + "}")
|
|
except ValueError:
|
|
return ""
|
|
if not isinstance(probe, dict):
|
|
return ""
|
|
equip_no = probe.get("equipNo", "")
|
|
key = generate_secret_key(str(equip_no))
|
|
key_int = _parse_key(key, radix)
|
|
prefix = s + ","
|
|
prefix_len = len(prefix.encode("utf-8"))
|
|
tail = raw[j + 1:]
|
|
plain = bytes(
|
|
(byte - k - prefix_len - key_int) & 0xFF
|
|
for k, byte in enumerate(tail)
|
|
)
|
|
try:
|
|
tail_str = plain.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return ""
|
|
return prefix + tail_str
|
|
except Exception:
|
|
return ""
|