mirror of
https://github.com/espressif/openthread.git
synced 2026-09-07 17:50:09 +00:00
[nexus] improve robustness of DBR verification in Nexus tests (#12741)
This commit refactors the verification logic for Distributed Border Router (DBR) tests in the Nexus framework to enhance robustness and reliability. Key changes include: - Introduced verify_utils.check_nwd_prefix_flags() to handle complex Thread Network Data structures, allowing for precise verification of Prefix TLV flags and Border Router sub-TLV flags even when multiple prefixes are present. - Updated verify_1_3_DBR_TC_1.py and verify_1_3_DBR_TC_2.py to use the new helper and improved the identification of OMR and ULA prefixes in Network Data by iterating through TLV types. - Added verification for Preferred and Valid Lifetimes in ICMPv6 Prefix Information Options (PIO) within Router Advertisements. - Enhanced pktverify to support icmpv6.opt.pio_valid_lifetime and ensured proper mapping of PIO lifetime fields. - Simplified MLE Data Response filtering in verify_1_3_DBR_TC_1.py for better maintainability.
This commit is contained in:
Jonathan Hui
@@ -51,56 +51,43 @@ BR_FLAG_DP_FALSE = 0
|
||||
|
||||
|
||||
def check_step3(p, omr_prefix, omr_prefix_len):
|
||||
try:
|
||||
prefixes = verify_utils.as_list(p.thread_nwd.tlv.prefix)
|
||||
types = verify_utils.as_list(p.thread_nwd.tlv.type)
|
||||
stables = verify_utils.as_list(p.thread_nwd.tlv.stable)
|
||||
except AttributeError:
|
||||
return False
|
||||
|
||||
# 1. Prefix TLV for OMR_1
|
||||
try:
|
||||
omr_idx = prefixes.index(omr_prefix)
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
# Check OMR prefix properties: 64 bits long and starts with ULA_PREFIX_START_BYTE
|
||||
if omr_prefix_len != 64 or omr_prefix[0] != ULA_PREFIX_START_BYTE:
|
||||
return False
|
||||
|
||||
# Find the NWD_PREFIX_TLV entry index corresponding to omr_idx
|
||||
prefix_indices = [i for i, t in enumerate(types) if t == consts.NWD_PREFIX_TLV]
|
||||
if omr_idx >= len(prefix_indices):
|
||||
return False
|
||||
|
||||
t_idx = prefix_indices[omr_idx]
|
||||
if stables[t_idx] != 1:
|
||||
return False
|
||||
|
||||
# Flags for OMR prefix
|
||||
# Note: We expect P_default (r flag) to be 0 for now to match OpenThread behavior
|
||||
try:
|
||||
if not (verify_utils.as_list(p.thread_nwd.tlv.border_router.pref)[0] == BR_PREFERENCE_LOW and \
|
||||
verify_utils.as_list(p.thread_nwd.tlv.border_router.flag.r)[0] == BR_FLAG_R_FALSE and \
|
||||
verify_utils.as_list(p.thread_nwd.tlv.border_router.flag.o)[0] == BR_FLAG_O_TRUE and \
|
||||
verify_utils.as_list(p.thread_nwd.tlv.border_router.flag.p)[0] == BR_FLAG_P_TRUE and \
|
||||
verify_utils.as_list(p.thread_nwd.tlv.border_router.flag.s)[0] == BR_FLAG_S_TRUE and \
|
||||
verify_utils.as_list(p.thread_nwd.tlv.border_router.flag.d)[0] == BR_FLAG_D_FALSE and \
|
||||
verify_utils.as_list(p.thread_nwd.tlv.border_router.flag.dp)[0] == BR_FLAG_DP_FALSE):
|
||||
return False
|
||||
except AttributeError:
|
||||
# 1. Prefix TLV for OMR_1 with stable=1 and BR sub-TLV with specific flags
|
||||
if not verify_utils.check_nwd_prefix_flags(p,
|
||||
omr_prefix,
|
||||
stable=1,
|
||||
pref=BR_PREFERENCE_LOW,
|
||||
r=BR_FLAG_R_FALSE,
|
||||
o=BR_FLAG_O_TRUE,
|
||||
p=BR_FLAG_P_TRUE,
|
||||
s=BR_FLAG_S_TRUE,
|
||||
d=BR_FLAG_D_FALSE,
|
||||
dp=BR_FLAG_DP_FALSE):
|
||||
return False
|
||||
|
||||
# 2. Prefix TLV for fc00::/7 with Has Route sub-TLV
|
||||
try:
|
||||
ula_idx = prefixes.index(Ipv6Addr("fc00::"))
|
||||
except ValueError:
|
||||
types = verify_utils.as_list(p.thread_nwd.tlv.type)
|
||||
prefixes = verify_utils.as_list(p.thread_nwd.tlv.prefix)
|
||||
except (AttributeError, IndexError):
|
||||
return False
|
||||
|
||||
if not hasattr(p.thread_nwd.tlv, 'has_route'):
|
||||
return False
|
||||
prefix_idx = 0
|
||||
is_ula_target = False
|
||||
for t in types:
|
||||
if t == consts.NWD_PREFIX_TLV:
|
||||
is_ula_target = (Ipv6Addr(prefixes[prefix_idx]) == Ipv6Addr("fc00::"))
|
||||
prefix_idx += 1
|
||||
elif t in (consts.NWD_COMMISSIONING_DATA_TLV, consts.NWD_SERVICE_TLV):
|
||||
is_ula_target = False
|
||||
elif t == consts.NWD_HAS_ROUTER_TLV:
|
||||
if is_ula_target:
|
||||
return True
|
||||
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def check_step4(p, pre_1_prefix, pre_1_prefix_len, omr_prefix, ext_pan_id):
|
||||
@@ -127,8 +114,13 @@ def check_step4(p, pre_1_prefix, pre_1_prefix_len, omr_prefix, ext_pan_id):
|
||||
if pre_1_prefix == omr_prefix:
|
||||
return False
|
||||
|
||||
# Check PIO A bit
|
||||
if verify_utils.as_list(p.icmpv6.opt.pio_flag.a)[0] != verify_utils.PIO_FLAG_A_TRUE:
|
||||
# Check PIO A bit and Preferred/Valid Lifetimes
|
||||
pio_index = pio_prefixes.index(pre_1_prefix)
|
||||
if verify_utils.as_list(p.icmpv6.opt.pio_flag.a)[pio_index] != verify_utils.PIO_FLAG_A_TRUE:
|
||||
return False
|
||||
if verify_utils.as_list(p.icmpv6.opt.pio_preferred_lifetime)[pio_index] == 0:
|
||||
return False
|
||||
if verify_utils.as_list(p.icmpv6.opt.pio_valid_lifetime)[pio_index] == 0:
|
||||
return False
|
||||
|
||||
# Check RIO contains OMR_1
|
||||
@@ -225,8 +217,7 @@ def verify(pv):
|
||||
# - OMR 1 MUST be 64 bits long and start with 0xFD.
|
||||
print("Step 3: BR 1 (DUT) registers itself as a Border Router.")
|
||||
|
||||
pkts.filter_wpan_src64(BR_1).\
|
||||
filter_mle_cmd(consts.MLE_DATA_RESPONSE).\
|
||||
pkts.filter(lambda p: hasattr(p, 'mle') and p.mle.cmd == consts.MLE_DATA_RESPONSE).\
|
||||
filter(lambda p: check_step3(p, OMR_PREFIX, OMR_PREFIX_LEN)).\
|
||||
must_next()
|
||||
|
||||
|
||||
@@ -44,17 +44,26 @@ def check_no_new_omr(p, omr_init):
|
||||
if not hasattr(p, 'thread_nwd'):
|
||||
return True
|
||||
try:
|
||||
types = verify_utils.as_list(p.thread_nwd.tlv.type)
|
||||
prefixes = verify_utils.as_list(p.thread_nwd.tlv.prefix)
|
||||
except AttributeError:
|
||||
except (AttributeError, IndexError):
|
||||
return True
|
||||
|
||||
for prefix in prefixes:
|
||||
if prefix is nullField:
|
||||
continue
|
||||
if prefix != omr_init and prefix[0] == 0xfd:
|
||||
# OMR prefixes start with 0xfd in this test.
|
||||
# Only OMR_INIT should be there.
|
||||
return False
|
||||
prefix_idx = 0
|
||||
for t in types:
|
||||
if t == consts.NWD_PREFIX_TLV:
|
||||
current_prefix = prefixes[prefix_idx]
|
||||
prefix_idx += 1
|
||||
if current_prefix is nullField:
|
||||
continue
|
||||
if current_prefix != omr_init and current_prefix[0] == 0xfd:
|
||||
# OMR prefixes start with 0xfd in this test.
|
||||
# Only OMR_INIT should be there.
|
||||
return False
|
||||
elif t == consts.NWD_BORDER_ROUTER_TLV:
|
||||
# Border Router sub-TLV belongs to the last seen Prefix TLV.
|
||||
pass
|
||||
|
||||
return True
|
||||
|
||||
|
||||
@@ -82,19 +91,53 @@ def check_step4(p, omr_init):
|
||||
return True
|
||||
|
||||
|
||||
# Step 12 BR constants
|
||||
BR_PREFERENCE_LOW = 3
|
||||
BR_FLAG_R_FALSE = 0
|
||||
BR_FLAG_O_TRUE = 1
|
||||
BR_FLAG_P_TRUE = 1
|
||||
BR_FLAG_S_TRUE = 1
|
||||
BR_FLAG_D_FALSE = 0
|
||||
BR_FLAG_DP_FALSE = 0
|
||||
|
||||
|
||||
def check_step12(p, omr_1, omr_init):
|
||||
if not hasattr(p, 'thread_nwd'):
|
||||
return False
|
||||
prefixes = verify_utils.as_list(p.thread_nwd.tlv.prefix)
|
||||
if omr_1 not in prefixes:
|
||||
return False
|
||||
# 1. New OMR prefix OMR_1 in Thread Network Data, not equal to OMR_init
|
||||
if omr_1 == omr_init:
|
||||
return False
|
||||
if Ipv6Addr("fc00::") not in prefixes:
|
||||
|
||||
if not verify_utils.check_nwd_prefix_flags(p,
|
||||
omr_1,
|
||||
stable=1,
|
||||
pref=BR_PREFERENCE_LOW,
|
||||
r=BR_FLAG_R_FALSE,
|
||||
o=BR_FLAG_O_TRUE,
|
||||
p=BR_FLAG_P_TRUE,
|
||||
s=BR_FLAG_S_TRUE,
|
||||
d=BR_FLAG_D_FALSE,
|
||||
dp=BR_FLAG_DP_FALSE):
|
||||
return False
|
||||
if not hasattr(p.thread_nwd.tlv, 'has_route'):
|
||||
|
||||
# 2. External route fc00::/7 in Network Data
|
||||
try:
|
||||
types = verify_utils.as_list(p.thread_nwd.tlv.type)
|
||||
prefixes = verify_utils.as_list(p.thread_nwd.tlv.prefix)
|
||||
except (AttributeError, IndexError):
|
||||
return False
|
||||
return True
|
||||
|
||||
prefix_idx = 0
|
||||
is_ula_target = False
|
||||
for t in types:
|
||||
if t == consts.NWD_PREFIX_TLV:
|
||||
is_ula_target = (Ipv6Addr(prefixes[prefix_idx]) == Ipv6Addr("fc00::"))
|
||||
prefix_idx += 1
|
||||
elif t in (consts.NWD_COMMISSIONING_DATA_TLV, consts.NWD_SERVICE_TLV):
|
||||
is_ula_target = False
|
||||
elif t == consts.NWD_HAS_ROUTER_TLV:
|
||||
if is_ula_target:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def check_step13(p, omr_1, ula_1, ext_pan_id):
|
||||
@@ -115,8 +158,13 @@ def check_step13(p, omr_1, ula_1, ext_pan_id):
|
||||
if ula_1 not in pio_prefixes:
|
||||
return False
|
||||
|
||||
# Check PIO A bit
|
||||
if verify_utils.as_list(p.icmpv6.opt.pio_flag.a)[0] != verify_utils.PIO_FLAG_A_TRUE:
|
||||
# Check PIO A bit and Preferred/Valid Lifetimes
|
||||
pio_index = pio_prefixes.index(ula_1)
|
||||
if verify_utils.as_list(p.icmpv6.opt.pio_flag.a)[pio_index] != verify_utils.PIO_FLAG_A_TRUE:
|
||||
return False
|
||||
if verify_utils.as_list(p.icmpv6.opt.pio_preferred_lifetime)[pio_index] == 0:
|
||||
return False
|
||||
if verify_utils.as_list(p.icmpv6.opt.pio_valid_lifetime)[pio_index] == 0:
|
||||
return False
|
||||
|
||||
# Check EXT_PAN_ID mapping in ULA_1
|
||||
|
||||
@@ -291,6 +291,73 @@ def get_ra_prefixes(p):
|
||||
return rio_prefixes, pio_prefixes
|
||||
|
||||
|
||||
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:
|
||||
current_prefix = prefixes[prefix_idx]
|
||||
current_stable = stables[i]
|
||||
prefix_idx += 1
|
||||
is_target = (Ipv6Addr(current_prefix) == Ipv6Addr(target_prefix))
|
||||
if is_target and stable is not None and current_stable != stable:
|
||||
is_target = False
|
||||
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 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)
|
||||
|
||||
@@ -509,6 +509,7 @@ _LAYER_FIELDS = {
|
||||
'icmpv6.opt.pio_flag.a': _list(_auto),
|
||||
'icmpv6.opt.pio_flag.l': _list(_auto),
|
||||
'icmpv6.opt.pio_preferred_lifetime': _list(_auto),
|
||||
'icmpv6.opt.pio_valid_lifetime': _list(_auto),
|
||||
'icmpv6.opt.length': _list(_auto),
|
||||
'icmpv6.opt.reserved': _str,
|
||||
'icmpv6.nd.ra.router_lifetime': _auto,
|
||||
@@ -754,10 +755,16 @@ def get_layer_field(packet: RawPacket, field_uri: str) -> Any:
|
||||
field_uri, orig_field_uri = 'icmpv6.opt.prefix.flag.a', 'icmpv6.opt.pio_flag.a'
|
||||
elif field_uri in ('icmpv6.opt.pio_flag.l', 'icmpv6.opt.prefix.flag.l'):
|
||||
field_uri, orig_field_uri = 'icmpv6.opt.prefix.flag.l', 'icmpv6.opt.pio_flag.l'
|
||||
elif field_uri in ('icmpv6.opt.pio_preferred_lifetime', 'icmpv6.opt.prefix.preferred_lifetime'):
|
||||
field_uri, orig_field_uri = 'icmpv6.opt.prefix.preferred_lifetime', 'icmpv6.opt.pio_preferred_lifetime'
|
||||
elif field_uri in ('icmpv6.opt.pio_valid_lifetime', 'icmpv6.opt.prefix.valid_lifetime'):
|
||||
field_uri, orig_field_uri = 'icmpv6.opt.prefix.valid_lifetime', 'icmpv6.opt.pio_valid_lifetime'
|
||||
elif field_uri == 'mle.tlv.addr_reg':
|
||||
field_uri = 'mle.tlv.addr_reg_ipv6'
|
||||
|
||||
if is_layer_field(field_uri) or field_uri in ('icmpv6.opt.prefix.flag.a', 'icmpv6.opt.prefix.flag.l'):
|
||||
if is_layer_field(field_uri) or field_uri in ('icmpv6.opt.prefix.flag.a', 'icmpv6.opt.prefix.flag.l',
|
||||
'icmpv6.opt.prefix.preferred_lifetime',
|
||||
'icmpv6.opt.prefix.valid_lifetime'):
|
||||
candidate_layers = _get_candidate_layers(packet, layer_name)
|
||||
for layers in candidate_layers:
|
||||
if layer_depth >= len(layers):
|
||||
|
||||
Reference in New Issue
Block a user