mirror of
https://github.com/espressif/openthread.git
synced 2026-09-29 20:37:40 +00:00
Jonathan Hui
ecd4c92465
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.
245 lines
9.6 KiB
Python
245 lines
9.6 KiB
Python
#!/usr/bin/env python3
|
|
#
|
|
# Copyright (c) 2019, The OpenThread Authors.
|
|
# All rights reserved.
|
|
#
|
|
# Redistribution and use in source and binary forms, with or without
|
|
# modification, are permitted provided that the following conditions are met:
|
|
# 1. Redistributions of source code must retain the above copyright
|
|
# notice, this list of conditions and the following disclaimer.
|
|
# 2. Redistributions in binary form must reproduce the above copyright
|
|
# notice, this list of conditions and the following disclaimer in the
|
|
# documentation and/or other materials provided with the distribution.
|
|
# 3. Neither the name of the copyright holder nor the
|
|
# names of its contributors may be used to endorse or promote products
|
|
# derived from this software without specific prior written permission.
|
|
#
|
|
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
|
|
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
|
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
|
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
|
|
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
|
|
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
|
|
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
|
|
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
|
|
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
|
|
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
|
# POSSIBILITY OF SUCH DAMAGE.
|
|
#
|
|
import logging
|
|
from typing import Tuple
|
|
|
|
from pktverify import consts
|
|
from pktverify.consts import MLE_CHILD_ID_REQUEST, MLE_ADVERTISEMENT, MLE_CHILD_ID_RESPONSE
|
|
from pktverify.pcap_reader import PcapReader
|
|
from pktverify.summary import Summary
|
|
from pktverify.test_info import TestInfo
|
|
from pktverify.verify_result import VerifyResult
|
|
|
|
|
|
class PacketVerifier(object):
|
|
"""
|
|
Base class for packet verifiers that runs the packet verification process
|
|
"""
|
|
NET_NAME = "OpenThread"
|
|
MC_PORT = 49191
|
|
MM_PORT = 61631
|
|
BB_PORT = 61631
|
|
LLANMA = 'ff02::1' # Link-Local All Nodes multicast address
|
|
LLARMA = 'ff02::2' # Link-Local All Routers multicast address
|
|
RLANMA = 'ff03::1' # realm-local all-nodes multicast address
|
|
RLARMA = 'ff03::2' # realm-local all-routers multicast address
|
|
RLAMFMA = 'ff03::fc' # realm-local ALL_MPL_FORWARDERS address
|
|
LLABMA = 'ff32:40:fd00:7d03:7d03:7d03:0:3' # Link-Local All BBRs multicast address
|
|
|
|
def __init__(self, test_info_path, wireshark_prefs=None):
|
|
logging.basicConfig(level=logging.INFO,
|
|
format='File "%(pathname)s", line %(lineno)d, in %(funcName)s\n'
|
|
'%(asctime)s - %(levelname)s - %(message)s')
|
|
|
|
ti = TestInfo(test_info_path)
|
|
if wireshark_prefs is not None:
|
|
pkts = PcapReader.read(ti.pcap_path, wireshark_prefs)
|
|
else:
|
|
pkts = PcapReader.read(ti.pcap_path)
|
|
print('loaded %d packets from %s' % (len(pkts), ti.pcap_path))
|
|
self.pkts = pkts
|
|
self.test_info = ti
|
|
|
|
self.summary = Summary(pkts, ti)
|
|
self._vars = {}
|
|
self._add_initial_vars()
|
|
|
|
def add_vars(self, **vars):
|
|
"""
|
|
Add new variables.
|
|
|
|
:param vars: The new variables.
|
|
"""
|
|
self._vars.update(vars)
|
|
|
|
@property
|
|
def vars(self):
|
|
"""
|
|
:return: the dict of all variables
|
|
"""
|
|
return self._vars
|
|
|
|
def add_common_vars(self):
|
|
"""
|
|
Add common variables that is needed by many test cases.
|
|
"""
|
|
self.add_vars(
|
|
NET_NAME=PacketVerifier.NET_NAME,
|
|
MM_PORT=PacketVerifier.MM_PORT,
|
|
MC_PORT=PacketVerifier.MC_PORT,
|
|
BB_PORT=PacketVerifier.BB_PORT,
|
|
LLANMA=PacketVerifier.LLANMA, # Link-Local All Nodes multicast address
|
|
LLARMA=PacketVerifier.LLARMA, # Link-Local All Routers multicast address
|
|
RLANMA=PacketVerifier.RLANMA, # realm-local all-nodes multicast address
|
|
RLARMA=PacketVerifier.RLARMA, # realm-local all-routers multicast address
|
|
RLAMFMA=PacketVerifier.RLAMFMA, # realm-local ALL_MPL_FORWARDERS address
|
|
LLABMA=PacketVerifier.LLABMA, # Link-Local All BBRs multicast address
|
|
MA1=consts.MA1,
|
|
MA2=consts.MA2,
|
|
MA3=consts.MA3,
|
|
MA4=consts.MA4,
|
|
MA5=consts.MA5,
|
|
MA6=consts.MA6,
|
|
MA1g=consts.MA1g,
|
|
MAe1=consts.MAe1,
|
|
MAe2=consts.MAe2,
|
|
MAe3=consts.MAe3,
|
|
)
|
|
|
|
def _add_initial_vars(self):
|
|
for i, addr in self.test_info.extaddrs.items():
|
|
name = self.test_info.get_node_name(i)
|
|
self._vars[name] = addr
|
|
|
|
for i, addr in self.test_info.ethaddrs.items():
|
|
name = self.test_info.get_node_name(i) + '_ETH'
|
|
self._vars[name] = addr
|
|
|
|
for i, addrs in self.test_info.ipaddrs.items():
|
|
name = self.test_info.get_node_name(i)
|
|
self._vars[name + '_IPADDRS'] = addrs
|
|
for addr in addrs:
|
|
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
|
|
key = name + '_BLLA'
|
|
elif addr.is_link_local:
|
|
key = name + '_LLA'
|
|
else:
|
|
logging.warning("IPv6 address ignored: name=%s, addr=%s, is_global=%s, is_link_local=%s", name,
|
|
addr, addr.is_global, addr.is_link_local)
|
|
continue
|
|
|
|
if key in self._vars:
|
|
logging.warning("duplicate IPv6 address type: name=%s, addr=%s,%s", name, addr, self._vars[key])
|
|
continue
|
|
|
|
self._vars[key] = addr
|
|
|
|
for i, addr in self.test_info.mleids.items():
|
|
name = self.test_info.get_node_name(i)
|
|
self._vars[name + '_MLEID'] = addr
|
|
|
|
for i, rloc16 in self.test_info.rloc16s.items():
|
|
key = self.test_info.get_node_name(i) + '_RLOC16'
|
|
self._vars[key] = rloc16
|
|
|
|
for i, rloc in self.test_info.rlocs.items():
|
|
key = self.test_info.get_node_name(i) + '_RLOC'
|
|
self._vars[key] = rloc
|
|
|
|
for i, omr in self.test_info.omrs.items():
|
|
key = self.test_info.get_node_name(i) + '_OMR'
|
|
self._vars[key] = omr
|
|
|
|
if self.test_info.leader_aloc:
|
|
self._vars['LEADER_ALOC'] = self.test_info.leader_aloc
|
|
|
|
for k, v in self.test_info.extra_vars.items():
|
|
assert k not in self._vars, k
|
|
logging.info("add extra var: %s = %s", k, v)
|
|
self._vars[k] = v
|
|
|
|
for i, topo in self.test_info.topology.items():
|
|
name = self.test_info.get_node_name(i)
|
|
if topo['version']:
|
|
self._vars[name + '_VERSION'] = {'1.1': 2, '1.2': 3, '1.3': 4, '1.4': 5}[topo['version']]
|
|
|
|
def verify_attached(self, child: str, parent: str = None, child_type: str = 'FTD', pkts=None) -> VerifyResult:
|
|
"""
|
|
Verify that the device attaches to the Thread network.
|
|
|
|
:param child: The child device name.
|
|
:param parent: The parent device name.
|
|
:param child_type: The child device type (FTD, FTD-ED, MTD).
|
|
"""
|
|
result = VerifyResult()
|
|
assert self.is_thread_device(child), child
|
|
assert child_type in ('FTD', 'FTD-ED', 'MTD'), child_type
|
|
pkts = pkts or self.pkts
|
|
child_extaddr = self.vars[child]
|
|
|
|
src_pkts = pkts.filter_wpan_src64(child_extaddr)
|
|
if parent:
|
|
assert self.is_thread_device(parent), parent
|
|
src_pkts = pkts.filter_wpan_src64(child_extaddr).\
|
|
filter_wpan_dst64(self.vars[parent])
|
|
src_pkts.filter_mle_cmd(MLE_CHILD_ID_REQUEST).must_next() # Child Id Request
|
|
result.record_last('child_id_request', pkts)
|
|
|
|
dst_pkts = pkts.filter_wpan_dst64(child_extaddr)
|
|
if parent:
|
|
dst_pkts = pkts.filter_wpan_src64(self.vars[parent]).\
|
|
filter_wpan_dst64(child_extaddr)
|
|
dst_pkts.filter_mle_cmd(MLE_CHILD_ID_RESPONSE).must_next() # Child Id Response
|
|
result.record_last('child_id_response', pkts)
|
|
|
|
with pkts.save_index():
|
|
if child_type == 'FTD':
|
|
src_pkts = pkts.filter_wpan_src64(child_extaddr)
|
|
src_pkts.filter_mle_cmd(MLE_ADVERTISEMENT).must_next() # MLE Advertisement
|
|
result.record_last('mle_advertisement', pkts)
|
|
logging.info(f"verify attached: d={child}, result={result}")
|
|
|
|
return result
|
|
|
|
def is_thread_device(self, name: str) -> bool:
|
|
"""
|
|
Returns if the device is an WPAN device.
|
|
|
|
:param name: The device name.
|
|
|
|
Note that device can be both a WPAN device and an Ethernet device.
|
|
"""
|
|
assert isinstance(name, str), name
|
|
|
|
return name in self.vars
|
|
|
|
def is_backbone_device(self, name: str) -> bool:
|
|
"""
|
|
Returns if the device s an Ethernet device.
|
|
|
|
:param name: The device name.
|
|
|
|
Note that device can be both a WPAN device and an Ethernet device.
|
|
"""
|
|
assert isinstance(name, str), name
|
|
|
|
return f'{name}_ETH' in self.vars
|
|
|
|
def max_index(self, *indexes: Tuple[int, int]) -> Tuple[int, int]:
|
|
wpan_idx = 0
|
|
eth_idx = 0
|
|
for wi, ei in indexes:
|
|
wpan_idx = max(wpan_idx, wi)
|
|
eth_idx = max(eth_idx, ei)
|
|
|
|
return wpan_idx, eth_idx
|