franklinwh-agate-client/franklinwh_agate_client/protocol/crypto.py

"""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 ""