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
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
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>
9"""
10ZigBee bindings for IEEE 802.15.4.
11"""
13import struct
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
21from scapy.layers.dot15d4 import dot15d4AddressField, Dot15d4Beacon, Dot15d4, \
22 Dot15d4FCS
23from scapy.layers.inet import UDP
24from scapy.layers.ntp import TimeStampField
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}
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}
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}
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}
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}
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}
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}
295# ZigBee #
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),
310 # ConditionalField(XLongField("ext_dst", 0), lambda pkt:pkt.flags & 8),
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
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 ]
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
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)
339class LinkStatusEntry(Packet):
340 name = "ZigBee Link Status Entry"
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 ]
352 def extract_padding(self, p):
353 return b"", p
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 }),
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
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
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
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
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),
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
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
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),
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
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
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),
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)
566 # StrField("data", ""),
567 ]
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
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 ]
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
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 ]
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)
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}
734_RequestKeyKeyTypes = {
735 0x02: "Application Link Key",
736 0x04: "Trust Center Link Key",
737}
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}
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 ]
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)
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 ]
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 ]
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)
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 ]
951# Zigbee Device Profile #
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 ]
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 ]
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 ]
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)
996# ZigBee Cluster Library #
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}
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
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 ]
1091 def extract_padding(self, s):
1092 return "", s
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 ]
1106 def extract_padding(self, s):
1107 return "", s
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 ]
1122 def extract_padding(self, s):
1123 return "", s
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 ]
1160 def extract_padding(self, s):
1161 return "", s
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 ]
1181 def extract_padding(self, s):
1182 return "", s
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 ]
1196 def extract_padding(self, s):
1197 return "", s
1200class ZCLGeneralReadAttributes(Packet):
1201 name = "General Domain: Command Frame Payload: read_attributes"
1202 fields_desc = [
1203 FieldListField("attribute_identifiers", [], XLEShortField("", 0x0000)),
1204 ]
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 ]
1214class ZCLGeneralWriteAttributes(Packet):
1215 name = "General Domain: Command Frame Payload: write_attributes"
1216 fields_desc = [
1217 PacketListField("write_records", [], ZCLWriteAttributeRecord),
1218 ]
1221class ZCLGeneralWriteAttributesResponse(Packet):
1222 name = "General Domain: Command Frame Payload: write_attributes_response"
1223 fields_desc = [
1224 PacketListField("status_records", [], ZCLWriteAttributeStatusRecord),
1225 ]
1228class ZCLGeneralConfigureReporting(Packet):
1229 name = "General Domain: Command Frame Payload: configure_reporting"
1230 fields_desc = [
1231 PacketListField("config_records", [], ZCLConfigureReportingRecord),
1232 ]
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 ]
1242class ZCLGeneralReportAttributes(Packet):
1243 name = "General Domain: Command Frame Payload: report_attributes"
1244 fields_desc = [
1245 PacketListField("attribute_reports", [], ZCLAttributeReport),
1246 ]
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 ]
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 ]
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 ]
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 ]
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 ]
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 ]
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 ]
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 ]
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 ]
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)
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)
1427# Zigbee Encapsulation Protocol
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 ]
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
1456 def guess_payload_class(self, payload):
1457 if self.lqi_mode:
1458 return Dot15d4
1459 else:
1460 return Dot15d4FCS
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 ]
1477# Bindings #
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)
1483bind_layers(ZigbeeAppDataPayload, ZigbeeAppCommandPayload, frametype=1)
1484bind_layers(Dot15d4Beacon, ZigBeeBeacon)
1486bind_bottom_up(UDP, ZEP2, sport=17754)
1487bind_bottom_up(UDP, ZEP2, sport=17754)
1488bind_layers(UDP, ZEP2, sport=17754, dport=17754)