"""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":25,"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 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. 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:: 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: * ``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. * ``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). """ 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 #: 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 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 ""