diff --git a/kubernetes/apps/otbr/kustomization.yaml b/kubernetes/apps/otbr/kustomization.yaml index 98d2f476..a8f1bd3a 100644 --- a/kubernetes/apps/otbr/kustomization.yaml +++ b/kubernetes/apps/otbr/kustomization.yaml @@ -7,3 +7,8 @@ resources: - network-attachment.yaml - pvc.yaml - service.yaml + - thread-migrate.yaml +configMapGenerator: + - name: thread-migrate + files: + - thread-migrate.py diff --git a/kubernetes/apps/otbr/thread-migrate.py b/kubernetes/apps/otbr/thread-migrate.py new file mode 100644 index 00000000..c199266a --- /dev/null +++ b/kubernetes/apps/otbr/thread-migrate.py @@ -0,0 +1,166 @@ +"""Move Matter-over-Thread nodes onto otbr's Thread network. + +Per node, over the existing CASE session (no re-commissioning): +ArmFailSafe -> RemoveNetwork(old) -> AddOrUpdateThreadNetwork(otbr dataset) +-> ConnectNetwork(new) -> CommissioningComplete over the new network. +If CommissioningComplete never arrives, the fail-safe expires and the node +reverts to its previous network configuration on its own. + +The dataset (incl. network key) is read from otbr's REST API and only ever +passed to matter-server; it is never logged. +""" + +import asyncio +import logging +import os +import sys + +import aiohttp +from chip.clusters import Objects as C +from matter_server.client import MatterClient + +OTBR_URL = os.environ["OTBR_URL"] +MATTER_SERVER_URL = os.environ["MATTER_SERVER_URL"] +NODE_IDS = [int(n) for n in os.environ["NODE_IDS"].split()] +FAILSAFE_SECONDS = int(os.environ.get("FAILSAFE_SECONDS", "180")) + +EXT_PAN_ATTR = "0/53/4" # ThreadNetworkDiagnostics.ExtendedPanId +MAX_NETWORKS_ATTR = "0/49/0" # NetworkCommissioning.MaxNetworks + +log = logging.getLogger("thread-migrate") + + +def extended_pan_id(tlv: bytes) -> bytes: + """Return the Extended PAN ID (TLV type 2) of an operational dataset.""" + i = 0 + while i + 2 <= len(tlv): + tlv_type, length = tlv[i], tlv[i + 1] + if tlv_type == 2: + return tlv[i + 2 : i + 2 + length] + i += 2 + length + raise ValueError("dataset has no Extended PAN ID") + + +async def read_dataset(session: aiohttp.ClientSession) -> bytes: + async with session.get( + f"{OTBR_URL}/node/dataset/active", headers={"Accept": "text/plain"} + ) as resp: + resp.raise_for_status() + return bytes.fromhex((await resp.text()).strip()) + + +async def read_value(client: MatterClient, node_id: int, path: str): + values = await client.read_attribute(node_id, path) + return values[path] + + +async def command(client: MatterClient, node_id: int, cmd, timeout: float = 60): + return await asyncio.wait_for( + client.send_device_command(node_id, 0, cmd), timeout=timeout + ) + + +async def migrate(client: MatterClient, node_id: int, tlv: bytes, new_ext: bytes) -> bool: + current = await read_value(client, node_id, EXT_PAN_ATTR) + log.info("node %s: extended PAN %016x", node_id, current) + if current == int.from_bytes(new_ext, "big"): + log.info("node %s: already on otbr's network", node_id) + return True + old_ext = current.to_bytes(8, "big") + max_networks = await read_value(client, node_id, MAX_NETWORKS_ATTR) + + log.info("node %s: ArmFailSafe %ss", node_id, FAILSAFE_SECONDS) + resp = await command( + client, + node_id, + C.GeneralCommissioning.Commands.ArmFailSafe( + expiryLengthSeconds=FAILSAFE_SECONDS, breadcrumb=1 + ), + ) + log.info("node %s: ArmFailSafe -> %s", node_id, resp) + + if max_networks <= 1: + resp = await command( + client, + node_id, + C.NetworkCommissioning.Commands.RemoveNetwork(networkID=old_ext, breadcrumb=2), + ) + log.info("node %s: RemoveNetwork(old) -> %s", node_id, resp) + + resp = await command( + client, + node_id, + C.NetworkCommissioning.Commands.AddOrUpdateThreadNetwork( + operationalDataset=tlv, breadcrumb=3 + ), + ) + log.info("node %s: AddOrUpdateThreadNetwork -> %s", node_id, resp) + + # The node leaves the old network while answering; a missing response + # is expected. + try: + resp = await command( + client, + node_id, + C.NetworkCommissioning.Commands.ConnectNetwork(networkID=new_ext, breadcrumb=4), + timeout=90, + ) + log.info("node %s: ConnectNetwork -> %s", node_id, resp) + except Exception as err: # noqa: BLE001 + log.info("node %s: ConnectNetwork without response (%r)", node_id, err) + + deadline = asyncio.get_running_loop().time() + FAILSAFE_SECONDS - 20 + while asyncio.get_running_loop().time() < deadline: + try: + resp = await command( + client, + node_id, + C.GeneralCommissioning.Commands.CommissioningComplete(), + timeout=30, + ) + log.info("node %s: CommissioningComplete -> %s", node_id, resp) + break + except Exception as err: # noqa: BLE001 + log.info("node %s: CommissioningComplete not yet (%r)", node_id, err) + await asyncio.sleep(10) + else: + log.error("node %s: no CommissioningComplete; fail-safe reverts it", node_id) + return False + + current = await read_value(client, node_id, EXT_PAN_ATTR) + ok = current == int.from_bytes(new_ext, "big") + log.info("node %s: extended PAN now %016x (%s)", node_id, current, "ok" if ok else "MISMATCH") + return ok + + +async def main() -> int: + async with aiohttp.ClientSession() as session: + tlv = await read_dataset(session) + new_ext = extended_pan_id(tlv) + log.info("otbr network extended PAN %s", new_ext.hex()) + + client = MatterClient(MATTER_SERVER_URL, session) + ready = asyncio.Event() + listener = asyncio.create_task(client.start_listening(ready)) + await ready.wait() + + failed = [] + for node_id in NODE_IDS: + try: + if not await migrate(client, node_id, tlv, new_ext): + failed.append(node_id) + except Exception: # noqa: BLE001 + log.exception("node %s: migration failed", node_id) + failed.append(node_id) + + listener.cancel() + await client.disconnect() + if failed: + log.error("failed nodes: %s", failed) + return 1 + return 0 + + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(message)s") + sys.exit(asyncio.run(main())) diff --git a/kubernetes/apps/otbr/thread-migrate.yaml b/kubernetes/apps/otbr/thread-migrate.yaml new file mode 100644 index 00000000..c638a0c8 --- /dev/null +++ b/kubernetes/apps/otbr/thread-migrate.yaml @@ -0,0 +1,48 @@ +--- +# Moves the Matter-over-Thread nodes in NODE_IDS from the Nest's Thread +# network onto otbr's, without re-commissioning (see thread-migrate.py). +# Idempotent: nodes already on otbr's network are skipped. NODE_IDS are +# matter-server node IDs. +apiVersion: batch/v1 +kind: Job +metadata: + name: thread-migrate + annotations: + # Changing NODE_IDS changes the pod template; recreate the Job for it. + kustomize.toolkit.fluxcd.io/force: enabled +spec: + # A failed node reverts by itself via its fail-safe; no blind retries. + backoffLimit: 0 + template: + metadata: + labels: + app.kubernetes.io/name: thread-migrate + spec: + restartPolicy: Never + containers: + - name: thread-migrate + # Same image as matter-server: ships matter_server.client and the + # chip cluster definitions matching the server. + image: ghcr.io/matter-js/python-matter-server:8.1.2 + command: ["python3", "-I", "/scripts/thread-migrate.py"] + env: + - name: NODE_IDS + value: "5" + - name: OTBR_URL + value: "http://otbr.home-assistant.svc.cluster.local:8081" + - name: MATTER_SERVER_URL + value: "ws://matter-server.home-assistant.svc.cluster.local:5580/ws" + resources: + requests: + cpu: 50m + memory: 128Mi + limits: + memory: 256Mi + volumeMounts: + - name: scripts + mountPath: /scripts + readOnly: true + volumes: + - name: scripts + configMap: + name: thread-migrate