diff --git a/SolixBLE/advertisement.py b/SolixBLE/advertisement.py new file mode 100644 index 0000000..89e04d1 --- /dev/null +++ b/SolixBLE/advertisement.py @@ -0,0 +1,116 @@ +"""Parsing of the Anker manufacturer advertisement record. + +Anker devices place a fixed-shape record under BLE company identifier +``0xffff`` in their advertisements. It carries the real MAC, the model +(``product_type`` and ``sku``), and a ``capability`` byte that declares which +negotiation path the device accepts, all readable before a connection is made. + +.. moduleauthor:: kb1ibt + +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from bleak.backends.scanner import AdvertisementData + +#: BLE company identifier the record is published under (a reserved/test id). +ANKER_COMPANY_ID = 0xFFFF + +#: Capability bit meaning the device accepts the encrypted ``4xxx`` negotiation +#: under the static key. Devices that set it also accept the cleartext path; +#: newer firmware may accept only the encrypted one. +CAPABILITY_ENCRYPTED_ECDH = 0x04 + +#: Byte offsets within the record. ``capability`` has no fixed offset -- it is +#: the byte after the sku when present, and absent on some models -- so it is +#: derived from the sku length rather than listed here. +_MAC = slice(1, 7) +_BIND_TYPE = 7 +_PRODUCT_TYPE = slice(8, 10) +_SKU_START = 10 + +#: Length of the ascii sku, keyed by ``version_code``. The record is otherwise +#: fixed up to the sku, and the capability byte (if any) follows the sku. +_SKU_LENGTH = {1: 3, 2: 4} + +#: Shortest valid record: everything up to the sku, plus the shortest sku. +_MIN_LENGTH = _SKU_START + min(_SKU_LENGTH.values()) + + +@dataclass(frozen=True) +class AnkerAdvertisement: + """Decoded Anker manufacturer record. + + :param version_code: Record layout version (selects the sku length). + :param mac: Device MAC as lowercase hex without separators. + :param bind_type: Provisioning-state byte (dynamic per device). + :param product_type: Two-byte model key as lowercase hex. + :param sku: Ascii sku, a substring of the device serial. + :param capability: Capability mask, or None when the record omits it. + """ + + version_code: int + mac: str + bind_type: int + product_type: str + sku: str + capability: int | None + + +def parse_manufacturer_record(data: bytes) -> AnkerAdvertisement | None: + """Decode an Anker ``0xffff`` manufacturer record. + + The layout is ``version_code(1) | mac(6) | bind_type(1) | product_type(2) | + sku(3-4 ascii) | capability(1, optional)``. The sku length follows the + version code, and the capability byte is present only on some models, so it + must be read relative to the sku rather than at a fixed offset. + + :param data: Raw manufacturer-data bytes for company id ``0xffff``. + :returns: The decoded record, or None if it is too short or malformed. + """ + if len(data) < _MIN_LENGTH: + return None + + version_code = data[0] + sku_length = _SKU_LENGTH.get(version_code) + if sku_length is None: + return None + + sku_end = _SKU_START + sku_length + if len(data) < sku_end: + return None + + try: + sku = data[_SKU_START:sku_end].decode("ascii") + except UnicodeDecodeError: + return None + + capability = data[sku_end] if len(data) > sku_end else None + + return AnkerAdvertisement( + version_code=version_code, + mac=data[_MAC].hex(), + bind_type=data[_BIND_TYPE], + product_type=data[_PRODUCT_TYPE].hex(), + sku=sku, + capability=capability, + ) + + +def capability_from_advertisement(advertisement: AdvertisementData) -> int | None: + """Read the capability byte from an advertisement, if it carries one. + + :param advertisement: Advertisement data from a bleak scan callback. + :returns: The capability mask, or None if there is no Anker record or the + record omits the byte. + """ + record = advertisement.manufacturer_data.get(ANKER_COMPANY_ID) + if record is None: + return None + + parsed = parse_manufacturer_record(record) + return parsed.capability if parsed is not None else None diff --git a/SolixBLE/device.py b/SolixBLE/device.py index 8e1d935..1930b71 100644 --- a/SolixBLE/device.py +++ b/SolixBLE/device.py @@ -9,6 +9,7 @@ import inspect import logging import time +import uuid from collections.abc import Callable from datetime import datetime from functools import partial @@ -26,8 +27,9 @@ ) from cryptography.hazmat.primitives.padding import PKCS7 +from SolixBLE.advertisement import CAPABILITY_ENCRYPTED_ECDH from SolixBLE.constructs import FragmentedPayload, Packet, ParameterDict, Parameters -from SolixBLE.utilities import _to_bytes, get_posix_tz +from SolixBLE.utilities import _offset_seconds_west, _to_bytes, get_posix_tz from .const import ( DEFAULT_METADATA_INT, @@ -49,6 +51,30 @@ #: The UUID sent to the device during negotiation UUID_STRING = "b2dc0b17-b75d-4abf-ba6e-ec7c997c23e7" +#: Static AES-GCM key, nonce and AAD for the encrypted negotiation, used before +#: the ECDH secret exists. The device derives the key and nonce from two DROM +#: constants at connect time (A1783 module firmware, confirmed 2026-09-08); the +#: results are fixed, so the values are inlined here. +NEGOTIATION_KEY = "b8ff7422955d4eb6d554a2c470280559" +NEGOTIATION_NONCE = "6ba3e3f2f3a60f2971ce5d1f" +NEGOTIATION_AAD = "3322110077665544bbaa9988ffeeddcc" + +#: The client's ECDH public key (uncompressed P-256 point without the ``04`` +#: prefix) matching ``const.PRIVATE_KEY``, sent in the ``4021`` exchange. +CLIENT_PUBLIC_KEY = ( + "060ea168f232aedb37fb2d120c49180329ac72ab5ec3eb8fd30a2f252dc5e151" + "dabccd9b1dc1e288704ca760a0d8c918e5c94823a1f609a4bf07fb4c33ee2190" +) + +#: Client's proposed MTU in the capability negotiation (``a4`` of ``0003``/``0005``). +#: ``u16`` little-endian ``0x00f0`` = 61440 = "no limit"; the device then streams at +#: ``min(this, its own ceiling)``. +NEGOTIATION_MTU_PROPOSAL = "00f0" + +#: encryptMethod the client confirms in ``0005`` (``a5``). The device selects ECDH when +#: ``a5 & 0x44`` is set; ``0x40`` is the base flag. +NEGOTIATION_ENCRYPT_METHOD = "40" + class SolixBLEDevice: """Solix BLE device object.""" @@ -61,8 +87,32 @@ class SolixBLEDevice: #: The maximum packet size an Anker device is able to send _mtu = 253 - def __init__(self, ble_device: BLEDevice) -> None: - """Initialise device object. Does not connect automatically.""" + #: Whether to negotiate on the encrypted path when the advertised + #: capability is unknown. A known capability byte overrides this; it is the + #: fallback for a device constructed without one. + _DEFAULT_ENCRYPTED_NEGOTIATION: bool = False + + def __init__( + self, + ble_device: BLEDevice, + capability: int | None = None, + client_token: str | None = None, + ) -> None: + """Initialise device object. Does not connect automatically. + + :param ble_device: The bleak device to wrap. + :param capability: The device's advertised capability byte, if known. + It selects the negotiation path (encrypted when the ECDH bit is + set). ``BLEDevice`` is slotted and cannot carry it, so the caller + reads it from the advertisement (see + :func:`SolixBLE.advertisement.capability_from_advertisement`) and + passes it here; None falls back to the class default. + :param client_token: Stable per-client identifier registered with the + device on the encrypted path. On hardened firmware the first use of + a new token needs a physical button press; the device then accepts + it on every later connection, so the caller should persist it and + pass the same value each time. A random one is generated if omitted. + """ _LOGGER.debug( f"Initializing Solix device '{ble_device.name}' with" @@ -83,6 +133,23 @@ def __init__(self, ble_device: BLEDevice) -> None: self._disconnect_event: asyncio.Event = asyncio.Event() self._connection_attempts: int = 0 self._shared_secret: bytes | None = None + self._capability: int | None = capability + self._authorized: bool = False + self._client_token: str = client_token or str(uuid.uuid4()) + self._auth_mode: bytes | None = None + + @property + def _encrypted_negotiation(self) -> bool: + """Whether to negotiate on the encrypted (GCM / ``4xxx``) path. + + Chosen from the advertised capability's ECDH bit when the byte is + known, else the class default. The device MCU's ``auth_mode`` is the + real policy, but it is only readable once stage 2 arrives, so the + pre-connect advert picks the initial cipher. + """ + if self._capability is not None: + return bool(self._capability & CAPABILITY_ENCRYPTED_ECDH) + return self._DEFAULT_ENCRYPTED_NEGOTIATION def add_callback(self, function: Callable[[], None]) -> None: """Register a callback to be run on state updates. @@ -103,7 +170,25 @@ def remove_callback(self, function: Callable[[], None]) -> None: self._state_changed_callbacks.remove(function) async def _initiate_negotiations(self) -> None: - """Send the negotiation initiation command.""" + """Send the negotiation initiation command. + + The encrypted path opens with ``4001`` under the static GCM key; the + cleartext path opens with ``0001`` carrying the client UUID. + """ + if self._encrypted_negotiation: + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4001", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + }, + ) + return + await self._send_packet(pattern=NEGOTIATION_PATTERN, cmd="0001", parameters={ "a1": { @@ -311,7 +396,11 @@ def negotiated(self) -> bool: :returns: True/False if session has been negotiated and connected. """ - return self.connected and self._shared_secret is not None + return ( + self.connected + and self._shared_secret is not None + and (not self._encrypted_negotiation or self._authorized) + ) @property def available(self) -> bool: @@ -378,8 +467,41 @@ def _parse_string(self, key: str, begin: int = None, end: int = None) -> str: else DEFAULT_METADATA_STRING ) + def _gcm_key_nonce(self) -> tuple[bytes, bytes]: + """Return the GCM (key, nonce): the ECDH secret if derived, else static. + + Before the ECDH exchange the encrypted path is keyed on the static + negotiation key and nonce; afterwards on the derived shared secret. + """ + if self._shared_secret is not None: + return self._shared_secret[:16], self._shared_secret[16:28] + return bytes.fromhex(NEGOTIATION_KEY), bytes.fromhex(NEGOTIATION_NONCE) + + def _decrypt_payload_gcm(self, payload: bytes) -> bytes: + """AES-GCM decrypt a payload on the encrypted negotiation path. + + The last 16 bytes are the authentication tag. + """ + key, nonce = self._gcm_key_nonce() + mac = payload[-16:] + body = payload[:-16] + cipher = AES.new(key, AES.MODE_GCM, nonce=nonce) + cipher.update(bytes.fromhex(NEGOTIATION_AAD)) + try: + return cipher.decrypt_and_verify(body, mac) + except ValueError: + _LOGGER.exception("GCM tag verify failed; decrypting without verify") + cipher = AES.new(key, AES.MODE_GCM, nonce=nonce) + return cipher.decrypt(body) + def _decrypt_payload(self, payload: bytes) -> bytes: - """Decrypt payload using negotiated shared secret and IV if available.""" + """Decrypt payload using negotiated shared secret and IV if available. + + The encrypted path uses AES-GCM (static key before the ECDH secret + exists); the cleartext path uses AES-CBC once the secret is derived. + """ + if self._encrypted_negotiation: + return self._decrypt_payload_gcm(payload) if self._shared_secret is None: _LOGGER.debug("Skipping decryption as key not negotiated...") @@ -394,7 +516,17 @@ def _decrypt_payload(self, payload: bytes) -> bytes: return unpadded_data + unpadder.finalize() def _encrypt_payload(self, payload: bytes) -> bytes: - """Encrypt payload using negotiated shared secret if available.""" + """Encrypt payload using negotiated shared secret if available. + + AES-GCM on the encrypted path (static key before the secret exists), + AES-CBC on the cleartext path. + """ + if self._encrypted_negotiation: + key, nonce = self._gcm_key_nonce() + cipher = AES.new(key, AES.MODE_GCM, nonce=nonce) + cipher.update(bytes.fromhex(NEGOTIATION_AAD)) + encrypted, mac = cipher.encrypt_and_digest(payload) + return encrypted + mac if self._shared_secret is None: _LOGGER.debug("Skipping encryption as key not negotiated...") @@ -557,6 +689,12 @@ async def _process_notification( else: _LOGGER.debug(f"Received unknown message of type: {cmd.hex()}") + # The unsolicited authorization grant the device pushes after a + # physical button press on the encrypted path. + case "030101": + _LOGGER.debug("Received authorization grant message!") + return await self._process_arm_grant(cmd, payload) + case _: _LOGGER.warning( f"Unexpected packet type '{pattern}' sent by device! Packet: {data.hex()}" @@ -605,6 +743,13 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: plain_text_payload = self._decrypt_payload(payload) _LOGGER.debug(f"Plain-text payload: {plain_text_payload.hex()}") + + # The encrypted path has its own stages and a status-9 reply that does + # not parse as parameters, so branch before the generic parse below. + if self._encrypted_negotiation: + await self._process_negotiation_encrypted(cmd, plain_text_payload) + return + parameters = Parameters.parse(plain_text_payload) _LOGGER.debug(f"Parameters: {parameters.to_str(verbose=True, types=False)}") @@ -638,7 +783,7 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: }, "a4": { "key": bytes.fromhex("a4"), "type": None, - "value": bytes.fromhex("00f0"), + "value": bytes.fromhex(NEGOTIATION_MTU_PROPOSAL), }, }, ) @@ -690,11 +835,11 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: }, "a4": { "key": bytes.fromhex("a4"), "type": None, - "value": bytes.fromhex("00f0"), + "value": bytes.fromhex(NEGOTIATION_MTU_PROPOSAL), }, "a5": { "key": bytes.fromhex("a5"), "type": None, - "value": bytes.fromhex("40"), + "value": bytes.fromhex(NEGOTIATION_ENCRYPT_METHOD), }, }, ) @@ -780,6 +925,225 @@ async def _process_negotiation(self, cmd: bytes, payload: bytes) -> None: f"Received unexpected negotiation request response from device! cmd: '{cmd}', parameters: '{parameters}'" ) + async def _process_negotiation_encrypted( + self, + cmd: bytes, + plaintext: bytes, + ) -> None: + """Drive the encrypted (GCM / ``4xxx``) negotiation and authorization. + + Built to the A1783 comms-module firmware: ``4005`` echoes the device's + own auth mode, the ``4022`` confer carries a signed UTC offset, and the + link is not authorized until a ``4027`` registration succeeds. That + succeeds at once for an already-registered client token; otherwise the + device replies status ``9`` and authorizes only after a physical button + press, which arrives as an unsolicited grant on pattern ``030101``. + + :param cmd: The negotiation response command code. + :param plaintext: The GCM-decrypted response payload. + """ + match cmd.hex(): + # Stage 1: propose capabilities. + case "4801": + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4003", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + "a3": { + "key": bytes.fromhex("a3"), + "type": None, + "value": bytes.fromhex("20"), + }, + "a4": { + "key": bytes.fromhex("a4"), + "type": None, + "value": bytes.fromhex(NEGOTIATION_MTU_PROPOSAL), + }, + }, + ) + + # Stage 2: record the device MTU and auth mode, ask for device info. + case "4803": + parameters = Parameters.parse(plaintext) + self._mtu = int.from_bytes( + parameters["a2"].value_legacy, + byteorder="little", + ) + self._auth_mode = parameters["a5"].value_legacy + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4029", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + }, + ) + + # Stage 3: set capabilities, echoing the device's declared auth mode. + case "4829": + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4005", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + "a3": { + "key": bytes.fromhex("a3"), + "type": None, + "value": bytes.fromhex("20"), + }, + # The firmware does not read the MTU echo; send the + # declared value for correctness. + "a4": { + "key": bytes.fromhex("a4"), + "type": None, + "value": self._mtu.to_bytes(2, byteorder="little"), + }, + # a5 selects the cipher; the 0x44 bits mean ECDH. + "a5": { + "key": bytes.fromhex("a5"), + "type": None, + "value": bytes.fromhex("44"), + }, + # a6 must equal the auth mode the device gave in 4803. + "a6": { + "key": bytes.fromhex("a6"), + "type": None, + "value": self._auth_mode or bytes.fromhex("02"), + }, + }, + ) + + # Stage 4: send our ECDH public key. + case "4805": + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4021", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": bytes.fromhex(CLIENT_PUBLIC_KEY), + }, + }, + ) + + # Stage 5: derive the shared secret, send the timezone confer. + case "4821": + parameters = Parameters.parse(plaintext) + self._negotiation_timestamp = time.time() + device_public_key_bytes = ( + bytes.fromhex("04") + parameters["a1"].value_legacy + ) + device_public_key = EllipticCurvePublicKey.from_encoded_point( + SECP256R1(), + device_public_key_bytes, + ) + private_value = int.from_bytes( + bytes.fromhex(PRIVATE_KEY), + byteorder="big", + ) + private_key = derive_private_key(private_value, SECP256R1()) + self._shared_secret = private_key.exchange( + ECDH(), + device_public_key, + ) + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4022", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + # a3: UTC offset, signed int32 LE, seconds west of UTC. + "a3": { + "key": bytes.fromhex("a3"), + "type": None, + "value": _offset_seconds_west(), + }, + "a5": { + "key": bytes.fromhex("a5"), + "type": None, + "value": (get_posix_tz() or FALLBACK_TZ).encode(), + }, + }, + ) + + # Stage 6: register the client token to authorize the link. + case "4822": + await self._send_packet( + pattern=NEGOTIATION_PATTERN, + cmd="4027", + parameters={ + "a1": { + "key": bytes.fromhex("a1"), + "type": None, + "value": lambda self: self._timestamp(), + }, + "a2": { + "key": bytes.fromhex("a2"), + "type": None, + "value": self._client_token.encode(), + }, + }, + ) + + # Stage 7: authorization result. + case "4827": + if plaintext[:1] == b"\x00": + _LOGGER.debug("Client registration accepted; link authorized!") + self._authorized = True + elif plaintext[:1] == b"\x09": + _LOGGER.info( + "Device is awaiting a physical button press to authorize " + "this client; press the button on the device.", + ) + else: + _LOGGER.warning( + "Unexpected 4027 registration status: %s", + plaintext[:1].hex(), + ) + + case _: + _LOGGER.warning( + "Received unexpected encrypted negotiation response! cmd: %s", + cmd.hex(), + ) + + async def _process_arm_grant(self, cmd: bytes, payload: bytes) -> None: + """Handle the unsolicited authorization grant on pattern ``030101``. + + The device pushes a ``4827`` with status ``0`` here once the operator + presses the button for a newly registered client token, authorizing the + link. + + :param cmd: The command code of the grant frame. + :param payload: The (still encrypted) frame payload. + """ + plaintext = self._decrypt_payload(payload) + if cmd.hex() == "4827" and plaintext[:1] == b"\x00": + _LOGGER.debug("Client authorized via button press!") + self._authorized = True + else: + _LOGGER.debug( + "Unexpected 030101 frame: cmd %s, payload %s", + cmd.hex(), + plaintext.hex(), + ) + def _timestamp(self) -> bytes: """Unix timestamp in byte form (4B).""" return int(time.time()).to_bytes(length=4, byteorder="little", signed=False) @@ -1053,6 +1417,8 @@ def _reset_session(self, reset_data: bool = True) -> None: self._fragment_buffers = {} self._fragment_totals = {} self._shared_secret = None + self._authorized = False + self._auth_mode = None self._last_packet_timestamp = None self._negotiation_timestamp = None self._packet_futures: dict[bytes, list[asyncio.Future]] = {} diff --git a/SolixBLE/utilities.py b/SolixBLE/utilities.py index 5438e88..0a8593f 100644 --- a/SolixBLE/utilities.py +++ b/SolixBLE/utilities.py @@ -8,6 +8,7 @@ import importlib.resources as resources import inspect import logging +import time from typing import Callable import tzlocal @@ -103,3 +104,16 @@ def get_posix_tz() -> str | None: return lines[-1].decode("ascii").strip() except Exception: _LOGGER.exception("Unable to determine system time zone!") + + +def _offset_seconds_west() -> bytes: + """UTC offset as the firmware reads it: signed int32 LE, seconds west. + + POSIX counts seconds *west* of UTC, so US-Eastern in summer is ``+14400`` + and zones east of UTC are negative. + + :returns: The signed 4-byte little-endian offset in seconds west of UTC. + """ + gmtoff = time.localtime().tm_gmtoff + seconds_west = -gmtoff if gmtoff is not None else 0 + return seconds_west.to_bytes(4, byteorder="little", signed=True) diff --git a/docs/source/encrypted_negotiation.rst b/docs/source/encrypted_negotiation.rst new file mode 100644 index 0000000..38ce55a --- /dev/null +++ b/docs/source/encrypted_negotiation.rst @@ -0,0 +1,64 @@ +================================= +Encrypted negotiation and pairing +================================= + +.. _BLEDevice: https://bleak.readthedocs.io/en/latest/api/index.html#bleak.backends.device.BLEDevice/ + + +Some Anker devices negotiate their session over an encrypted handshake rather +than the cleartext one, and some firmware additionally requires the client to +be *paired* to the device -- confirmed by a physical button press -- before it +will stream telemetry or accept commands. Newer firmware in particular enforces +this. Both are handled automatically once the device is told which path to use +and given a stable client token. + + +Choosing the negotiation path +----------------------------- + +Each device advertises a capability byte that says whether it accepts the +encrypted negotiation, readable at scan time before connecting. Read it from the +manufacturer record in the advertisement and pass it to the device when you +construct it; the correct path is then chosen automatically -- encrypted when +the capability's ECDH bit is set, cleartext otherwise:: + + from SolixBLE.advertisement import capability_from_advertisement + + capability = capability_from_advertisement(advertising_data) + device = C1000G2(ble_device, capability=capability) + +.. note:: + + A `BLEDevice`_ cannot carry the capability itself, so it is passed to the + constructor rather than attached to the device. When it is not supplied the + class default is used. The ``advertising_data`` comes from your own Bleak + scan (for example a Home Assistant Bluetooth callback); the capability is a + property of the device model and does not change once a unit is bound. + + +Pairing with a client token +--------------------------- + +On the encrypted path the device authorizes a specific *client*, identified by a +token you provide. Pass a stable value and **persist it**, so the same client is +recognised on every later connection:: + + device = C1000G2(ble_device, capability=capability, client_token=my_token) + +If you do not pass one a random token is generated, which means a new client on +every run. On firmware that enforces pairing, the **first** connection with a +new token needs a one-time physical confirmation: + +#. Call :py:meth:`.connect`. The device reports that it is awaiting confirmation. +#. Press the button on the device (the same one used to wake its Bluetooth). +#. The connection completes and telemetry begins. + +After that first pairing the token is remembered, and every later connection is +authorized immediately with no button press. The whole process is local -- no +Anker account or cloud service is involved. + +.. note:: + + A device that does not enforce pairing authorizes as soon as the handshake + completes, so no button press is needed; the token is still registered for + later connections. diff --git a/docs/source/index.rst b/docs/source/index.rst index e1ac02f..9227fc7 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -186,6 +186,7 @@ Contents Home examples usage + encrypted_negotiation api limitations new_devices diff --git a/tests/test_advertisement.py b/tests/test_advertisement.py new file mode 100644 index 0000000..ca27a8d --- /dev/null +++ b/tests/test_advertisement.py @@ -0,0 +1,99 @@ +"""Tests for the Anker manufacturer advertisement record parser. + +Vectors are the real passive-scan records for the bench devices. + +.. moduleauthor:: kb1ibt +""" + +from unittest import mock + +import pytest + +from SolixBLE.advertisement import ( + ANKER_COMPANY_ID, + CAPABILITY_ENCRYPTED_ECDH, + AnkerAdvertisement, + capability_from_advertisement, + parse_manufacturer_record, +) + + +@pytest.mark.parametrize( + ("record", "expected"), + [ + pytest.param( + "01f49d8a2519b200b4014a544200", + AnkerAdvertisement(1, "f49d8a2519b2", 0x00, "b401", "JTB", 0x00), + id="a91b2_station_capability_0", + ), + pytest.param( + "01f49d8a2f05f000b402514a4204", + AnkerAdvertisement(1, "f49d8a2f05f0", 0x00, "b402", "QJB", 0x04), + id="a2345_charger_capability_4", + ), + pytest.param( + "01007f1d44e79e01b402514a4204", + AnkerAdvertisement(1, "007f1d44e79e", 0x01, "b402", "QJB", 0x04), + id="a2345_sealed_first_boot", + ), + pytest.param( + "027ce91346c50c00b11a444b4b4504", + AnkerAdvertisement(2, "7ce91346c50c", 0x00, "b11a", "DKKE", 0x04), + id="c2000g2_a1783_capability_4", + ), + pytest.param( + "01aabbccddeeff02b106373434", + AnkerAdvertisement(1, "aabbccddeeff", 0x02, "b106", "744", None), + id="f3800_no_capability_byte", + ), + ], +) +def test_parse_manufacturer_record(record: str, expected: AnkerAdvertisement) -> None: + """The record decodes to the app's own field values. + + :param record: Hex of the raw ``0xffff`` manufacturer record. + :param expected: The fields the app reports for that record. + """ + assert parse_manufacturer_record(bytes.fromhex(record)) == expected + + +@pytest.mark.parametrize( + ("record", "reason"), + [ + pytest.param("01f49d8a25", "too_short", id="too_short"), + pytest.param( + "03f49d8a2519b200b4014a544200", + "unknown_version", + id="unknown_version", + ), + pytest.param( + "01f49d8a2519b200b401ffffff00", + "non_ascii_sku", + id="non_ascii_sku", + ), + ], +) +def test_parse_manufacturer_record_rejects(record: str, reason: str) -> None: + """A malformed record decodes to None rather than a wrong guess. + + :param record: Hex of a record that cannot be trusted. + :param reason: Why the record is rejected (documentation only). + """ + assert reason + assert parse_manufacturer_record(bytes.fromhex(record)) is None + + +def test_capability_from_advertisement() -> None: + """The capability byte is read from the ``0xffff`` record on an advert.""" + advertisement = mock.Mock() + advertisement.manufacturer_data = { + ANKER_COMPANY_ID: bytes.fromhex("027ce91346c50c00b11a444b4b4504"), + } + assert capability_from_advertisement(advertisement) == CAPABILITY_ENCRYPTED_ECDH + + +def test_capability_from_advertisement_absent() -> None: + """No Anker record means no capability, not an error.""" + advertisement = mock.Mock() + advertisement.manufacturer_data = {0x004C: b"\x02\x15"} + assert capability_from_advertisement(advertisement) is None diff --git a/tests/test_encrypted_negotiation.py b/tests/test_encrypted_negotiation.py new file mode 100644 index 0000000..ff9c3e3 --- /dev/null +++ b/tests/test_encrypted_negotiation.py @@ -0,0 +1,121 @@ +"""Tests for the capability-driven encrypted negotiation and authorization. + +The device response plaintexts and frames below were captured from a live +C2000 Gen 2 (A1783) on comms-module firmware v0.3.3.0. + +.. moduleauthor:: kb1ibt +""" + +from unittest import mock + +import pytest + +from SolixBLE.constructs import Packet +from SolixBLE.device import SolixBLEDevice +from tests.const import MOCK_BLE_DEVICE + +#: Static-key GCM negotiation frames as sent by the device, and their plaintexts. +FRAME_4801 = "ff091e000300014801ab273ed3e27270c3f4d676ac7d69a00572793732a6" +FRAME_4803 = ( + "ff092b000300014803ab273ed04438d4b25db54c6d4a6ec3d481f5ad58ff7cc2be8bc8369" + "fd98c0b914e03" +) +PLAIN_4801 = "00a10101" +PLAIN_4803 = "00a10102a202fd00a30144a40101a50102" +PLAIN_4829 = ( + "00a10103a2054553503332a307302e302e302e33a411415043444b4b4530463339363030" + "303131a5067ce91346c50c" +) +PLAIN_4805 = "00" +PLAIN_4821 = ( + "00a1405dff69533d15aae7194ccfce70978889ed3b090f0ea76c9d1b44bfcb145c80f8eb5" + "59e5734fd9a17ea03a903eb6024786c009faa14d837031c9636c42910e490" +) +PLAIN_4822 = "00" +PLAIN_4827_OK = "00" +PLAIN_4827_BUTTON = "09a1021e00" + +EXPECTED_MTU = 253 +AUTH_MODE_ENCRYPTED = b"\x02" + + +def _device(token: str = "test-token-0001") -> SolixBLEDevice: # noqa: S107 + """Build a base device on the encrypted path with a fixed client token.""" + return SolixBLEDevice(MOCK_BLE_DEVICE, capability=4, client_token=token) + + +async def _feed(device: SolixBLEDevice, cmd: str, plaintext: str) -> None: + """Feed one decrypted negotiation response into the state machine.""" + await device._process_negotiation_encrypted( # noqa: SLF001 + bytes.fromhex(cmd), + bytes.fromhex(plaintext), + ) + + +def test_encrypted_path_selected_from_capability() -> None: + """A capability with the ECDH bit set selects the encrypted path.""" + assert _device()._encrypted_negotiation is True # noqa: SLF001 + cleartext = SolixBLEDevice(MOCK_BLE_DEVICE, capability=0) + assert cleartext._encrypted_negotiation is False # noqa: SLF001 + + +@pytest.mark.parametrize( + ("frame", "plaintext"), + [ + pytest.param(FRAME_4801, PLAIN_4801, id="stage1"), + pytest.param(FRAME_4803, PLAIN_4803, id="stage2"), + ], +) +def test_static_key_gcm_decrypt(frame: str, plaintext: str) -> None: + """The base GCM decrypt recovers the real device frames under the static key.""" + payload = Packet.parse(bytes.fromhex(frame)).payload_bytes + assert _device()._decrypt_payload(payload).hex() == plaintext # noqa: SLF001 + + +@pytest.mark.asyncio +async def test_encrypted_negotiation_reaches_authorized() -> None: + """Driving the stages emits the right commands and authorizes at 4827/00.""" + device = _device() + with mock.patch.object(device, "_send_packet", new=mock.AsyncMock()) as send: + await _feed(device, "4801", PLAIN_4801) + await _feed(device, "4803", PLAIN_4803) + assert device._mtu == EXPECTED_MTU # noqa: SLF001 + assert device._auth_mode == AUTH_MODE_ENCRYPTED # noqa: SLF001 + await _feed(device, "4829", PLAIN_4829) + await _feed(device, "4805", PLAIN_4805) + await _feed(device, "4821", PLAIN_4821) + assert device._shared_secret is not None # noqa: SLF001 + await _feed(device, "4822", PLAIN_4822) + + # The 4005 echo must carry the ECDH cipher bits and the device auth mode. + cmd_4005 = next(c for c in send.await_args_list if c.kwargs["cmd"] == "4005") + params = cmd_4005.kwargs["parameters"] + assert params["a5"]["value"] == bytes.fromhex("44") + assert params["a6"]["value"] == AUTH_MODE_ENCRYPTED + # The 4027 registration carries the client token. + cmd_4027 = next(c for c in send.await_args_list if c.kwargs["cmd"] == "4027") + assert cmd_4027.kwargs["parameters"]["a2"]["value"] == b"test-token-0001" + + sent = [c.kwargs["cmd"] for c in send.await_args_list] + assert sent == ["4003", "4029", "4005", "4021", "4022", "4027"] + + assert device._authorized is False # noqa: SLF001 + await _feed(device, "4827", PLAIN_4827_OK) + assert device._authorized is True # noqa: SLF001 + + +@pytest.mark.asyncio +async def test_encrypted_negotiation_awaits_button() -> None: + """A 4827 status-9 reply does not authorize; it waits for the button.""" + device = _device() + await _feed(device, "4827", PLAIN_4827_BUTTON) + assert device._authorized is False # noqa: SLF001 + + +@pytest.mark.asyncio +async def test_button_press_grant_authorizes() -> None: + """The unsolicited 030101 grant authorizes the link.""" + device = _device() + payload = device._encrypt_payload(b"\x00") # noqa: SLF001 -- GCM, static key + await device._process_arm_grant(bytes.fromhex("4827"), payload) # noqa: SLF001 + assert device._authorized is True # noqa: SLF001