Implement key-seed auth handshake (service 0010)

Root cause of immediate disconnect after write: panel requires auth on
service 0010 before accepting commands on 0030. Protocol found by
decompiling Plugin.BLE.dll (BleDeviceUnlockManager.PerformKeySeedExchange).

Auth flow:
1. Read 4-byte seed from char 0012 (00000012-0200-a58e-e411-afe28044e62c)
2. Apply modified TEA with RV-specific cypher (612643285 / 0x248431D5)
3. Write 4-byte little-endian result to char 0013
4. Retries up to 3x if seed < 4 bytes (panel not yet ready)

connect() now calls perform_auth() before enabling 0034 notifications.
Also cleans up experimental auth attempts from campsite debugging session.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
wes
2026-04-03 19:56:36 -04:00
co-authored by Claude Sonnet 4.6
parent 72f415e051
commit b6d498c829
+76 -8
View File
@@ -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("<I", seed_bytes, 0)[0]
if seed == 0:
print("Auth: seed is zero, retrying...")
await asyncio.sleep(5)
continue
key = _tea_encrypt(RV_LINK_CYPHER, seed)
key_bytes = struct.pack("<I", key)
print(f"Auth: seed=0x{seed:08x} → key=0x{key:08x}")
try:
await client.write_gatt_char(KEY_CHAR, key_bytes)
await asyncio.sleep(0.5)
print("Auth: key written, handshake complete")
return True
except Exception as e:
print(f"Auth: failed to write key (attempt {attempt+1}): {e}")
return False
print("Auth: all attempts exhausted")
return False
class CommandType(IntEnum):
GET_DEVICES = 1
ACTION_SWITCH = 64
@@ -44,9 +112,12 @@ class OneControlClient:
self._pending_commands = {} # seq -> 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)