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

[dua] completely remove DUA features and configurations (#13191)

Browse Source 
This commit removes the OPENTHREAD_CONFIG_TMF_PROXY_DUA_ENABLE feature
and all associated code, tests, CLI commands, and harness references.

Changes:
- Removed OPENTHREAD_CONFIG_TMF_PROXY_DUA_ENABLE definition and all
  assert/preprocessor checks.
- Completely deleted dua_manager.cpp and dua_manager.hpp.
- Removed DUA registration notifying and request URI paths.
- Cleaned up all references to Domain Unicast Address (DUA) across
  child management, notifier, time ticker, and MLE.
- Removed DUA commands and logic from the CLI and Python cert tests
  (including packet verifier).
- Verified that the entire codebase compiles clean and all tests
  successfully pass using the Nexus test suite.
This commit is contained in:
Jonathan Hui
2026-06-03 12:29:18 -07:00
committed by GitHub
parent b69c905763
commit ecd4c92465
36 changed files with 56 additions and 1128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/scripts/thread-cert/config.py
-2
View File
@@ -94,7 +94,6 @@ class ADDRESS_TYPE(Enum):
RLOC = 'RLOC'
ALOC = 'ALOC'
ML_EID = 'ML_EID'
DUA = 'DUA'
BACKBONE_GUA = 'BACKBONE_GUA'
BACKBONE_LINK_LOCAL = 'BACKBONE_LINK_LOCAL'
OMR = 'OMR'
@@ -139,7 +138,6 @@ ADDRESS_QUERY_INITIAL_RETRY_DELAY = 15
DEFAULT_CHILD_TIMEOUT = 6
VIRTUAL_TIME = int(os.getenv('VIRTUAL_TIME', 0))
PARENT_AGGREGATIOIN_DELAY = 5
DUA_DAD_DELAY = 5
DEFAULT_BBR_REGISTRATION_JITTER = 2
DEFAULT_ROUTER_SELECTION_JITTER = 1
tests/scripts/thread-cert/node.py
+1 -36
View File
@@ -1616,32 +1616,6 @@ class NodeImpl:
self.remove_prefix(prefix)
self.register_netdata()
def set_next_dua_response(self, status: Union[str, int], iid=None):
# Convert 5.00 to COAP CODE 160
if isinstance(status, str):
assert '.' in status
status = status.split('.')
status = (int(status[0]) << 5) + int(status[1])
cmd = 'bbr mgmt dua {}'.format(status)
if iid is not None:
cmd += ' ' + str(iid)
self.send_command(cmd)
self._expect_done()
def set_dua_iid(self, iid: str):
assert len(iid) == 16
int(iid, 16)
cmd = 'dua iid {}'.format(iid)
self.send_command(cmd)
self._expect_done()
def clear_dua_iid(self):
cmd = 'dua iid clear'
self.send_command(cmd)
self._expect_done()
def multicast_listener_list(self) -> Dict[IPv6Address, int]:
cmd = 'bbr mgmt mlr listener'
self.send_command(cmd)
@@ -2153,7 +2127,7 @@ class NodeImpl:
omr_addrs = []
for addr in self.get_addrs():
for prefix in prefixes:
if (addr.startswith(prefix)) and (addr != self.__getDua()):
if addr.startswith(prefix):
omr_addrs.append(addr)
break
@@ -2202,13 +2176,6 @@ class NodeImpl:
return None
def __getDua(self) -> Optional[str]:
for ip6Addr in self.get_addrs():
if re.match(config.DOMAIN_PREFIX_REGEX_PATTERN, ip6Addr, re.I):
return ip6Addr
return None
def get_ip6_address_by_prefix(self, prefix: Union[str, IPv6Network]) -> List[IPv6Address]:
"""Get addresses matched with given prefix.
@@ -2244,8 +2211,6 @@ class NodeImpl:
return self.__getAloc()
elif address_type == config.ADDRESS_TYPE.ML_EID:
return self.__getMleid()
elif address_type == config.ADDRESS_TYPE.DUA:
return self.__getDua()
elif address_type == config.ADDRESS_TYPE.BACKBONE_GUA:
return self._getBackboneGua()
elif address_type == config.ADDRESS_TYPE.OMR:
tests/scripts/thread-cert/pktverify/addrs.py
-8
View File
@@ -109,14 +109,6 @@ class Ipv6Addr(Bytes):
return True
@property
def is_dua(self) -> bool:
"""
Returns if the Ip6 address is Domain Unicast Address.
"""
from pktverify import consts
return self.startswith(consts.DOMAIN_PREFIX)
@property
def is_backbone_gua(self) -> bool:
"""
tests/scripts/thread-cert/pktverify/consts.py
-9
View File
@@ -30,7 +30,6 @@
from pktverify.addrs import Ipv6Addr
from pktverify.bytes import Bytes
DOMAIN_PREFIX = Bytes('fd00:7d03:7d03:7d03')
BACKBONE_IPV6_PREFIX = Bytes('91')
LINK_LOCAL_ALL_THREAD_NODES_MULTICAST_ADDRESS = Ipv6Addr('ff32:40:fd00:db8::1')
@@ -279,8 +278,6 @@ NWD_SERVER_TLV = 6
LM_FORWARD_PROBING_REGISTRATION_SUB_TLV = 3
LM_ENHANCED_ACK_CONFIGURATION_SUB_TLV = 7
# DUA related constants
ADDRESS_QUERY_INITIAL_RETRY_DELAY = 15
ADDRESS_QUERY_MAX_RETRY_DELAY = 8
ADDRESS_QUERY_TIMEOUT = 3
@@ -291,10 +288,6 @@ CONTEXT_ID_REUSE_DELAY = 48
DATA_RESUBMIT_DELAY = 300
DUA_DAD_PERIOD = 100
DUA_DAD_QUERY_TIMEOUT = 1.0
DUA_DAD_REPEATS = 2
DUA_RECENT_TIME = 20
FAILED_ROUTER_TRANSMISSIONS = 4
ID_REUSE_DELAY = 100
ID_SEQUENCE_PERIOD = 10
@@ -377,8 +370,6 @@ WIRESHARK_DECODE_AS_ENTRIES = {
}
TIMEOUT_JOIN_NETWORK = 10
TIMEOUT_DUA_REGISTRATION = 10
TIMEOUT_DUA_DAD = 15
TIMEOUT_HOST_READY = 10
TIMEOUT_CHILD_DETACH = 120
TIMEOUT_REGISTER_MA = 5
tests/scripts/thread-cert/pktverify/packet_verifier.py
+1 -115
View File
@@ -125,9 +125,7 @@ class PacketVerifier(object):
name = self.test_info.get_node_name(i)
self._vars[name + '_IPADDRS'] = addrs
for addr in addrs:
if addr.is_dua:
key = name + '_DUA'
elif addr.is_backbone_gua:
if addr.is_backbone_gua:
key = name + '_BGUA'
elif addr.is_link_local and (name + '_BGUA') in self._vars:
# FIXME: assume the link-local address after Backbone GUA is the Backbone Link Local address
@@ -161,10 +159,6 @@ class PacketVerifier(object):
key = self.test_info.get_node_name(i) + '_OMR'
self._vars[key] = omr
for i, dua in self.test_info.duas.items():
key = self.test_info.get_node_name(i) + '_DUA'
self._vars[key] = dua
if self.test_info.leader_aloc:
self._vars['LEADER_ALOC'] = self.test_info.leader_aloc
@@ -216,72 +210,6 @@ class PacketVerifier(object):
return result
def verify_ping(self, src: str, dst: str, bbr: str = None, pkts: 'PacketVerifier' = None) -> VerifyResult:
"""
Verify the ping process.
:param src: The source device name.
:param dst: The destination device name.
:param bbr: The Backbone Router name.
If specified, this method also verifies that the ping request and reply be forwarded by the Backbone Router.
:param pkts: The PacketFilter to search.
:return: The verification result.
"""
if bbr:
assert not (self.is_thread_device(src) and self.is_thread_device(dst)), \
f"both {src} and {dst} are WPAN devices"
assert not (self.is_backbone_device(src) and self.is_backbone_device(dst)), \
f"both {src} and {dst} are ETH devices"
if pkts is None:
pkts = self.pkts
src_dua = self.vars[src + '_DUA']
dst_dua = self.vars[dst + '_DUA']
if bbr:
bbr_ext = self.vars[bbr]
bbr_eth = self.vars[bbr + '_ETH']
result = VerifyResult()
ping_req = pkts.filter_ping_request().filter_ipv6_dst(dst_dua)
if self.is_backbone_device(src):
p = ping_req.filter_eth_src(self.vars[src + '_ETH']).must_next()
else:
p = ping_req.filter_wpan_src64(self.vars[src]).must_next()
# pkts.last().show()
ping_id = p.icmpv6.echo.identifier
logging.info("verify_ping: ping_id=%x", ping_id)
result.record_last('ping_request', pkts)
ping_req = ping_req.filter(lambda p: p.icmpv6.echo.identifier == ping_id)
# BBR unicasts the ping packet to TD.
if bbr:
if self.is_backbone_device(src):
ping_req.filter_wpan_src64(bbr_ext).must_next()
else:
ping_req.filter_eth_src(bbr_eth).must_next()
ping_reply = pkts.filter_ping_reply().filter_ipv6_dst(src_dua).filter(
lambda p: p.icmpv6.echo.identifier == ping_id)
# TD receives ping packet and responds back to Host via SBBR.
if self.is_thread_device(dst):
ping_reply.filter_wpan_src64(self.vars[dst]).must_next()
else:
ping_reply.filter_eth_src(self.vars[dst + '_ETH']).must_next()
result.record_last('ping_reply', pkts)
if bbr:
# SBBR forwards the ping response packet to Host.
if self.is_thread_device(dst):
ping_reply.filter_eth_src(bbr_eth).must_next()
else:
ping_reply.filter_wpan_src64(bbr_ext).must_next()
return result
def is_thread_device(self, name: str) -> bool:
"""
Returns if the device is an WPAN device.
@@ -314,45 +242,3 @@ class PacketVerifier(object):
eth_idx = max(eth_idx, ei)
return wpan_idx, eth_idx
def verify_dua_registration(self, src64, dua, *, pbbr_eth, sbbr_eth=None, pbbr_src64=None):
pv, pkts = self, self.pkts
MM = pv.vars['MM_PORT']
BB = pv.vars['BB_PORT']
# Router1 should send /n/dr for DUA registration
dr = pkts.filter_wpan_src64(src64).filter_coap_request('/n/dr', port=MM).filter(
'thread_nm.tlv.target_eid == {ROUTER1_DUA}', ROUTER1_DUA=dua).must_next()
# SBBR should not send /b/bq for Router1's DUA
if sbbr_eth is not None:
pkts.filter_backbone_query(dua, eth_src=sbbr_eth, port=BB).must_not_next()
# PBBR should respond to /n/dr
if pbbr_src64 is not None:
pkts.filter_wpan_src64(pbbr_src64).filter_coap_ack(
'/n/dr', port=MM).must_next().must_verify('thread_nm.tlv.status == 0')
# PBBR should send /b/bq for Router1's DUA (1st time)
bq1 = pkts.filter_backbone_query(dua, eth_src=pbbr_eth, port=BB).must_next()
bq1_index = pkts.index
assert bq1.sniff_timestamp - dr.sniff_timestamp <= 1.01, bq1.sniff_timestamp - dr.sniff_timestamp
# PBBR should send /b/bq for Router1's DUA (2nd time)
bq2 = pkts.filter_backbone_query(dua, eth_src=pbbr_eth, port=BB).must_next()
assert 0.9 < bq2.sniff_timestamp - bq1.sniff_timestamp < 1.1, bq2.sniff_timestamp - bq1.sniff_timestamp
# PBBR should send /b/bq for Router1's DUA (3rd time)
bq3 = pkts.filter_backbone_query(dua, eth_src=pbbr_eth, port=BB).must_next()
assert 0.9 < bq3.sniff_timestamp - bq2.sniff_timestamp < 1.1, bq3.sniff_timestamp - bq2.sniff_timestamp
# PBBR should send PRO_BB.ntf for Router's DUA when DAD completed
pkts.filter_eth_src(pbbr_eth).filter_backbone_answer(dua, port=BB, confirmable=False).must_next().show()
# PBBR should not recv /b/ba response from other BBRs during this period
pkts.range(bq1_index, pkts.index,
cascade=False).filter('eth.src != {PBBR_ETH}',
PBBR_ETH=pbbr_eth).filter_backbone_answer(dua, port=BB).must_not_next()
tests/scripts/thread-cert/pktverify/test_info.py
-1
View File
@@ -54,7 +54,6 @@ class TestInfo(object):
self.rlocs = {int(k): Ipv6Addr(v) for k, v in test_info.get('rlocs', {}).items()}
self.rloc16s = self._convert_hex_values(self._convert_keys_to_ints(test_info.get('rloc16s', {})))
self.omrs = {int(k): [Ipv6Addr(x) for x in l] for k, l in test_info.get('omrs', {}).items()}
self.duas = {int(k): Ipv6Addr(v) for k, v in test_info.get('duas', {}).items()}
self.extra_vars = test_info.get('extra_vars', {})
self.leader_aloc = Ipv6Addr(test_info.get('leader_aloc')) if 'leader_aloc' in test_info else ''
tests/scripts/thread-cert/thread_cert.py
-13
View File
@@ -410,19 +410,6 @@ class TestCase(NcpSupportMixin, unittest.TestCase):
test_info['omrs'][i] = node.get_ip6_address(config.ADDRESS_TYPE.OMR)
def collect_duas(self):
if not self._do_packet_verification:
return
test_info = self._test_info
test_info['duas'] = {}
for i, node in self.nodes.items():
if node.is_host:
continue
test_info['duas'][i] = node.get_ip6_address(config.ADDRESS_TYPE.DUA)
def collect_leader_aloc(self, node):
if not self._do_packet_verification:
return