"""Classes to interact with zigpy network backups, including JSON serialization.""" from __future__ import annotations import asyncio import copy import dataclasses from datetime import UTC, datetime import logging from typing import TYPE_CHECKING, Any import zigpy.config as conf import zigpy.state import zigpy.types as t from zigpy.util import ListenableMixin if TYPE_CHECKING: import zigpy.application LOGGER = logging.getLogger(__name__) BACKUP_FORMAT_VERSION = 1 @dataclasses.dataclass class NetworkBackup(t.BaseDataclassMixin): version: int = dataclasses.field(default=BACKUP_FORMAT_VERSION) backup_time: datetime = dataclasses.field(default_factory=lambda: datetime.now(UTC)) network_info: zigpy.state.NetworkInfo = dataclasses.field( default_factory=zigpy.state.NetworkInfo ) node_info: zigpy.state.NodeInfo = dataclasses.field( default_factory=zigpy.state.NodeInfo ) def is_compatible_with(self, backup: NetworkBackup) -> bool: """Two backups are compatible if, ignoring frame counters, the same external device will be able to join either network. """ return ( self.node_info.nwk == backup.node_info.nwk and self.node_info.logical_type == backup.node_info.logical_type and self.node_info.ieee == backup.node_info.ieee and self.network_info.extended_pan_id == backup.network_info.extended_pan_id and self.network_info.pan_id == backup.network_info.pan_id and self.network_info.nwk_update_id == backup.network_info.nwk_update_id and self.network_info.nwk_manager_id == backup.network_info.nwk_manager_id and self.network_info.channel == backup.network_info.channel and self.network_info.security_level == backup.network_info.security_level and self.network_info.tc_link_key.key == backup.network_info.tc_link_key.key and self.network_info.network_key.key == backup.network_info.network_key.key ) def supersedes(self, backup: NetworkBackup) -> bool: """Checks if this network backup is more recent than another backup.""" return ( self.is_compatible_with(backup) and ( self.network_info.network_key.tx_counter > backup.network_info.network_key.tx_counter ) and self.network_info.nwk_update_id >= backup.network_info.nwk_update_id ) def is_complete(self) -> bool: """Checks if this backup captures enough network state to recreate the network.""" return ( self.node_info.ieee != t.EUI64.UNKNOWN # noqa: PLR1714 and self.network_info.extended_pan_id != t.EUI64.UNKNOWN and self.network_info.pan_id not in (0x0000, 0xFFFF) and self.network_info.channel in range(11, 26 + 1) and self.network_info.network_key.key != t.KeyData.UNKNOWN ) def as_dict(self) -> dict[str, Any]: return { "version": self.version, "backup_time": self.backup_time.isoformat(), "network_info": self.network_info.as_dict(), "node_info": self.node_info.as_dict(), } @classmethod def from_dict(cls, obj: dict[str, Any]) -> NetworkBackup: if "metadata" in obj: return cls.from_open_coordinator_json(obj) elif "network_info" in obj: version = obj.get("version", 0) if version > BACKUP_FORMAT_VERSION: LOGGER.warning( "Network backup has version %d but current backup format" " version is %d. Downgrading is not recommended.", version, BACKUP_FORMAT_VERSION, ) # Version 1 introduced the `model`, `manufacturer`, and `version` fields if version == 0: obj = copy.deepcopy(obj) obj["node_info"]["model"] = None obj["node_info"]["manufacturer"] = None obj["node_info"]["version"] = None version = 1 # Incrementing the version number raised an assertion error in older # versions of zigpy. The backup format version will be incremented from 1 # to 2 in 2025.12.0, to allow one release cycle to pass before making a # backwards-incompatible breaking change. if "route_table" not in obj["network_info"]: obj = copy.deepcopy(obj) obj["network_info"]["route_table"] = {} if "tx_power" not in obj["network_info"]: obj = copy.deepcopy(obj) obj["network_info"]["tx_power"] = None return cls( version=BACKUP_FORMAT_VERSION, backup_time=datetime.fromisoformat(obj["backup_time"]), network_info=zigpy.state.NetworkInfo.from_dict(obj["network_info"]), node_info=zigpy.state.NodeInfo.from_dict(obj["node_info"]), ) else: raise ValueError(f"Invalid network backup object: {obj!r}") def as_open_coordinator_json(self) -> dict[str, Any]: return _network_backup_to_open_coordinator_backup(self) @classmethod def from_open_coordinator_json(cls, obj: dict[str, Any]) -> NetworkBackup: return _open_coordinator_backup_to_network_backup(obj) class BackupManager(ListenableMixin): def __init__(self, app: zigpy.application.ControllerApplication): super().__init__() self.app: zigpy.application.ControllerApplication = app self.backups: list[NetworkBackup] = [] self._backup_task: asyncio.Task | None = None def most_recent_backup(self) -> NetworkBackup | None: """Most recent network backup""" return self.backups[-1] if self.backups else None def from_network_state(self) -> NetworkBackup: """Create a backup object from the current network's state.""" return NetworkBackup( network_info=self.app.state.network_info, node_info=self.app.state.node_info, ) async def create_backup(self, *, load_devices: bool = True) -> NetworkBackup: await self.app.load_network_info(load_devices=load_devices) backup = self.from_network_state() self.add_backup(backup) return backup async def restore_backup( self, backup: NetworkBackup, *, counter_increment: int = 10000, allow_incomplete: bool = False, create_new: bool = True, ) -> None: LOGGER.debug("Restoring backup %s", backup) if not backup.is_complete() and not allow_incomplete: raise ValueError("Backup is incomplete, it is not possible to restore") key = backup.network_info.network_key new_backup = NetworkBackup( network_info=backup.network_info.replace( network_key=key.replace(tx_counter=key.tx_counter + counter_increment) ), node_info=backup.node_info, ) await self.app.write_network_info( network_info=new_backup.network_info, node_info=new_backup.node_info, ) if create_new: await self.create_backup() def add_backup( self, backup: NetworkBackup, *, suppress_event: bool = False ) -> None: """Adds a new backup to the database, superseding older ones if necessary.""" LOGGER.debug("Adding a new backup %s", backup) if not backup.is_complete(): LOGGER.debug("Backup is incomplete, ignoring") return # Only delete the most recent backup if the frame counter doesn't roll back. # 1. Old Conbee backups replace one another: the FC never increments # 2. EZSP -> old Conbee: create bad backup for Conbee # 3. Old Conbee -> EZSP: replace Conbee backup, its FC is always zero for old_backup in self.backups[:]: if backup.is_compatible_with(old_backup) and ( backup.network_info.network_key.tx_counter >= old_backup.network_info.network_key.tx_counter ): if not suppress_event: self.listener_event("network_backup_removed", old_backup) self.backups.remove(old_backup) if not suppress_event: self.listener_event("network_backup_created", backup) self.backups.append(backup) def start_periodic_backups(self, period: float) -> None: self.stop_periodic_backups() self._backup_task = asyncio.create_task(self._backup_loop(period)) def stop_periodic_backups(self): if self._backup_task is not None: self._backup_task.cancel() async def _backup_loop(self, period: float): while True: try: await self.create_backup() except Exception: # noqa: BLE001 LOGGER.warning("Failed to create a network backup", exc_info=True) LOGGER.debug("Waiting for %ss before backing up again", period) await asyncio.sleep(period) def __getitem__(self, key) -> NetworkBackup: return self.backups[key] def _network_backup_to_open_coordinator_backup(backup: NetworkBackup) -> dict[str, Any]: """Converts a `NetworkBackup` to an Open Coordinator Backup-compatible dictionary.""" node_info = backup.node_info network_info = backup.network_info devices = {} for ieee, nwk in network_info.nwk_addresses.items(): devices[ieee] = { "ieee_address": ieee.serialize()[::-1].hex(), "nwk_address": nwk.serialize()[::-1].hex(), "is_child": False, } for ieee in network_info.children: if ieee not in devices: devices[ieee] = { "ieee_address": ieee.serialize()[::-1].hex(), "nwk_address": None, "is_child": True, } else: devices[ieee]["is_child"] = True for key in network_info.key_table: if key.partner_ieee not in devices: devices[key.partner_ieee] = { "ieee_address": key.partner_ieee.serialize()[::-1].hex(), "nwk_address": None, "is_child": False, } devices[key.partner_ieee]["link_key"] = { "key": key.key.serialize().hex(), "tx_counter": key.tx_counter, "rx_counter": key.rx_counter, } return { "metadata": { "version": 1, "format": "zigpy/open-coordinator-backup", "source": network_info.source, "internal": { "creation_time": backup.backup_time.isoformat(), "node": { "ieee": node_info.ieee.serialize()[::-1].hex(), "nwk": node_info.nwk.serialize()[::-1].hex(), "type": zigpy.state.LOGICAL_TYPE_TO_JSON[node_info.logical_type], "model": node_info.model, "manufacturer": node_info.manufacturer, "version": node_info.version, }, "network": { "tc_link_key": { "key": network_info.tc_link_key.key.serialize().hex(), "frame_counter": network_info.tc_link_key.tx_counter, }, "tc_address": network_info.tc_link_key.partner_ieee.serialize()[ ::-1 ].hex(), "nwk_manager": network_info.nwk_manager_id.serialize()[::-1].hex(), }, "link_key_seqs": { key.partner_ieee.serialize()[::-1].hex(): key.seq for key in network_info.key_table }, "route_table": { str(t.NWK(dst))[2:]: str(t.NWK(next_hop))[2:] for dst, next_hop in network_info.route_table.items() }, "tx_power": network_info.tx_power, **network_info.metadata, }, }, "stack_specific": network_info.stack_specific, "coordinator_ieee": node_info.ieee.serialize()[::-1].hex(), "pan_id": network_info.pan_id.serialize()[::-1].hex(), "extended_pan_id": network_info.extended_pan_id.serialize()[::-1].hex(), "nwk_update_id": network_info.nwk_update_id, "security_level": network_info.security_level, "channel": network_info.channel, "channel_mask": list(network_info.channel_mask), "network_key": { "key": network_info.network_key.key.serialize().hex(), "sequence_number": network_info.network_key.seq or 0, "frame_counter": network_info.network_key.tx_counter or 0, }, "devices": sorted(devices.values(), key=lambda d: d["ieee_address"]), } def _open_coordinator_backup_to_network_backup(obj: dict[str, Any]) -> NetworkBackup: """Creates a `NetworkBackup` from an Open Coordinator Backup dictionary.""" internal = obj["metadata"].get("internal", {}) node_info = zigpy.state.NodeInfo() node_meta = internal.get("node", {}) if "nwk" in node_meta: node_info.nwk, _ = t.NWK.deserialize(bytes.fromhex(node_meta["nwk"])[::-1]) else: node_info.nwk = t.NWK(0x0000) node_info.logical_type = zigpy.state.JSON_TO_LOGICAL_TYPE[ node_meta.get("type", "coordinator") ] # Should be identical to `metadata.internal.node.ieee` node_info.ieee, _ = t.EUI64.deserialize( bytes.fromhex(obj["coordinator_ieee"])[::-1] ) node_info.model = node_meta.get("model") node_info.manufacturer = node_meta.get("manufacturer") node_info.version = node_meta.get("version") network_info = zigpy.state.NetworkInfo() network_info.source = obj["metadata"]["source"] network_info.metadata = { k: v for k, v in internal.items() if k not in ( "node", "network", "link_key_seqs", "creation_time", "route_table", "tx_power", ) } network_info.pan_id, _ = t.NWK.deserialize(bytes.fromhex(obj["pan_id"])[::-1]) network_info.extended_pan_id, _ = t.EUI64.deserialize( bytes.fromhex(obj["extended_pan_id"])[::-1] ) network_info.nwk_update_id = obj["nwk_update_id"] network_meta = internal.get("network", {}) if "nwk_manager" in network_meta: network_info.nwk_manager_id, _ = t.NWK.deserialize( bytes.fromhex(network_meta["nwk_manager"]) ) else: network_info.nwk_manager_id = t.NWK(0x0000) network_info.channel = obj["channel"] network_info.channel_mask = t.Channels.from_channel_list(obj["channel_mask"]) network_info.security_level = obj["security_level"] if obj.get("stack_specific"): network_info.stack_specific = obj.get("stack_specific") network_info.tc_link_key = zigpy.state.Key() if "tc_link_key" in network_meta: network_info.tc_link_key.key, _ = t.KeyData.deserialize( bytes.fromhex(network_meta["tc_link_key"]["key"]) ) network_info.tc_link_key.tx_counter = network_meta["tc_link_key"].get( "frame_counter", 0 ) network_info.tc_link_key.partner_ieee, _ = t.EUI64.deserialize( bytes.fromhex(network_meta["tc_address"])[::-1] ) else: network_info.tc_link_key.key = conf.CONF_NWK_TC_LINK_KEY_DEFAULT network_info.tc_link_key.partner_ieee = node_info.ieee network_info.network_key = zigpy.state.Key() network_info.network_key.key, _ = t.KeyData.deserialize( bytes.fromhex(obj["network_key"]["key"]) ) network_info.network_key.tx_counter = obj["network_key"]["frame_counter"] network_info.network_key.seq = obj["network_key"]["sequence_number"] network_info.children = [] network_info.nwk_addresses = {} for device in obj["devices"]: if device["nwk_address"] is not None: # zfill(4) is used because Z2M backups include 0x0ABC as `abc`, not `0abc` nwk, _ = t.NWK.deserialize( bytes.fromhex(device["nwk_address"].zfill(4))[::-1] ) else: nwk = None ieee, _ = t.EUI64.deserialize(bytes.fromhex(device["ieee_address"])[::-1]) # The `is_child` key is currently optional if device.get("is_child", True): network_info.children.append(ieee) if nwk is not None: network_info.nwk_addresses[ieee] = nwk if "link_key" in device: key = zigpy.state.Key() key.key, _ = t.KeyData.deserialize(bytes.fromhex(device["link_key"]["key"])) key.tx_counter = device["link_key"]["tx_counter"] key.rx_counter = device["link_key"]["rx_counter"] key.partner_ieee = ieee try: key.seq = obj["metadata"]["internal"]["link_key_seqs"][ device["ieee_address"] ] except KeyError: key.seq = 0 network_info.key_table.append(key) # XXX: Devices that are not children, have no NWK address, and have no link key # are effectively ignored, since there is no place to write them for dst, next_hop in obj["metadata"]["internal"].get("route_table", {}).items(): network_info.route_table[t.NWK.convert(dst)] = t.NWK.convert(next_hop) network_info.tx_power = obj["metadata"]["internal"].get("tx_power") if "date" in internal: # Z2M format creation_time = internal["date"].replace("Z", "+00:00") else: # Zigpy format creation_time = internal.get("creation_time", "1970-01-01T00:00:00+00:00") return NetworkBackup( version=BACKUP_FORMAT_VERSION, backup_time=datetime.fromisoformat(creation_time), network_info=network_info, node_info=node_info, )