From 84bce0f5879c48a36131f597608de1ff662f101c Mon Sep 17 00:00:00 2001 From: Jonathan Hui Date: Wed, 25 Mar 2026 00:49:52 -0500 Subject: [PATCH] [nexus] add helper methods to verify_utils.py and update tests (#12756) This commit adds several new helper methods to tests/nexus/verify_utils.py to simplify common verification tasks in Nexus tests: - check_ra_has_rio: verify presence and preference of RIO in RA. - check_ra_has_pio: verify presence of PIO in RA. - check_nwd_has_route: verify presence and preference of external route in Network Data. Existing tests (1_3_DBR_TC_7A/B/C and 1_3_DBR_TC_8) are updated to use these new helper methods, which improves code readability and consistency across the Nexus test suite. --- tests/nexus/verify_1_3_DBR_TC_7A.py | 111 ++++++---------------------- tests/nexus/verify_1_3_DBR_TC_7B.py | 111 ++++++---------------------- tests/nexus/verify_1_3_DBR_TC_7C.py | 92 +++++------------------ tests/nexus/verify_1_3_DBR_TC_8.py | 92 +++++------------------ tests/nexus/verify_utils.py | 102 +++++++++++++++++++++++-- 5 files changed, 176 insertions(+), 332 deletions(-) diff --git a/tests/nexus/verify_1_3_DBR_TC_7A.py b/tests/nexus/verify_1_3_DBR_TC_7A.py index 8a1248309..5b612d57a 100644 --- a/tests/nexus/verify_1_3_DBR_TC_7A.py +++ b/tests/nexus/verify_1_3_DBR_TC_7A.py @@ -37,7 +37,6 @@ sys.path.append(CUR_DIR) import verify_utils from pktverify import consts from pktverify.addrs import Ipv6Addr -from pktverify.null_field import nullField # Constants ULA_PREFIX_START_BYTE = 0xfd @@ -54,102 +53,24 @@ BR_FLAG_DP_FALSE = 0 ICMPV6_TYPE_ROUTER_SOLICITATION = 133 -def get_val(field_values, index): - vals = verify_utils.as_list(field_values) - assert index < len(vals), f"Index {index} is out of bounds for field values with length {len(vals)}" - return vals[index] - - -def check_step3(p, target_prefix, target_prefix_len): - # Check OMR prefix properties: 64 bits long and starts with ULA_PREFIX_START_BYTE - if target_prefix_len != 64 or target_prefix[0] != ULA_PREFIX_START_BYTE: - return False - - try: - types = verify_utils.as_list(p.thread_nwd.tlv.type) - prefixes = verify_utils.as_list(p.thread_nwd.tlv.prefix) - stables = verify_utils.as_list(p.thread_nwd.tlv.stable) - except (AttributeError, IndexError): - return False - - prefix_idx = 0 - br_idx = 0 - is_target = False - - for i, t in enumerate(types): - if t == consts.NWD_PREFIX_TLV: - current_prefix = prefixes[prefix_idx] - current_stable = stables[i] - prefix_idx += 1 - if current_prefix is nullField: - is_target = False - continue - is_target = (Ipv6Addr(current_prefix) == Ipv6Addr(target_prefix)) - if is_target and current_stable != 1: - is_target = False - elif t in (consts.NWD_COMMISSIONING_DATA_TLV, consts.NWD_SERVICE_TLV): - is_target = False - elif t == consts.NWD_BORDER_ROUTER_TLV: - if is_target: - # This BR sub-TLV belongs to our target prefix! - try: - actual_pref = get_val(p.thread_nwd.tlv.border_router.pref, br_idx) - actual_r = get_val(p.thread_nwd.tlv.border_router.flag.r, br_idx) - actual_o = get_val(p.thread_nwd.tlv.border_router.flag.o, br_idx) - actual_p = get_val(p.thread_nwd.tlv.border_router.flag.p, br_idx) - actual_s = get_val(p.thread_nwd.tlv.border_router.flag.s, br_idx) - actual_d = get_val(p.thread_nwd.tlv.border_router.flag.d, br_idx) - actual_dp = get_val(p.thread_nwd.tlv.border_router.flag.dp, br_idx) - - if actual_pref == BR_PREFERENCE_LOW and\ - actual_r == BR_FLAG_R_TRUE and\ - actual_o == BR_FLAG_O_TRUE and\ - actual_p == BR_FLAG_P_TRUE and\ - actual_s == BR_FLAG_S_TRUE and\ - actual_d == BR_FLAG_D_FALSE and\ - actual_dp == BR_FLAG_DP_FALSE: - return True - except (AttributeError, IndexError): - pass - - br_idx += 1 - - return False - - def check_step4(p, omr_prefix, pre_1_prefix, expect_high_pref=True): if p.icmpv6.type != verify_utils.ICMPV6_TYPE_ROUTER_ADVERTISEMENT: return False - # Check PIO (type ICMPV6_OPT_TYPE_PIO) and RIO (type ICMPV6_OPT_TYPE_RIO) - opts = verify_utils.as_list(p.icmpv6.opt.type) - if verify_utils.ICMPV6_OPT_TYPE_RIO not in opts: - return False + _, pio_prefixes = verify_utils.get_ra_prefixes(p) # MUST NOT contain a PIO with a ULA prefix. - if verify_utils.ICMPV6_OPT_TYPE_PIO in opts: - _, pio_prefixes = verify_utils.get_ra_prefixes(p) - for prefix in pio_prefixes: - if prefix[0] == ULA_PREFIX_START_BYTE: - return False - - rio_prefixes, _ = verify_utils.get_ra_prefixes(p) - - # MUST contain a Route Information Option (RIO) with OMR_1. - if omr_prefix not in rio_prefixes: - return False - - # Check preference of OMR_1 RIO - if expect_high_pref: - # P_preference = 01 (High) or 00 (Medium). - # In RA RIO, pref is 2 bits. 01 is High, 00 is Medium, 11 is Low. - rio_idx = rio_prefixes.index(omr_prefix) - actual_pref = get_val(p.icmpv6.opt.route_info.flag.route_preference, rio_idx) - if actual_pref not in (BR_PREFERENCE_HIGH, BR_PREFERENCE_MEDIUM): + for prefix in pio_prefixes: + if prefix[0] == ULA_PREFIX_START_BYTE: return False # MUST contain a Route Information Option (RIO) with PRE_1. - if pre_1_prefix not in rio_prefixes: + if not verify_utils.check_ra_has_rio(p, pre_1_prefix): + return False + + # MUST contain a Route Information Option (RIO) with OMR_1 with correct preference. + prf_values = (BR_PREFERENCE_HIGH, BR_PREFERENCE_MEDIUM) if expect_high_pref else None + if not verify_utils.check_ra_has_rio(p, omr_prefix, prf=prf_values): return False return True @@ -241,9 +162,21 @@ def verify(pv): # - 8. P_dp=false # - OMR_1 MUST be 64 bits long and start with 0xFD. print("Step 3: BR 1 (DUT) registers OMR_1 in Network Data.") + if OMR_1_LEN != 64 or OMR_1[0] != ULA_PREFIX_START_BYTE: + raise verify_utils.VerificationError(f"OMR_1 ({OMR_1}) must be 64 bits long and start with 0xFD") + pkts.filter(lambda p: hasattr(p, 'mle') and\ p.mle.cmd in (consts.MLE_DATA_RESPONSE, consts.MLE_ADVERTISEMENT)).\ - filter(lambda p: check_step3(p, OMR_1, OMR_1_LEN)).\ + filter(lambda p: verify_utils.check_nwd_prefix_flags(p, + OMR_1, + stable=1, + pref=BR_PREFERENCE_LOW, + r=BR_FLAG_R_TRUE, + 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)).\ must_next() # Step 4 diff --git a/tests/nexus/verify_1_3_DBR_TC_7B.py b/tests/nexus/verify_1_3_DBR_TC_7B.py index 42885cd4b..f729da0df 100644 --- a/tests/nexus/verify_1_3_DBR_TC_7B.py +++ b/tests/nexus/verify_1_3_DBR_TC_7B.py @@ -37,7 +37,6 @@ sys.path.append(CUR_DIR) import verify_utils from pktverify import consts from pktverify.addrs import Ipv6Addr -from pktverify.null_field import nullField # Constants ULA_PREFIX_START_BYTE = 0xfd @@ -54,102 +53,24 @@ BR_FLAG_DP_FALSE = 0 ICMPV6_TYPE_ROUTER_SOLICITATION = 133 -def get_val(field_values, index): - vals = verify_utils.as_list(field_values) - assert index < len(vals), f"Index {index} is out of bounds for field values with length {len(vals)}" - return vals[index] - - -def check_step3(p, target_prefix, target_prefix_len): - # Check OMR prefix properties: 64 bits long and starts with ULA_PREFIX_START_BYTE - if target_prefix_len != 64 or target_prefix[0] != ULA_PREFIX_START_BYTE: - return False - - try: - types = verify_utils.as_list(p.thread_nwd.tlv.type) - prefixes = verify_utils.as_list(p.thread_nwd.tlv.prefix) - stables = verify_utils.as_list(p.thread_nwd.tlv.stable) - except (AttributeError, IndexError): - return False - - prefix_idx = 0 - br_idx = 0 - is_target = False - - for i, t in enumerate(types): - if t == consts.NWD_PREFIX_TLV: - current_prefix = prefixes[prefix_idx] - current_stable = stables[i] - prefix_idx += 1 - if current_prefix is nullField: - is_target = False - continue - is_target = (Ipv6Addr(current_prefix) == Ipv6Addr(target_prefix)) - if is_target and current_stable != 1: - is_target = False - elif t in (consts.NWD_COMMISSIONING_DATA_TLV, consts.NWD_SERVICE_TLV): - is_target = False - elif t == consts.NWD_BORDER_ROUTER_TLV: - if is_target: - # This BR sub-TLV belongs to our target prefix! - try: - actual_pref = get_val(p.thread_nwd.tlv.border_router.pref, br_idx) - actual_r = get_val(p.thread_nwd.tlv.border_router.flag.r, br_idx) - actual_o = get_val(p.thread_nwd.tlv.border_router.flag.o, br_idx) - actual_p = get_val(p.thread_nwd.tlv.border_router.flag.p, br_idx) - actual_s = get_val(p.thread_nwd.tlv.border_router.flag.s, br_idx) - actual_d = get_val(p.thread_nwd.tlv.border_router.flag.d, br_idx) - actual_dp = get_val(p.thread_nwd.tlv.border_router.flag.dp, br_idx) - - if actual_pref == BR_PREFERENCE_LOW and\ - actual_r == BR_FLAG_R_TRUE and\ - actual_o == BR_FLAG_O_TRUE and\ - actual_p == BR_FLAG_P_TRUE and\ - actual_s == BR_FLAG_S_TRUE and\ - actual_d == BR_FLAG_D_FALSE and\ - actual_dp == BR_FLAG_DP_FALSE: - return True - except (AttributeError, IndexError): - pass - - br_idx += 1 - - return False - - def check_step4(p, omr_prefix, pre_1_prefix, expect_high_pref=True): if p.icmpv6.type != verify_utils.ICMPV6_TYPE_ROUTER_ADVERTISEMENT: return False - # Check PIO (type ICMPV6_OPT_TYPE_PIO) and RIO (type ICMPV6_OPT_TYPE_RIO) - opts = verify_utils.as_list(p.icmpv6.opt.type) - if verify_utils.ICMPV6_OPT_TYPE_RIO not in opts: - return False + _, pio_prefixes = verify_utils.get_ra_prefixes(p) # MUST NOT contain a PIO with a ULA prefix. - if verify_utils.ICMPV6_OPT_TYPE_PIO in opts: - _, pio_prefixes = verify_utils.get_ra_prefixes(p) - for prefix in pio_prefixes: - if prefix[0] == ULA_PREFIX_START_BYTE: - return False - - rio_prefixes, _ = verify_utils.get_ra_prefixes(p) - - # MUST contain a Route Information Option (RIO) with OMR_1. - if omr_prefix not in rio_prefixes: - return False - - # Check preference of OMR_1 RIO - if expect_high_pref: - # P_preference = 01 (High) or 00 (Medium). - # In RA RIO, pref is 2 bits. 01 is High, 00 is Medium, 11 is Low. - rio_idx = rio_prefixes.index(omr_prefix) - actual_pref = get_val(p.icmpv6.opt.route_info.flag.route_preference, rio_idx) - if actual_pref not in (BR_PREFERENCE_HIGH, BR_PREFERENCE_MEDIUM): + for prefix in pio_prefixes: + if prefix[0] == ULA_PREFIX_START_BYTE: return False # MUST contain a Route Information Option (RIO) with PRE_1. - if pre_1_prefix not in rio_prefixes: + if not verify_utils.check_ra_has_rio(p, pre_1_prefix): + return False + + # MUST contain a Route Information Option (RIO) with OMR_1 with correct preference. + prf_values = (BR_PREFERENCE_HIGH, BR_PREFERENCE_MEDIUM) if expect_high_pref else None + if not verify_utils.check_ra_has_rio(p, omr_prefix, prf=prf_values): return False return True @@ -241,9 +162,21 @@ def verify(pv): # - 16. P_dp = false # - OMR_1 MUST be 64 bits long and start with 0xFD. print("Step 3: BR 1 (DUT) registers OMR_1 in Network Data.") + if OMR_1_LEN != 64 or OMR_1[0] != ULA_PREFIX_START_BYTE: + raise verify_utils.VerificationError(f"OMR_1 ({OMR_1}) must be 64 bits long and start with 0xFD") + pkts.filter(lambda p: hasattr(p, 'mle') and\ p.mle.cmd in (consts.MLE_DATA_RESPONSE, consts.MLE_ADVERTISEMENT)).\ - filter(lambda p: check_step3(p, OMR_1, OMR_1_LEN)).\ + filter(lambda p: verify_utils.check_nwd_prefix_flags(p, + OMR_1, + stable=1, + pref=BR_PREFERENCE_LOW, + r=BR_FLAG_R_TRUE, + 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)).\ must_next() # Step 4 diff --git a/tests/nexus/verify_1_3_DBR_TC_7C.py b/tests/nexus/verify_1_3_DBR_TC_7C.py index 81af228f7..773137ae9 100644 --- a/tests/nexus/verify_1_3_DBR_TC_7C.py +++ b/tests/nexus/verify_1_3_DBR_TC_7C.py @@ -53,86 +53,16 @@ BR_FLAG_DP_FALSE = 0 ICMPV6_TYPE_ROUTER_SOLICITATION = 133 -def get_val(field_values, index): - vals = verify_utils.as_list(field_values) - assert index < len(vals), f"Index {index} is out of bounds for field values with length {len(vals)}" - return vals[index] - - -def check_step3(p, target_prefix, target_prefix_len): - # Check OMR prefix properties: 64 bits long and starts with ULA_PREFIX_START_BYTE - if target_prefix_len != 64 or target_prefix[0] != ULA_PREFIX_START_BYTE: - return False - - try: - types = verify_utils.as_list(p.thread_nwd.tlv.type) - prefixes = verify_utils.as_list(p.thread_nwd.tlv.prefix) - stables = verify_utils.as_list(p.thread_nwd.tlv.stable) - except (AttributeError, IndexError): - return False - - prefix_idx = 0 - br_idx = 0 - is_target = False - - for i, t in enumerate(types): - if t == consts.NWD_PREFIX_TLV: - current_prefix = prefixes[prefix_idx] - current_stable = stables[i] - prefix_idx += 1 - if current_prefix is nullField: - is_target = False - continue - is_target = (Ipv6Addr(current_prefix) == Ipv6Addr(target_prefix)) - if is_target and current_stable != 1: - is_target = False - elif t in (consts.NWD_COMMISSIONING_DATA_TLV, consts.NWD_SERVICE_TLV): - is_target = False - elif t == consts.NWD_BORDER_ROUTER_TLV: - if is_target: - # This BR sub-TLV belongs to our target prefix! - try: - actual_pref = get_val(p.thread_nwd.tlv.border_router.pref, br_idx) - actual_r = get_val(p.thread_nwd.tlv.border_router.flag.r, br_idx) - actual_o = get_val(p.thread_nwd.tlv.border_router.flag.o, br_idx) - actual_p = get_val(p.thread_nwd.tlv.border_router.flag.p, br_idx) - actual_s = get_val(p.thread_nwd.tlv.border_router.flag.s, br_idx) - actual_d = get_val(p.thread_nwd.tlv.border_router.flag.d, br_idx) - actual_dp = get_val(p.thread_nwd.tlv.border_router.flag.dp, br_idx) - - if actual_pref == BR_PREFERENCE_LOW and\ - actual_r == BR_FLAG_R_TRUE and\ - actual_o == BR_FLAG_O_TRUE and\ - actual_p == BR_FLAG_P_TRUE and\ - actual_s == BR_FLAG_S_TRUE and\ - actual_d == BR_FLAG_D_FALSE and\ - actual_dp == BR_FLAG_DP_FALSE: - return True - except (AttributeError, IndexError): - pass - - br_idx += 1 - - return False - - def check_step4(p, omr_prefix, pre_1_prefix): if p.icmpv6.type != verify_utils.ICMPV6_TYPE_ROUTER_ADVERTISEMENT: return False - # Check PIO (type ICMPV6_OPT_TYPE_PIO) and RIO (type ICMPV6_OPT_TYPE_RIO) - opts = verify_utils.as_list(p.icmpv6.opt.type) - if verify_utils.ICMPV6_OPT_TYPE_RIO not in opts: - return False + rio_prefixes, pio_prefixes = verify_utils.get_ra_prefixes(p) # MUST NOT contain a PIO with a ULA prefix. - if verify_utils.ICMPV6_OPT_TYPE_PIO in opts: - _, pio_prefixes = verify_utils.get_ra_prefixes(p) - for prefix in pio_prefixes: - if prefix[0] == ULA_PREFIX_START_BYTE: - return False - - rio_prefixes, _ = verify_utils.get_ra_prefixes(p) + for prefix in pio_prefixes: + if prefix[0] == ULA_PREFIX_START_BYTE: + return False # MUST contain a Route Information Option (RIO) with OMR_1. if omr_prefix not in rio_prefixes: @@ -231,9 +161,21 @@ def verify(pv): # - 24. P_dp = false # - OMR_1 MUST be 64 bits long and start with 0xFD. print("Step 3: BR 1 (DUT) registers OMR_1 in Network Data.") + if OMR_1_LEN != 64 or OMR_1[0] != ULA_PREFIX_START_BYTE: + raise verify_utils.VerificationError(f"OMR_1 ({OMR_1}) must be 64 bits long and start with 0xFD") + pkts.filter(lambda p: hasattr(p, 'mle')).\ filter(lambda p: p.mle.cmd in (consts.MLE_DATA_RESPONSE, consts.MLE_ADVERTISEMENT)).\ - filter(lambda p: check_step3(p, OMR_1, OMR_1_LEN)).\ + filter(lambda p: verify_utils.check_nwd_prefix_flags(p, + OMR_1, + stable=1, + pref=BR_PREFERENCE_LOW, + r=BR_FLAG_R_TRUE, + 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)).\ must_next() # Step 4 diff --git a/tests/nexus/verify_1_3_DBR_TC_8.py b/tests/nexus/verify_1_3_DBR_TC_8.py index 0ec38d10f..0dc38dac2 100644 --- a/tests/nexus/verify_1_3_DBR_TC_8.py +++ b/tests/nexus/verify_1_3_DBR_TC_8.py @@ -37,7 +37,6 @@ sys.path.append(CUR_DIR) import verify_utils from pktverify.addrs import Ipv6Addr from pktverify import consts -from pktverify.null_field import nullField # Protocol Constants ULA_PREFIX_START_BYTE = 0xfd @@ -52,63 +51,6 @@ BR_FLAG_D_FALSE = 0 BR_FLAG_DP_FALSE = 0 -def check_nwd_has_route(packet, prefix): - """Checks if Network Data has an External Route with the given prefix.""" - try: - if not hasattr(packet, 'thread_nwd'): - return False - types = verify_utils.as_list(packet.thread_nwd.tlv.type) - prefixes = verify_utils.as_list(packet.thread_nwd.tlv.prefix) - except (AttributeError, IndexError): - return False - - prefix_idx = 0 - is_target = False - - for t in types: - if t == consts.NWD_PREFIX_TLV: - if prefix_idx < len(prefixes): - current_prefix = prefixes[prefix_idx] - prefix_idx += 1 - if current_prefix: - is_target = (Ipv6Addr(current_prefix) == Ipv6Addr(prefix)) - else: - is_target = False - elif t == consts.NWD_HAS_ROUTER_TLV: - if is_target: - return True - elif t in (consts.NWD_COMMISSIONING_DATA_TLV, consts.NWD_SERVICE_TLV): - is_target = False - - return False - - -def check_ra_rio(packet, prefix, not_prefixes=None): - """Checks if an ICMPv6 RA contains a Route Information Option (RIO) with the given prefix.""" - rio_prefixes, _ = verify_utils.get_ra_prefixes(packet) - target_addr = Ipv6Addr(prefix) - found = any(Ipv6Addr(p) == target_addr for p in rio_prefixes) - if not found: - return False - if not_prefixes: - for np in not_prefixes: - if any(Ipv6Addr(p) == Ipv6Addr(np) for p in rio_prefixes): - return False - return True - - -def check_nwd_contains_prefix(packet, prefix): - """Simple check 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 = verify_utils.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 verify(pv): pkts = pv.pkts @@ -175,7 +117,7 @@ def verify(pv): print("Step 3: BR_1 (DUT) registers route to GUA_1.") pkts.filter_wpan_src64(BR_1).\ filter(lambda p: hasattr(p, 'mle') and p.mle.cmd == consts.MLE_DATA_RESPONSE).\ - filter(lambda p: check_nwd_has_route(p, "::")).\ + filter(lambda p: verify_utils.check_nwd_has_route(p, "::")).\ must_next() # Step 4 @@ -191,7 +133,7 @@ def verify(pv): print("Step 4: BR_1 (DUT) announces OMR_1 on AIL.") pkts.filter_eth_src(pv.vars['BR_1_ETH']).\ filter_ipv6_dst("ff02::1").\ - filter(lambda p: check_ra_rio(p, OMR_1_PREFIX)).\ + filter(lambda p: verify_utils.check_ra_has_rio(p, OMR_1_PREFIX)).\ filter(lambda p: not any(Ipv6Addr(pref)[0] == ULA_PREFIX_START_BYTE for pref in verify_utils.get_ra_prefixes(p)[1])).\ must_next() @@ -254,7 +196,7 @@ def verify(pv): s=BR_FLAG_S_TRUE, d=BR_FLAG_D_FALSE, dp=BR_FLAG_DP_FALSE)).\ - filter(lambda p: check_nwd_has_route(p, "::")).\ + filter(lambda p: verify_utils.check_nwd_has_route(p, "::")).\ must_next() # Step 7 @@ -271,7 +213,8 @@ def verify(pv): print("Step 7: BR_1 (DUT) multicasts ND RA with OMR_2, no OMR_1.") pkts.filter_eth_src(pv.vars['BR_1_ETH']).\ filter_ipv6_dst("ff02::1").\ - filter(lambda p: check_ra_rio(p, OMR_2_PREFIX, not_prefixes=[OMR_1_PREFIX])).\ + filter(lambda p: verify_utils.check_ra_has_rio(p, OMR_2_PREFIX)).\ + filter(lambda p: not verify_utils.check_ra_has_rio(p, OMR_1_PREFIX)).\ must_next() # Step 8 @@ -329,7 +272,7 @@ def verify(pv): # Search from beginning of pcap as it might appear early due to propagation. _pkt_step10 = pkts.copy().filter(lambda p: (hasattr(p, 'thread_nwd') and\ verify_utils.check_nwd_prefix_flags(p, OMR_3_PREFIX)) or\ - check_ra_rio(p, OMR_3_PREFIX)).\ + verify_utils.check_ra_has_rio(p, OMR_3_PREFIX)).\ must_next() # Step 11 @@ -347,7 +290,8 @@ def verify(pv): print("Step 11: BR_1 (DUT) withdraws OMR_2 from Network Data.") # Step 11 must happen after Step 10. pkts.index = (_pkt_step10.number, _pkt_step10.number) - pkts.filter(lambda p: hasattr(p, 'thread_nwd') and not check_nwd_contains_prefix(p, OMR_2_PREFIX)).must_next() + pkts.filter( + lambda p: hasattr(p, 'thread_nwd') and not verify_utils.check_nwd_has_prefix(p, OMR_2_PREFIX)).must_next() # Step 12 # - Device: BR_1 (DUT) @@ -367,8 +311,8 @@ def verify(pv): print("Step 12: BR_1 (DUT) multicasts ND RA with OMR_2 and OMR_3.") pkts.filter_eth_src(pv.vars['BR_1_ETH']).\ filter_ipv6_dst("ff02::1").\ - filter(lambda p: check_ra_rio(p, OMR_2_PREFIX)).\ - filter(lambda p: check_ra_rio(p, OMR_3_PREFIX)).\ + filter(lambda p: verify_utils.check_ra_has_rio(p, OMR_2_PREFIX)).\ + filter(lambda p: verify_utils.check_ra_has_rio(p, OMR_3_PREFIX)).\ must_next() # Step 13 @@ -477,8 +421,8 @@ def verify(pv): print("Step 17: BR_1 (DUT) multicasts ND RA with OMR_2, no OMR_3.") pkts.filter_eth_src(pv.vars['BR_1_ETH']).\ filter_ipv6_dst("ff02::1").\ - filter(lambda p: check_ra_rio(p, OMR_2_PREFIX)).\ - filter(lambda p: not check_ra_rio(p, OMR_3_PREFIX)).\ + filter(lambda p: verify_utils.check_ra_has_rio(p, OMR_2_PREFIX)).\ + filter(lambda p: not verify_utils.check_ra_has_rio(p, OMR_3_PREFIX)).\ must_next() # Step 18 @@ -489,7 +433,7 @@ def verify(pv): # - Pass Criteria: # - N/A print("Step 18: BR_2 adds OMR_4.") - pkts.filter(lambda p: check_nwd_contains_prefix(p, OMR_4_PREFIX)).\ + pkts.filter(lambda p: verify_utils.check_nwd_prefix_flags(p, OMR_4_PREFIX)).\ must_next() # Step 19 @@ -511,8 +455,8 @@ def verify(pv): print("Step 19: BR_1 (DUT) continues OMR_2, adds OMR_4 on AIL.") pkts.filter_eth_src(pv.vars['BR_1_ETH']).\ filter_ipv6_dst("ff02::1").\ - filter(lambda p: check_ra_rio(p, OMR_2_PREFIX)).\ - filter(lambda p: check_ra_rio(p, OMR_4_PREFIX)).\ + filter(lambda p: verify_utils.check_ra_has_rio(p, OMR_2_PREFIX)).\ + filter(lambda p: verify_utils.check_ra_has_rio(p, OMR_4_PREFIX)).\ must_next() # Step 20 @@ -541,7 +485,7 @@ def verify(pv): # - "Prf" bits MUST be 00 (medium) or 11 (low). # - MUST NOT contain a Route Information Option (RIO) with OMR_1 or OMR_3. print("Step 21: BR_1 (DUT) withdraws OMR_2.") - pkts.filter(lambda p: not check_nwd_contains_prefix(p, OMR_2_PREFIX)).\ + pkts.filter(lambda p: hasattr(p, 'thread_nwd') and not verify_utils.check_nwd_has_prefix(p, OMR_2_PREFIX)).\ must_next() # Step 22 @@ -611,8 +555,8 @@ def verify(pv): print("Step 24: BR_1 (DUT) multicasts ND RA with OMR_2 and OMR_4.") pkts.filter_eth_src(pv.vars['BR_1_ETH']).\ filter_ipv6_dst("ff02::1").\ - filter(lambda p: check_ra_rio(p, OMR_2_PREFIX)).\ - filter(lambda p: check_ra_rio(p, OMR_4_PREFIX)).\ + filter(lambda p: verify_utils.check_ra_has_rio(p, OMR_2_PREFIX)).\ + filter(lambda p: verify_utils.check_ra_has_rio(p, OMR_4_PREFIX)).\ must_next() # Step 25 diff --git a/tests/nexus/verify_utils.py b/tests/nexus/verify_utils.py index 3b83d0ae6..0eac8f261 100644 --- a/tests/nexus/verify_utils.py +++ b/tests/nexus/verify_utils.py @@ -291,6 +291,42 @@ def get_ra_prefixes(p): 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. @@ -310,12 +346,17 @@ def check_nwd_prefix_flags(packet, target_prefix, stable=None, **expected_flags) # 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] + 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 - 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 @@ -358,6 +399,57 @@ def check_nwd_prefix_flags(packet, target_prefix, stable=None, **expected_flags) 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)