espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-08-12 05:37:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

[nexus] add helper methods to verify_utils.py and update tests (#12756)

Browse Source 
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.
This commit is contained in:
Jonathan Hui
2026-03-25 00:49:52 -05:00
committed by GitHub
parent 81d49bf906
commit 84bce0f587
5 changed files with 176 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/nexus/verify_1_3_DBR_TC_7A.py
+22 -89
View File
@@ -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
tests/nexus/verify_1_3_DBR_TC_7B.py
+22 -89
View File
@@ -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
tests/nexus/verify_1_3_DBR_TC_7C.py
+17 -75
View File
@@ -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
tests/nexus/verify_1_3_DBR_TC_8.py
+18 -74
View File
@@ -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
tests/nexus/verify_utils.py
+97 -5
View File
@@ -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)