espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-10-07 16:27:35 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
main
openthread/tests/nexus/verify_utils.py
T
Jonathan Hui 635d2ffaca [nexus] implement test 1_4_DNS_TC_5 for DNS record types (#12836) 
This commit implements the Nexus test specification 1_4_DNS_TC_5 for
DNS record types and special cases in OpenThread 1.4.

The test verifies that the Border Router:
- Can resolve A and AAAA records from upstream DNS servers.
- Does not perform IPv6 AAAA synthesis from A records when not
  specifically requested or configured.
- Can resolve mDNS records on the Adjacent Infrastructure Link (AIL).
- Supports non-typical record types (RRTypes) and "Private Use"
  ranges (0xFF00-0xFFFE).
- Correctly blocks and responds with NXDomain for "ipv4only.arpa"
  queries, ensuring they are not forwarded upstream.

Test Implementation:
- Created test_1_4_DNS_TC_5.cpp to simulate the network topology and
  DNS query/response sequences.
- Created verify_1_4_DNS_TC_5.py to perform packet-level verification
  of the DNS interactions and BR behavior.
- Integrated the new test into the Nexus build and test execution
  scripts.
2026-04-08 20:01:15 -05:00

759 lines
34 KiB
Python
Raw Permalink Blame History

#!/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]} <json_file>")
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:<MeshLocalPrefix>::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)
Reference in New Issue View Git Blame Copy Permalink