Coverage for src / bluetooth_sig / types / mesh.py: 60%

1"""Bluetooth Mesh protocol types per Bluetooth Mesh Profile Specification. 

2 

3This module contains types for Bluetooth Mesh protocol structures including 

4beacon types, provisioning PDUs, and mesh network data. 

5 

6Reference: Bluetooth Mesh Profile Specification v1.1 

7""" 

8 

9from __future__ import annotations 

10 

11import secrets 

12import struct 

13from enum import IntEnum, IntFlag 

14 

15import msgspec 

16 

17# ============================================================================= 

18# Bluetooth Mesh Protocol Sizes (bytes) 

19# ============================================================================= 

20NETWORK_ID_LENGTH = 8 

21NETWORK_KEY_LENGTH = 16 

22DEVICE_UUID_LENGTH = 16 

23AUTHENTICATION_VALUE_LENGTH = 8 

24OOB_INFO_LENGTH = 2 

25URI_HASH_LENGTH = 4 

26UNPROVISIONED_BEACON_MIN_LENGTH = DEVICE_UUID_LENGTH + OOB_INFO_LENGTH 

27UNPROVISIONED_BEACON_WITH_HASH_LENGTH = UNPROVISIONED_BEACON_MIN_LENGTH + URI_HASH_LENGTH 

28 

29# Network MIC lengths per Bluetooth Mesh Profile 3.4.3 

30ACCESS_MESSAGE_MIC_LENGTH = 4 

31CONTROL_MESSAGE_MIC_LENGTH = 8 

32 

33# Mesh Provisioning algorithm constants 

34ALGORITHM_FIPS_P256 = 0x0001 

35PUBLIC_KEY_TYPE_NONE = 0x00 

36STATIC_OOB_TYPE_NONE = 0x00 

37 

38# ============================================================================= 

39# Enums and Flags 

40# ============================================================================= 

41 

42 

43class SecureNetworkBeaconFlags(IntFlag): 

44 """Secure Network Beacon flags per Bluetooth Mesh Profile 3.10.3. 

45 

46 These flags are combined in the beacon's flags byte. 

47 """ 

48 

49 NONE = 0x00 

50 KEY_REFRESH = 0x01 

51 IV_UPDATE = 0x02 

52 

53 

54class MeshBeaconType(IntEnum): 

55 """Mesh beacon types per Bluetooth Mesh Profile 3.10. 

56 

57 These identify the type of mesh beacon being broadcast. 

58 """ 

59 

60 UNPROVISIONED_DEVICE = 0x00 

61 SECURE_NETWORK = 0x01 

62 

63 

64class ProvisioningPDUType(IntEnum): 

65 """Mesh Provisioning PDU types per Bluetooth Mesh Profile 5.4. 

66 

67 These identify the provisioning protocol message type. 

68 """ 

69 

70 INVITE = 0x00 

71 CAPABILITIES = 0x01 

72 START = 0x02 

73 PUBLIC_KEY = 0x03 

74 INPUT_COMPLETE = 0x04 

75 CONFIRMATION = 0x05 

76 RANDOM = 0x06 

77 DATA = 0x07 

78 COMPLETE = 0x08 

79 FAILED = 0x09 

80 

81 

82class MeshCapabilities(msgspec.Struct, frozen=True, kw_only=True): 

83 """Mesh Provisioning Capabilities per Bluetooth Mesh Profile 5.4.1.2. 

84 

85 This structure describes the device's provisioning capabilities 

86 and is sent in response to a Provisioning Invite PDU. 

87 

88 Attributes: 

89 num_elements: Number of elements supported by the device 

90 algorithms: Supported provisioning algorithms (bitmask) 

91 public_key_type: Supported public key types 

92 static_oob_type: Supported static OOB types 

93 output_oob_size: Maximum size of Output OOB 

94 output_oob_action: Supported Output OOB actions 

95 input_oob_size: Maximum size of Input OOB 

96 input_oob_action: Supported Input OOB actions 

97 

98 """ 

99 

100 num_elements: int = 1 

101 algorithms: int = ALGORITHM_FIPS_P256 

102 public_key_type: int = PUBLIC_KEY_TYPE_NONE 

103 static_oob_type: int = STATIC_OOB_TYPE_NONE 

104 output_oob_size: int = 0 

105 output_oob_action: int = 0 

106 input_oob_size: int = 0 

107 input_oob_action: int = 0 

108 

109 def encode(self) -> bytearray: 

110 """Encode capabilities to provisioning PDU format.""" 

111 return bytearray( 

112 struct.pack( 

113 "<BBHBBBBBB", 

114 ProvisioningPDUType.CAPABILITIES, 

115 self.num_elements, 

116 self.algorithms, 

117 self.public_key_type, 

118 self.static_oob_type, 

119 self.output_oob_size, 

120 self.output_oob_action, 

121 self.input_oob_size, 

122 self.input_oob_action, 

123 ) 

124 ) 

125 

126 

127class SecureNetworkBeacon(msgspec.Struct, frozen=True, kw_only=True): 

128 """Mesh Secure Network Beacon per Bluetooth Mesh Profile 3.10.3. 

129 

130 This beacon is broadcast by provisioned nodes to announce 

131 network presence and IV index. 

132 

133 Attributes: 

134 network_id: 8-byte Network ID derived from network key 

135 iv_index: Current IV Index 

136 key_refresh_flag: Key Refresh Flag 

137 iv_update_flag: IV Update Flag 

138 authentication_value: 8-byte authentication value 

139 

140 """ 

141 

142 network_id: bytes # 8 bytes, derived from network key 

143 iv_index: int = 0 

144 key_refresh_flag: bool = False 

145 iv_update_flag: bool = False 

146 authentication_value: bytes = b"" # 8 bytes when parsed 

147 

148 def encode(self) -> bytearray: 

149 """Encode to beacon format. 

150 

151 Returns: 

152 Encoded beacon bytes 

153 

154 Raises: 

155 ValueError: If network_id is not 8 bytes 

156 

157 """ 

158 if len(self.network_id) != NETWORK_ID_LENGTH: 

159 msg = f"network_id must be {NETWORK_ID_LENGTH} bytes, got {len(self.network_id)}" 

160 raise ValueError(msg) 

161 

162 flags = SecureNetworkBeaconFlags.NONE 

163 if self.key_refresh_flag: 

164 flags |= SecureNetworkBeaconFlags.KEY_REFRESH 

165 if self.iv_update_flag: 

166 flags |= SecureNetworkBeaconFlags.IV_UPDATE 

167 

168 # Use provided auth value or generate random for encoding 

169 auth_value = ( 

170 self.authentication_value 

171 if len(self.authentication_value) == AUTHENTICATION_VALUE_LENGTH 

172 else secrets.token_bytes(AUTHENTICATION_VALUE_LENGTH) 

173 ) 

174 

175 return bytearray( 

176 struct.pack( 

177 "<BB8sI8s", 

178 MeshBeaconType.SECURE_NETWORK, 

179 flags, 

180 self.network_id, 

181 self.iv_index, 

182 auth_value, 

183 ) 

184 ) 

185 

186 @classmethod 

187 def decode(cls, data: bytes | bytearray) -> SecureNetworkBeacon: 

188 """Decode from beacon bytes. 

189 

190 Args: 

191 data: Raw beacon bytes (without beacon type byte) 

192 

193 Returns: 

194 Parsed SecureNetworkBeacon 

195 

196 Raises: 

197 ValueError: If data is too short 

198 

199 """ 

200 min_length = 1 + NETWORK_ID_LENGTH + 4 + AUTHENTICATION_VALUE_LENGTH # flags + netid + iv + auth 

201 if len(data) < min_length: 

202 msg = f"Expected at least {min_length} bytes, got {len(data)}" 

203 raise ValueError(msg) 

204 

205 flags = SecureNetworkBeaconFlags(data[0]) 

206 network_id = bytes(data[1:9]) 

207 iv_index = struct.unpack("<I", data[9:13])[0] 

208 auth_value = bytes(data[13:21]) 

209 

210 return cls( 

211 network_id=network_id, 

212 iv_index=iv_index, 

213 key_refresh_flag=bool(flags & SecureNetworkBeaconFlags.KEY_REFRESH), 

214 iv_update_flag=bool(flags & SecureNetworkBeaconFlags.IV_UPDATE), 

215 authentication_value=auth_value, 

216 ) 

217 

218 

219class UnprovisionedDeviceBeacon(msgspec.Struct, frozen=True, kw_only=True): 

220 """Mesh Unprovisioned Device Beacon per Bluetooth Mesh Profile 3.10.2. 

221 

222 This beacon is broadcast by unprovisioned devices to announce 

223 their presence and provisioning capabilities. 

224 

225 Attributes: 

226 device_uuid: 16-byte device UUID 

227 oob_info: OOB information flags 

228 uri_hash: Optional 4-byte URI hash 

229 

230 """ 

231 

232 device_uuid: bytes # 16 bytes 

233 oob_info: int = 0x0000 

234 uri_hash: bytes | None = None # 4 bytes, optional 

235 

236 def encode(self) -> bytearray: 

237 """Encode to beacon format. 

238 

239 Returns: 

240 Encoded beacon bytes 

241 

242 Raises: 

243 ValueError: If device_uuid is not 16 bytes 

244 

245 """ 

246 if len(self.device_uuid) != DEVICE_UUID_LENGTH: 

247 msg = f"device_uuid must be {DEVICE_UUID_LENGTH} bytes, got {len(self.device_uuid)}" 

248 raise ValueError(msg) 

249 

250 result = bytearray( 

251 struct.pack( 

252 "<B16sH", 

253 MeshBeaconType.UNPROVISIONED_DEVICE, 

254 self.device_uuid, 

255 self.oob_info, 

256 ) 

257 ) 

258 

259 if self.uri_hash is not None: 

260 result.extend(self.uri_hash[:4]) 

261 

262 return result 

263 

264 @classmethod 

265 def decode(cls, data: bytes | bytearray) -> UnprovisionedDeviceBeacon: 

266 """Decode from beacon bytes. 

267 

268 Args: 

269 data: Raw beacon bytes (without beacon type byte) 

270 

271 Returns: 

272 Parsed UnprovisionedDeviceBeacon 

273 

274 Raises: 

275 ValueError: If data is too short 

276 

277 """ 

278 if len(data) < UNPROVISIONED_BEACON_MIN_LENGTH: 

279 msg = f"Expected at least {UNPROVISIONED_BEACON_MIN_LENGTH} bytes, got {len(data)}" 

280 raise ValueError(msg) 

281 

282 device_uuid = bytes(data[0:DEVICE_UUID_LENGTH]) 

283 oob_info = struct.unpack("<H", data[DEVICE_UUID_LENGTH:UNPROVISIONED_BEACON_MIN_LENGTH])[0] 

284 uri_hash = ( 

285 bytes(data[UNPROVISIONED_BEACON_MIN_LENGTH:UNPROVISIONED_BEACON_WITH_HASH_LENGTH]) 

286 if len(data) >= UNPROVISIONED_BEACON_WITH_HASH_LENGTH 

287 else None 

288 ) 

289 

290 return cls( 

291 device_uuid=device_uuid, 

292 oob_info=oob_info, 

293 uri_hash=uri_hash, 

294 ) 

295 

296 

297# MeshMessage constants 

298MESH_MESSAGE_MIN_LENGTH = 9 # ivi/nid (1) + ctl/ttl (1) + seq (3) + src (2) + dst (2) 

299MESH_MESSAGE_IVI_MASK = 0x80 

300MESH_MESSAGE_NID_MASK = 0x7F 

301MESH_MESSAGE_CTL_MASK = 0x80 

302MESH_MESSAGE_TTL_MASK = 0x7F 

303 

304 

305class MeshMessage(msgspec.Struct, frozen=True, kw_only=True): 

306 """Mesh Network PDU message per Bluetooth Mesh Profile 3.4. 

307 

308 Attributes: 

309 ivi: IV Index least significant bit 

310 nid: Network ID (7 bits) 

311 ctl: Control message flag 

312 ttl: Time To Live 

313 seq: Sequence number (24 bits) 

314 src: Source address 

315 dst: Destination address 

316 transport_pdu: Encrypted transport PDU 

317 net_mic: Network MIC (32 or 64 bits) 

318 

319 """ 

320 

321 ivi: int = 0 

322 nid: int = 0 

323 ctl: bool = False 

324 ttl: int = 0 

325 seq: int = 0 

326 src: int = 0 

327 dst: int = 0 

328 transport_pdu: bytes = b"" 

329 net_mic: bytes = b"" 

330 

331 @classmethod 

332 def decode(cls, data: bytes | bytearray) -> MeshMessage: 

333 """Decode from Network PDU bytes. 

334 

335 Args: 

336 data: Raw Network PDU bytes 

337 

338 Returns: 

339 Parsed MeshMessage 

340 

341 Raises: 

342 ValueError: If data is too short 

343 

344 """ 

345 if len(data) < MESH_MESSAGE_MIN_LENGTH: 

346 msg = f"Expected at least {MESH_MESSAGE_MIN_LENGTH} bytes, got {len(data)}" 

347 raise ValueError(msg) 

348 

349 # First byte: IVI (1 bit) + NID (7 bits) 

350 ivi = (data[0] & MESH_MESSAGE_IVI_MASK) >> 7 

351 nid = data[0] & MESH_MESSAGE_NID_MASK 

352 

353 # Second byte: CTL (1 bit) + TTL (7 bits) 

354 ctl = bool(data[1] & MESH_MESSAGE_CTL_MASK) 

355 ttl = data[1] & MESH_MESSAGE_TTL_MASK 

356 

357 # SEQ: 3 bytes big-endian 

358 seq = (data[2] << 16) | (data[3] << 8) | data[4] 

359 

360 # SRC: 2 bytes big-endian 

361 src = (data[5] << 8) | data[6] 

362 

363 # DST: 2 bytes big-endian 

364 dst = (data[7] << 8) | data[8] 

365 

366 # Rest is transport PDU + NetMIC 

367 # NetMIC length depends on message type per Bluetooth Mesh Profile 3.4.3 

368 mic_length = CONTROL_MESSAGE_MIC_LENGTH if ctl else ACCESS_MESSAGE_MIC_LENGTH 

369 transport_pdu = bytes(data[9:-mic_length]) if len(data) > MESH_MESSAGE_MIN_LENGTH + mic_length else b"" 

370 net_mic = bytes(data[-mic_length:]) if len(data) >= MESH_MESSAGE_MIN_LENGTH + mic_length else b"" 

371 

372 return cls( 

373 ivi=ivi, 

374 nid=nid, 

375 ctl=ctl, 

376 ttl=ttl, 

377 seq=seq, 

378 src=src, 

379 dst=dst, 

380 transport_pdu=transport_pdu, 

381 net_mic=net_mic, 

382 ) 

383 

384 

385# ProvisioningBearerData constants 

386PB_ADV_MIN_LENGTH = 6 # link_id (4) + transaction (1) + pdu_type (1) 

387 

388 

389class ProvisioningBearerData(msgspec.Struct, frozen=True, kw_only=True): 

390 """Provisioning Bearer (PB-ADV) data per Bluetooth Mesh Profile 5.3. 

391 

392 Attributes: 

393 link_id: Link identifier (32 bits) 

394 transaction_number: Transaction number 

395 pdu_type: Provisioning PDU type 

396 pdu_data: Raw PDU payload 

397 

398 """ 

399 

400 link_id: int = 0 

401 transaction_number: int = 0 

402 pdu_type: ProvisioningPDUType = ProvisioningPDUType.INVITE 

403 pdu_data: bytes = b"" 

404 

405 @classmethod 

406 def decode(cls, data: bytes | bytearray) -> ProvisioningBearerData: 

407 """Decode from PB-ADV bytes. 

408 

409 Args: 

410 data: Raw PB-ADV bytes 

411 

412 Returns: 

413 Parsed ProvisioningBearerData 

414 

415 Raises: 

416 ValueError: If data is too short 

417 

418 """ 

419 if len(data) < PB_ADV_MIN_LENGTH: 

420 msg = f"Expected at least {PB_ADV_MIN_LENGTH} bytes, got {len(data)}" 

421 raise ValueError(msg) 

422 

423 # Link ID: 4 bytes big-endian 

424 link_id = struct.unpack(">I", data[0:4])[0] 

425 

426 # Transaction number: 1 byte 

427 transaction_number = data[4] 

428 

429 # PDU type: 1 byte 

430 try: 

431 pdu_type = ProvisioningPDUType(data[5]) 

432 except ValueError: 

433 pdu_type = ProvisioningPDUType.INVITE # Default for unknown types 

434 

435 # Rest is PDU data 

436 pdu_data = bytes(data[6:]) if len(data) > PB_ADV_MIN_LENGTH else b"" 

437 

438 return cls( 

439 link_id=link_id, 

440 transaction_number=transaction_number, 

441 pdu_type=pdu_type, 

442 pdu_data=pdu_data, 

443 ) 

444 

445 

446__all__ = [ 

447 # Constants 

448 "ACCESS_MESSAGE_MIC_LENGTH", 

449 "ALGORITHM_FIPS_P256", 

450 "AUTHENTICATION_VALUE_LENGTH", 

451 "CONTROL_MESSAGE_MIC_LENGTH", 

452 "DEVICE_UUID_LENGTH", 

453 "MESH_MESSAGE_CTL_MASK", 

454 "MESH_MESSAGE_IVI_MASK", 

455 "MESH_MESSAGE_MIN_LENGTH", 

456 "MESH_MESSAGE_NID_MASK", 

457 "MESH_MESSAGE_TTL_MASK", 

458 "NETWORK_ID_LENGTH", 

459 "NETWORK_KEY_LENGTH", 

460 "OOB_INFO_LENGTH", 

461 "PB_ADV_MIN_LENGTH", 

462 "PUBLIC_KEY_TYPE_NONE", 

463 "STATIC_OOB_TYPE_NONE", 

464 "UNPROVISIONED_BEACON_MIN_LENGTH", 

465 "UNPROVISIONED_BEACON_WITH_HASH_LENGTH", 

466 "URI_HASH_LENGTH", 

467 # Enums 

468 "MeshBeaconType", 

469 "ProvisioningPDUType", 

470 "SecureNetworkBeaconFlags", 

471 # Structs 

472 "MeshCapabilities", 

473 "MeshMessage", 

474 "ProvisioningBearerData", 

475 "SecureNetworkBeacon", 

476 "UnprovisionedDeviceBeacon", 

477]