#!/usr/bin/env python3 # # Copyright (c) 2026, The OpenThread Authors. # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are met: # 1. Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # 2. Redistributions in binary form must reproduce the above copyright # notice, this list of conditions and the following disclaimer in the # documentation and/or other materials provided with the distribution. # 3. Neither the name of the copyright holder nor the # names of its contributors may be used to endorse or promote products # derived from this software without specific prior written permission. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. # import sys import os import json import traceback import struct import logging import ipaddress # Add the thread-cert directory to sys.path to find pktverify CUR_DIR = os.path.dirname(os.path.abspath(__file__)) OT_ROOT = os.path.abspath(os.path.join(CUR_DIR, '../..')) THREAD_CERT_DIR = os.path.join(OT_ROOT, 'tests/scripts/thread-cert') if THREAD_CERT_DIR not in sys.path: sys.path.append(THREAD_CERT_DIR) from pktverify import consts from pktverify.packet_verifier import PacketVerifier from pktverify import utils as pvutils from pktverify import coap from pktverify.coap import CoapTlvParser from pktverify.addrs import Ipv6Addr from pktverify.bytes import Bytes # Constants CSL_PERIOD_500MS = 500000 // consts.US_PER_TEN_SYMBOLS CSL_PERIOD_3300MS = 3300000 // consts.US_PER_TEN_SYMBOLS CSL_PERIOD_400MS = 400000 // consts.US_PER_TEN_SYMBOLS # Monkey-patch CoapTlvParser to parse Thread TLVs in CoAP payload def thread_coap_tlv_parse(t, v, layer=None): kvs = [] uri_path = None if layer is not None: uri_path = layer.uri_path # If the URI starts with '/d/', it is likely a Diagnostic message. # Otherwise, we assume it's MeshCoP or other Thread TLVs. is_diag = uri_path is not None and uri_path.startswith('/d/') # MeshCoP TLVs (often overlap with Diagnostic TLVs) if t == consts.NM_COMMISSIONER_SESSION_ID_TLV and len(v) == 2 and not is_diag: kvs.append(('comm_sess_id', struct.unpack('>H', v)[0])) elif t == consts.NM_STATE_TLV and len(v) == 1 and not is_diag: kvs.append(('state', v[0])) elif t == consts.NM_STEERING_DATA_TLV and not is_diag: # DG_IPV6_ADDRESS_LIST_TLV is 16*n kvs.append(('steering_data', v)) elif t == consts.NM_BORDER_AGENT_LOCATOR_TLV and len(v) == 2 and not is_diag: # DG_MAC_COUNTERS_TLV is 4*n kvs.append(('border_agent_rloc16', struct.unpack('>H', v)[0])) elif t == consts.TLV_REQUEST_TLV: kvs.append(('tlv_request', v)) elif t == consts.NM_CHANNEL_TLV and len(v) == 3 and not is_diag: # DG_MAC_EXTENDED_ADDRESS_TLV is 8 kvs.append(('channel', struct.unpack('>H', v[1:3])[0])) elif t == consts.NL_TIMEOUT_TLV and not is_diag: if len(v) == 4: kvs.append(('timeout', struct.unpack('>I', v)[0])) elif t == consts.NL_ACTIVE_TIMESTAMP_TLV and not is_diag: if len(v) == 8: kvs.append(('active_timestamp', struct.unpack('>Q', v)[0] >> 16)) elif len(v) % 16 == 0: for i in range(0, len(v), 16): kvs.append(('ipv6_address', str(Ipv6Addr(v[i:i + 16])))) elif t == consts.NM_PENDING_TIMESTAMP_TLV and len(v) == 8 and not is_diag: kvs.append(('pending_timestamp', struct.unpack('>Q', v)[0] >> 16)) elif t == consts.NM_DELAY_TIMER_TLV and len(v) == 4 and not is_diag: kvs.append(('delay_timer', struct.unpack('>I', v)[0])) elif t == consts.NM_CHANNEL_MASK_TLV and not is_diag: kvs.append(('channel_mask', v)) elif t == consts.NM_COUNT_TLV and len(v) == 1 and not is_diag: kvs.append(('count', v[0])) elif t == consts.NM_PERIOD_TLV and len(v) == 2 and not is_diag: kvs.append(('period', struct.unpack('>H', v)[0])) elif t == consts.NM_SCAN_DURATION and len(v) == 2 and not is_diag: kvs.append(('scan_duration', struct.unpack('>H', v)[0])) elif t == consts.NM_ENERGY_LIST_TLV and not is_diag: kvs.append(('energy_list', v)) elif t == consts.NM_EXTENDED_PAN_ID_TLV and len(v) == 8 and not is_diag: kvs.append(('ext_pan_id', v)) elif t == consts.NM_NETWORK_NAME_TLV and not is_diag: kvs.append(('network_name', v.decode('utf-8', errors='replace'))) elif t == consts.NM_PSKC_TLV and len(v) == 16 and not is_diag: kvs.append(('pskc', v)) elif t == consts.NM_SECURITY_POLICY_TLV and not is_diag: kvs.append(('security_policy', v)) if len(v) >= 3: # v[0:2] is rotation time # v[2] is the first byte of flags # Bits in first byte of flags (v[2]): # Bits in first byte of flags (v[2]): # 7: o (obtaining network key) # 6: n (native commissioning) # 5: r (routers) # 4: c (external commissioner) # 3: b (beacons) # 2: C (commercial commissioning) # 1: e (autonomous enrollment) # 0: p (network key provisioning) flags = v[2] kvs.append(('sec_policy_o', (flags >> 7) & 1)) kvs.append(('sec_policy_n', (flags >> 6) & 1)) kvs.append(('sec_policy_r', (flags >> 5) & 1)) kvs.append(('sec_policy_c', (flags >> 4) & 1)) kvs.append(('sec_policy_b', (flags >> 3) & 1)) kvs.append(('sec_policy_C', (flags >> 2) & 1)) kvs.append(('sec_policy_e', (flags >> 1) & 1)) kvs.append(('sec_policy_p', flags & 1)) elif t == consts.NM_NETWORK_KEY_TLV and len(v) == 16 and not is_diag: kvs.append(('network_key', v)) elif t == consts.NM_PAN_ID_TLV and len(v) == 2 and not is_diag: kvs.append(('pan_id', struct.unpack('>H', v)[0])) elif t == consts.NM_NETWORK_MESH_LOCAL_PREFIX_TLV and len(v) == 8 and not is_diag: kvs.append(('mesh_local_prefix', v)) elif t == consts.NM_JOINER_DTLS_ENCAPSULATION_TLV and not is_diag: kvs.append(('joiner_dtls_encap', v)) elif t == consts.NM_JOINER_UDP_PORT_TLV and len(v) == 2 and not is_diag: kvs.append(('joiner_udp_port', struct.unpack('>H', v)[0])) elif t == consts.NM_JOINER_IID_TLV and len(v) == 8 and not is_diag: kvs.append(('joiner_iid', v)) elif t == consts.NM_JOINER_ROUTER_LOCATOR_TLV and len(v) == 2 and not is_diag: kvs.append(('joiner_router_locator', struct.unpack('>H', v)[0])) elif t == consts.NM_JOINER_ROUTER_KEK_TLV and len(v) == 16 and not is_diag: kvs.append(('joiner_router_kek', v)) elif t == consts.NM_PROVISIONING_URL_TLV and not is_diag: kvs.append(('provisioning_url', v.decode('utf-8', errors='replace'))) elif t == consts.NM_FUTURE_TLV: kvs.append(('future_tlv', v)) # Network Data TLV (often in SVR_DATA.ntf) elif t == consts.NL_THREAD_NETWORK_DATA_TLV: # Value is Network Data # We only care about Service TLV (Type 5) -> Server TLV (Type 6) # In Network Data, TLV type is 5 bits + 1 bit Stable flag. pos = 0 while pos + 2 <= len(v): nwd_t = v[pos] nwd_l = v[pos + 1] nwd_v = v[pos + 2:pos + 2 + nwd_l] if pos + 2 + nwd_l > len(v): break if (nwd_t >> 1) == consts.NWD_SERVICE_TLV: # Service TLV if len(nwd_v) >= 2: # nwd_v[0] is T bit (1) and Service ID (7 bits) s_data_len = nwd_v[1] if len(nwd_v) >= 2 + s_data_len: s_data = nwd_v[2:2 + s_data_len] if s_data_len >= 1 and s_data[0] == consts.THREAD_SERVICE_DATA_BBR: # THREAD_SERVICE_DATA_BBR # Sub-TLVs start after S_data sub_pos = 2 + s_data_len while sub_pos + 2 <= len(nwd_v): sub_t = nwd_v[sub_pos] sub_l = nwd_v[sub_pos + 1] sub_v = nwd_v[sub_pos + 2:sub_pos + 2 + sub_l] if sub_pos + 2 + sub_l > len(nwd_v): break if (sub_t >> 1) == consts.NWD_SERVER_TLV: # Server TLV # sub_v is [RLOC16(2), SeqNo(1), ReregDelay(2), MLRTimeout(4)] if len(sub_v) >= 3: kvs.append(('bbr_seqno', sub_v[2])) sub_pos += 2 + sub_l pos += 2 + nwd_l # Other Thread TLVs elif t == consts.NL_TARGET_EID_TLV and len(v) == 16: kvs.append(('target_eid', str(Ipv6Addr(v)))) elif t == consts.DG_MAC_EXTENDED_ADDRESS_TLV and len(v) == 8 and is_diag: kvs.append(('mac_addr', v.hex())) elif t == consts.DG_MAC_ADDRESS_TLV and len(v) == 2: kvs.append(('rloc16', hex(struct.unpack('>H', v)[0]))) elif t == consts.NL_MAC_EXTENDED_ADDRESS_TLV and len(v) == 8: kvs.append(('mac_addr', v.hex())) elif t == consts.NL_ML_EID_TLV and len(v) == 8: kvs.append(('ml_eid', Bytes(v).format_hextets())) elif t == consts.NL_RLOC16_TLV and len(v) == 2: kvs.append(('rloc16', hex(struct.unpack('>H', v)[0]))) elif t == consts.NL_STATUS_TLV and len(v) == 1: kvs.append(('status', str(v[0]))) elif t == consts.NL_ROUTER_MASK_TLV: kvs.append(('router_mask', v.hex())) elif t == consts.DG_MAC_COUNTERS_TLV and is_diag: # MAC counters are a list of 4-byte values for i in range(0, len(v), 4): if i + 4 <= len(v): val = struct.unpack('>I', v[i:i + 4])[0] kvs.append(('mac_counter', str(val))) elif t == consts.DG_MODE_TLV and len(v) == 1: kvs.append(('mode', hex(v[0]))) elif t == consts.DG_IPV6_ADDRESS_LIST_TLV and is_diag: for i in range(0, len(v), 16): if i + 16 <= len(v): kvs.append(('ipv6_address', str(Ipv6Addr(v[i:i + 16])))) elif t == consts.DG_LEADER_DATA_TLV and len(v) == 8: # Leader data contains Partition ID (4), Weighting (1), Data Version (1), # Stable Data Version (1), Leader Router ID (1) kvs.append(('partition_id', hex(struct.unpack('>I', v[0:4])[0]))) kvs.append(('leader_router_id', str(v[7]))) elif t == consts.DG_ROUTE64_TLV: # Route64 contains Router ID Sequence (1), and Router ID Mask (8), then link qualities kvs.append(('router_id_sequence', str(v[0]))) kvs.append(('router_id_mask', v[1:9].hex())) elif t == consts.DG_CHILD_TABLE_TLV and is_diag: # Child table contains a list of child entries. # Each entry: [Timeout(5 bits), LQI(2 bits), Child ID(9 bits), Mode(8 bits)] -> total 3 bytes for i in range(0, len(v), 3): if i + 3 <= len(v): timeout_child_id = struct.unpack('>H', v[i:i + 2])[0] child_id = timeout_child_id & 0x1ff mode = v[i + 2] kvs.append(('child_id', hex(child_id))) kvs.append(('child_mode', hex(mode))) elif t == consts.DG_CHANNEL_PAGES_TLV: kvs.append(('channel_pages', v)) elif t == consts.NM_DISCOVERY_RESPONSE_TLV: # Discovery Response TLV # Bits 7-4: Version, Bit 3: Native Commissioner if len(v) >= 1: kvs.append(('discovery_version', (v[0] >> 4) & 0xf)) kvs.append(('discovery_native_commissioner', (v[0] >> 3) & 1)) return kvs # RA constants ICMPV6_TYPE_ROUTER_ADVERTISEMENT = 134 RA_FLAG_M_FALSE = 0 RA_FLAG_O_FALSE = 0 RA_ROUTER_LIFETIME_ZERO = 0 ICMPV6_OPT_TYPE_PIO = 3 ICMPV6_OPT_TYPE_RIO = 24 PIO_FLAG_A_TRUE = 1 # EXT_PAN_ID mapping offsets and lengths EXT_PAN_ID_GLOBAL_ID_OFFSET = 1 EXT_PAN_ID_GLOBAL_ID_LEN = 5 EXT_PAN_ID_SUBNET_ID_OFFSET = 6 EXT_PAN_ID_SUBNET_ID_LEN = 2 def as_list(x): return x if isinstance(x, list) else [x] def get_ra_prefixes(p): rio_prefixes = [] pio_prefixes = [] try: opts = as_list(p.icmpv6.opt.type) all_prefixes = as_list(p.icmpv6.opt.prefix) except (AttributeError, IndexError): return rio_prefixes, pio_prefixes prefix_idx = 0 for opt_type in opts: if opt_type in (ICMPV6_OPT_TYPE_PIO, ICMPV6_OPT_TYPE_RIO): if prefix_idx < len(all_prefixes): if opt_type == ICMPV6_OPT_TYPE_RIO: rio_prefixes.append(Ipv6Addr(all_prefixes[prefix_idx])) else: # PIO pio_prefixes.append(Ipv6Addr(all_prefixes[prefix_idx])) prefix_idx += 1 return rio_prefixes, pio_prefixes def check_ra_has_rio(packet, prefix, prf=None): """ Check if an ICMPv6 RA contains a Route Information Option (RIO) with the given prefix. """ if packet.icmpv6.type != ICMPV6_TYPE_ROUTER_ADVERTISEMENT: return False target_addr = Ipv6Addr(prefix) try: all_prefixes = as_list(packet.icmpv6.opt.prefix) all_types = as_list(packet.icmpv6.opt.type) all_prfs = as_list(packet.icmpv6.opt.route_info.flag.route_preference) except (AttributeError, IndexError): return False ri_idx = 0 prefix_idx = 0 for opt_type in all_types: if opt_type == ICMPV6_OPT_TYPE_RIO: if prefix_idx < len(all_prefixes) and Ipv6Addr(all_prefixes[prefix_idx]) == target_addr: if prf is not None: if not isinstance(prf, (list, tuple)): prf = [prf] if ri_idx < len(all_prfs) and int(all_prfs[ri_idx]) in prf: return True else: return True ri_idx += 1 prefix_idx += 1 elif opt_type == ICMPV6_OPT_TYPE_PIO: prefix_idx += 1 return False def check_nwd_prefix_flags(packet, target_prefix, stable=None, **expected_flags): """ Robustly check flags for a specific OMR prefix in Network Data. Handles cases with multiple prefixes and different sub-TLVs. """ try: types = as_list(packet.thread_nwd.tlv.type) prefixes = as_list(packet.thread_nwd.tlv.prefix) stables = as_list(packet.thread_nwd.tlv.stable) except (AttributeError, IndexError): return False prefix_idx = 0 br_idx = 0 is_target = False # We iterate through all TLVs to find the target prefix and its BR sub-TLV for i, t in enumerate(types): if t == consts.NWD_PREFIX_TLV: is_target = False if prefix_idx < len(prefixes) and prefixes[prefix_idx]: current_prefix = prefixes[prefix_idx] is_target = (Ipv6Addr(current_prefix) == Ipv6Addr(target_prefix)) if is_target and stable is not None: try: if stables[i] != stable: is_target = False except IndexError: is_target = False prefix_idx += 1 elif t in (consts.NWD_COMMISSIONING_DATA_TLV, consts.NWD_SERVICE_TLV): # These are also top-level TLVs, reset target prefix is_target = False elif t == consts.NWD_BORDER_ROUTER_TLV: if is_target: # This BR sub-TLV belongs to our target prefix! try: # Map expected_flags keys to pktverify field names field_map = { 'pref': packet.thread_nwd.tlv.border_router.pref, 'r': packet.thread_nwd.tlv.border_router.flag.r, 'o': packet.thread_nwd.tlv.border_router.flag.o, 'p': packet.thread_nwd.tlv.border_router.flag.p, 's': packet.thread_nwd.tlv.border_router.flag.s, 'd': packet.thread_nwd.tlv.border_router.flag.d, 'dp': packet.thread_nwd.tlv.border_router.flag.dp, 'n': packet.thread_nwd.tlv.border_router.flag.n, } match = True for key, expected_val in expected_flags.items(): if key in field_map: field_values = field_map[key] if field_values is not None: actual_val = as_list(field_values)[br_idx] if actual_val != expected_val: match = False break else: match = False break if match: return True # Found it and all specified flags match! except (AttributeError, IndexError): pass br_idx += 1 return False def check_nwd_has_prefix(packet, prefix): """Checks if Network Data contains the given prefix in a Prefix TLV.""" try: if not hasattr(packet, 'thread_nwd') or not hasattr(packet.thread_nwd.tlv, 'prefix'): return False prefixes = as_list(packet.thread_nwd.tlv.prefix) target_addr = Ipv6Addr(prefix) return any(p and Ipv6Addr(p) == target_addr for p in prefixes) except (AttributeError, IndexError): return False def check_nwd_has_route(packet, target_prefix, pref=None): """ Robustly check if Network Data has an External Route with the given prefix. """ try: types = as_list(packet.thread_nwd.tlv.type) prefixes = as_list(packet.thread_nwd.tlv.prefix) except (AttributeError, IndexError): return False prefix_idx = 0 hr_idx = 0 is_target = False for i, t in enumerate(types): if t == consts.NWD_PREFIX_TLV: is_target = False if prefix_idx < len(prefixes) and prefixes[prefix_idx]: is_target = (Ipv6Addr(prefixes[prefix_idx]) == Ipv6Addr(target_prefix)) prefix_idx += 1 elif t in (consts.NWD_COMMISSIONING_DATA_TLV, consts.NWD_SERVICE_TLV): is_target = False elif t == consts.NWD_HAS_ROUTER_TLV: if is_target: if pref is not None: try: all_prefs = as_list(packet.thread_nwd.tlv.has_route.pref) allowed_prefs = pref if isinstance(pref, (list, tuple)) else [pref] if hr_idx < len(all_prefs) and int(all_prefs[hr_idx]) in allowed_prefs: return True except (AttributeError, IndexError, ValueError): pass else: return True hr_idx += 1 return False def is_leader_aloc_or_rloc(addr_str: str) -> bool: """Checks if an IPv6 address is a Leader ALOC or an RLOC.""" addr = ipaddress.ip_address(addr_str) iid = addr.packed[8:] # Leader ALOC IID is 0000:00ff:fe00:fc00 is_aloc = (iid == b'\x00\x00\x00\xff\xfe\x00\xfc\x00') # RLOC IID is 0000:00ff:fe00:xxxx is_rloc = (iid[:6] == b'\x00\x00\x00\xff\xfe\x00') return is_aloc or is_rloc def is_dns_compression_used(p) -> bool: """ Check if any DNS name in the packet is compressed. """ dns = getattr(p, 'dns', None) if not dns: dns = getattr(p, 'mdns', None) if not dns: return False try: raw_layer = dns._layer for field_name, field in raw_layer._all_fields.items(): # Check fields that might contain DNS names if '.name' in field_name or '.target' in field_name or '.domain_name' in field_name: fields = field.fields if hasattr(field, 'fields') else [field] for f in fields: rv = f.raw_value if rv: for i in range(0, len(rv), 2): if int(rv[i:i + 2], 16) & 0xc0 == 0xc0: return True except Exception: pass return False def apply_patches(): CoapTlvParser.parse = staticmethod(thread_coap_tlv_parse) from pktverify import consts, layer_fields consts.VALID_LAYER_NAMES.add('wpan_tap') consts.VALID_LAYER_NAMES.add('wpan-tap') consts.VALID_LAYER_NAMES.add('trel') consts.REAL_LAYER_NAMES.add('trel') # Patch _get_candidate_layers to map wpan_tap to wpan-tap old_get_candidate_layers = layer_fields._get_candidate_layers def patched_get_candidate_layers(packet, layer_name): if layer_name == 'wpan_tap': layer_name = 'wpan-tap' return old_get_candidate_layers(packet, layer_name) layer_fields._get_candidate_layers = patched_get_candidate_layers # Patch Layer.get_field to handle wpan_tap.ch_num from pyshark.packet.layers.base import BaseLayer as Layer old_get_field = Layer.get_field def patched_get_field(self, name): v = old_get_field(self, name) if v is None: if name == 'wpan_tap.ch_num': v = old_get_field(self, 'wpan-tap.ch_num') elif name.startswith('mle.aux_sec.'): # Map MLE security fields from WPAN layer if missing in MLE layer try: wpan_layer = self._packet.wpan wpan_name = name.replace('mle.aux_sec.', 'wpan.aux_sec.') v = wpan_layer.get_field(wpan_name) except AttributeError: pass elif name == 'mle.sec_suite': try: v = self._packet.wpan.get_field('wpan.aux_sec.sec_suite') except AttributeError: pass return v Layer.get_field = patched_get_field layer_fields._LAYER_FIELDS['mle.tlv.link_forward_series'] = layer_fields._list(layer_fields._auto) layer_fields._LAYER_FIELDS['mle.tlv.addr_reg_ipv6'] = layer_fields._list(layer_fields._ipv6_addr) layer_fields._LAYER_FIELDS['mle.tlv.link_forward_series_flags'] = layer_fields._auto layer_fields._LAYER_FIELDS['mle.tlv.link_status_sub_tlv'] = layer_fields._auto layer_fields._LAYER_FIELDS['mle.tlv.query_id'] = layer_fields._auto layer_fields._LAYER_FIELDS['mle.tlv.link_requested_type_id_flags'] = layer_fields._bytes layer_fields._LAYER_FIELDS['mle.tlv.link_enh_ack_flags'] = layer_fields._auto layer_fields._LAYER_FIELDS['wpan.payload_ie.vendor.variable'] = layer_fields._bytes layer_fields._LAYER_FIELDS['mle.tlv.metric_type_id_flags.type'] = layer_fields._list(layer_fields._hex) layer_fields._LAYER_FIELDS['mle.tlv.metric_type_id_flags.metric'] = layer_fields._list(layer_fields._hex) layer_fields._LAYER_FIELDS['mle.tlv.metric_type_id_flags.l'] = layer_fields._list(layer_fields._hex) layer_fields._LAYER_FIELDS['coap.tlv.tlv_request'] = layer_fields._bytes layer_fields._LAYER_FIELDS['mle.tlv.active_operational_dataset'] = layer_fields._bytes layer_fields._LAYER_FIELDS['mle.tlv.pending_operational_dataset'] = layer_fields._bytes layer_fields._LAYER_FIELDS['mle.aux_sec.key_id_mode'] = layer_fields._auto layer_fields._LAYER_FIELDS['mle.aux_sec.key_source'] = layer_fields._auto layer_fields._LAYER_FIELDS['mle.aux_sec.key_index'] = layer_fields._auto layer_fields._layer_containers.add('mle.aux_sec') layer_fields._LAYER_FIELDS['coap.tlv.ipv6_address'] = layer_fields._list(layer_fields._ipv6_addr) layer_fields._LAYER_FIELDS['coap.tlv.timeout'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.status'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.rloc16'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.mode'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.leader_router_id'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.child_id'] = layer_fields._list(layer_fields._auto) layer_fields._LAYER_FIELDS['coap.tlv.child_mode'] = layer_fields._list(layer_fields._auto) layer_fields._LAYER_FIELDS['coap.tlv.provisioning_url'] = layer_fields._str layer_fields._LAYER_FIELDS['coap.tlv.channel_pages'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.steering_data'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.future_tlv'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.comm_sess_id'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.state'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.border_agent_rloc16'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.channel'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.active_timestamp'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.pending_timestamp'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.delay_timer'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.channel_mask'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.count'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.period'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.scan_duration'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.energy_list'] = layer_fields._bytes layer_fields._LAYER_FIELDS['wpan.header_ie.csl.phase'] = layer_fields._auto layer_fields._LAYER_FIELDS['wpan_tap.ch_num'] = layer_fields._auto layer_fields._LAYER_FIELDS['wpan-tap.ch_num'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.ext_pan_id'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.network_name'] = layer_fields._str layer_fields._LAYER_FIELDS['coap.tlv.pskc'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.joiner_dtls_encap'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.joiner_udp_port'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.joiner_iid'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.joiner_router_locator'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.joiner_router_kek'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.security_policy'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.sec_policy_o'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.sec_policy_n'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.sec_policy_r'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.sec_policy_c'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.sec_policy_b'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.sec_policy_C'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.sec_policy_e'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.sec_policy_p'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.network_key'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.tlv.pan_id'] = layer_fields._auto layer_fields._LAYER_FIELDS['coap.tlv.mesh_local_prefix'] = layer_fields._bytes layer_fields._LAYER_FIELDS['coap.opt_content_format'] = layer_fields._auto layer_fields._LAYER_FIELDS['dtls.alert_message_level'] = layer_fields._auto layer_fields._LAYER_FIELDS['dtls.alert_message_desc'] = layer_fields._auto layer_fields._LAYER_FIELDS['dtls.handshake.cookie'] = layer_fields._bytes layer_fields._LAYER_FIELDS['thread_meshcop.tlv.discovery_version'] = layer_fields._dec layer_fields._LAYER_FIELDS['thread_meshcop.tlv.discovery_native_commissioner'] = layer_fields._dec layer_fields._LAYER_FIELDS['thread_meshcop.tlv.active_tstamp'] = layer_fields._list(layer_fields._thread_timestamp) def _parse_next_tlv_patched(payload, read_pos, layer=None) -> tuple: assert read_pos <= len(payload) if read_pos == len(payload): return None, None, read_pos t = payload[read_pos] if read_pos + 1 >= len(payload): return None, None, len(payload) len_ = payload[read_pos + 1] val_pos = read_pos + 2 if len_ == 255: if read_pos + 3 >= len(payload): return None, None, len(payload) len_ = (payload[read_pos + 2] << 8) | payload[read_pos + 3] val_pos = read_pos + 4 if len(payload) - val_pos < len_: return None, None, len(payload) kvs = coap.CoapTlvParser.parse(t, payload[val_pos:val_pos + len_], layer=layer) return t, kvs, val_pos + len_ coap.CoapLayer._parse_next_tlv = staticmethod(_parse_next_tlv_patched) layer_fields._LAYER_FIELDS['coap.tlv.bbr_seqno'] = layer_fields._auto layer_fields._LAYER_FIELDS['thread_meshcop.tlv.delay_timer'] = layer_fields._auto layer_fields._LAYER_FIELDS['mle.tlv.link_query_options'] = layer_fields._bytes layer_fields._LAYER_FIELDS['dns.opt.data'] = layer_fields._list(layer_fields._bytes) layer_fields._LAYER_FIELDS['dns.count.queries'] = layer_fields._auto layer_fields._layer_containers.add('dns.opt') layer_fields._layer_containers.add('dns.count') layer_fields._LAYER_FIELDS['mdns.nsec'] = layer_fields._list(layer_fields._bytes) layer_fields._LAYER_FIELDS['dns.a'] = layer_fields._list(layer_fields._str) layer_fields._LAYER_FIELDS['mdns.a'] = layer_fields._list(layer_fields._str) layer_fields._LAYER_FIELDS['dns.data'] = layer_fields._list(layer_fields._bytes) layer_fields._LAYER_FIELDS['mdns.data'] = layer_fields._list(layer_fields._bytes) def which_tshark_patch(): default_path = '/tmp/thread-wireshark/tshark' if os.path.exists(default_path): return default_path import shutil return shutil.which('tshark') or '/usr/local/bin/tshark' pvutils.which_tshark = which_tshark_patch def get_vars(pv, prefix, count, suffix=''): return [pv.vars[f'{prefix}_{i}{suffix}'] for i in range(1, count + 1)] def run_main(verify_func): if len(sys.argv) < 2: print(f"Usage: python3 {sys.argv[0]} ") sys.exit(1) apply_patches() json_file = os.path.abspath(sys.argv[1]) with open(json_file, 'rt') as f: data = json.load(f) try: wireshark_prefs = consts.WIRESHARK_OVERRIDE_PREFS.copy() # Clean up existing keys from consts.py to avoid formatting issues existing_keys_str = wireshark_prefs.get('uat:ieee802154_keys', '') keys = [] for line in existing_keys_str.split('\n'): line = line.strip() if line: keys.append(line) network_key = data.get('network_key') if network_key: new_key = f'"{network_key}","1","Thread hash"' if not any(network_key in k for k in keys): keys.append(new_key) # Add all network keys specified in the test info network_keys = data.get('network_keys', []) for k in network_keys: new_key = f'"{k}","1","Thread hash"' if not any(k in existing for existing in keys): keys.append(new_key) wireshark_prefs['uat:ieee802154_keys'] = '\n'.join(keys) mesh_local_prefix = data.get('extra_vars', {}).get('mesh_local_prefix') if mesh_local_prefix: prefix_addr = mesh_local_prefix.split('/')[0] wireshark_prefs['6lowpan.context0'] = f'{prefix_addr}/64' # Update the Link-Local All Thread Nodes multicast address constant # FF32:40:::1 prefix = Ipv6Addr(prefix_addr) all_thread_nodes_mcast_addr = bytearray(Ipv6Addr('ff32:40::1')) all_thread_nodes_mcast_addr[4:12] = prefix[0:8] consts.LINK_LOCAL_ALL_THREAD_NODES_MULTICAST_ADDRESS = Ipv6Addr(all_thread_nodes_mcast_addr) # Add TREL UDP port variables before creating PacketVerifier (so PcapReader uses them) # We only do this for TREL tests to avoid interference with other tests # using similar UDP ports for regular Thread traffic. testcase = data.get('testcase', '') if 'TREL' in testcase: for node_id, port in data.get('trel_udp_ports', {}).items(): port = int(port) if port > 0: consts.WIRESHARK_DECODE_AS_ENTRIES[f'udp.port=={port}'] = 'trel' pv = PacketVerifier(json_file, wireshark_prefs=wireshark_prefs) pv.add_common_vars() # Add RLOC16 variables for convenience for node_id, rloc16 in data.get('rloc16s', {}).items(): name = pv.test_info.get_node_name(int(node_id)) pv.add_vars(**{f'{name}_RLOC16': int(rloc16, 16)}) # Add TREL UDP port variables to pv.vars for node_id, port in data.get('trel_udp_ports', {}).items(): name = pv.test_info.get_node_name(int(node_id)) port = int(port) if port > 0: pv.add_vars(**{f'{name}_TREL_PORT': port}) # Add channel variables for node_id, channel in data.get('channels', {}).items(): name = pv.test_info.get_node_name(int(node_id)) pv.add_vars(**{f'{name}_CHANNEL': int(channel)}) # Add OMR prefix variables omr_prefixes = data.get('omr_prefixes', {}) for node_id, omr_prefix in omr_prefixes.items(): if omr_prefix: name = pv.test_info.get_node_name(int(node_id)) pv.add_vars(**{f'{name}_OMR_PREFIX': omr_prefix}) # If all valid OMR prefixes are the same, add a generic OMR_PREFIX variable valid_omr_prefixes = [p for p in omr_prefixes.values() if p] if len(valid_omr_prefixes) > 0 and all(p == valid_omr_prefixes[0] for p in valid_omr_prefixes): pv.add_vars(OMR_PREFIX=valid_omr_prefixes[0]) verify_func(pv) print("Verification PASSED") except Exception as e: print(f"Verification FAILED: {e}") traceback.print_exc() sys.exit(1)