diff --git a/src/infuse_iot/database.py b/src/infuse_iot/database.py index 78298d75..2b4a2a63 100644 --- a/src/infuse_iot/database.py +++ b/src/infuse_iot/database.py @@ -2,6 +2,10 @@ import base64 import binascii +import pathlib + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import x25519 from infuse_iot.api_client import Client from infuse_iot.api_client.api.key import get_shared_secret @@ -42,16 +46,18 @@ class DeviceState: def __init__( self, - address: int, + infuse_id: int, network_id: int | None = None, - device_id: int | None = None, + device_key_id: int | None = None, ): - self.address = address + self.infuse_id = infuse_id self.network_id = network_id - self.device_id = device_id + self.device_key_id = device_key_id self.bt_addr: InterfaceAddress.BluetoothLeAddr | None = None - self.public_key: bytes | None = None + self.device_public_key: bytes | None = None self.shared_key: bytes | None = None + self.secondary_device_key_id: int | None = None + self.local_shared_key: bytes | None = None self._tx_gatt_seq = 0 def gatt_sequence_num(self): @@ -59,50 +65,84 @@ def gatt_sequence_num(self): self._tx_gatt_seq += 1 return self._tx_gatt_seq - def __init__(self) -> None: + def __init__(self, local_root: pathlib.Path | None) -> None: self.gateway: int | None = None self.devices: dict[int, DeviceDatabase.DeviceState] = {} self.bt_addr: dict[InterfaceAddress.BluetoothLeAddr, int] = {} + self._local_root: x25519.X25519PrivateKey | None = None + self._local_root_public: bytes | None = None + if local_root: + with local_root.open() as f: + private_key = serialization.load_pem_private_key(f.read().encode("utf-8"), password=None) + assert isinstance(private_key, x25519.X25519PrivateKey) + self._local_root = private_key + self._local_root_public = self._local_root.public_key().public_bytes_raw() + + @property + def has_local_root(self) -> bool: + return self._local_root is not None + + def is_local_root(self, public_key: bytes) -> bool: + return public_key == self._local_root_public def observe_device( self, - address: int, + infuse_id: int, network_id: int | None = None, - device_id: int | None = None, + device_key_id: int | None = None, bt_addr: InterfaceAddress.BluetoothLeAddr | None = None, ) -> None: """Update device state based on observed packet""" if self.gateway is None: - self.gateway = address - if address not in self.devices: - self.devices[address] = self.DeviceState(address) + self.gateway = infuse_id + if infuse_id not in self.devices: + self.devices[infuse_id] = self.DeviceState(infuse_id) + dev = self.devices[infuse_id] if network_id is not None: - self.devices[address].network_id = network_id - if device_id is not None: - if self.devices[address].device_id is not None and self.devices[address].device_id != device_id: - raise DeviceKeyChangedError(f"Device key for {address:016x} has changed") - self.devices[address].device_id = device_id + dev.network_id = network_id + if device_key_id is not None: + if device_key_id == dev.secondary_device_key_id: + pass + elif dev.device_key_id is not None and dev.device_key_id != device_key_id: + raise DeviceKeyChangedError(f"Device key for {infuse_id:016x} has changed") + else: + dev.device_key_id = device_key_id if bt_addr is not None: - self.bt_addr[bt_addr] = address - self.devices[address].bt_addr = bt_addr - - def observe_security_state(self, address: int, cloud_key: bytes, device_key: bytes, network_id: int) -> None: + self.bt_addr[bt_addr] = infuse_id + dev.bt_addr = bt_addr + + def observe_secondary_remote_public_key(self, infuse_id: int, secondary_pub_key: bytes): + if not self.is_local_root(secondary_pub_key): + return + if infuse_id not in self.devices: + return + dev = self.devices[infuse_id] + assert self._local_root is not None + assert self._local_root_public is not None + assert dev.device_public_key is not None + device_public_key = x25519.X25519PublicKey.from_public_bytes(dev.device_public_key) + dev.secondary_device_key_id = binascii.crc32(self._local_root_public + dev.device_public_key) & 0xFFFFFF + dev.local_shared_key = self._local_root.exchange(device_public_key) + + def observe_security_state( + self, infuse_id: int, cloud_pub_key: bytes, device_pub_key: bytes, network_id: int + ) -> None: """Update device state based on security_state response""" - if address not in self.devices: - self.devices[address] = self.DeviceState(address) - device_id = binascii.crc32(cloud_key + device_key) & 0x00FFFFFF - self.devices[address].device_id = device_id - self.devices[address].network_id = network_id - self.devices[address].public_key = device_key + if infuse_id not in self.devices: + self.devices[infuse_id] = self.DeviceState(infuse_id) + device_key_id = binascii.crc32(cloud_pub_key + device_pub_key) & 0x00FFFFFF + self.devices[infuse_id].device_key_id = device_key_id + self.devices[infuse_id].network_id = network_id + self.devices[infuse_id].device_public_key = device_pub_key client = Client(base_url="https://api.infuse-iot.com").with_headers({"x-api-key": f"Bearer {get_api_key()}"}) with client as client: - body = Key(base64.b64encode(device_key).decode("utf-8")) + body = Key(base64.b64encode(device_pub_key).decode("utf-8")) response = get_shared_secret.sync(client=client, body=body) if response is not None: key = base64.b64decode(response.key) - self.devices[address].shared_key = key + self.devices[infuse_id].shared_key = key def _network_key(self, network_id: int, interface: bytes, gps_time: int) -> bytes: if network_id not in self._network_keys: @@ -120,126 +160,83 @@ def _network_key(self, network_id: int, interface: bytes, gps_time: int) -> byte return self._derived_keys[key_id] - def _serial_key(self, base: bytes, time_idx: int) -> bytes: - return hkdf_derive(base, time_idx.to_bytes(4, "little"), b"serial") - - def _bt_adv_key(self, base: bytes, time_idx: int) -> bytes: - return hkdf_derive(base, time_idx.to_bytes(4, "little"), b"bt_adv") - - def _bt_gatt_key(self, base: bytes, time_idx: int) -> bytes: - return hkdf_derive(base, time_idx.to_bytes(4, "little"), b"bt_gatt") - - def _udp_key(self, base: bytes, time_idx: int) -> bytes: - return hkdf_derive(base, time_idx.to_bytes(4, "little"), b"udp") - - def has_public_key(self, address: int) -> bool: + def has_public_key(self, infuse_id: int) -> bool: """Does the database have the public key for this device?""" - if address not in self.devices: + if infuse_id not in self.devices: return False - return self.devices[address].public_key is not None + return self.devices[infuse_id].device_public_key is not None - def has_network_id(self, address: int) -> bool: + def has_network_id(self, infuse_id: int) -> bool: """Does the database know the network ID for this device?""" - if address not in self.devices: + if infuse_id not in self.devices: return False - return self.devices[address].network_id is not None + return self.devices[infuse_id].network_id is not None def infuse_id_from_bluetooth(self, bt_addr: InterfaceAddress.BluetoothLeAddr) -> int | None: - """Get Bluetooth address associated with device""" + """Get Bluetooth infuse_id associated with device""" return self.bt_addr.get(bt_addr, None) - def serial_network_key(self, address: int, gps_time: int) -> bytes: - """Network key for serial interface""" - if address not in self.devices: + def _get_network_key(self, infuse_id: int, name: bytes, gps_time: int) -> bytes: + if infuse_id not in self.devices: raise DeviceUnknownNetworkKey - network_id = self.devices[address].network_id + network_id = self.devices[infuse_id].network_id if network_id is None: raise DeviceUnknownNetworkKey + return self._network_key(network_id, name, gps_time) - return self._network_key(network_id, b"serial", gps_time) - - def serial_device_key(self, address: int, gps_time: int) -> bytes: - """Device key for serial interface""" - if address not in self.devices: + def _get_device_key( + self, infuse_id: int, name: bytes, gps_time: int, key_id: int | None = None + ) -> tuple[int, bytes]: + if infuse_id not in self.devices: raise DeviceUnknownDeviceKey - d = self.devices[address] - if d.device_id is None: + d = self.devices[infuse_id] + if key_id is None: + if d.secondary_device_key_id: + key_id = d.secondary_device_key_id + base = d.local_shared_key + else: + key_id = d.device_key_id + base = d.shared_key + elif key_id == d.device_key_id: + base = d.shared_key + elif key_id == d.secondary_device_key_id: + base = d.local_shared_key + else: raise DeviceUnknownDeviceKey - base = self.devices[address].shared_key if base is None: raise DeviceUnknownDeviceKey + assert key_id is not None time_idx = gps_time // (60 * 60 * 24) + return key_id, hkdf_derive(base, time_idx.to_bytes(4, "little"), name) + + def serial_network_key(self, infuse_id: int, gps_time: int) -> bytes: + """Network key for serial interface""" + return self._get_network_key(infuse_id, b"serial", gps_time) - return self._serial_key(base, time_idx) + def serial_device_key(self, infuse_id: int, gps_time: int, key_id: int | None = None) -> tuple[int, bytes]: + """Device key for serial interface""" + return self._get_device_key(infuse_id, b"serial", gps_time, key_id) - def bt_adv_network_key(self, address: int, gps_time: int) -> bytes: + def bt_adv_network_key(self, infuse_id: int, gps_time: int) -> bytes: """Network key for Bluetooth advertising interface""" - if address not in self.devices: - raise DeviceUnknownNetworkKey - network_id = self.devices[address].network_id - if network_id is None: - raise DeviceUnknownNetworkKey - - return self._network_key(network_id, b"bt_adv", gps_time) + return self._get_network_key(infuse_id, b"bt_adv", gps_time) - def bt_adv_device_key(self, address: int, gps_time: int) -> bytes: + def bt_adv_device_key(self, infuse_id: int, gps_time: int, key_id: int | None = None) -> tuple[int, bytes]: """Device key for Bluetooth advertising interface""" - if address not in self.devices: - raise DeviceUnknownDeviceKey - d = self.devices[address] - if d.device_id is None: - raise DeviceUnknownDeviceKey - base = self.devices[address].shared_key - if base is None: - raise DeviceUnknownDeviceKey - time_idx = gps_time // (60 * 60 * 24) + return self._get_device_key(infuse_id, b"bt_adv", gps_time, key_id) - return self._bt_adv_key(base, time_idx) - - def bt_gatt_network_key(self, address: int, gps_time: int) -> bytes: + def bt_gatt_network_key(self, infuse_id: int, gps_time: int) -> bytes: """Network key for Bluetooth advertising interface""" - if address not in self.devices: - raise DeviceUnknownNetworkKey - network_id = self.devices[address].network_id - if network_id is None: - raise DeviceUnknownNetworkKey - - return self._network_key(network_id, b"bt_gatt", gps_time) + return self._get_network_key(infuse_id, b"bt_gatt", gps_time) - def bt_gatt_device_key(self, address: int, gps_time: int) -> bytes: + def bt_gatt_device_key(self, infuse_id: int, gps_time: int, key_id: int | None = None) -> tuple[int, bytes]: """Device key for Bluetooth advertising interface""" - if address not in self.devices: - raise DeviceUnknownDeviceKey - d = self.devices[address] - if d.device_id is None: - raise DeviceUnknownDeviceKey - base = self.devices[address].shared_key - if base is None: - raise DeviceUnknownDeviceKey - time_idx = gps_time // (60 * 60 * 24) - - return self._bt_gatt_key(base, time_idx) + return self._get_device_key(infuse_id, b"bt_gatt", gps_time, key_id) - def udp_network_key(self, address: int, gps_time: int) -> bytes: + def udp_network_key(self, infuse_id: int, gps_time: int) -> bytes: """Network key for UDP interface""" - if address not in self.devices: - raise DeviceUnknownNetworkKey - network_id = self.devices[address].network_id - if network_id is None: - raise DeviceUnknownNetworkKey + return self._get_network_key(infuse_id, b"udp", gps_time) - return self._network_key(network_id, b"udp", gps_time) - - def udp_device_key(self, address: int, gps_time: int) -> bytes: + def udp_device_key(self, infuse_id: int, gps_time: int, key_id: int | None = None) -> tuple[int, bytes]: """Device key for UDP interface""" - if address not in self.devices: - raise DeviceUnknownDeviceKey - d = self.devices[address] - if d.device_id is None: - raise DeviceUnknownDeviceKey - base = self.devices[address].shared_key - if base is None: - raise DeviceUnknownDeviceKey - time_idx = gps_time // (60 * 60 * 24) - - return self._udp_key(base, time_idx) + return self._get_device_key(infuse_id, b"udp", gps_time, key_id) diff --git a/src/infuse_iot/epacket/packet.py b/src/infuse_iot/epacket/packet.py index b8cdb078..7d297cb8 100644 --- a/src/infuse_iot/epacket/packet.py +++ b/src/infuse_iot/epacket/packet.py @@ -258,8 +258,7 @@ def to_serial(self, database: DeviceDatabase) -> bytes: key = database.serial_network_key(serial.infuse_id, gps_time) else: flags = Flags.ENCR_DEVICE - key_metadata = database.devices[serial.infuse_id].device_id - key = database.serial_device_key(serial.infuse_id, gps_time) + key_metadata, key = database.serial_device_key(serial.infuse_id, gps_time) # Validation assert key_metadata is not None @@ -420,8 +419,8 @@ def hop_received(self) -> HopReceived: def decrypt(cls, database: DeviceDatabase, frame: bytes): header = cls.from_buffer_copy(frame) if header.flags & Flags.ENCR_DEVICE: - database.observe_device(header.device_id, device_id=header.key_metadata) - key = database.serial_device_key(header.device_id, header.gps_time) + database.observe_device(header.device_id, device_key_id=header.key_metadata) + _, key = database.serial_device_key(header.device_id, header.gps_time, header.key_metadata) else: database.observe_device(header.device_id, network_id=header.key_metadata) key = database.serial_network_key(header.device_id, header.gps_time) @@ -460,11 +459,11 @@ def encrypt( ) -> bytes: dev_state = database.devices[infuse_id] gps_time = InfuseTime.gps_seconds_from_unix(int(time.time())) + key_meta: int | None flags = 0 if auth == Auth.DEVICE: - key_meta = dev_state.device_id - key = database.bt_gatt_device_key(infuse_id, gps_time) + key_meta, key = database.bt_gatt_device_key(infuse_id, gps_time) flags |= Flags.ENCR_DEVICE else: key_meta = dev_state.network_id @@ -493,8 +492,8 @@ def encrypt( def decrypt(cls, database: DeviceDatabase, bt_addr: Address.BluetoothLeAddr | None, frame: bytes): header = cls.from_buffer_copy(frame) if header.flags & Flags.ENCR_DEVICE: - database.observe_device(header.device_id, device_id=header.key_metadata, bt_addr=bt_addr) - key = database.bt_gatt_device_key(header.device_id, header.gps_time) + database.observe_device(header.device_id, device_key_id=header.key_metadata, bt_addr=bt_addr) + _, key = database.bt_gatt_device_key(header.device_id, header.gps_time, header.key_metadata) else: database.observe_device(header.device_id, network_id=header.key_metadata, bt_addr=bt_addr) key = database.bt_gatt_network_key(header.device_id, header.gps_time) @@ -508,8 +507,8 @@ class CtypeUdpFrame(CtypeV0UnversionedFrame): def decrypt(cls, database: DeviceDatabase, frame: bytes): header = cls.from_buffer_copy(frame) if header.flags & Flags.ENCR_DEVICE: - database.observe_device(header.device_id, device_id=header.key_metadata) - key = database.udp_device_key(header.device_id, header.gps_time) + database.observe_device(header.device_id, device_key_id=header.key_metadata) + _, key = database.udp_device_key(header.device_id, header.gps_time, header.key_metadata) else: database.observe_device(header.device_id, network_id=header.key_metadata) key = database.udp_network_key(header.device_id, header.gps_time) diff --git a/src/infuse_iot/generated/rpc_definitions.py b/src/infuse_iot/generated/rpc_definitions.py index b9f9e89f..18e362cb 100644 --- a/src/infuse_iot/generated/rpc_definitions.py +++ b/src/infuse_iot/generated/rpc_definitions.py @@ -222,6 +222,16 @@ class rpc_struct_data_logger_chunk(VLACompatLittleEndianStruct): _pack_ = 1 +class rpc_struct_public_key_info_256bit(VLACompatLittleEndianStruct): + """256 bit public key and identifier""" + + _fields_ = [ + ("id", ctypes.c_uint8), + ("key", 32 * ctypes.c_uint8), + ] + _pack_ = 1 + + class rpc_enum_bt_le_addr_type(enum.IntEnum): """Bluetooth LE address type""" @@ -273,6 +283,8 @@ class rpc_enum_key_id(enum.IntEnum): NETWORK_KEY = 0 SECONDARY_NETWORK_KEY = 1 SECONDARY_REMOTE_PUBLIC_KEY = 2 + DEVICE_PUBLIC_KEY = 3 + CLOUD_PUBLIC_KEY = 4 class rpc_enum_key_action(enum.IntEnum): @@ -1212,6 +1224,30 @@ class response(VLACompatLittleEndianStruct): _pack_ = 1 +class security_public_keys(RPCDefinitionBase): + """Query device public keys""" + + NAME = "security_public_keys" + HELP = "Query device public keys" + DESCRIPTION = "Query device public keys" + COMMAND_ID = 30002 + + class request(VLACompatLittleEndianStruct): + _fields_ = [ + ("skip", ctypes.c_uint8), + ] + _pack_ = 1 + + class response(VLACompatLittleEndianStruct): + _fields_ = [ + ("keys_total", ctypes.c_uint8), + ("keys_included", ctypes.c_uint8), + ] + vla_field = ("public_keys", 0 * rpc_struct_public_key_info_256bit) + vla_counted_by = "keys_included" + _pack_ = 1 + + class data_sender(RPCDefinitionBase): """Send multiple INFUSE_RPC_DATA packets""" @@ -1310,6 +1346,7 @@ class response(VLACompatLittleEndianStruct): ubx_assist_now_ztp_creds.COMMAND_ID: ubx_assist_now_ztp_creds, security_state.COMMAND_ID: security_state, security_key_update.COMMAND_ID: security_key_update, + security_public_keys.COMMAND_ID: security_public_keys, data_sender.COMMAND_ID: data_sender, data_receiver.COMMAND_ID: data_receiver, echo.COMMAND_ID: echo, @@ -1333,6 +1370,7 @@ class response(VLACompatLittleEndianStruct): "rpc_struct_sockaddr", "rpc_struct_heap_info", "rpc_struct_data_logger_chunk", + "rpc_struct_public_key_info_256bit", "rpc_enum_bt_le_addr_type", "rpc_enum_file_action", "rpc_enum_infuse_bt_characteristic", @@ -1380,6 +1418,7 @@ class response(VLACompatLittleEndianStruct): "ubx_assist_now_ztp_creds", "security_state", "security_key_update", + "security_public_keys", "data_sender", "data_receiver", "echo", diff --git a/src/infuse_iot/rpc_wrappers/security_key_update.py b/src/infuse_iot/rpc_wrappers/security_key_update.py new file mode 100644 index 00000000..e6eb612b --- /dev/null +++ b/src/infuse_iot/rpc_wrappers/security_key_update.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import x25519 + +import infuse_iot.definitions.rpc as defs +from infuse_iot.commands import InfuseRpcCommand +from infuse_iot.epacket.packet import Auth +from infuse_iot.util.argparse import ValidFile +from infuse_iot.util.ctypes import bytes_to_uint8 +from infuse_iot.zephyr.errno import errno + + +class security_key_update(InfuseRpcCommand, defs.security_key_update): + @classmethod + def add_parser(cls, parser): + parser.add_argument("--network-auth", action="store_true", help="Use network auth instead of device") + parser.add_argument("--delete", action="store_true", help="Delete instead of writing key") + parser.add_argument("--delay", type=int, default=2, help="Reboot delay (seconds)") + key_group = parser.add_mutually_exclusive_group(required=True) + key_group.add_argument("--secondary-root", type=ValidFile) + + def __init__(self, args): + self._auth = Auth.NETWORK if args.network_auth else Auth.DEVICE + self._key_action = defs.rpc_enum_key_action.KEY_DELETE if args.delete else defs.rpc_enum_key_action.KEY_WRITE + self._delay = args.delay + if args.secondary_root: + self._key_id = defs.rpc_enum_key_id.SECONDARY_REMOTE_PUBLIC_KEY + self._global_key_id = 0 + with args.secondary_root.open("r") as f: + private_key = serialization.load_pem_private_key(f.read().encode("utf-8"), password=None) + assert isinstance(private_key, x25519.X25519PrivateKey) + public_key = private_key.public_key() + self._key_bytes = public_key.public_bytes_raw() + else: + raise NotImplementedError("Unimplemented key type") + + def auth_level(self): + return self._auth + + def request_struct(self): + return self.request( + self._key_id, + self._key_action, + self._global_key_id, + bytes_to_uint8(self._key_bytes), + self._delay, + ) + + def handle_response(self, return_code, response): + if return_code != 0: + print(f"Failed to update key ({errno.strerror(-return_code)}, {-return_code})") + return + print(f"Updated key {self._key_id.name} on device") diff --git a/src/infuse_iot/rpc_wrappers/security_public_keys.py b/src/infuse_iot/rpc_wrappers/security_public_keys.py new file mode 100644 index 00000000..9d5903f8 --- /dev/null +++ b/src/infuse_iot/rpc_wrappers/security_public_keys.py @@ -0,0 +1,29 @@ +#!/usr/bin/env python3 + +import infuse_iot.definitions.rpc as defs +from infuse_iot.commands import InfuseRpcCommand +from infuse_iot.zephyr.errno import errno + + +class security_public_keys(InfuseRpcCommand, defs.security_public_keys): + @classmethod + def add_parser(cls, parser): + parser.add_argument("--skip", type=int, default=0, help="Skip first N keys") + + def __init__(self, args): + self._skip = args.skip + + def request_struct(self): + return self.request( + self._skip, + ) + + def handle_response(self, return_code, response): + if return_code != 0: + print(f"Failed to update key ({errno.strerror(-return_code)}, {-return_code})") + return + + for key in response.public_keys: + key_id = defs.rpc_enum_key_id(key.id) + key_str = bytes(key.key).hex() + print(f"{key_id.name:>30}: {key_str}") diff --git a/src/infuse_iot/tools/gateway.py b/src/infuse_iot/tools/gateway.py index 3a50dfe7..0d9a5a90 100644 --- a/src/infuse_iot/tools/gateway.py +++ b/src/infuse_iot/tools/gateway.py @@ -140,15 +140,34 @@ def security_state_done(pkt: PacketReceived, _: int, response: bytes): if cb_event is not None: cb_event.set() - # Generate security_state RPC - cmd_pkt = self.rpc.generate(30000, random.randbytes(16), Auth.NETWORK, security_state_done) - encrypted = cmd_pkt.to_serial(self.ddb) - # Write to serial port - Console.log_tx(cmd_pkt.ptype, len(encrypted)) - self.port.write(encrypted) - if cb_event is not None: - # Wait for the response - cb_event.wait(1.0) + def public_keys_done(pkt: PacketReceived, rc: int, response: bytes): + if rc != 0: + return + decoded = defs.security_public_keys.response.vla_from_buffer_copy(response) + for key in decoded.public_keys: + if key.id == defs.rpc_enum_key_id.SECONDARY_REMOTE_PUBLIC_KEY: + infuse_id = pkt.route[0].infuse_id + self.ddb.observe_secondary_remote_public_key(infuse_id, bytes(key.key)) + + def run_cmd_pkt(cmd: PacketOutputRouted): + encrypted = cmd_pkt.to_serial(self.ddb) + # Write to serial port + Console.log_tx(cmd_pkt.ptype, len(encrypted)) + self.port.write(encrypted) + if cb_event is not None: + # Wait for the response + cb_event.wait(1.0) + + # Run security_state RPC + cmd_pkt = self.rpc.generate( + defs.security_state.COMMAND_ID, random.randbytes(16), Auth.NETWORK, security_state_done + ) + run_cmd_pkt(cmd_pkt) + + if self.ddb.has_local_root: + # Query other public keys from the device + cmd_pkt = self.rpc.generate(defs.security_public_keys.COMMAND_ID, b"\x00", Auth.NETWORK, public_keys_done) + run_cmd_pkt(cmd_pkt) class SerialRxThread(SignaledThread): @@ -401,8 +420,8 @@ def _handle_observed_devices(self): info = {} if state.network_id is not None: info["network_id"] = state.network_id - if state.device_id is not None: - info["device_id"] = state.device_id + if state.device_key_id is not None: + info["device_key_id"] = state.device_key_id if self._common.ddb.gateway == device: info["gateway"] = True observed_devices[device] = info @@ -456,6 +475,7 @@ def add_parser(cls, parser): help="Save serial output to file", ) parser.add_argument("--baud", type=int, default=115200, help="Baudrate for serial port") + parser.add_argument("--root", type=ValidFile, help="Root identity certificate to use instead of cloud") def __init__(self, args: argparse.Namespace): self.port: SerialLike @@ -467,7 +487,7 @@ def __init__(self, args: argparse.Namespace): self.port = RttPort(args.rtt) elif args.pyocd is not None: self.port = PyOcdPort(args.pyocd) - self.ddb = DeviceDatabase() + self.ddb = DeviceDatabase(args.root) if args.display_only: self.server = None else: diff --git a/src/infuse_iot/tools/native_bt.py b/src/infuse_iot/tools/native_bt.py index dad447e8..a12ed19a 100644 --- a/src/infuse_iot/tools/native_bt.py +++ b/src/infuse_iot/tools/native_bt.py @@ -42,7 +42,7 @@ LocalServer, default_multicast_address, ) -from infuse_iot.util.argparse import BtLeAddress +from infuse_iot.util.argparse import BtLeAddress, ValidFile from infuse_iot.util.console import Console @@ -204,11 +204,11 @@ class SubCommand(InfuseCommand): @classmethod def add_parser(cls, parser): - pass + parser.add_argument("--root", type=ValidFile, help="Root identity certificate to use instead of cloud") def __init__(self, args: argparse.Namespace): self.infuse_manu = 0x0DE4 - self.database = DeviceDatabase() + self.database = DeviceDatabase(args.root) self.server = LocalServer(default_multicast_address()) self.bleak_mapping: dict[int, BLEDevice] = {} self.unknown_networks: set[int] = set() diff --git a/src/infuse_iot/util/ctypes.py b/src/infuse_iot/util/ctypes.py index 489770c0..21cd1eae 100644 --- a/src/infuse_iot/util/ctypes.py +++ b/src/infuse_iot/util/ctypes.py @@ -52,8 +52,8 @@ class property is not `None`, it will consume the remainder of if issubclass(vla_field_type, ctypes.Array): array_base: ctypes._PyCSimpleType = vla_field_type._type_ # type: ignore - if hasattr(array_base, "vla_counted_by"): - # This is an array of VLA arrays where the sub-arrys define their own length + if hasattr(array_base, "vla_counted_by") and array_base.vla_counted_by: + # This is an array of VLA arrays where the sub-arrays define their own length vla_val = [] # Consume all remaining buffer bytes while len(remainder) > 0: