diff --git a/src/onecontrol_client.py b/src/onecontrol_client.py index f081ef5..62284e7 100644 --- a/src/onecontrol_client.py +++ b/src/onecontrol_client.py @@ -9,6 +9,74 @@ from cobs_protocol import CobsEncoder, CobsDecoder from enum import IntEnum +# --------------------------------------------------------------------------- +# Auth / Key-Seed Exchange (service 0010) +# Reverse-engineered from Plugin.BLE.dll :: BleDeviceUnlockManager +# --------------------------------------------------------------------------- + +AUTH_SERVICE = "00000010-0200-a58e-e411-afe28044e62c" +SEED_CHAR = "00000012-0200-a58e-e411-afe28044e62c" # read seed from here +KEY_CHAR = "00000013-0200-a58e-e411-afe28044e62c" # write response here +RV_LINK_CYPHER = 612643285 # MyRvLinkBleGatewayScanResult.RvLinkKeySeedCypher + + +def _tea_encrypt(cypher: int, seed: int) -> int: + """Modified TEA used by Lippert OneControl for BLE auth.""" + DELTA = 2654435769 # 0x9E3779B9 + num = DELTA + for _ in range(32): + seed = (seed + (((cypher << 4) + 1131376761) ^ (cypher + num) ^ ((cypher >> 5) + 1919510376))) & 0xFFFFFFFF + cypher = (cypher + (((seed << 4) + 1948272964) ^ (seed + num) ^ ((seed >> 5) + 1400073827))) & 0xFFFFFFFF + num = (num + DELTA) & 0xFFFFFFFF + return seed + + +async def perform_auth(client: BleakClient) -> bool: + """ + Perform key-seed exchange on service 0010 before any commands on 0030. + Returns True on success, False on failure. + """ + for attempt in range(3): + try: + seed_bytes = await client.read_gatt_char(SEED_CHAR) + except Exception as e: + print(f"Auth: failed to read seed (attempt {attempt+1}): {e}") + await asyncio.sleep(5) + continue + + # If device returns "unlocked" it's already authenticated + if len(seed_bytes) == len(b"unlocked") and seed_bytes.lower() == b"unlocked": + print("Auth: already unlocked") + return True + + if len(seed_bytes) < 4: + print(f"Auth: seed too short ({len(seed_bytes)} bytes), retrying...") + await asyncio.sleep(5) + continue + + seed = struct.unpack_from(" cmd_type async def connect(self): - """Connect to OneControl device""" + """Connect to OneControl device and perform key-seed auth handshake.""" self.client = BleakClient(self.address) - await self.client.connect() + await self.client.connect(timeout=20.0) + if not await perform_auth(self.client): + await self.client.disconnect() + raise RuntimeError("Auth handshake failed") await self.client.start_notify(self.READ_CHAR, self._notification_handler) async def disconnect(self): @@ -168,7 +239,7 @@ async def scan_for_onecontrol(): found = [] for address, (device, adv) in devices.items(): - if device.name and "OneControl" in device.name: + if device.name and "LCIRemote" in device.name: print(f"Found by name: {device.name} ({address})") found.append(device) elif any("00000030-0200" in uuid.lower() for uuid in adv.service_uuids): @@ -192,12 +263,9 @@ async def main(): print(f" {d.name} ({d.address})") return else: - # No device found via scan — set address manually if you already know it - ADDRESS = "XX:XX:XX:XX:XX:XX" - if ADDRESS == "XX:XX:XX:XX:XX:XX": - print("No device found. Set ADDRESS manually if you know the BT address.") - return + ADDRESS = "30:C6:F7:E0:80:8E" address = ADDRESS + print(f"No device found via scan, trying known address {address}") client = OneControlClient(address)