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

"""Envelope encryption — ``communication_center/core/encryption.dart``.

Reconstructed from the blutter disassembly; see
``docs/RE_DART_FINDINGS.md`` §5 and Open items #2 / #5.

Wire form of an envelope::

    s  = compact JSON, e.g.  {"cmdType":0,"equipNo":"SN",...,"data":""}
    a  = UTF-8( parts[0] + "," + parts[1] )     # {"cmdType":..,"equipNo":..
    b  = UTF-8( ",".join(parts[2:]) )           # the rest of the JSON
    key_int = int.tryParse(key, radix) or 0
    wire  = bytes( (b[i] + i + len(a) + key_int) & 0xFF ) || a

i.e. everything after the second comma is transformed byte-wise (position-
dependent additive stream) and the two leading JSON parts are appended
**unchanged** at the end.

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`` (null -> 0) with a ``radix``
  argument; a literal ``#0x20`` (32) is materialised right before the call in
  both encrypt and decrypt paths.  Hex is self-consistent with the key
  encoding, so **radix 16 is the default** here — pass ``radix=32`` to try the
  other candidate.  Confirm against a live capture.
"""

from __future__ import annotations

from typing import Optional, Tuple

#: Default key radix — see module docstring for the 16-vs-32 note.
DEFAULT_KEY_RADIX = 16

#: Alternate candidate observed next to the ``tryParse`` call in the asm.
ALT_KEY_RADIX = 32


def generate_secret_key(equip_no: str) -> str:
    """``generateSecretKey(equipNo)``: sum of UTF-8 bytes, ``& 0xFF``, hex.

    Example: ``generate_secret_key("01234567890123456789")`` is the lowercase
    hex of ``(sum of those 20 bytes) & 0xFF`` (1- or 2-digit string).
    """
    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)."""
    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 (``transform(b) || a``)."""
    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 transformed + a


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 can use to locate the unencrypted
    tail of an incoming frame.
    """
    part0 = '"cmdType":%d' % int(cmd_type)
    part1 = '"equipNo":"%s"' % equip_no
    return len(part0) + 1 + len(part1)


def decrypt_object(
    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.

    ``header_len`` is ``len(a)`` (the unencrypted tail).  When omitted it is
    derived from ``cmd_type`` + ``equip_no`` via :func:`header_len_for` —
    both must be known in that case.
    """
    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[len(wire) - header_len:] if header_len else b""
    b = wire[: len(wire) - header_len] if header_len else bytes(wire)
    key_int = _parse_key(key, radix)
    plain_b = bytes(
        (byte - i - len(a) - key_int) & 0xFF for i, byte in enumerate(b)
    )
    a_str = a.decode("utf-8")
    b_str = plain_b.decode("utf-8")
    if not a_str:
        return b_str
    if not b_str:
        return a_str
    return a_str + "," + b_str