Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/layers/zigbee.py: 64%

Shortcuts on this page

r m x   toggle line displays

j k   next/prev highlighted chunk

0   (zero) top of page

1   (one) first highlighted chunk

268 statements  

1# SPDX-License-Identifier: GPL-2.0-only 

2# This file is part of Scapy 

3# See https://scapy.net/ for more information 

4# Copyright (C) Ryan Speers <ryan@rmspeers.com> 2011-2012 

5# Copyright (C) Roger Meyer <roger.meyer@csus.edu>: 2012-03-10 Added frames 

6# Copyright (C) Gabriel Potter <gabriel[]potter[]fr>: 2018 

7# Copyright (C) 2020-2021 Dimitrios-Georgios Akestoridis <akestoridis@cmu.edu> 

8 

9""" 

10ZigBee bindings for IEEE 802.15.4. 

11""" 

12 

13import struct 

14 

15from scapy.packet import bind_layers, bind_bottom_up, Packet 

16from scapy.fields import BitField, ByteField, XLEIntField, ConditionalField, \ 

17 ByteEnumField, EnumField, BitEnumField, FieldListField, FlagsField, \ 

18 IntField, PacketListField, ShortField, StrField, StrFixedLenField, \ 

19 StrLenField, XLEShortField, XStrField 

20 

21from scapy.layers.dot15d4 import dot15d4AddressField, Dot15d4Beacon, Dot15d4, \ 

22 Dot15d4FCS 

23from scapy.layers.inet import UDP 

24from scapy.layers.ntp import TimeStampField 

25 

26 

27# APS Profile Identifiers 

28_aps_profile_identifiers = { 

29 0x0000: "Zigbee_Device_Profile", 

30 0x0101: "IPM_Industrial_Plant_Monitoring", 

31 0x0104: "HA_Home_Automation", 

32 0x0105: "CBA_Commercial_Building_Automation", 

33 0x0107: "TA_Telecom_Applications", 

34 0x0108: "HC_Health_Care", 

35 0x0109: "SE_Smart_Energy_Profile", 

36} 

37 

38# ZigBee Cluster Library Identifiers, Table 2.2 ZCL 

39_zcl_cluster_identifier = { 

40 # Functional Domain: General 

41 0x0000: "basic", 

42 0x0001: "power_configuration", 

43 0x0002: "device_temperature_configuration", 

44 0x0003: "identify", 

45 0x0004: "groups", 

46 0x0005: "scenes", 

47 0x0006: "on_off", 

48 0x0007: "on_off_switch_configuration", 

49 0x0008: "level_control", 

50 0x0009: "alarms", 

51 0x000a: "time", 

52 0x000b: "rssi_location", 

53 0x000c: "analog_input", 

54 0x000d: "analog_output", 

55 0x000e: "analog_value", 

56 0x000f: "binary_input", 

57 0x0010: "binary_output", 

58 0x0011: "binary_value", 

59 0x0012: "multistate_input", 

60 0x0013: "multistate_output", 

61 0x0014: "multistate_value", 

62 0x0015: "commissioning", 

63 # 0x0016 - 0x00ff reserved 

64 # Functional Domain: Closures 

65 0x0100: "shade_configuration", 

66 # 0x0101 - 0x01ff reserved 

67 # Functional Domain: HVAC 

68 0x0200: "pump_configuration_and_control", 

69 0x0201: "thermostat", 

70 0x0202: "fan_control", 

71 0x0203: "dehumidification_control", 

72 0x0204: "thermostat_user_interface_configuration", 

73 # 0x0205 - 0x02ff reserved 

74 # Functional Domain: Lighting 

75 0x0300: "color_control", 

76 0x0301: "ballast_configuration", 

77 # Functional Domain: Measurement and sensing 

78 0x0400: "illuminance_measurement", 

79 0x0401: "illuminance_level_sensing", 

80 0x0402: "temperature_measurement", 

81 0x0403: "pressure_measurement", 

82 0x0404: "flow_measurement", 

83 0x0405: "relative_humidity_measurement", 

84 0x0406: "occupancy_sensing", 

85 # Functional Domain: Security and safethy 

86 0x0500: "ias_zone", 

87 0x0501: "ias_ace", 

88 0x0502: "ias_wd", 

89 # Functional Domain: Protocol Interfaces 

90 0x0600: "generic_tunnel", 

91 0x0601: "bacnet_protocol_tunnel", 

92 0x0602: "analog_input_regular", 

93 0x0603: "analog_input_extended", 

94 0x0604: "analog_output_regular", 

95 0x0605: "analog_output_extended", 

96 0x0606: "analog_value_regular", 

97 0x0607: "analog_value_extended", 

98 0x0608: "binary_input_regular", 

99 0x0609: "binary_input_extended", 

100 0x060a: "binary_output_regular", 

101 0x060b: "binary_output_extended", 

102 0x060c: "binary_value_regular", 

103 0x060d: "binary_value_extended", 

104 0x060e: "multistate_input_regular", 

105 0x060f: "multistate_input_extended", 

106 0x0610: "multistate_output_regular", 

107 0x0611: "multistate_output_extended", 

108 0x0612: "multistate_value_regular", 

109 0x0613: "multistate_value", 

110 # Smart Energy Profile Clusters 

111 0x0700: "price", 

112 0x0701: "demand_response_and_load_control", 

113 0x0702: "metering", 

114 0x0703: "messaging", 

115 0x0704: "smart_energy_tunneling", 

116 0x0705: "prepayment", 

117 # Functional Domain: General 

118 # Key Establishment 

119 0x0800: "key_establishment", 

120} 

121 

122# ZigBee Cluster Library, Table 2.8 ZCL Command Frames 

123_zcl_command_frames = { 

124 0x00: "read_attributes", 

125 0x01: "read_attributes_response", 

126 0x02: "write_attributes", 

127 0x03: "write_attributes_undivided", 

128 0x04: "write_attributes_response", 

129 0x05: "write_attributes_no_response", 

130 0x06: "configure_reporting", 

131 0x07: "configure_reporting_response", 

132 0x08: "read_reporting_configuration", 

133 0x09: "read_reporting_configuration_response", 

134 0x0a: "report_attributes", 

135 0x0b: "default_response", 

136 0x0c: "discover_attributes", 

137 0x0d: "discover_attributes_response", 

138 0x0e: "read_attributes_structured", 

139 0x0f: "write_attributes_structured", 

140 0x10: "write_attributes_structured_response", 

141 0x11: "discover_commands_received", 

142 0x12: "discover_commands_received_response", 

143 0x13: "discover_commands_generated", 

144 0x14: "discover_commands_generated_response", 

145 0x15: "discover_attributes_extended", 

146 0x16: "discover_attributes_extended_response", 

147 # 0x17 - 0xff Reserved 

148} 

149 

150# ZigBee Cluster Library, Table 2.16 Enumerated Status Values 

151_zcl_enumerated_status_values = { 

152 0x00: "SUCCESS", 

153 0x01: "FAILURE", 

154 # 0x02 - 0x7d Reserved 

155 0x7e: "NOT_AUTHORIZED", 

156 0x7f: "RESERVED_FIELD_NOT_ZERO", 

157 0x80: "MALFORMED_COMMAND", 

158 0x81: "UNSUP_CLUSTER_COMMAND", 

159 0x82: "UNSUP_GENERAL_COMMAND", 

160 0x83: "UNSUP_MANUF_CLUSTER_COMMAND", 

161 0x84: "UNSUP_MANUF_GENERAL_COMMAND", 

162 0x85: "INVALID_FIELD", 

163 0x86: "UNSUPPORTED_ATTRIBUTE", 

164 0x87: "INVALID_VALUE", 

165 0x88: "READ_ONLY", 

166 0x89: "INSUFFICIENT_SPACE", 

167 0x8a: "DUPLICATE_EXISTS", 

168 0x8b: "NOT_FOUND", 

169 0x8c: "UNREPORTABLE_ATTRIBUTE", 

170 0x8d: "INVALID_DATA_TYPE", 

171 0x8e: "INVALID_SELECTOR", 

172 0x8f: "WRITE_ONLY", 

173 0x90: "INCONSISTENT_STARTUP_STATE", 

174 0x91: "DEFINED_OUT_OF_BAND", 

175 0x92: "INCONSISTENT", 

176 0x93: "ACTION_DENIED", 

177 0x94: "TIMEOUT", 

178 0x95: "ABORT", 

179 0x96: "INVALID_IMAGE", 

180 0x97: "WAIT_FOR_DATA", 

181 0x98: "NO_IMAGE_AVAILABLE", 

182 0x99: "REQUIRE_MORE_IMAGE", 

183 0x9a: "NOTIFICATION_PENDING", 

184 # 0x9b - 0xbf Reserved 

185 0xc0: "HARDWARE_FAILURE", 

186 0xc1: "SOFTWARE_FAILURE", 

187 0xc2: "CALIBRATION_ERROR", 

188 0xc3: "UNSUPPORTED_CLUSTER", 

189 # 0xc4 - 0xff Reserved 

190} 

191 

192# ZigBee Cluster Library, Table 2.15 Data Types 

193_zcl_attribute_data_types = { 

194 0x00: "no_data", 

195 # General data 

196 0x08: "8-bit_data", 

197 0x09: "16-bit_data", 

198 0x0a: "24-bit_data", 

199 0x0b: "32-bit_data", 

200 0x0c: "40-bit_data", 

201 0x0d: "48-bit_data", 

202 0x0e: "56-bit_data", 

203 0x0f: "64-bit_data", 

204 # Logical 

205 0x10: "boolean", 

206 # Bitmap 

207 0x18: "8-bit_bitmap", 

208 0x19: "16-bit_bitmap", 

209 0x1a: "24-bit_bitmap", 

210 0x1b: "32-bit_bitmap", 

211 0x1c: "40-bit_bitmap", 

212 0x1d: "48-bit_bitmap", 

213 0x1e: "56-bit_bitmap", 

214 0x1f: "64-bit_bitmap", 

215 # Unsigned integer 

216 0x20: "Unsigned_8-bit_integer", 

217 0x21: "Unsigned_16-bit_integer", 

218 0x22: "Unsigned_24-bit_integer", 

219 0x23: "Unsigned_32-bit_integer", 

220 0x24: "Unsigned_40-bit_integer", 

221 0x25: "Unsigned_48-bit_integer", 

222 0x26: "Unsigned_56-bit_integer", 

223 0x27: "Unsigned_64-bit_integer", 

224 # Signed integer 

225 0x28: "Signed_8-bit_integer", 

226 0x29: "Signed_16-bit_integer", 

227 0x2a: "Signed_24-bit_integer", 

228 0x2b: "Signed_32-bit_integer", 

229 0x2c: "Signed_40-bit_integer", 

230 0x2d: "Signed_48-bit_integer", 

231 0x2e: "Signed_56-bit_integer", 

232 0x2f: "Signed_64-bit_integer", 

233 # Enumeration 

234 0x30: "8-bit_enumeration", 

235 0x31: "16-bit_enumeration", 

236 # Floating point 

237 0x38: "semi_precision", 

238 0x39: "single_precision", 

239 0x3a: "double_precision", 

240 # String 

241 0x41: "octet-string", 

242 0x42: "character_string", 

243 0x43: "long_octet_string", 

244 0x44: "long_character_string", 

245 # Ordered sequence 

246 0x48: "array", 

247 0x4c: "structure", 

248 # Collection 

249 0x50: "set", 

250 0x51: "bag", 

251 # Time 

252 0xe0: "time_of_day", 

253 0xe1: "date", 

254 0xe2: "utc_time", 

255 # Identifier 

256 0xe8: "cluster_id", 

257 0xe9: "attribute_id", 

258 0xea: "bacnet_oid", 

259 # Miscellaneous 

260 0xf0: "ieee_address", 

261 0xf1: "128-bit_security_key", 

262 # Unknown 

263 0xff: "unknown", 

264} 

265 

266# Zigbee Cluster Library, IAS Zone, Enroll Response Codes 

267_zcl_ias_zone_enroll_response_codes = { 

268 0x00: "Success", 

269 0x01: "Not supported", 

270 0x02: "No enroll permit", 

271 0x03: "Too many zones", 

272} 

273 

274# Zigbee Cluster Library, IAS Zone, Zone Types 

275_zcl_ias_zone_zone_types = { 

276 0x0000: "Standard CIE", 

277 0x000d: "Motion sensor", 

278 0x0015: "Contact switch", 

279 0x0028: "Fire sensor", 

280 0x002a: "Water sensor", 

281 0x002b: "Carbon Monoxide (CO) sensor", 

282 0x002c: "Personal emergency device", 

283 0x002d: "Vibration/Movement sensor", 

284 0x010f: "Remote Control", 

285 0x0115: "Key fob", 

286 0x021d: "Keypad", 

287 0x0225: "Standard Warning Device", 

288 0x0226: "Glass break sensor", 

289 0x0229: "Security repeater", 

290 # 0x8000 - 0xfffe Manufacturer-specific types 

291 0xffff: "Invalid Zone Type", 

292} 

293 

294 

295# ZigBee # 

296 

297class ZigbeeNWK(Packet): 

298 name = "Zigbee Network Layer" 

299 fields_desc = [ 

300 BitField("discover_route", 0, 2), 

301 BitField("proto_version", 2, 4), 

302 BitEnumField("frametype", 0, 2, 

303 {0: 'data', 1: 'command', 3: 'Inter-PAN'}), 

304 FlagsField("flags", 0, 8, ['multicast', 'security', 'source_route', 'extended_dst', 'extended_src', 'reserved1', 'reserved2', 'reserved3']), # noqa: E501 

305 XLEShortField("destination", 0), 

306 XLEShortField("source", 0), 

307 ByteField("radius", 0), 

308 ByteField("seqnum", 1), 

309 

310 # ConditionalField(XLongField("ext_dst", 0), lambda pkt:pkt.flags & 8), 

311 

312 ConditionalField(dot15d4AddressField("ext_dst", 0, adjust=lambda pkt, x: 8), lambda pkt:pkt.flags & 8), # noqa: E501 

313 ConditionalField(dot15d4AddressField("ext_src", 0, adjust=lambda pkt, x: 8), lambda pkt:pkt.flags & 16), # noqa: E501 

314 

315 ConditionalField(ByteField("relay_count", 1), lambda pkt:pkt.flags & 0x04), # noqa: E501 

316 ConditionalField(ByteField("relay_index", 0), lambda pkt:pkt.flags & 0x04), # noqa: E501 

317 ConditionalField(FieldListField("relays", [], XLEShortField("", 0x0000), count_from=lambda pkt:pkt.relay_count), lambda pkt:pkt.flags & 0x04), # noqa: E501 

318 ] 

319 

320 @classmethod 

321 def dispatch_hook(cls, _pkt=None, *args, **kargs): 

322 if _pkt and len(_pkt) >= 2: 

323 frametype = ord(_pkt[:1]) & 3 

324 if frametype == 3: 

325 return ZigbeeNWKStub 

326 return cls 

327 

328 def guess_payload_class(self, payload): 

329 if self.flags.security: 

330 return ZigbeeSecurityHeader 

331 elif self.frametype == 0: 

332 return ZigbeeAppDataPayload 

333 elif self.frametype == 1: 

334 return ZigbeeNWKCommandPayload 

335 else: 

336 return Packet.guess_payload_class(self, payload) 

337 

338 

339class LinkStatusEntry(Packet): 

340 name = "ZigBee Link Status Entry" 

341 

342 fields_desc = [ 

343 # Neighbor network address (2 octets) 

344 XLEShortField("neighbor_network_address", 0x0000), 

345 # Link status (1 octet) 

346 BitField("reserved1", 0, 1), 

347 BitField("outgoing_cost", 0, 3), 

348 BitField("reserved2", 0, 1), 

349 BitField("incoming_cost", 0, 3), 

350 ] 

351 

352 def extract_padding(self, p): 

353 return b"", p 

354 

355 

356class ZigbeeNWKCommandPayload(Packet): 

357 name = "Zigbee Network Layer Command Payload" 

358 fields_desc = [ 

359 ByteEnumField("cmd_identifier", 1, { 

360 1: "route request", 

361 2: "route reply", 

362 3: "network status", 

363 4: "leave", 

364 5: "route record", 

365 6: "rejoin request", 

366 7: "rejoin response", 

367 8: "link status", 

368 9: "network report", 

369 10: "network update", 

370 11: "end device timeout request", 

371 12: "end device timeout response" 

372 # 0x0d - 0xff reserved 

373 }), 

374 

375 # - Route Request Command - # 

376 # Command options (1 octet) 

377 ConditionalField(BitField("res1", 0, 1), 

378 lambda pkt: pkt.cmd_identifier in [1, 2]), 

379 ConditionalField(BitField("multicast", 0, 1), 

380 lambda pkt: pkt.cmd_identifier in [1, 2]), 

381 ConditionalField(BitField("dest_addr_bit", 0, 1), lambda pkt: pkt.cmd_identifier == 1), # noqa: E501 

382 ConditionalField( 

383 BitEnumField("many_to_one", 0, 2, { 

384 0: "not_m2one", 1: "m2one_support_rrt", 2: "m2one_no_support_rrt", 3: "reserved"} # noqa: E501 

385 ), lambda pkt: pkt.cmd_identifier == 1), 

386 ConditionalField(BitField("res2", 0, 3), lambda pkt: pkt.cmd_identifier == 1), # noqa: E501 

387 

388 # - Route Reply Command - # 

389 # Command options (1 octet) 

390 ConditionalField(BitField("responder_addr_bit", 0, 1), lambda pkt: pkt.cmd_identifier == 2), # noqa: E501 

391 ConditionalField(BitField("originator_addr_bit", 0, 1), lambda pkt: pkt.cmd_identifier == 2), # noqa: E501 

392 ConditionalField(BitField("res3", 0, 4), lambda pkt: pkt.cmd_identifier == 2), # noqa: E501 

393 # Route request identifier (1 octet) 

394 ConditionalField(ByteField("route_request_identifier", 0), 

395 lambda pkt: pkt.cmd_identifier in [1, 2]), # noqa: E501 

396 # Originator address (2 octets) 

397 ConditionalField(XLEShortField("originator_address", 0x0000), lambda pkt: pkt.cmd_identifier == 2), # noqa: E501 

398 # Responder address (2 octets) 

399 ConditionalField(XLEShortField("responder_address", 0x0000), lambda pkt: pkt.cmd_identifier == 2), # noqa: E501 

400 

401 # - Network Status Command - # 

402 # Status code (1 octet) 

403 ConditionalField(ByteEnumField("status_code", 0, { 

404 0x00: "No route available", 

405 0x01: "Tree link failure", 

406 0x02: "Non-tree link failure", 

407 0x03: "Low battery level", 

408 0x04: "No routing capacity", 

409 0x05: "No indirect capacity", 

410 0x06: "Indirect transaction expiry", 

411 0x07: "Target device unavailable", 

412 0x08: "Target address unallocated", 

413 0x09: "Parent link failure", 

414 0x0a: "Validate route", 

415 0x0b: "Source route failure", 

416 0x0c: "Many-to-one route failure", 

417 0x0d: "Address conflict", 

418 0x0e: "Verify addresses", 

419 0x0f: "PAN identifier update", 

420 0x10: "Network address update", 

421 0x11: "Bad frame counter", 

422 0x12: "Bad key sequence number", 

423 # 0x13 - 0xff Reserved 

424 }), lambda pkt: pkt.cmd_identifier == 3), 

425 # Destination address (2 octets) 

426 ConditionalField(XLEShortField("destination_address", 0x0000), 

427 lambda pkt: pkt.cmd_identifier in [1, 3]), 

428 # Path cost (1 octet) 

429 ConditionalField(ByteField("path_cost", 0), 

430 lambda pkt: pkt.cmd_identifier in [1, 2]), # noqa: E501 

431 # Destination IEEE Address (0/8 octets), only present when dest_addr_bit has a value of 1 # noqa: E501 

432 ConditionalField(dot15d4AddressField("ext_dst", 0, adjust=lambda pkt, x: 8), # noqa: E501 

433 lambda pkt: (pkt.cmd_identifier == 1 and pkt.dest_addr_bit == 1)), # noqa: E501 

434 # Originator IEEE address (0/8 octets) 

435 ConditionalField(dot15d4AddressField("originator_addr", 0, adjust=lambda pkt, x: 8), # noqa: E501 

436 lambda pkt: (pkt.cmd_identifier == 2 and pkt.originator_addr_bit == 1)), # noqa: E501 

437 # Responder IEEE address (0/8 octets) 

438 ConditionalField(dot15d4AddressField("responder_addr", 0, adjust=lambda pkt, x: 8), # noqa: E501 

439 lambda pkt: (pkt.cmd_identifier == 2 and pkt.responder_addr_bit == 1)), # noqa: E501 

440 

441 # - Leave Command - # 

442 # Command options (1 octet) 

443 # Bit 7: Remove children 

444 ConditionalField(BitField("remove_children", 0, 1), lambda pkt: pkt.cmd_identifier == 4), # noqa: E501 

445 # Bit 6: Request 

446 ConditionalField(BitField("request", 0, 1), lambda pkt: pkt.cmd_identifier == 4), # noqa: E501 

447 # Bit 5: Rejoin 

448 ConditionalField(BitField("rejoin", 0, 1), lambda pkt: pkt.cmd_identifier == 4), # noqa: E501 

449 # Bit 0 - 4: Reserved 

450 ConditionalField(BitField("res4", 0, 5), lambda pkt: pkt.cmd_identifier == 4), # noqa: E501 

451 

452 # - Route Record Command - # 

453 # Relay count (1 octet) 

454 ConditionalField(ByteField("rr_relay_count", 0), lambda pkt: pkt.cmd_identifier == 5), # noqa: E501 

455 # Relay list (variable in length) 

456 ConditionalField( 

457 FieldListField("rr_relay_list", [], XLEShortField("", 0x0000), count_from=lambda pkt:pkt.rr_relay_count), # noqa: E501 

458 lambda pkt:pkt.cmd_identifier == 5), 

459 

460 # - Rejoin Request Command - # 

461 # Capability Information (1 octet) 

462 ConditionalField(BitField("allocate_address", 0, 1), lambda pkt:pkt.cmd_identifier == 6), # Allocate Address # noqa: E501 

463 ConditionalField(BitField("security_capability", 0, 1), lambda pkt:pkt.cmd_identifier == 6), # Security Capability # noqa: E501 

464 ConditionalField(BitField("reserved2", 0, 1), lambda pkt:pkt.cmd_identifier == 6), # bit 5 is reserved # noqa: E501 

465 ConditionalField(BitField("reserved1", 0, 1), lambda pkt:pkt.cmd_identifier == 6), # bit 4 is reserved # noqa: E501 

466 ConditionalField(BitField("receiver_on_when_idle", 0, 1), lambda pkt:pkt.cmd_identifier == 6), # Receiver On When Idle # noqa: E501 

467 ConditionalField(BitField("power_source", 0, 1), lambda pkt:pkt.cmd_identifier == 6), # Power Source # noqa: E501 

468 ConditionalField(BitField("device_type", 0, 1), lambda pkt:pkt.cmd_identifier == 6), # Device Type # noqa: E501 

469 ConditionalField(BitField("alternate_pan_coordinator", 0, 1), lambda pkt:pkt.cmd_identifier == 6), # Alternate PAN Coordinator # noqa: E501 

470 

471 # - Rejoin Response Command - # 

472 # Network address (2 octets) 

473 ConditionalField(XLEShortField("network_address", 0xFFFF), lambda pkt:pkt.cmd_identifier == 7), # noqa: E501 

474 # Rejoin status (1 octet) 

475 ConditionalField(ByteField("rejoin_status", 0), lambda pkt:pkt.cmd_identifier == 7), # noqa: E501 

476 

477 # - Link Status Command - # 

478 # Command options (1 octet) 

479 ConditionalField(BitField("res5", 0, 1), lambda pkt:pkt.cmd_identifier == 8), # Reserved # noqa: E501 

480 ConditionalField(BitField("last_frame", 0, 1), lambda pkt:pkt.cmd_identifier == 8), # Last frame # noqa: E501 

481 ConditionalField(BitField("first_frame", 0, 1), lambda pkt:pkt.cmd_identifier == 8), # First frame # noqa: E501 

482 ConditionalField(BitField("entry_count", 0, 5), lambda pkt:pkt.cmd_identifier == 8), # Entry count # noqa: E501 

483 # Link status list (variable size) 

484 ConditionalField( 

485 PacketListField("link_status_list", [], LinkStatusEntry, count_from=lambda pkt:pkt.entry_count), # noqa: E501 

486 lambda pkt:pkt.cmd_identifier == 8), 

487 

488 # - Network Report Command - # 

489 # Command options (1 octet) 

490 ConditionalField( 

491 BitEnumField("report_command_identifier", 0, 3, {0: "PAN identifier conflict"}), # 0x01 - 0x07 Reserved # noqa: E501 

492 lambda pkt: pkt.cmd_identifier == 9), 

493 ConditionalField(BitField("report_information_count", 0, 5), lambda pkt: pkt.cmd_identifier == 9), # noqa: E501 

494 

495 # - Network Update Command - # 

496 # Command options (1 octet) 

497 ConditionalField( 

498 BitEnumField("update_command_identifier", 0, 3, {0: "PAN Identifier Update"}), # 0x01 - 0x07 Reserved # noqa: E501 

499 lambda pkt: pkt.cmd_identifier == 10), 

500 ConditionalField(BitField("update_information_count", 0, 5), lambda pkt: pkt.cmd_identifier == 10), # noqa: E501 

501 # EPID: Extended PAN ID (8 octets) 

502 ConditionalField( 

503 dot15d4AddressField("epid", 0, adjust=lambda pkt, x: 8), 

504 lambda pkt: pkt.cmd_identifier in [9, 10] 

505 ), 

506 # Report information (variable length) 

507 # Only present if we have a PAN Identifier Conflict Report 

508 ConditionalField( 

509 FieldListField("PAN_ID_conflict_report", [], XLEShortField("", 0x0000), # noqa: E501 

510 count_from=lambda pkt:pkt.report_information_count), 

511 lambda pkt:(pkt.cmd_identifier == 9 and pkt.report_command_identifier == 0) # noqa: E501 

512 ), 

513 # Update Id (1 octet) 

514 ConditionalField(ByteField("update_id", 0), lambda pkt: pkt.cmd_identifier == 10), # noqa: E501 

515 # Update Information (Variable) 

516 # Only present if we have a PAN Identifier Update 

517 # New PAN ID (2 octets) 

518 ConditionalField(XLEShortField("new_PAN_ID", 0x0000), 

519 lambda pkt: (pkt.cmd_identifier == 10 and pkt.update_command_identifier == 0)), # noqa: E501 

520 

521 # - End Device Timeout Request Command - # 

522 # Requested Timeout (1 octet) 

523 ConditionalField( 

524 ByteEnumField("req_timeout", 3, { 

525 0: "10 seconds", 

526 1: "2 minutes", 

527 2: "4 minutes", 

528 3: "8 minutes", 

529 4: "16 minutes", 

530 5: "32 minutes", 

531 6: "64 minutes", 

532 7: "128 minutes", 

533 8: "256 minutes", 

534 9: "512 minutes", 

535 10: "1024 minutes", 

536 11: "2048 minutes", 

537 12: "4096 minutes", 

538 13: "8192 minutes", 

539 14: "16384 minutes" 

540 }), 

541 lambda pkt: pkt.cmd_identifier == 11), 

542 # End Device Configuration (1 octet) 

543 ConditionalField( 

544 ByteField("ed_conf", 0), 

545 lambda pkt: pkt.cmd_identifier == 11), 

546 

547 # - End Device Timeout Response Command - # 

548 # Status (1 octet) 

549 ConditionalField( 

550 ByteEnumField("status", 0, { 

551 0: "Success", 

552 1: "Incorrect Value" 

553 }), 

554 lambda pkt: pkt.cmd_identifier == 12), 

555 # Parent Information (1 octet) 

556 ConditionalField( 

557 BitField("res6", 0, 6), 

558 lambda pkt: pkt.cmd_identifier == 12), 

559 ConditionalField( 

560 BitField("ed_timeout_req_keepalive", 0, 1), 

561 lambda pkt: pkt.cmd_identifier == 12), 

562 ConditionalField( 

563 BitField("mac_data_poll_keepalive", 0, 1), 

564 lambda pkt: pkt.cmd_identifier == 12) 

565 

566 # StrField("data", ""), 

567 ] 

568 

569 

570def util_mic_len(pkt): 

571 ''' Calculate the length of the attribute value field ''' 

572 if (pkt.nwk_seclevel == 0): # no encryption, no mic 

573 return 0 

574 elif (pkt.nwk_seclevel == 1): # MIC-32 

575 return 4 

576 elif (pkt.nwk_seclevel == 2): # MIC-64 

577 return 8 

578 elif (pkt.nwk_seclevel == 3): # MIC-128 

579 return 16 

580 elif (pkt.nwk_seclevel == 4): # ENC 

581 return 0 

582 elif (pkt.nwk_seclevel == 5): # ENC-MIC-32 

583 return 4 

584 elif (pkt.nwk_seclevel == 6): # ENC-MIC-64 

585 return 8 

586 elif (pkt.nwk_seclevel == 7): # ENC-MIC-128 

587 return 16 

588 else: 

589 return 0 

590 

591 

592class ZigbeeSecurityHeader(Packet): 

593 name = "Zigbee Security Header" 

594 fields_desc = [ 

595 # Security control (1 octet) 

596 FlagsField("reserved1", 0, 2, ['reserved1', 'reserved2']), 

597 BitField("extended_nonce", 1, 1), # set to 1 if the sender address field is present (source) # noqa: E501 

598 # Key identifier 

599 BitEnumField("key_type", 1, 2, { 

600 0: 'data_key', 

601 1: 'network_key', 

602 2: 'key_transport_key', 

603 3: 'key_load_key' 

604 }), 

605 # Security level (3 bits) 

606 BitEnumField("nwk_seclevel", 0, 3, { 

607 0: "None", 

608 1: "MIC-32", 

609 2: "MIC-64", 

610 3: "MIC-128", 

611 4: "ENC", 

612 5: "ENC-MIC-32", 

613 6: "ENC-MIC-64", 

614 7: "ENC-MIC-128" 

615 }), 

616 # Frame counter (4 octets) 

617 XLEIntField("fc", 0), # provide frame freshness and prevent duplicate frames # noqa: E501 

618 # Source address (0/8 octets) 

619 ConditionalField(dot15d4AddressField("source", 0, adjust=lambda pkt, x: 8), lambda pkt: pkt.extended_nonce), # noqa: E501 

620 # Key sequence number (0/1 octet): only present when key identifier is 1 (network key) # noqa: E501 

621 ConditionalField(ByteField("key_seqnum", 0), lambda pkt: pkt.getfieldval("key_type") == 1), # noqa: E501 

622 # Payload 

623 # the length of the encrypted data is the payload length minus the MIC 

624 StrField("data", ""), # noqa: E501 

625 # Message Integrity Code (0/variable in size), length depends on nwk_seclevel # noqa: E501 

626 XStrField("mic", ""), 

627 ] 

628 

629 def post_dissect(self, s): 

630 # Get the mic dissected correctly 

631 mic_length = util_mic_len(self) 

632 if mic_length > 0: # Slice "data" into "data + mic" 

633 _data, _mic = self.data[:-mic_length], self.data[-mic_length:] 

634 self.data, self.mic = _data, _mic 

635 return s 

636 

637 

638class ZigbeeAppDataPayload(Packet): 

639 name = "Zigbee Application Layer Data Payload (General APS Frame Format)" 

640 fields_desc = [ 

641 # Frame control (1 octet) 

642 FlagsField("frame_control", 2, 4, 

643 ['ack_format', 'security', 'ack_req', 'extended_hdr']), 

644 BitEnumField("delivery_mode", 0, 2, 

645 {0: 'unicast', 1: 'indirect', 

646 2: 'broadcast', 3: 'group_addressing'}), 

647 BitEnumField("aps_frametype", 0, 2, 

648 {0: 'data', 1: 'command', 2: 'ack'}), 

649 # Destination endpoint (0/1 octet) 

650 ConditionalField( 

651 ByteField("dst_endpoint", 10), 

652 lambda pkt: ((pkt.aps_frametype == 0 and 

653 pkt.delivery_mode in [0, 2]) or 

654 (pkt.aps_frametype == 2 and not 

655 pkt.frame_control.ack_format)) 

656 ), 

657 # Group address (0/2 octets) 

658 ConditionalField( 

659 XLEShortField("group_addr", 0x0000), 

660 lambda pkt: (pkt.aps_frametype == 0 and pkt.delivery_mode == 3) 

661 ), 

662 # Cluster identifier (0/2 octets) 

663 ConditionalField( 

664 # unsigned short (little-endian) 

665 XLEShortField("cluster", 0x0000), 

666 lambda pkt: ((pkt.aps_frametype == 0) or 

667 (pkt.aps_frametype == 2 and not 

668 pkt.frame_control.ack_format)) 

669 ), 

670 # Profile identifier (0/2 octets) 

671 ConditionalField( 

672 EnumField("profile", 0, _aps_profile_identifiers, fmt="<H"), 

673 lambda pkt: ((pkt.aps_frametype == 0) or 

674 (pkt.aps_frametype == 2 and not 

675 pkt.frame_control.ack_format)) 

676 ), 

677 # Source endpoint (0/1 octets) 

678 ConditionalField( 

679 ByteField("src_endpoint", 10), 

680 lambda pkt: ((pkt.aps_frametype == 0) or 

681 (pkt.aps_frametype == 2 and not 

682 pkt.frame_control.ack_format)) 

683 ), 

684 # APS counter (1 octet) 

685 ByteField("counter", 0), 

686 # Extended header (0/1/2 octets) 

687 # cribbed from https://github.com/wireshark/wireshark/blob/master/epan/dissectors/packet-zbee-aps.c # noqa: E501 

688 ConditionalField( 

689 ByteEnumField( 

690 "fragmentation", 0, 

691 {0: "none", 1: "first_block", 2: "middle_block"}), 

692 lambda pkt: (pkt.aps_frametype in [0, 2] and 

693 pkt.frame_control.extended_hdr) 

694 ), 

695 ConditionalField( 

696 ByteField("block_number", 0), 

697 lambda pkt: (pkt.aps_frametype in [0, 2] and 

698 pkt.fragmentation in [1, 2]) 

699 ), 

700 ConditionalField( 

701 ByteField("ack_bitfield", 0), 

702 lambda pkt: (pkt.aps_frametype == 2 and 

703 pkt.fragmentation in [1, 2]) 

704 ), 

705 # variable length frame payload: 

706 # 3 frame types: data, APS command, and acknowledgement 

707 # ConditionalField(StrField("data", ""), lambda pkt:pkt.aps_frametype == 0), # noqa: E501 

708 ] 

709 

710 def guess_payload_class(self, payload): 

711 if self.frame_control & 0x02: # we have a security header 

712 return ZigbeeSecurityHeader 

713 elif self.aps_frametype == 0: # data 

714 if self.profile == 0x0000: 

715 return ZigbeeDeviceProfile 

716 else: 

717 return ZigbeeClusterLibrary 

718 elif self.aps_frametype == 1: # command 

719 return ZigbeeAppCommandPayload 

720 else: 

721 return Packet.guess_payload_class(self, payload) 

722 

723 

724_TransportKeyKeyTypes = { 

725 0x00: "Trust Center Master Key", 

726 0x01: "Standard Network Key", 

727 0x02: "Application Master Key", 

728 0x03: "Application Link Key", 

729 0x04: "Trust Center Link Key", 

730 0x05: "High-Security Network Key", 

731} 

732 

733 

734_RequestKeyKeyTypes = { 

735 0x02: "Application Link Key", 

736 0x04: "Trust Center Link Key", 

737} 

738 

739 

740_ApsStatusValues = { 

741 0x00: "SUCCESS", 

742 0xa0: "ASDU_TOO_LONG", 

743 0xa1: "DEFRAG_DEFERRED", 

744 0xa2: "DEFRAG_UNSUPPORTED", 

745 0xa3: "ILLEGAL_REQUEST", 

746 0xa4: "INVALID_BINDING", 

747 0xa5: "INVALID_GROUP", 

748 0xa6: "INVALID_PARAMETER", 

749 0xa7: "NO_ACK", 

750 0xa8: "NO_BOUND_DEVICE", 

751 0xa9: "NO_SHORT_ADDRESS", 

752 0xaa: "NOT_SUPPORTED", 

753 0xab: "SECURED_LINK_KEY", 

754 0xac: "SECURED_NWK_KEY", 

755 0xad: "SECURITY_FAIL", 

756 0xae: "TABLE_FULL", 

757 0xaf: "UNSECURED", 

758 0xb0: "UNSUPPORTED_ATTRIBUTE" 

759} 

760 

761 

762class ZigbeeAppCommandPayload(Packet): 

763 name = "Zigbee Application Layer Command Payload" 

764 fields_desc = [ 

765 ByteEnumField("cmd_identifier", 1, { 

766 1: "APS_CMD_SKKE_1", 

767 2: "APS_CMD_SKKE_2", 

768 3: "APS_CMD_SKKE_3", 

769 4: "APS_CMD_SKKE_4", 

770 5: "APS_CMD_TRANSPORT_KEY", 

771 6: "APS_CMD_UPDATE_DEVICE", 

772 7: "APS_CMD_REMOVE_DEVICE", 

773 8: "APS_CMD_REQUEST_KEY", 

774 9: "APS_CMD_SWITCH_KEY", 

775 # TODO: implement 10 to 13 

776 10: "APS_CMD_EA_INIT_CHLNG", 

777 11: "APS_CMD_EA_RSP_CHLNG", 

778 12: "APS_CMD_EA_INIT_MAC_DATA", 

779 13: "APS_CMD_EA_RSP_MAC_DATA", 

780 14: "APS_CMD_TUNNEL", 

781 15: "APS_CMD_VERIFY_KEY", 

782 16: "APS_CMD_CONFIRM_KEY" 

783 }), 

784 # SKKE Commands 

785 ConditionalField(dot15d4AddressField("initiator", 0, 

786 adjust=lambda pkt, x: 8), 

787 lambda pkt: pkt.cmd_identifier in [1, 2, 3, 4]), 

788 ConditionalField(dot15d4AddressField("responder", 0, 

789 adjust=lambda pkt, x: 8), 

790 lambda pkt: pkt.cmd_identifier in [1, 2, 3, 4]), 

791 ConditionalField(StrFixedLenField("data", 0, length=16), 

792 lambda pkt: pkt.cmd_identifier in [1, 2, 3, 4]), 

793 # Confirm-key command 

794 ConditionalField( 

795 ByteEnumField("status", 0, _ApsStatusValues), 

796 lambda pkt: pkt.cmd_identifier == 16), 

797 # Common fields 

798 ConditionalField( 

799 ByteEnumField("key_type", 0, _TransportKeyKeyTypes), 

800 lambda pkt: pkt.cmd_identifier in [5, 8, 15, 16]), 

801 ConditionalField(dot15d4AddressField("address", 0, 

802 adjust=lambda pkt, x: 8), 

803 lambda pkt: pkt.cmd_identifier in [6, 7, 15, 16]), 

804 # Transport-key Command 

805 ConditionalField( 

806 StrFixedLenField("key", None, 16), 

807 lambda pkt: pkt.cmd_identifier == 5), 

808 ConditionalField( 

809 ByteField("key_seqnum", 0), 

810 lambda pkt: (pkt.cmd_identifier == 5 and 

811 pkt.key_type in [0x01, 0x05])), 

812 ConditionalField( 

813 dot15d4AddressField("dest_addr", 0, adjust=lambda pkt, x: 8), 

814 lambda pkt: ((pkt.cmd_identifier == 5 and 

815 pkt.key_type not in [0x02, 0x03]) or 

816 pkt.cmd_identifier == 14)), 

817 ConditionalField( 

818 dot15d4AddressField("src_addr", 0, adjust=lambda pkt, x: 8), 

819 lambda pkt: (pkt.cmd_identifier == 5 and 

820 pkt.key_type not in [0x02, 0x03])), 

821 ConditionalField( 

822 dot15d4AddressField("partner_addr", 0, adjust=lambda pkt, x: 8), 

823 lambda pkt: ((pkt.cmd_identifier == 5 and 

824 pkt.key_type in [0x02, 0x03]) or 

825 (pkt.cmd_identifier == 8 and pkt.key_type == 0x02))), 

826 ConditionalField( 

827 ByteField("initiator_flag", 0), 

828 lambda pkt: (pkt.cmd_identifier == 5 and 

829 pkt.key_type in [0x02, 0x03])), 

830 # Update-Device Command 

831 ConditionalField(XLEShortField("short_address", 0), 

832 lambda pkt: pkt.cmd_identifier == 6), 

833 ConditionalField(ByteField("update_status", 0), 

834 lambda pkt: pkt.cmd_identifier == 6), 

835 # Switch-Key Command 

836 ConditionalField(StrFixedLenField("seqnum", None, 8), 

837 lambda pkt: pkt.cmd_identifier == 9), 

838 # Un-implemented: 10-13 (+?) 

839 ConditionalField(StrField("unimplemented", ""), 

840 lambda pkt: (pkt.cmd_identifier >= 10 and 

841 pkt.cmd_identifier <= 13)), 

842 # Tunnel Command 

843 ConditionalField( 

844 FlagsField("frame_control", 2, 4, [ 

845 "ack_format", 

846 "security", 

847 "ack_req", 

848 "extended_hdr" 

849 ]), 

850 lambda pkt: pkt.cmd_identifier == 14), 

851 ConditionalField( 

852 BitEnumField("delivery_mode", 0, 2, { 

853 0: "unicast", 

854 1: "indirect", 

855 2: "broadcast", 

856 3: "group_addressing" 

857 }), 

858 lambda pkt: pkt.cmd_identifier == 14), 

859 ConditionalField( 

860 BitEnumField("aps_frametype", 1, 2, { 

861 0: "data", 

862 1: "command", 

863 2: "ack" 

864 }), 

865 lambda pkt: pkt.cmd_identifier == 14), 

866 ConditionalField( 

867 ByteField("counter", 0), 

868 lambda pkt: pkt.cmd_identifier == 14), 

869 # Verify-Key Command 

870 ConditionalField( 

871 StrFixedLenField("key_hash", None, 16), 

872 lambda pkt: pkt.cmd_identifier == 15), 

873 ] 

874 

875 def guess_payload_class(self, payload): 

876 if self.cmd_identifier == 14: 

877 # Tunneled APS Auxiliary Header 

878 return ZigbeeSecurityHeader 

879 else: 

880 return Packet.guess_payload_class(self, payload) 

881 

882 

883class ZigBeeBeacon(Packet): 

884 name = "ZigBee Beacon Payload" 

885 fields_desc = [ 

886 # Protocol ID (1 octet) 

887 ByteField("proto_id", 0), 

888 # nwkcProtocolVersion (4 bits) 

889 BitField("nwkc_protocol_version", 0, 4), 

890 # Stack profile (4 bits) 

891 BitField("stack_profile", 0, 4), 

892 # End device capacity (1 bit) 

893 BitField("end_device_capacity", 0, 1), 

894 # Device depth (4 bits) 

895 BitField("device_depth", 0, 4), 

896 # Router capacity (1 bit) 

897 BitField("router_capacity", 0, 1), 

898 # Reserved (2 bits) 

899 BitField("reserved", 0, 2), 

900 # Extended PAN ID (8 octets) 

901 dot15d4AddressField("extended_pan_id", 0, adjust=lambda pkt, x: 8), 

902 # Tx offset (3 bytes) 

903 # In ZigBee 2006 the Tx-Offset is optional, while in the 2007 and later versions, the Tx-Offset is a required value. # noqa: E501 

904 BitField("tx_offset", 0, 24), 

905 # Update ID (1 octet) 

906 ByteField("update_id", 0), 

907 ] 

908 

909 

910# Inter-PAN Transmission # 

911class ZigbeeNWKStub(Packet): 

912 name = "Zigbee Network Layer for Inter-PAN Transmission" 

913 fields_desc = [ 

914 # NWK frame control 

915 BitField("res1", 0, 2), # remaining subfields shall have a value of 0 # noqa: E501 

916 BitField("proto_version", 2, 4), 

917 BitField("frametype", 0b11, 2), # 0b11 (3) is a reserved frame type 

918 BitField("res2", 0, 8), # remaining subfields shall have a value of 0 # noqa: E501 

919 ] 

920 

921 def guess_payload_class(self, payload): 

922 if self.frametype == 0b11: 

923 return ZigbeeAppDataPayloadStub 

924 else: 

925 return Packet.guess_payload_class(self, payload) 

926 

927 

928class ZigbeeAppDataPayloadStub(Packet): 

929 name = "Zigbee Application Layer Data Payload for Inter-PAN Transmission" 

930 fields_desc = [ 

931 FlagsField("frame_control", 0, 4, ['reserved1', 'security', 'ack_req', 'extended_hdr']), # noqa: E501 

932 BitEnumField("delivery_mode", 0, 2, {0: 'unicast', 2: 'broadcast', 3: 'group'}), # noqa: E501 

933 BitField("frametype", 3, 2), # value 0b11 (3) is a reserved frame type 

934 # Group Address present only when delivery mode field has a value of 0b11 (group delivery mode) # noqa: E501 

935 ConditionalField( 

936 XLEShortField("group_addr", 0x0), # 16-bit identifier of the group 

937 lambda pkt: pkt.getfieldval("delivery_mode") == 0b11 

938 ), 

939 # Cluster identifier 

940 XLEShortField("cluster", 0x0000), 

941 # Profile identifier 

942 EnumField("profile", 0, _aps_profile_identifiers, fmt="<H"), 

943 # ZigBee Payload 

944 ConditionalField( 

945 StrField("data", ""), 

946 lambda pkt: pkt.frametype == 3 

947 ), 

948 ] 

949 

950 

951# Zigbee Device Profile # 

952 

953 

954class ZDPActiveEPReq(Packet): 

955 name = "ZDP Transaction Data: Active_EP_req" 

956 fields_desc = [ 

957 # NWK Address (2 octets) 

958 XLEShortField("nwk_addr", 0), 

959 ] 

960 

961 

962class ZDPDeviceAnnce(Packet): 

963 name = "ZDP Transaction Data: Device_annce" 

964 fields_desc = [ 

965 # NWK Address (2 octets) 

966 XLEShortField("nwk_addr", 0), 

967 # IEEE Address (8 octets) 

968 dot15d4AddressField("ieee_addr", 0, adjust=lambda pkt, x: 8), 

969 # Capability Information (1 octet) 

970 BitField("allocate_address", 0, 1), 

971 BitField("security_capability", 0, 1), 

972 BitField("reserved2", 0, 1), 

973 BitField("reserved1", 0, 1), 

974 BitField("receiver_on_when_idle", 0, 1), 

975 BitField("power_source", 0, 1), 

976 BitField("device_type", 0, 1), 

977 BitField("alternate_pan_coordinator", 0, 1), 

978 ] 

979 

980 

981class ZigbeeDeviceProfile(Packet): 

982 name = "Zigbee Device Profile (ZDP) Frame" 

983 fields_desc = [ 

984 # Transaction Sequence Number (1 octet) 

985 ByteField("trans_seqnum", 0), 

986 ] 

987 

988 def guess_payload_class(self, payload): 

989 if self.underlayer.cluster == 0x0005: 

990 return ZDPActiveEPReq 

991 elif self.underlayer.cluster == 0x0013: 

992 return ZDPDeviceAnnce 

993 return Packet.guess_payload_class(self, payload) 

994 

995 

996# ZigBee Cluster Library # 

997 

998 

999_ZCL_attr_length = { 

1000 0x00: 0, # no data 

1001 0x08: 1, # 8-bit data 

1002 0x09: 2, # 16-bit data 

1003 0x0a: 3, # 24-bit data 

1004 0x0b: 4, # 32-bit data 

1005 0x0c: 5, # 40-bit data 

1006 0x0d: 6, # 48-bit data 

1007 0x0e: 7, # 56-bit data 

1008 0x0f: 8, # 64-bit data 

1009 0x10: 1, # boolean 

1010 0x18: 1, # 8-bit bitmap 

1011 0x19: 2, # 16-bit bitmap 

1012 0x1a: 3, # 24-bit bitmap 

1013 0x1b: 4, # 32-bit bitmap 

1014 0x1c: 5, # 40-bit bitmap 

1015 0x1d: 6, # 48-bit bitmap 

1016 0x1e: 7, # 46-bit bitmap 

1017 0x1f: 8, # 64-bit bitmap 

1018 0x20: 1, # Unsigned 8-bit integer 

1019 0x21: 2, # Unsigned 16-bit integer 

1020 0x22: 3, # Unsigned 24-bit integer 

1021 0x23: 4, # Unsigned 32-bit integer 

1022 0x24: 5, # Unsigned 40-bit integer 

1023 0x25: 6, # Unsigned 48-bit integer 

1024 0x26: 7, # Unsigned 56-bit integer 

1025 0x27: 8, # Unsigned 64-bit integer 

1026 0x28: 1, # Signed 8-bit integer 

1027 0x29: 2, # Signed 16-bit integer 

1028 0x2a: 3, # Signed 24-bit integer 

1029 0x2b: 4, # Signed 32-bit integer 

1030 0x2c: 5, # Signed 40-bit integer 

1031 0x2d: 6, # Signed 48-bit integer 

1032 0x2e: 7, # Signed 56-bit integer 

1033 0x2f: 8, # Signed 64-bit integer 

1034 0x30: 1, # 8-bit enumeration 

1035 0x31: 2, # 16-bit enumeration 

1036 0x38: 2, # Semi-precision 

1037 0x39: 4, # Single precision 

1038 0x3a: 8, # Double precision 

1039 0x41: (1, "!B"), # Octet string 

1040 0x42: (1, "!B"), # Character string 

1041 0x43: (2, "!H"), # Long octet string 

1042 0x44: (2, "!H"), # Long character string 

1043 # TODO (implement Ordered sequence & collection 

1044 0xe0: 4, # Time of day 

1045 0xe1: 4, # Date 

1046 0xe2: 4, # UTCTime 

1047 0xe8: 2, # Cluster ID 

1048 0xe9: 2, # Attribute ID 

1049 0xea: 4, # BACnet OID 

1050 0xf0: 8, # IEEE address 

1051 0xf1: 16, # 128-bit security key 

1052 0xff: 0, # Unknown 

1053} 

1054 

1055 

1056class _DiscreteString(StrLenField): 

1057 def getfield(self, pkt, s): 

1058 dtype = pkt.attribute_data_type 

1059 length = _ZCL_attr_length.get(dtype, None) 

1060 if length is None: 

1061 return b"", self.m2i(pkt, s) 

1062 elif isinstance(length, tuple): # Variable length 

1063 size, fmt = length 

1064 # We add size as we include the length tag in the string 

1065 length = struct.unpack(fmt, s[:size])[0] + size 

1066 if isinstance(length, int): 

1067 self.length_from = lambda x: length 

1068 return StrLenField.getfield(self, pkt, s) 

1069 return s 

1070 

1071 

1072class ZCLReadAttributeStatusRecord(Packet): 

1073 name = "ZCL Read Attribute Status Record" 

1074 fields_desc = [ 

1075 # Attribute Identifier 

1076 XLEShortField("attribute_identifier", 0), 

1077 # Status 

1078 ByteEnumField("status", 0, _zcl_enumerated_status_values), 

1079 # Attribute data type (0/1 octet), and data (0/variable size) 

1080 # are only included if status == 0x00 (SUCCESS) 

1081 ConditionalField( 

1082 ByteEnumField("attribute_data_type", 0, _zcl_attribute_data_types), 

1083 lambda pkt:pkt.status == 0x00 

1084 ), 

1085 ConditionalField( 

1086 _DiscreteString("attribute_value", ""), 

1087 lambda pkt:pkt.status == 0x00 

1088 ), 

1089 ] 

1090 

1091 def extract_padding(self, s): 

1092 return "", s 

1093 

1094 

1095class ZCLWriteAttributeRecord(Packet): 

1096 name = "ZCL Write Attribute Record" 

1097 fields_desc = [ 

1098 # Attribute Identifier (2 octets) 

1099 XLEShortField("attribute_identifier", 0), 

1100 # Attribute Data Type (1 octet) 

1101 ByteEnumField("attribute_data_type", 0, _zcl_attribute_data_types), 

1102 # Attribute Data (variable) 

1103 _DiscreteString("attribute_data", ""), 

1104 ] 

1105 

1106 def extract_padding(self, s): 

1107 return "", s 

1108 

1109 

1110class ZCLWriteAttributeStatusRecord(Packet): 

1111 name = "ZCL Write Attribute Status Record" 

1112 fields_desc = [ 

1113 # Status (1 octet) 

1114 ByteEnumField("status", 0, _zcl_enumerated_status_values), 

1115 # Attribute Identifier (0/2 octets) 

1116 ConditionalField( 

1117 XLEShortField("attribute_identifier", 0), 

1118 lambda pkt:pkt.status != 0x00 

1119 ), 

1120 ] 

1121 

1122 def extract_padding(self, s): 

1123 return "", s 

1124 

1125 

1126class ZCLConfigureReportingRecord(Packet): 

1127 name = "ZCL Configure Reporting Record" 

1128 fields_desc = [ 

1129 # Direction (1 octet) 

1130 ByteField("attribute_direction", 0), 

1131 # Attribute Identifier (2 octets) 

1132 XLEShortField("attribute_identifier", 0), 

1133 # Attribute Data Type (0/1 octet) 

1134 ConditionalField( 

1135 ByteEnumField("attribute_data_type", 0, _zcl_attribute_data_types), 

1136 lambda pkt:pkt.attribute_direction == 0x00 

1137 ), 

1138 # Minimum Reporting Interval (0/2 octets) 

1139 ConditionalField( 

1140 XLEShortField("min_reporting_interval", 0), 

1141 lambda pkt:pkt.attribute_direction == 0x00 

1142 ), 

1143 # Maximum Reporting Interval (0/2 octets) 

1144 ConditionalField( 

1145 XLEShortField("max_reporting_interval", 0), 

1146 lambda pkt:pkt.attribute_direction == 0x00 

1147 ), 

1148 # Reportable Change (variable) 

1149 ConditionalField( 

1150 _DiscreteString("reportable_change", ""), 

1151 lambda pkt:pkt.attribute_direction == 0x00 

1152 ), 

1153 # Timeout Period (0/2 octets) 

1154 ConditionalField( 

1155 XLEShortField("timeout_period", 0), 

1156 lambda pkt:pkt.attribute_direction == 0x01 

1157 ), 

1158 ] 

1159 

1160 def extract_padding(self, s): 

1161 return "", s 

1162 

1163 

1164class ZCLConfigureReportingResponseRecord(Packet): 

1165 name = "ZCL Configure Reporting Response Record" 

1166 fields_desc = [ 

1167 # Status (1 octet) 

1168 ByteEnumField("status", 0, _zcl_enumerated_status_values), 

1169 # Direction (0/1 octet) 

1170 ConditionalField( 

1171 ByteField("attribute_direction", 0), 

1172 lambda pkt:pkt.status != 0x00 

1173 ), 

1174 # Attribute Identifier (0/2 octets) 

1175 ConditionalField( 

1176 XLEShortField("attribute_identifier", 0), 

1177 lambda pkt:pkt.status != 0x00 

1178 ), 

1179 ] 

1180 

1181 def extract_padding(self, s): 

1182 return "", s 

1183 

1184 

1185class ZCLAttributeReport(Packet): 

1186 name = "ZCL Attribute Report" 

1187 fields_desc = [ 

1188 # Attribute Identifier (2 octets) 

1189 XLEShortField("attribute_identifier", 0), 

1190 # Attribute Data Type (1 octet) 

1191 ByteEnumField("attribute_data_type", 0, _zcl_attribute_data_types), 

1192 # Attribute Data (variable) 

1193 _DiscreteString("attribute_data", ""), 

1194 ] 

1195 

1196 def extract_padding(self, s): 

1197 return "", s 

1198 

1199 

1200class ZCLGeneralReadAttributes(Packet): 

1201 name = "General Domain: Command Frame Payload: read_attributes" 

1202 fields_desc = [ 

1203 FieldListField("attribute_identifiers", [], XLEShortField("", 0x0000)), 

1204 ] 

1205 

1206 

1207class ZCLGeneralReadAttributesResponse(Packet): 

1208 name = "General Domain: Command Frame Payload: read_attributes_response" 

1209 fields_desc = [ 

1210 PacketListField("read_attribute_status_record", [], ZCLReadAttributeStatusRecord), # noqa: E501 

1211 ] 

1212 

1213 

1214class ZCLGeneralWriteAttributes(Packet): 

1215 name = "General Domain: Command Frame Payload: write_attributes" 

1216 fields_desc = [ 

1217 PacketListField("write_records", [], ZCLWriteAttributeRecord), 

1218 ] 

1219 

1220 

1221class ZCLGeneralWriteAttributesResponse(Packet): 

1222 name = "General Domain: Command Frame Payload: write_attributes_response" 

1223 fields_desc = [ 

1224 PacketListField("status_records", [], ZCLWriteAttributeStatusRecord), 

1225 ] 

1226 

1227 

1228class ZCLGeneralConfigureReporting(Packet): 

1229 name = "General Domain: Command Frame Payload: configure_reporting" 

1230 fields_desc = [ 

1231 PacketListField("config_records", [], ZCLConfigureReportingRecord), 

1232 ] 

1233 

1234 

1235class ZCLGeneralConfigureReportingResponse(Packet): 

1236 name = "General Domain: Command Frame Payload: configure_reporting_response" # noqa: E501 

1237 fields_desc = [ 

1238 PacketListField("status_records", [], ZCLConfigureReportingResponseRecord), # noqa: E501 

1239 ] 

1240 

1241 

1242class ZCLGeneralReportAttributes(Packet): 

1243 name = "General Domain: Command Frame Payload: report_attributes" 

1244 fields_desc = [ 

1245 PacketListField("attribute_reports", [], ZCLAttributeReport), 

1246 ] 

1247 

1248 

1249class ZCLGeneralDefaultResponse(Packet): 

1250 name = "General Domain: Command Frame Payload: default_response" 

1251 fields_desc = [ 

1252 # Response Command Identifier (1 octet) 

1253 ByteField("response_command_identifier", 0), 

1254 # Status (1 octet) 

1255 ByteEnumField("status", 0, _zcl_enumerated_status_values), 

1256 ] 

1257 

1258 

1259class ZCLIASZoneZoneEnrollResponse(Packet): 

1260 name = "IAS Zone Cluster: Zone Enroll Response Command (Server: Received)" 

1261 fields_desc = [ 

1262 # Enroll Response Code (1 octet) 

1263 ByteEnumField("rsp_code", 0, _zcl_ias_zone_enroll_response_codes), 

1264 # Zone ID (1 octet) 

1265 ByteField("zone_id", 0), 

1266 ] 

1267 

1268 

1269class ZCLIASZoneZoneStatusChangeNotification(Packet): 

1270 name = "IAS Zone Cluster: Zone Status Change Notification Command (Server: Generated)" # noqa: E501 

1271 fields_desc = [ 

1272 # Zone Status (2 octets) 

1273 StrFixedLenField("zone_status", b'\x00\x00', length=2), 

1274 # Extended Status (1 octet) 

1275 StrFixedLenField("extended_status", b'\x00', length=1), 

1276 # Zone ID (1 octet) 

1277 ByteField("zone_id", 0), 

1278 # Delay (2 octets) 

1279 XLEShortField("delay", 0), 

1280 ] 

1281 

1282 

1283class ZCLIASZoneZoneEnrollRequest(Packet): 

1284 name = "IAS Zone Cluster: Zone Enroll Request Command (Server: Generated)" 

1285 fields_desc = [ 

1286 # Zone Type (2 octets) 

1287 EnumField("zone_type", 0, _zcl_ias_zone_zone_types, fmt="<H"), 

1288 # Manufacturer Code (2 octets) 

1289 XLEShortField("manuf_code", 0), 

1290 ] 

1291 

1292 

1293class ZCLMeteringGetProfile(Packet): 

1294 name = "Metering Cluster: Get Profile Command (Server: Received)" 

1295 fields_desc = [ 

1296 # Interval Channel (8-bit Enumeration): 1 octet 

1297 ByteField("Interval_Channel", 0), # 0 == Consumption Delivered ; 1 == Consumption Received # noqa: E501 

1298 # End Time (UTCTime): 4 octets 

1299 XLEIntField("End_Time", 0x00000000), 

1300 # NumberOfPeriods (Unsigned 8-bit Integer): 1 octet 

1301 ByteField("NumberOfPeriods", 1), # Represents the number of intervals being requested. # noqa: E501 

1302 ] 

1303 

1304 

1305class ZCLPriceGetCurrentPrice(Packet): 

1306 name = "Price Cluster: Get Current Price Command (Server: Received)" 

1307 fields_desc = [ 

1308 BitField("reserved", 0, 7), 

1309 BitField("Requestor_Rx_On_When_Idle", 0, 1), 

1310 ] 

1311 

1312 

1313class ZCLPriceGetScheduledPrices(Packet): 

1314 name = "Price Cluster: Get Scheduled Prices Command (Server: Received)" 

1315 fields_desc = [ 

1316 XLEIntField("start_time", 0x00000000), # UTCTime (4 octets) 

1317 ByteField("number_of_events", 0), # Number of Events (1 octet) 

1318 ] 

1319 

1320 

1321class ZCLPricePublishPrice(Packet): 

1322 name = "Price Cluster: Publish Price Command (Server: Generated)" 

1323 fields_desc = [ 

1324 XLEIntField("provider_id", 0x00000000), # Unsigned 32-bit Integer (4 octets) # noqa: E501 

1325 # Rate Label is a UTF-8 encoded Octet String (0-12 octets). The first Octet indicates the length. # noqa: E501 

1326 StrLenField("rate_label", "", length_from=lambda pkt:int(pkt.rate_label[0])), # TODO verify # noqa: E501 

1327 XLEIntField("issuer_event_id", 0x00000000), # Unsigned 32-bit Integer (4 octets) # noqa: E501 

1328 XLEIntField("current_time", 0x00000000), # UTCTime (4 octets) 

1329 ByteField("unit_of_measure", 0), # 8 bits enumeration (1 octet) 

1330 XLEShortField("currency", 0x0000), # Unsigned 16-bit Integer (2 octets) # noqa: E501 

1331 ByteField("price_trailing_digit", 0), # 8-bit BitMap (1 octet) 

1332 ByteField("number_of_price_tiers", 0), # 8-bit BitMap (1 octet) 

1333 XLEIntField("start_time", 0x00000000), # UTCTime (4 octets) 

1334 XLEShortField("duration_in_minutes", 0x0000), # Unsigned 16-bit Integer (2 octets) # noqa: E501 

1335 XLEIntField("price", 0x00000000), # Unsigned 32-bit Integer (4 octets) 

1336 ByteField("price_ratio", 0), # Unsigned 8-bit Integer (1 octet) 

1337 XLEIntField("generation_price", 0x00000000), # Unsigned 32-bit Integer (4 octets) # noqa: E501 

1338 ByteField("generation_price_ratio", 0), # Unsigned 8-bit Integer (1 octet) # noqa: E501 

1339 XLEIntField("alternate_cost_delivered", 0x00000000), # Unsigned 32-bit Integer (4 octets) # noqa: E501 

1340 ByteField("alternate_cost_unit", 0), # 8-bit enumeration (1 octet) 

1341 ByteField("alternate_cost_trailing_digit", 0), # 8-bit BitMap (1 octet) # noqa: E501 

1342 ByteField("number_of_block_thresholds", 0), # 8-bit BitMap (1 octet) 

1343 ByteField("price_control", 0), # 8-bit BitMap (1 octet) 

1344 ] 

1345 

1346 

1347class ZigbeeClusterLibrary(Packet): 

1348 name = "Zigbee Cluster Library (ZCL) Frame" 

1349 deprecated_fields = { 

1350 "direction": ("command_direction", "2.5.0"), 

1351 } 

1352 fields_desc = [ 

1353 # Frame control (8 bits) 

1354 BitField("reserved", 0, 3), 

1355 BitField("disable_default_response", 0, 1), # 0 default response command will be returned # noqa: E501 

1356 BitField("command_direction", 0, 1), # 0 command sent from client to server; 1 command sent from server to client # noqa: E501 

1357 BitField("manufacturer_specific", 0, 1), # 0 manufacturer code shall not be included in the ZCL frame # noqa: E501 

1358 # Frame Type 

1359 # 0b00 command acts across the entire profile 

1360 # 0b01 command is specific to a cluster 

1361 # 0b10 - 0b11 reserved 

1362 BitEnumField("zcl_frametype", 0, 2, {0: 'profile-wide', 1: 'cluster-specific', 2: 'reserved2', 3: 'reserved3'}), # noqa: E501 

1363 # Manufacturer code (0/16 bits) only present then manufacturer_specific field is set to 1 # noqa: E501 

1364 ConditionalField(XLEShortField("manufacturer_code", 0x0), 

1365 lambda pkt: pkt.getfieldval("manufacturer_specific") == 1 # noqa: E501 

1366 ), 

1367 # Transaction sequence number (8 bits) 

1368 ByteField("transaction_sequence", 0), 

1369 # Command identifier (8 bits): the cluster command 

1370 ByteEnumField("command_identifier", 0, _zcl_command_frames), 

1371 ] 

1372 

1373 def guess_payload_class(self, payload): 

1374 if self.zcl_frametype == 0x00: 

1375 # Profile-wide command 

1376 if (self.command_identifier in 

1377 {0x00, 0x01, 0x02, 0x04, 0x06, 0x07, 0x0a, 0x0b}): 

1378 # done in bind_layers 

1379 pass 

1380 elif self.zcl_frametype == 0x01: 

1381 # Cluster-specific command 

1382 if self.underlayer.cluster == 0x0500: 

1383 # IAS Zone 

1384 if self.command_direction == 0: 

1385 # Client-to-Server command 

1386 if self.command_identifier == 0x00: 

1387 return ZCLIASZoneZoneEnrollResponse 

1388 elif self.command_direction == 1: 

1389 # Server-to-Client command 

1390 if self.command_identifier == 0x00: 

1391 return ZCLIASZoneZoneStatusChangeNotification 

1392 elif self.command_identifier == 0x01: 

1393 return ZCLIASZoneZoneEnrollRequest 

1394 elif self.underlayer.cluster == 0x0700: 

1395 # Price cluster 

1396 if self.command_direction == 0: 

1397 # Client-to-Server command 

1398 if self.command_identifier == 0x00: 

1399 return ZCLPriceGetCurrentPrice 

1400 elif self.command_identifier == 0x01: 

1401 return ZCLPriceGetScheduledPrices 

1402 elif self.command_direction == 1: 

1403 # Server-to-Client command 

1404 if self.command_identifier == 0x00: 

1405 return ZCLPricePublishPrice 

1406 return Packet.guess_payload_class(self, payload) 

1407 

1408 

1409bind_layers(ZigbeeClusterLibrary, ZCLGeneralReadAttributes, 

1410 zcl_frametype=0x00, command_identifier=0x00) 

1411bind_layers(ZigbeeClusterLibrary, ZCLGeneralReadAttributesResponse, 

1412 zcl_frametype=0x00, command_identifier=0x01) 

1413bind_layers(ZigbeeClusterLibrary, ZCLGeneralWriteAttributes, 

1414 zcl_frametype=0x00, command_identifier=0x02) 

1415bind_layers(ZigbeeClusterLibrary, ZCLGeneralWriteAttributesResponse, 

1416 zcl_frametype=0x00, command_identifier=0x04) 

1417bind_layers(ZigbeeClusterLibrary, ZCLGeneralConfigureReporting, 

1418 zcl_frametype=0x00, command_identifier=0x06) 

1419bind_layers(ZigbeeClusterLibrary, ZCLGeneralConfigureReportingResponse, 

1420 zcl_frametype=0x00, command_identifier=0x07) 

1421bind_layers(ZigbeeClusterLibrary, ZCLGeneralReportAttributes, 

1422 zcl_frametype=0x00, command_identifier=0x0a) 

1423bind_layers(ZigbeeClusterLibrary, ZCLGeneralDefaultResponse, 

1424 zcl_frametype=0x00, command_identifier=0x0b) 

1425 

1426 

1427# Zigbee Encapsulation Protocol 

1428 

1429 

1430class ZEP2(Packet): 

1431 name = "Zigbee Encapsulation Protocol (V2)" 

1432 fields_desc = [ 

1433 StrFixedLenField("preamble", "EX", length=2), 

1434 ByteField("ver", 0), 

1435 ByteField("type", 0), 

1436 ByteField("channel", 0), 

1437 ShortField("device", 0), 

1438 ByteField("lqi_mode", 1), 

1439 ByteField("lqi_val", 0), 

1440 TimeStampField("timestamp", 0), 

1441 IntField("seq", 0), 

1442 BitField("res", 0, 80), # 10 bytes reserved field 

1443 ByteField("length", 0), 

1444 ] 

1445 

1446 @classmethod 

1447 def dispatch_hook(cls, _pkt=b"", *args, **kargs): 

1448 if _pkt and len(_pkt) >= 4: 

1449 v = _pkt[2] 

1450 if v == 1: 

1451 return ZEP1 

1452 elif v == 2: 

1453 return ZEP2 

1454 return cls 

1455 

1456 def guess_payload_class(self, payload): 

1457 if self.lqi_mode: 

1458 return Dot15d4 

1459 else: 

1460 return Dot15d4FCS 

1461 

1462 

1463class ZEP1(ZEP2): 

1464 name = "Zigbee Encapsulation Protocol (V1)" 

1465 fields_desc = [ 

1466 StrFixedLenField("preamble", "EX", length=2), 

1467 ByteField("ver", 0), 

1468 ByteField("channel", 0), 

1469 ShortField("device", 0), 

1470 ByteField("lqi_mode", 0), 

1471 ByteField("lqi_val", 0), 

1472 BitField("res", 0, 56), # 7 bytes reserved field 

1473 ByteField("len", 0), 

1474 ] 

1475 

1476 

1477# Bindings # 

1478 

1479# TODO: find a way to chose between ZigbeeNWK and SixLoWPAN (cf. sixlowpan.py) 

1480# Currently: use conf.dot15d4_protocol value 

1481# bind_layers( Dot15d4Data, ZigbeeNWK) 

1482 

1483bind_layers(ZigbeeAppDataPayload, ZigbeeAppCommandPayload, frametype=1) 

1484bind_layers(Dot15d4Beacon, ZigBeeBeacon) 

1485 

1486bind_bottom_up(UDP, ZEP2, sport=17754) 

1487bind_bottom_up(UDP, ZEP2, sport=17754) 

1488bind_layers(UDP, ZEP2, sport=17754, dport=17754)