From c854c63126a07e4b55fee32c673cd943530d01d3 Mon Sep 17 00:00:00 2001 From: Jonathan Hui Date: Mon, 23 Mar 2026 12:38:14 -0500 Subject: [PATCH] [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. --- tests/nexus/verify_1_3_DBR_TC_1.py | 77 ++++++++--------- tests/nexus/verify_1_3_DBR_TC_2.py | 84 +++++++++++++++---- tests/nexus/verify_utils.py | 67 +++++++++++++++ .../thread-cert/pktverify/layer_fields.py | 9 +- 4 files changed, 175 insertions(+), 62 deletions(-) diff --git a/tests/nexus/verify_1_3_DBR_TC_1.py b/tests/nexus/verify_1_3_DBR_TC_1.py index b44de1c5a..0228dd6d5 100644 --- a/tests/nexus/verify_1_3_DBR_TC_1.py +++ b/tests/nexus/verify_1_3_DBR_TC_1.py @@ -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() diff --git a/tests/nexus/verify_1_3_DBR_TC_2.py b/tests/nexus/verify_1_3_DBR_TC_2.py index cc5b8537b..5493371a2 100644 --- a/tests/nexus/verify_1_3_DBR_TC_2.py +++ b/tests/nexus/verify_1_3_DBR_TC_2.py @@ -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 diff --git a/tests/nexus/verify_utils.py b/tests/nexus/verify_utils.py index 222e3f615..3b83d0ae6 100644 --- a/tests/nexus/verify_utils.py +++ b/tests/nexus/verify_utils.py @@ -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) diff --git a/tests/scripts/thread-cert/pktverify/layer_fields.py b/tests/scripts/thread-cert/pktverify/layer_fields.py index 866944abc..59a624217 100644 --- a/tests/scripts/thread-cert/pktverify/layer_fields.py +++ b/tests/scripts/thread-cert/pktverify/layer_fields.py @@ -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):