From 9ade175d5c9677be2e6286692cf7c270a9c7ed19 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Mon, 10 Nov 2025 15:48:37 +1000 Subject: [PATCH 1/8] util: soc: move default provisioning struct Move the default provisioning struct to a common location. Signed-off-by: Jordan Yates --- src/infuse_iot/tools/provision.py | 11 ++--------- src/infuse_iot/util/soc/soc.py | 7 +++++++ 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/src/infuse_iot/tools/provision.py b/src/infuse_iot/tools/provision.py index c3d2506..79ab9e8 100644 --- a/src/infuse_iot/tools/provision.py +++ b/src/infuse_iot/tools/provision.py @@ -23,13 +23,6 @@ from infuse_iot.util.soc import nrf, soc, stm -class ProvisioningStruct(ctypes.LittleEndianStructure): - _fields_ = [ - ("device_id", ctypes.c_uint64), - ] - _pack_ = 1 - - class SubCommand(InfuseCommand): NAME = "provision" HELP = "Provision device on Infuse Cloud" @@ -171,8 +164,8 @@ def run(self): assert isinstance(response.parsed.device_id, str) # Compare current flash contents to desired flash contents cloud_id = int(response.parsed.device_id, 16) - current_bytes = interface.read_provisioned_data(ctypes.sizeof(ProvisioningStruct)) - desired = ProvisioningStruct(cloud_id) + current_bytes = interface.read_provisioned_data(ctypes.sizeof(interface.DefaultProvisioningStruct)) + desired = interface.DefaultProvisioningStruct(cloud_id) desired_bytes = bytes(desired) if current_bytes == desired_bytes: diff --git a/src/infuse_iot/util/soc/soc.py b/src/infuse_iot/util/soc/soc.py index 2ddd7f8..923e1f1 100644 --- a/src/infuse_iot/util/soc/soc.py +++ b/src/infuse_iot/util/soc/soc.py @@ -1,11 +1,18 @@ #!/usr/bin/env +import ctypes from abc import ABCMeta, abstractmethod class ProvisioningInterface(metaclass=ABCMeta): "Generic SoC provisioning interface" + class DefaultProvisioningStruct(ctypes.LittleEndianStructure): + _fields_ = [ + ("device_id", ctypes.c_uint64), + ] + _pack_ = 1 + @property @abstractmethod def soc_name(self) -> str: From 08e773b2fe41586467acab197d61fde8cdc3984a Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Sat, 8 Nov 2025 20:53:34 +1000 Subject: [PATCH 2/8] util: soc: nrf: convert `sys.exit` to exceptions Convert the `sys.exit` calls to exceptions that can be caught by higher levels. Signed-off-by: Jordan Yates --- src/infuse_iot/util/soc/nrf.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/infuse_iot/util/soc/nrf.py b/src/infuse_iot/util/soc/nrf.py index 4ffea4e..a7d8ece 100644 --- a/src/infuse_iot/util/soc/nrf.py +++ b/src/infuse_iot/util/soc/nrf.py @@ -98,12 +98,12 @@ def __init__(self, snr: int | None): self.snr = snr devices = self._exec(["device-info"]) if len(devices) == 0: - sys.exit() + raise RuntimeError("No devices found") devices_info = devices[0]["devices"] if len(devices_info) > 1: serials = ",".join([d["serialNumber"] for d in devices_info]) - sys.exit(f"Multiple devices found without a SNR provided (Found: {serials})") + raise RuntimeError(f"Multiple devices found without a SNR provided (Found: {serials})") self.snr = devices_info[0]["serialNumber"] self.device_info = devices_info[0]["deviceInfo"] self.core_info = self._exec(["core-info"]) From 0fbc5120861ef8e49577f2f085d0ec825facb180 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Mon, 10 Nov 2025 15:49:46 +1000 Subject: [PATCH 3/8] util: soc: nrf: handle JLink serial number properly Fix handling of the serial number argument when multiple adapters are present. Signed-off-by: Jordan Yates --- src/infuse_iot/util/soc/nrf.py | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/src/infuse_iot/util/soc/nrf.py b/src/infuse_iot/util/soc/nrf.py index a7d8ece..5b841ad 100644 --- a/src/infuse_iot/util/soc/nrf.py +++ b/src/infuse_iot/util/soc/nrf.py @@ -101,15 +101,19 @@ def __init__(self, snr: int | None): raise RuntimeError("No devices found") devices_info = devices[0]["devices"] - if len(devices_info) > 1: - serials = ",".join([d["serialNumber"] for d in devices_info]) - raise RuntimeError(f"Multiple devices found without a SNR provided (Found: {serials})") - self.snr = devices_info[0]["serialNumber"] - self.device_info = devices_info[0]["deviceInfo"] + if snr is None: + if len(devices_info) > 1: + serials = ",".join([d["serialNumber"] for d in devices_info]) + raise RuntimeError(f"Multiple devices found without a SNR provided (Found: {serials})") + self.snr = int(devices_info[0]["serialNumber"]) + infos = [info for info in devices_info if int(info["serialNumber"]) == self.snr] + if len(infos) == 0: + raise RuntimeError(f"Devices with SNR {self.snr} not found") + self.device_info = infos[0]["deviceInfo"] self.core_info = self._exec(["core-info"]) self.family = DEVICE_FAMILY_MAPPING[self.device_info["jlink"]["deviceFamily"]] - self.uicr_base = self.core_info[0]["devices"][0]["uicrAddress"] - self._soc_name = self.family.soc(self.device_info) + self.uicr_base: int = self.core_info[0]["devices"][0]["uicrAddress"] + self._soc_name: str = self.family.soc(self.device_info) def _exec(self, args: list[str]): jout_all = [] From 7b231825e01f99144718662c34f17ce356eb18a6 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Sat, 8 Nov 2025 20:46:40 +1000 Subject: [PATCH 4/8] serial_comms: optional JLink serial number Add the ability to specify the JLink that will be used. Signed-off-by: Jordan Yates --- src/infuse_iot/serial_comms.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/infuse_iot/serial_comms.py b/src/infuse_iot/serial_comms.py index 568a818..63ceb62 100644 --- a/src/infuse_iot/serial_comms.py +++ b/src/infuse_iot/serial_comms.py @@ -2,6 +2,7 @@ import time from abc import ABCMeta, abstractmethod +from io import BufferedWriter import pylink import serial @@ -108,14 +109,15 @@ def __str__(self) -> str: class RttPort(SerialLike): """Segger RTT handling""" - def __init__(self, rtt_device): + def __init__(self, rtt_device: str, serial_number: str | None = None): self._jlink = pylink.JLink() self._name = rtt_device - self._modem_trace = None + self._serial_number = serial_number + self._modem_trace: BufferedWriter | None = None self._modem_trace_buf = 0 def open(self): - self._jlink.open() + self._jlink.open(serial_no=self._serial_number) self._jlink.set_tif(pylink.enums.JLinkInterfaces.SWD) self._jlink.connect(self._name, 4000) self._jlink.rtt_start() From 99b8ce29a520fa00c6b0e3a1942f63f35e5198a2 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Sat, 8 Nov 2025 20:48:40 +1000 Subject: [PATCH 5/8] serial_comms: optional timeout on `open` Add an optional timeout to the open call for backends that want it. Signed-off-by: Jordan Yates --- src/infuse_iot/serial_comms.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/infuse_iot/serial_comms.py b/src/infuse_iot/serial_comms.py index 63ceb62..fe9b1fe 100644 --- a/src/infuse_iot/serial_comms.py +++ b/src/infuse_iot/serial_comms.py @@ -44,7 +44,7 @@ def reconstructor(cls): class SerialLike(metaclass=ABCMeta): @abstractmethod - def open(self) -> None: + def open(self, timeout: float | None = None) -> None: """Open serial port""" @abstractmethod @@ -80,7 +80,7 @@ def __init__(self, serial_port, baudrate=115200): # receivers (STM32) time to wake up on RX before real data arrives. self._prefix = b"\x00\x00" if baudrate > 115200 else b"" - def open(self): + def open(self, timeout: float | None = None): self._ser.open() def read_bytes(self, num) -> bytes: @@ -116,14 +116,18 @@ def __init__(self, rtt_device: str, serial_number: str | None = None): self._modem_trace: BufferedWriter | None = None self._modem_trace_buf = 0 - def open(self): + def open(self, timeout: float | None = None): self._jlink.open(serial_no=self._serial_number) self._jlink.set_tif(pylink.enums.JLinkInterfaces.SWD) self._jlink.connect(self._name, 4000) self._jlink.rtt_start() + end_time = time.time() + timeout if timeout else None + # Loop until JLink initialised properly while True: + if end_time and time.time() > end_time: + raise TimeoutError("RTT port never initialised") try: num_up = self._jlink.rtt_get_num_up_buffers() _num_down = self._jlink.rtt_get_num_down_buffers() @@ -181,7 +185,7 @@ def __init__(self, target: str): self._target = self._session.target self._rtt = GenericRTTControlBlock(self._target) - def open(self): + def open(self, timeout: float | None = None): self._session.open() self._target.resume() self._rtt.start() From 22cc1de3a938eb5ef9ba3fbe31d4813f9f8de836 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Sat, 8 Nov 2025 20:52:22 +1000 Subject: [PATCH 6/8] database: UDP key functions Add functions to derive UDP keys. Signed-off-by: Jordan Yates --- src/infuse_iot/database.py | 27 +++++++++++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/src/infuse_iot/database.py b/src/infuse_iot/database.py index e32e0d6..53f46ca 100644 --- a/src/infuse_iot/database.py +++ b/src/infuse_iot/database.py @@ -132,6 +132,9 @@ def _bt_adv_key(self, base: bytes, time_idx: int) -> bytes: 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: """Does the database have the public key for this device?""" if address not in self.devices: @@ -219,3 +222,27 @@ def bt_gatt_device_key(self, address: int, gps_time: int) -> bytes: time_idx = gps_time // (60 * 60 * 24) return self._bt_gatt_key(base, time_idx) + + def udp_network_key(self, address: 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._network_key(network_id, b"udp", gps_time) + + def udp_device_key(self, address: int, gps_time: 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) From 8a563294e6fe1373232a70c3cd5e83c0d9abace7 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Sat, 8 Nov 2025 20:49:40 +1000 Subject: [PATCH 7/8] epacket: packet: define V0 unversioned frame Define the V0 unversioned frame type. Signed-off-by: Jordan Yates --- src/infuse_iot/epacket/packet.py | 43 +++++++++++++++++++++++--------- 1 file changed, 31 insertions(+), 12 deletions(-) diff --git a/src/infuse_iot/epacket/packet.py b/src/infuse_iot/epacket/packet.py index 660ccb7..7aa119d 100644 --- a/src/infuse_iot/epacket/packet.py +++ b/src/infuse_iot/epacket/packet.py @@ -337,18 +337,8 @@ class CtypeForwardHeaderBtGatt(ctypes.LittleEndianStructure): _pack_ = 1 -class CtypeV0VersionedFrame(ctypes.LittleEndianStructure): - _fields_ = [ - ("version", ctypes.c_uint8), - ("_type", ctypes.c_uint8), - ("flags", ctypes.c_uint16), - ("_key_metadata", ctypes.c_uint8 * 3), - ("_device_id_upper", ctypes.c_uint32), - ("_device_id_lower", ctypes.c_uint32), - ("gps_time", ctypes.c_uint32), - ("sequence", ctypes.c_uint16), - ("entropy", ctypes.c_uint16), - ] +class CtypeV0Frame(ctypes.LittleEndianStructure): + _fields_ = [] _pack_ = 1 @property @@ -381,6 +371,35 @@ def parse(cls, frame: bytes) -> tuple[Self, int]: ) +class CtypeV0VersionedFrame(CtypeV0Frame): + _fields_ = [ + ("version", ctypes.c_uint8), + ("_type", ctypes.c_uint8), + ("flags", ctypes.c_uint16), + ("_key_metadata", ctypes.c_uint8 * 3), + ("_device_id_upper", ctypes.c_uint32), + ("_device_id_lower", ctypes.c_uint32), + ("gps_time", ctypes.c_uint32), + ("sequence", ctypes.c_uint16), + ("entropy", ctypes.c_uint16), + ] + _pack_ = 1 + + +class CtypeV0UnversionedFrame(CtypeV0Frame): + _fields_ = [ + ("_type", ctypes.c_uint8), + ("flags", ctypes.c_uint16), + ("_key_metadata", ctypes.c_uint8 * 3), + ("_device_id_upper", ctypes.c_uint32), + ("_device_id_lower", ctypes.c_uint32), + ("gps_time", ctypes.c_uint32), + ("sequence", ctypes.c_uint16), + ("entropy", ctypes.c_uint16), + ] + _pack_ = 1 + + class CtypeSerialFrame(CtypeV0VersionedFrame): """Serial packet header""" From 81feaacd7c74fcb7f46723fe83fb284f67aa0335 Mon Sep 17 00:00:00 2001 From: Jordan Yates Date: Sat, 8 Nov 2025 20:52:42 +1000 Subject: [PATCH 8/8] epacket: packet: UDP decryption function Add the ability to decrypt UDP packets for debugging purposes. Signed-off-by: Jordan Yates --- src/infuse_iot/epacket/packet.py | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/infuse_iot/epacket/packet.py b/src/infuse_iot/epacket/packet.py index 7aa119d..b8cdb07 100644 --- a/src/infuse_iot/epacket/packet.py +++ b/src/infuse_iot/epacket/packet.py @@ -503,6 +503,21 @@ def decrypt(cls, database: DeviceDatabase, bt_addr: Address.BluetoothLeAddr | No return header, decrypted +class CtypeUdpFrame(CtypeV0UnversionedFrame): + @classmethod + 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) + else: + database.observe_device(header.device_id, network_id=header.key_metadata) + key = database.udp_network_key(header.device_id, header.gps_time) + + decrypted = chachapoly_decrypt(key, frame[:10], frame[10:22], frame[22:]) + return header, decrypted + + class CtypePacketReceived: class CommonHeader(ctypes.Structure): _fields_ = [