mirror of
https://github.com/espressif/openthread.git
synced 2026-06-05 21:14:49 +00:00
[tests] remove DUA and ND proxy backbone tests (#13135)
Remove obsolete backbone test cases for Domain Unicast Address (DUA) Duplicate Address Detection (DAD), DUA routing, DUA routing for Minimal End Devices (MED), and Neighbor Discovery (ND) Proxy. These features and their corresponding tests are no longer needed.
This commit is contained in:
@@ -1,312 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
#
|
||||
# Copyright (c) 2020, 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.
|
||||
#
|
||||
# This test verifies that PBBR sets DUA routes correctly.
|
||||
#
|
||||
import ipaddress
|
||||
import re
|
||||
import unittest
|
||||
|
||||
import config
|
||||
import thread_cert
|
||||
from pktverify.packet_verifier import PacketVerifier
|
||||
|
||||
# Use two channels
|
||||
CH1 = 11
|
||||
CH2 = 22
|
||||
|
||||
PBBR = 1
|
||||
SBBR = 2
|
||||
ROUTER1 = 3
|
||||
HOST = 4
|
||||
PBBR2 = 5
|
||||
ROUTER2 = 6
|
||||
|
||||
REREG_DELAY = 5 # Seconds
|
||||
MLR_TIMEOUT = 300 # Seconds
|
||||
WAIT_REDUNDANCE = 3
|
||||
|
||||
DUPLICATE_DUA = 'fd00:7d03:7d03:7d03:11:2233:4455:6677'
|
||||
assert (re.match(config.DOMAIN_PREFIX_REGEX_PATTERN, DUPLICATE_DUA))
|
||||
|
||||
DUPLICATE_IID = ''.join('%02x' % c for c in ipaddress.IPv6Address(DUPLICATE_DUA).packed[-8:])
|
||||
|
||||
print('DUPLICATE_DUA: ', DUPLICATE_DUA)
|
||||
print('DUPLICATE_IID: ', DUPLICATE_IID)
|
||||
|
||||
|
||||
class TestDuaDad(thread_cert.TestCase):
|
||||
USE_MESSAGE_FACTORY = False
|
||||
|
||||
# Topology:
|
||||
# ------(eth)----------------------
|
||||
# | | | |
|
||||
# PBBR----SBBR HOST PBBR2
|
||||
# \ CH1 / | CH2
|
||||
# ROUTER1 ROUTER2
|
||||
#
|
||||
# PBBR2 is in the secondary channel
|
||||
#
|
||||
TOPOLOGY = {
|
||||
PBBR: {
|
||||
'name': 'PBBR',
|
||||
'allowlist': [SBBR, ROUTER1],
|
||||
'is_otbr': True,
|
||||
'version': '1.2',
|
||||
'channel': CH1,
|
||||
'prefer_router_id': 0x1, # Use prefer_router_id to avoid Router ID conflicts in the two channels.
|
||||
},
|
||||
SBBR: {
|
||||
'name': 'SBBR',
|
||||
'allowlist': [PBBR, ROUTER1],
|
||||
'is_otbr': True,
|
||||
'version': '1.2',
|
||||
'channel': CH1,
|
||||
'prefer_router_id': 0x2,
|
||||
},
|
||||
ROUTER1: {
|
||||
'name': 'ROUTER1',
|
||||
'allowlist': [PBBR, SBBR],
|
||||
'version': '1.2',
|
||||
'channel': CH1,
|
||||
'prefer_router_id': 0x3,
|
||||
},
|
||||
HOST: {
|
||||
'name': 'HOST',
|
||||
'is_host': True
|
||||
},
|
||||
PBBR2: {
|
||||
'name': 'PBBR2',
|
||||
'is_otbr': True,
|
||||
'version': '1.2',
|
||||
'channel': CH2,
|
||||
'prefer_router_id': 0x4,
|
||||
},
|
||||
ROUTER2: {
|
||||
'name': 'ROUTER2',
|
||||
'version': '1.2',
|
||||
'channel': CH2,
|
||||
'prefer_router_id': 0x5,
|
||||
},
|
||||
}
|
||||
|
||||
def _bootstrap(self):
|
||||
# Bring up HOST
|
||||
self.nodes[HOST].start()
|
||||
|
||||
# Bring up PBBR
|
||||
self.nodes[PBBR].start()
|
||||
self.simulator.go(config.LEADER_STARTUP_DELAY)
|
||||
self.assertEqual('leader', self.nodes[PBBR].get_state())
|
||||
self.wait_node_state(PBBR, 'leader', 10)
|
||||
|
||||
self.nodes[PBBR].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT)
|
||||
self.nodes[PBBR].enable_backbone_router()
|
||||
self.nodes[PBBR].set_domain_prefix(config.DOMAIN_PREFIX, 'prosD')
|
||||
self.simulator.go(5)
|
||||
self.assertTrue(self.nodes[PBBR].is_primary_backbone_router)
|
||||
self.assertIsNotNone(self.nodes[PBBR].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up SBBR
|
||||
self.nodes[SBBR].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('router', self.nodes[SBBR].get_state())
|
||||
|
||||
self.nodes[SBBR].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT)
|
||||
self.nodes[SBBR].enable_backbone_router()
|
||||
self.simulator.go(5)
|
||||
self.assertFalse(self.nodes[SBBR].is_primary_backbone_router)
|
||||
self.assertIsNotNone(self.nodes[SBBR].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up ROUTER1
|
||||
self.nodes[ROUTER1].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('router', self.nodes[ROUTER1].get_state())
|
||||
self.simulator.go(config.DUA_DAD_DELAY + WAIT_REDUNDANCE)
|
||||
self.assertIsNotNone(self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up PBBR2
|
||||
self.nodes[PBBR2].start()
|
||||
self.simulator.go(config.LEADER_STARTUP_DELAY)
|
||||
self.assertEqual('leader', self.nodes[PBBR2].get_state())
|
||||
self.wait_node_state(PBBR2, 'leader', 10)
|
||||
|
||||
self.nodes[PBBR2].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT)
|
||||
self.nodes[PBBR2].enable_backbone_router()
|
||||
self.nodes[PBBR2].set_domain_prefix(config.DOMAIN_PREFIX, 'prosD')
|
||||
self.simulator.go(config.DUA_DAD_DELAY)
|
||||
self.assertTrue(self.nodes[PBBR2].is_primary_backbone_router)
|
||||
self.assertIsNotNone(self.nodes[PBBR2].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up ROUTER2
|
||||
self.nodes[ROUTER2].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('router', self.nodes[ROUTER2].get_state())
|
||||
self.simulator.go(config.DUA_DAD_DELAY + WAIT_REDUNDANCE)
|
||||
self.assertIsNotNone(self.nodes[ROUTER2].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
def test(self):
|
||||
self._bootstrap()
|
||||
|
||||
self.collect_ipaddrs()
|
||||
self.collect_rloc16s()
|
||||
self.collect_rlocs()
|
||||
|
||||
self._test_extend_backbone_query()
|
||||
|
||||
self._test_dad_duplicate()
|
||||
|
||||
self._send_fake_pro_bb_ntf()
|
||||
|
||||
def _test_extend_backbone_query(self):
|
||||
self.nodes[ROUTER2].ping(self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
def _test_dad_duplicate(self):
|
||||
self.nodes[ROUTER2].set_dua_iid(DUPLICATE_IID)
|
||||
self.simulator.go(config.DUA_DAD_DELAY + WAIT_REDUNDANCE)
|
||||
self.nodes[ROUTER1].set_dua_iid(DUPLICATE_IID)
|
||||
self.simulator.go(config.DUA_DAD_DELAY + WAIT_REDUNDANCE)
|
||||
|
||||
# now Router1 should have generated new DUA
|
||||
self.assertNotEqual(DUPLICATE_DUA, self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
def _send_fake_pro_bb_ntf(self):
|
||||
router1_dua = self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA)
|
||||
self.nodes[PBBR2].send_proactive_backbone_notification(router1_dua, DUPLICATE_IID, 0)
|
||||
self.simulator.go(config.DUA_DAD_DELAY + WAIT_REDUNDANCE)
|
||||
|
||||
def verify(self, pv: PacketVerifier):
|
||||
pkts = pv.pkts
|
||||
pv.add_common_vars()
|
||||
pv.summary.show()
|
||||
|
||||
PBBR = pv.vars['PBBR']
|
||||
ROUTER1 = pv.vars['ROUTER1']
|
||||
ROUTER1_DUA = pv.vars['ROUTER1_DUA']
|
||||
ROUTER2 = pv.vars['ROUTER2']
|
||||
ROUTER2_DUA = pv.vars['ROUTER2_DUA']
|
||||
PBBR_ETH = pv.vars['PBBR_ETH']
|
||||
SBBR_ETH = pv.vars['SBBR_ETH']
|
||||
PBBR2 = pv.vars['PBBR2']
|
||||
PBBR2_ETH = pv.vars['PBBR2_ETH']
|
||||
|
||||
MM = pv.vars['MM_PORT']
|
||||
BB = pv.vars['BB_PORT']
|
||||
|
||||
# Verify the whole DAD process for ROUTER1
|
||||
pv.verify_dua_registration(ROUTER1, ROUTER1_DUA, pbbr_eth=PBBR_ETH, pbbr_src64=PBBR, sbbr_eth=SBBR_ETH)
|
||||
|
||||
# Verify the whole DAD process for ROUTER2
|
||||
pv.verify_dua_registration(ROUTER2, ROUTER2_DUA, pbbr_eth=PBBR2_ETH, pbbr_src64=PBBR2)
|
||||
|
||||
###############################################################################################################
|
||||
# Now we verify Extending ADDR.qry to BB.qry works for the Ping Request from Router2 to Router1
|
||||
###############################################################################################################
|
||||
|
||||
# Router2 should send ADDR.qry in the Thread network for Router1's DUA
|
||||
pkts.filter_wpan_src64(ROUTER2).filter_coap_request('/a/aq', port=MM).filter(
|
||||
'thread_address.tlv.target_eid == {ROUTER1_DUA}', ROUTER1_DUA=ROUTER1_DUA).must_next()
|
||||
# PBBR2 should extend ADDR.qry to BB.qry
|
||||
pkts.filter_backbone_query(ROUTER1_DUA, eth_src=PBBR2_ETH, port=BB).must_next()
|
||||
# SBBR should not answer with BB.ans
|
||||
pkts.filter_backbone_query(ROUTER1_DUA, eth_src=SBBR_ETH, port=BB).must_not_next()
|
||||
# PBBR1 should answer with BB.ans
|
||||
pkts.filter_backbone_answer(ROUTER1_DUA, eth_src=PBBR_ETH, port=BB).must_next()
|
||||
# PBBR2 should send ADDR.ntf to Router2
|
||||
pkts.filter_wpan_src64(PBBR2).filter_coap_request('/a/an', port=MM).filter(
|
||||
'thread_address.tlv.target_eid == {ROUTER1_DUA}', ROUTER1_DUA=ROUTER1_DUA).must_next()
|
||||
|
||||
# Now, Router2 should send the Ping Request to PBBR2
|
||||
pkts.filter_wpan_src64(ROUTER2).filter_ping_request().filter_ipv6_dst(ROUTER1_DUA).must_next()
|
||||
|
||||
###############################################################################################################
|
||||
# Now we start to verify that DAD duplicate is handled correctly
|
||||
###############################################################################################################
|
||||
|
||||
# PBBR should send /b/bq for DUPLICATE_DUA
|
||||
pkts.filter_backbone_query(DUPLICATE_DUA, eth_src=PBBR_ETH, port=BB).must_next()
|
||||
|
||||
# PBBR2 should send /b/ba for DUPLICATE_DUA
|
||||
ba = pkts.filter_backbone_answer(DUPLICATE_DUA, eth_src=PBBR2_ETH, port=BB).must_next()
|
||||
ba.must_verify("""
|
||||
set(thread_bl.tlv.type) == {tlvs}
|
||||
and thread_bl.tlv.last_transaction_time > 0
|
||||
and thread_meshcop.tlv.net_name == [{NET_NAME}]
|
||||
""",
|
||||
tlvs={0, 3, 6, 12},
|
||||
**pv.vars)
|
||||
|
||||
# PBBR should NOT send /b/bq for DUPLICATE_DUA anymore
|
||||
pkts.filter_backbone_query(DUPLICATE_DUA, eth_src=PBBR_ETH, port=BB).must_not_next()
|
||||
|
||||
# PBBR should send ADDR_ERR.ntf to Router1
|
||||
ROUTER1_RLOC = pv.vars['ROUTER1_RLOC']
|
||||
pkts.filter_wpan_src64(PBBR).filter_ipv6_dst(ROUTER1_RLOC).filter_coap_request('/a/ae', port=MM).must_next() \
|
||||
.must_verify("""
|
||||
thread_address.tlv.target_eid == {DUPLICATE_DUA}
|
||||
and thread_address.tlv.ml_eid == {ml_eid}
|
||||
""", DUPLICATE_DUA=DUPLICATE_DUA, ml_eid=ba.thread_bl.tlv.ml_eid)
|
||||
|
||||
# Router1 should generate new DUA and register again
|
||||
# Router1 should generate and register a new DUA
|
||||
with pkts.save_index():
|
||||
new_dr = pkts.filter_wpan_src64(ROUTER1).filter_coap_request('/n/dr', port=MM).must_next().must_verify(
|
||||
"""
|
||||
thread_nm.tlv.target_eid != {DUPLICATE_DUA}
|
||||
""",
|
||||
DUPLICATE_DUA=DUPLICATE_DUA)
|
||||
router1_dua2 = new_dr.thread_nm.tlv.target_eid
|
||||
|
||||
# Verify the DAD process for the new DUA
|
||||
pv.verify_dua_registration(ROUTER1, router1_dua2, pbbr_eth=PBBR_ETH, pbbr_src64=PBBR, sbbr_eth=SBBR_ETH)
|
||||
|
||||
###############################################################################################################
|
||||
# Now we start to verify that PRO_BB.ntf is handled correctly
|
||||
###############################################################################################################
|
||||
|
||||
# Scripted: PBBR2 should send PRO_BB.ntf for Router1's DUA
|
||||
pkts.filter_eth_src(PBBR2_ETH).filter_backbone_answer(router1_dua2,
|
||||
port=BB,
|
||||
confirmable=False,
|
||||
mliid=DUPLICATE_IID).must_next()
|
||||
|
||||
# PBBR should broadcaset /a/ae to the Thread network
|
||||
pkts.filter_wpan_src64(PBBR).filter_coap_request('/a/ae', port=MM,
|
||||
confirmable=False).filter_RLARMA().must_next()
|
||||
|
||||
# Router1 should generate and register a new DUA
|
||||
with pkts.save_index():
|
||||
new_dr = pkts.filter_wpan_src64(ROUTER1).filter_coap_request('/n/dr', port=MM).must_next()
|
||||
router1_dua3 = new_dr.thread_nm.tlv.target_eid
|
||||
|
||||
# Verify the DAD process for the new DUA
|
||||
pv.verify_dua_registration(ROUTER1, router1_dua3, pbbr_eth=PBBR_ETH, pbbr_src64=PBBR, sbbr_eth=SBBR_ETH)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -1,249 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
#
|
||||
# Copyright (c) 2020, 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.
|
||||
#
|
||||
# This test verifies that PBBR sets DUA routes correctly.
|
||||
#
|
||||
import ipaddress
|
||||
import unittest
|
||||
|
||||
import config
|
||||
import thread_cert
|
||||
from pktverify.packet_verifier import PacketVerifier
|
||||
|
||||
# Use two channels
|
||||
CH1 = 11
|
||||
CH2 = 22
|
||||
|
||||
PBBR = 1
|
||||
SBBR = 2
|
||||
ROUTER1 = 3
|
||||
HOST = 4
|
||||
PBBR2 = 5
|
||||
ROUTER2 = 6
|
||||
|
||||
REREG_DELAY = 5 # Seconds
|
||||
MLR_TIMEOUT = 300 # Seconds
|
||||
WAIT_REDUNDANCE = 3
|
||||
|
||||
|
||||
class TestNdProxy(thread_cert.TestCase):
|
||||
USE_MESSAGE_FACTORY = False
|
||||
|
||||
# Topology:
|
||||
# ------(eth)----------------------
|
||||
# | | | |
|
||||
# PBBR----SBBR HOST PBBR2
|
||||
# \ CH1 / | CH2
|
||||
# ROUTER1 ROUTER2
|
||||
#
|
||||
# PBBR2 is in the secondary channel
|
||||
#
|
||||
TOPOLOGY = {
|
||||
PBBR: {
|
||||
'name': 'PBBR',
|
||||
'allowlist': [SBBR, ROUTER1],
|
||||
'is_otbr': True,
|
||||
'version': '1.2',
|
||||
'channel': CH1,
|
||||
'router_id_range': [0, 30],
|
||||
},
|
||||
SBBR: {
|
||||
'name': 'SBBR',
|
||||
'allowlist': [PBBR, ROUTER1],
|
||||
'is_otbr': True,
|
||||
'version': '1.2',
|
||||
'channel': CH1,
|
||||
},
|
||||
ROUTER1: {
|
||||
'name': 'ROUTER1',
|
||||
'allowlist': [PBBR, SBBR],
|
||||
'version': '1.2',
|
||||
'channel': CH1,
|
||||
},
|
||||
HOST: {
|
||||
'name': 'HOST',
|
||||
'is_host': True
|
||||
},
|
||||
PBBR2: {
|
||||
'name': 'PBBR2',
|
||||
'is_otbr': True,
|
||||
'version': '1.2',
|
||||
'channel': CH2,
|
||||
'router_id_range': [31, 60],
|
||||
},
|
||||
ROUTER2: {
|
||||
'name': 'ROUTER2',
|
||||
'version': '1.2',
|
||||
'channel': CH2,
|
||||
},
|
||||
}
|
||||
|
||||
def _bootstrap(self):
|
||||
# Bring up HOST
|
||||
self.nodes[HOST].start()
|
||||
|
||||
# Bring up PBBR
|
||||
self.nodes[PBBR].start()
|
||||
self.simulator.go(config.LEADER_STARTUP_DELAY)
|
||||
self.assertEqual('leader', self.nodes[PBBR].get_state())
|
||||
self.wait_node_state(PBBR, 'leader', 10)
|
||||
|
||||
self.nodes[PBBR].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT)
|
||||
self.nodes[PBBR].enable_backbone_router()
|
||||
self.nodes[PBBR].set_domain_prefix(config.DOMAIN_PREFIX, 'prosD')
|
||||
self.simulator.go(5)
|
||||
self.assertTrue(self.nodes[PBBR].is_primary_backbone_router)
|
||||
self.assertIsNotNone(self.nodes[PBBR].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up SBBR
|
||||
self.nodes[SBBR].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('router', self.nodes[SBBR].get_state())
|
||||
|
||||
self.nodes[SBBR].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT)
|
||||
self.nodes[SBBR].enable_backbone_router()
|
||||
self.simulator.go(5)
|
||||
self.assertFalse(self.nodes[SBBR].is_primary_backbone_router)
|
||||
self.assertIsNotNone(self.nodes[SBBR].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up ROUTER1
|
||||
self.nodes[ROUTER1].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('router', self.nodes[ROUTER1].get_state())
|
||||
self.assertIsNotNone(self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up PBBR2
|
||||
self.nodes[PBBR2].start()
|
||||
self.simulator.go(config.LEADER_STARTUP_DELAY)
|
||||
self.assertEqual('leader', self.nodes[PBBR2].get_state())
|
||||
self.wait_node_state(PBBR2, 'leader', 10)
|
||||
|
||||
self.nodes[PBBR2].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT)
|
||||
self.nodes[PBBR2].enable_backbone_router()
|
||||
self.nodes[PBBR2].set_domain_prefix(config.DOMAIN_PREFIX, 'prosD')
|
||||
self.simulator.go(5)
|
||||
self.assertTrue(self.nodes[PBBR2].is_primary_backbone_router)
|
||||
self.assertIsNotNone(self.nodes[PBBR2].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up ROUTER2
|
||||
self.nodes[ROUTER2].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('router', self.nodes[ROUTER2].get_state())
|
||||
self.assertIsNotNone(self.nodes[ROUTER2].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
def test(self):
|
||||
self._bootstrap()
|
||||
|
||||
self.collect_ipaddrs()
|
||||
self.collect_rloc16s()
|
||||
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
# Step 1: Host pings PBBR's DUA
|
||||
self.assertTrue(self.nodes[HOST].ping(self.nodes[PBBR].get_ip6_address(config.ADDRESS_TYPE.DUA),
|
||||
backbone=True))
|
||||
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
# Step 2: Host pings SBBR's DUA
|
||||
self.assertTrue(self.nodes[HOST].ping(self.nodes[SBBR].get_ip6_address(config.ADDRESS_TYPE.DUA),
|
||||
backbone=True))
|
||||
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
# Step 3: Host pings ROUTER1's DUA
|
||||
self.assertTrue(self.nodes[HOST].ping(self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA),
|
||||
backbone=True))
|
||||
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
# Step 4: ROUTER2 pings ROUTER1 which should succeed
|
||||
ROUTER1_DUA = self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA)
|
||||
self.assertTrue(self.nodes[ROUTER2].ping(ROUTER1_DUA))
|
||||
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
def _get_mliid(self, nodeid: int) -> str:
|
||||
mleid = self.nodes[nodeid].get_ip6_address(config.ADDRESS_TYPE.ML_EID)
|
||||
mliid = ipaddress.IPv6Address(mleid).packed[-8:]
|
||||
return ''.join(['%02x' % b for b in mliid])
|
||||
|
||||
def verify(self, pv: PacketVerifier):
|
||||
pkts = pv.pkts
|
||||
pv.add_common_vars()
|
||||
pv.summary.show()
|
||||
|
||||
PBBR = pv.vars['PBBR']
|
||||
PBBR2 = pv.vars['PBBR2']
|
||||
ROUTER1 = pv.vars['ROUTER1']
|
||||
ROUTER2 = pv.vars['ROUTER2']
|
||||
SBBR = pv.vars['SBBR']
|
||||
PBBR_DUA = pv.vars['PBBR_DUA']
|
||||
SBBR_DUA = pv.vars['SBBR_DUA']
|
||||
ROUTER1_DUA = pv.vars['ROUTER1_DUA']
|
||||
ROUTER2_DUA = pv.vars['ROUTER2_DUA']
|
||||
HOST_BGUA = pv.vars['HOST_BGUA']
|
||||
PBBR_ETH = pv.vars['PBBR_ETH']
|
||||
PBBR2_ETH = pv.vars['PBBR2_ETH']
|
||||
|
||||
# Verify that SBBR should not forward any Ping Request to the Thread network.
|
||||
# Use `ipv6.hlim == 63` to avoid false fails because SBBR might still forward Ping Request from PBBR to ROUTER1
|
||||
pkts.filter_wpan_src64(SBBR).filter_ping_request().filter('ipv6.hlim == 63').must_not_next()
|
||||
|
||||
# Step 1: Host pings PBBR's DUA
|
||||
# PBBR should not forward the Ping Request to Thread network
|
||||
ping1_pkts = pkts.filter_ipv6_src_dst(HOST_BGUA, PBBR_DUA).filter_ping_request()
|
||||
ping1_pkts.filter_eth().must_next()
|
||||
ping1_pkts.filter_wpan().must_not_next()
|
||||
|
||||
# Step 2: Host pings SBBR's DUA
|
||||
# PBBR should forward the Ping Request to Thread network according to DUA routes
|
||||
ping2_pkts = pkts.filter_ipv6_src_dst(HOST_BGUA, SBBR_DUA).filter_ping_request()
|
||||
ping2_pkts.filter_eth().must_next()
|
||||
ping2_pkts.filter_wpan().must_next()
|
||||
|
||||
# Step 3: Host pings ROUTER1's DUA
|
||||
# PBBR should forward the Ping Request to Thread network according to DUA routes
|
||||
ping3_pkts = pkts.filter_ipv6_src_dst(HOST_BGUA, ROUTER1_DUA).filter_ping_request()
|
||||
ping3_pkts.filter_eth().must_next()
|
||||
ping3_pkts.filter_wpan().must_next()
|
||||
|
||||
# Step 4: ROUTER2 pings ROUTER1
|
||||
# Verify Ping Request: ROUTER2 -> PBBR2 -> PBBR -> ROUTER1
|
||||
ping4_request_pkts = pkts.filter_ipv6_src_dst(ROUTER2_DUA, ROUTER1_DUA)
|
||||
ping4_id = ping4_request_pkts.filter_wpan_src64(
|
||||
ROUTER2).filter_ping_request().must_next().icmpv6.echo.identifier
|
||||
ping4_request_pkts.filter_eth_src(PBBR2_ETH).filter_ping_request(identifier=ping4_id).must_next()
|
||||
ping4_request_pkts.filter_wpan_src64(PBBR).filter_ping_request(identifier=ping4_id).must_next()
|
||||
# Verify Ping Reply: ROUTER1 -> PBBR -> PBBR2 -> ROUTER2
|
||||
ping4_reply_pkts = pkts.filter_ipv6_src_dst(ROUTER1_DUA, ROUTER2_DUA).filter_ping_reply(identifier=ping4_id)
|
||||
ping4_reply_pkts.filter_eth_src(PBBR_ETH).must_next()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -1,275 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
#
|
||||
# Copyright (c) 2021, 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.
|
||||
#
|
||||
# This test verifies that PBBR sets DUA routes correctly.
|
||||
#
|
||||
import ipaddress
|
||||
import unittest
|
||||
|
||||
import config
|
||||
import thread_cert
|
||||
from pktverify.consts import ADDR_QRY_URI, BACKBONE_QUERY_URI, BACKBONE_ANSWER_URI, ADDR_NTF_URI
|
||||
from pktverify.packet_verifier import PacketVerifier
|
||||
|
||||
# Use two channels
|
||||
CH1 = 11
|
||||
CH2 = 22
|
||||
|
||||
PBBR = 1
|
||||
ROUTER1 = 2
|
||||
MED1 = 3
|
||||
HOST = 4
|
||||
PBBR2 = 5
|
||||
MED2 = 6
|
||||
|
||||
REREG_DELAY = 5 # Seconds
|
||||
MLR_TIMEOUT = 300 # Seconds
|
||||
WAIT_REDUNDANCE = 3
|
||||
|
||||
|
||||
class TestDuaRoutingMed(thread_cert.TestCase):
|
||||
USE_MESSAGE_FACTORY = False
|
||||
|
||||
# Topology:
|
||||
# ------(eth)----------------------
|
||||
# | | |
|
||||
# PBBR HOST PBBR2
|
||||
# / CH1\ | CH2
|
||||
# MED1 ROUTER1 MED2
|
||||
#
|
||||
# PBBR2 is in the secondary channel
|
||||
#
|
||||
TOPOLOGY = {
|
||||
PBBR: {
|
||||
'name': 'PBBR',
|
||||
'allowlist': [MED1, ROUTER1],
|
||||
'is_otbr': True,
|
||||
'version': '1.2',
|
||||
'channel': CH1,
|
||||
},
|
||||
ROUTER1: {
|
||||
'name': 'ROUTER1',
|
||||
'allowlist': [PBBR],
|
||||
'version': '1.2',
|
||||
'channel': CH1,
|
||||
},
|
||||
MED1: {
|
||||
'name': 'MED1',
|
||||
'allowlist': [PBBR],
|
||||
'version': '1.2',
|
||||
'channel': CH1,
|
||||
'mode': 'rn',
|
||||
},
|
||||
HOST: {
|
||||
'name': 'HOST',
|
||||
'is_host': True
|
||||
},
|
||||
PBBR2: {
|
||||
'name': 'PBBR2',
|
||||
'is_otbr': True,
|
||||
'version': '1.2',
|
||||
'channel': CH2,
|
||||
},
|
||||
MED2: {
|
||||
'name': 'MED2',
|
||||
'version': '1.2',
|
||||
'channel': CH2,
|
||||
'mode': 'rn',
|
||||
},
|
||||
}
|
||||
|
||||
def _bootstrap(self):
|
||||
# Bring up HOST
|
||||
self.nodes[HOST].start()
|
||||
|
||||
# Bring up PBBR
|
||||
self.nodes[PBBR].start()
|
||||
self.simulator.go(config.LEADER_STARTUP_DELAY)
|
||||
self.assertEqual('leader', self.nodes[PBBR].get_state())
|
||||
self.wait_node_state(PBBR, 'leader', 10)
|
||||
|
||||
self.nodes[PBBR].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT)
|
||||
self.nodes[PBBR].enable_backbone_router()
|
||||
self.nodes[PBBR].set_domain_prefix(config.DOMAIN_PREFIX, 'prosD')
|
||||
self.simulator.go(5)
|
||||
self.assertTrue(self.nodes[PBBR].is_primary_backbone_router)
|
||||
self.assertIsNotNone(self.nodes[PBBR].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up ROUTER1
|
||||
self.nodes[ROUTER1].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('router', self.nodes[ROUTER1].get_state())
|
||||
self.assertIsNotNone(self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up MED1
|
||||
self.nodes[MED1].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('child', self.nodes[MED1].get_state())
|
||||
self.assertIsNotNone(self.nodes[MED1].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up PBBR2
|
||||
self.nodes[PBBR2].start()
|
||||
self.simulator.go(config.LEADER_STARTUP_DELAY)
|
||||
self.assertEqual('leader', self.nodes[PBBR2].get_state())
|
||||
self.wait_node_state(PBBR2, 'leader', 10)
|
||||
|
||||
self.nodes[PBBR2].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT)
|
||||
self.nodes[PBBR2].enable_backbone_router()
|
||||
self.nodes[PBBR2].set_domain_prefix(config.DOMAIN_PREFIX, 'prosD')
|
||||
self.simulator.go(5)
|
||||
self.assertTrue(self.nodes[PBBR2].is_primary_backbone_router)
|
||||
self.assertIsNotNone(self.nodes[PBBR2].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up MED2
|
||||
self.nodes[MED2].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('child', self.nodes[MED2].get_state())
|
||||
self.assertIsNotNone(self.nodes[MED2].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
def test(self):
|
||||
self._bootstrap()
|
||||
|
||||
self.collect_ipaddrs()
|
||||
self.collect_rloc16s()
|
||||
|
||||
self.simulator.go(10)
|
||||
|
||||
# PBBR2 pings MED2
|
||||
self.assertTrue(self.nodes[PBBR2].ping(self.nodes[MED2].get_ip6_address(config.ADDRESS_TYPE.DUA)))
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
# MED2 pings PBBR2
|
||||
self.assertTrue(self.nodes[MED2].ping(self.nodes[PBBR2].get_ip6_address(config.ADDRESS_TYPE.DUA)))
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
# PBBR pings PBBR2
|
||||
self.nodes[PBBR].ping(self.nodes[PBBR2].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
# MED1 pings ROUTER1 should succeed
|
||||
self.assertTrue(self.nodes[MED1].ping(self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA)))
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
# MED1 pings MED2 which should succeed
|
||||
MED2_DUA = self.nodes[MED2].get_ip6_address(config.ADDRESS_TYPE.DUA)
|
||||
self.assertTrue(self.nodes[MED1].ping(MED2_DUA))
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
def _get_mliid(self, nodeid: int) -> str:
|
||||
mleid = self.nodes[nodeid].get_ip6_address(config.ADDRESS_TYPE.ML_EID)
|
||||
mliid = ipaddress.IPv6Address(mleid).packed[-8:]
|
||||
return ''.join(['%02x' % b for b in mliid])
|
||||
|
||||
def verify(self, pv: PacketVerifier):
|
||||
pkts = pv.pkts
|
||||
pv.add_common_vars()
|
||||
pv.summary.show()
|
||||
|
||||
PBBR = pv.vars['PBBR']
|
||||
ROUTER1 = pv.vars['ROUTER1']
|
||||
PBBR2 = pv.vars['PBBR2']
|
||||
MED1 = pv.vars['MED1']
|
||||
MED2 = pv.vars['MED2']
|
||||
MED1_DUA = pv.vars['MED1_DUA']
|
||||
ROUTER1_DUA = pv.vars['ROUTER1_DUA']
|
||||
PBBR_DUA = pv.vars['PBBR_DUA']
|
||||
PBBR2_DUA = pv.vars['PBBR2_DUA']
|
||||
MED2_DUA = pv.vars['MED2_DUA']
|
||||
PBBR_ETH = pv.vars['PBBR_ETH']
|
||||
PBBR2_ETH = pv.vars['PBBR2_ETH']
|
||||
|
||||
# PBBR should never send Address Query for MED1 (Child)
|
||||
pkts.filter_wpan_src64(PBBR).filter_RLARMA().filter_coap_request(ADDR_QRY_URI) \
|
||||
.filter('thread_address.tlv.target_eid == {eid}', eid=MED1_DUA) \
|
||||
.must_not_next()
|
||||
|
||||
# PBBR2 should never send Address Query for MED2 (Child)
|
||||
pkts.filter_wpan_src64(PBBR2).filter_RLARMA().filter_coap_request(ADDR_QRY_URI) \
|
||||
.filter('thread_address.tlv.target_eid == {eid}', eid=MED2_DUA) \
|
||||
.must_not_next()
|
||||
|
||||
# MEDs should never send Address Query
|
||||
pkts.filter_wpan_src64(MED1).filter_RLARMA().filter_coap_request(ADDR_QRY_URI).must_not_next()
|
||||
pkts.filter_wpan_src64(MED2).filter_RLARMA().filter_coap_request(ADDR_QRY_URI).must_not_next()
|
||||
|
||||
# PBBR pings PBBR2 should succeed
|
||||
ping_id = pkts.filter_ipv6_src_dst(
|
||||
PBBR_DUA, PBBR2_DUA).filter_eth_src(PBBR_ETH).filter_ping_request().must_next().icmpv6.echo.identifier
|
||||
pkts.filter_ipv6_src_dst(PBBR2_DUA,
|
||||
PBBR_DUA).filter_eth_src(PBBR2_ETH).filter_ping_reply(identifier=ping_id).must_next()
|
||||
|
||||
# MED1 pings ROUTER1
|
||||
ping_request_pkts = pkts.filter_ipv6_src_dst(MED1_DUA, ROUTER1_DUA)
|
||||
# MED1 sends the Ping Request
|
||||
ping_pkt = ping_request_pkts.filter_wpan_src64(MED1).filter_ping_request().must_next()
|
||||
ping_id = ping_pkt.icmpv6.echo.identifier
|
||||
# PBBR sends Address Query for ROUTER1_DUA
|
||||
pkts.filter_wpan_src64(PBBR).filter_RLARMA().filter_coap_request(ADDR_QRY_URI) \
|
||||
.filter('thread_address.tlv.target_eid == {eid}', eid=ROUTER1_DUA) \
|
||||
.must_next()
|
||||
# PBBR sends Backbone Query for ROUTER1_DUA
|
||||
pkts.filter_eth_src(PBBR_ETH).filter_coap_request(BACKBONE_QUERY_URI) \
|
||||
.filter('thread_bl.tlv.target_eid == {eid}', eid=ROUTER1_DUA) \
|
||||
.must_next()
|
||||
# ROUTER1 sends Address Answer for ROUTER1_DUA
|
||||
pkts.filter_wpan_src64(ROUTER1).filter_coap_request(ADDR_NTF_URI, confirmable=True) \
|
||||
.filter('thread_address.tlv.target_eid == {eid}', eid=ROUTER1_DUA) \
|
||||
.must_next()
|
||||
# PBBR forwards Ping Request to ROUTER1, decreasing TTL by 1
|
||||
ping_request_pkts.filter_wpan_src64(PBBR).filter_ping_request(identifier=ping_id).must_next().must_verify(
|
||||
'ipv6.hlim == {hlim}', hlim=ping_pkt.ipv6.hlim - 1)
|
||||
|
||||
# MED1 pings MED2
|
||||
# Verify Ping Request: MED1 -> PBBR -> PBBR2 -> MED2
|
||||
ping_request_pkts = pkts.filter_ipv6_src_dst(MED1_DUA, MED2_DUA)
|
||||
# MED1 sends the Ping Request
|
||||
ping_pkt = ping_request_pkts.filter_wpan_src64(MED1).filter_ping_request().must_next()
|
||||
ping_id = ping_pkt.icmpv6.echo.identifier
|
||||
# PBBR sends Address Query for MED2_DUA
|
||||
pkts.filter_wpan_src64(PBBR).filter_RLARMA().filter_coap_request(ADDR_QRY_URI) \
|
||||
.filter('thread_address.tlv.target_eid == {eid}', eid=MED2_DUA) \
|
||||
.must_next()
|
||||
# PBBR sends Backbone Query for MED2_DUA
|
||||
pkts.filter_eth_src(PBBR_ETH).filter_coap_request(BACKBONE_QUERY_URI) \
|
||||
.filter('thread_bl.tlv.target_eid == {eid}', eid=MED2_DUA) \
|
||||
.must_next()
|
||||
# PBBR2 sends Backbone Answer for MED2_DUA
|
||||
pkts.filter_eth_src(PBBR2_ETH).filter_coap_request(BACKBONE_ANSWER_URI, confirmable=True) \
|
||||
.filter('thread_bl.tlv.target_eid == {eid}', eid=MED2_DUA) \
|
||||
.must_next()
|
||||
# PBBR forwards Ping Request to Backbone link, decreasing TTL by 1
|
||||
ping_request_pkts.filter_eth_src(PBBR_ETH).filter_ping_request(identifier=ping_id).must_next().must_verify(
|
||||
'ipv6.hlim == {hlim}', hlim=ping_pkt.ipv6.hlim - 1)
|
||||
ping_request_pkts.filter_wpan_src64(PBBR2).filter_ping_request(identifier=ping_id).must_next()
|
||||
# Verify Ping Reply: MED2 -> PBBR2 -> PBBR -> MED1
|
||||
ping_reply_pkts = pkts.filter_ipv6_src_dst(MED2_DUA, MED1_DUA).filter_ping_reply(identifier=ping_id)
|
||||
ping_reply_pkts.filter_eth_src(PBBR2_ETH).must_next()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -1,222 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
#
|
||||
# Copyright (c) 2020, 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.
|
||||
#
|
||||
# This test verifies PBBR's ND Proxy feature:
|
||||
# 1. Sends ICMPv6 Neighbor Advertisement (NA) for ND Proxy
|
||||
# 2. Receive and handle ICMPv6 Neighbor Solicitation (NS) and sends NA accordingly.
|
||||
#
|
||||
import ipaddress
|
||||
import unittest
|
||||
|
||||
import config
|
||||
import thread_cert
|
||||
from pktverify.packet_verifier import PacketVerifier
|
||||
from pktverify.consts import DUA_RECENT_TIME
|
||||
|
||||
PBBR = 1
|
||||
SBBR = 2
|
||||
ROUTER1 = 3
|
||||
HOST = 4
|
||||
|
||||
REREG_DELAY = 5 # Seconds
|
||||
MLR_TIMEOUT = 300 # Seconds
|
||||
WAIT_REDUNDANCE = 3
|
||||
|
||||
|
||||
class TestNdProxy(thread_cert.TestCase):
|
||||
USE_MESSAGE_FACTORY = False
|
||||
|
||||
# Topology:
|
||||
# ---- -(eth)------------
|
||||
# | | |
|
||||
# PBBR----SBBR HOST
|
||||
# \ /
|
||||
# Router1
|
||||
#
|
||||
TOPOLOGY = {
|
||||
PBBR: {
|
||||
'name': 'PBBR',
|
||||
'allowlist': [SBBR, ROUTER1],
|
||||
'is_otbr': True,
|
||||
'version': '1.2',
|
||||
},
|
||||
SBBR: {
|
||||
'name': 'SBBR',
|
||||
'allowlist': [PBBR, ROUTER1],
|
||||
'is_otbr': True,
|
||||
'version': '1.2',
|
||||
},
|
||||
ROUTER1: {
|
||||
'name': 'ROUTER1',
|
||||
'allowlist': [PBBR, SBBR],
|
||||
'version': '1.2',
|
||||
},
|
||||
HOST: {
|
||||
'name': 'HOST',
|
||||
'is_host': True
|
||||
},
|
||||
}
|
||||
|
||||
def test(self):
|
||||
# Bring up HOST
|
||||
self.nodes[HOST].start()
|
||||
|
||||
# Bring up PBBR
|
||||
self.nodes[PBBR].start()
|
||||
self.simulator.go(config.LEADER_STARTUP_DELAY)
|
||||
self.assertEqual('leader', self.nodes[PBBR].get_state())
|
||||
self.wait_node_state(PBBR, 'leader', 10)
|
||||
|
||||
self.nodes[PBBR].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT)
|
||||
self.nodes[PBBR].enable_backbone_router()
|
||||
self.nodes[PBBR].set_domain_prefix(config.DOMAIN_PREFIX, 'prosD')
|
||||
self.simulator.go(5)
|
||||
self.assertTrue(self.nodes[PBBR].is_primary_backbone_router)
|
||||
self.assertIsNotNone(self.nodes[PBBR].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up SBBR
|
||||
self.nodes[SBBR].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('router', self.nodes[SBBR].get_state())
|
||||
|
||||
self.nodes[SBBR].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT)
|
||||
self.nodes[SBBR].enable_backbone_router()
|
||||
self.simulator.go(5)
|
||||
self.assertFalse(self.nodes[SBBR].is_primary_backbone_router)
|
||||
self.assertIsNotNone(self.nodes[SBBR].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
# Bring up ROUTER1
|
||||
self.nodes[ROUTER1].start()
|
||||
self.simulator.go(5)
|
||||
self.assertEqual('router', self.nodes[ROUTER1].get_state())
|
||||
self.assertIsNotNone(self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA))
|
||||
|
||||
self.collect_ipaddrs()
|
||||
self.collect_rloc16s()
|
||||
|
||||
# Wait for DUA_RECENT_TIME
|
||||
self.simulator.go(DUA_RECENT_TIME + WAIT_REDUNDANCE)
|
||||
|
||||
self._test_neighbor_solicitation(PBBR)
|
||||
self._test_neighbor_solicitation(SBBR)
|
||||
self._test_neighbor_solicitation(ROUTER1)
|
||||
|
||||
def _test_neighbor_solicitation(self, nodeid):
|
||||
dua = self.nodes[nodeid].get_ip6_address(config.ADDRESS_TYPE.DUA)
|
||||
print('Node DUA: ', dua)
|
||||
self.nodes[HOST].ip_neighbors_flush()
|
||||
self.nodes[HOST].ping(dua, backbone=True)
|
||||
self.simulator.go(WAIT_REDUNDANCE)
|
||||
|
||||
def verify(self, pv: PacketVerifier):
|
||||
pkts = pv.pkts
|
||||
pv.add_common_vars()
|
||||
pv.summary.show()
|
||||
|
||||
PBBR_DUA = pv.vars['PBBR_DUA']
|
||||
SBBR_DUA = pv.vars['SBBR_DUA']
|
||||
ROUTER1_DUA = pv.vars['ROUTER1_DUA']
|
||||
PBBR_ETH = pv.vars['PBBR_ETH']
|
||||
SBBR_ETH = pv.vars['SBBR_ETH']
|
||||
HOST_ETH = pv.vars['HOST_ETH']
|
||||
HOST_BGUA = pv.vars['HOST_BGUA']
|
||||
|
||||
# SBBR must not send NA for any DUA
|
||||
pkts.filter_eth_src(SBBR_ETH).filter_LLANMA().filter_icmpv6_nd_na(PBBR_DUA).must_not_next()
|
||||
pkts.filter_eth_src(SBBR_ETH).filter_LLANMA().filter_icmpv6_nd_na(SBBR_DUA).must_not_next()
|
||||
pkts.filter_eth_src(SBBR_ETH).filter_LLANMA().filter_icmpv6_nd_na(ROUTER1_DUA).must_not_next()
|
||||
|
||||
# PBBR must send unsolicited NA for PBBR's DUA
|
||||
pkts.filter_eth_src(PBBR_ETH).filter_LLANMA().filter_icmpv6_nd_na(PBBR_DUA).filter(
|
||||
'icmpv6.nd.na.flag.s == 0').must_next().must_verify("""
|
||||
icmpv6.opt.linkaddr == {PBBR_ETH}
|
||||
and icmpv6.nd.na.flag.r == 1
|
||||
and icmpv6.nd.na.flag.o == 1
|
||||
and icmpv6.nd.na.flag.rsv == 0
|
||||
""",
|
||||
PBBR_ETH=PBBR_ETH)
|
||||
|
||||
# PBBR must send unsolicited NA for SBBR's DUA
|
||||
pkts.filter_eth_src(PBBR_ETH).filter_LLANMA().filter_icmpv6_nd_na(SBBR_DUA).filter(
|
||||
'icmpv6.nd.na.flag.s == 0').must_next().must_verify("""
|
||||
icmpv6.opt.linkaddr == {PBBR_ETH}
|
||||
and icmpv6.nd.na.flag.r == 1
|
||||
and icmpv6.nd.na.flag.o == 1
|
||||
and icmpv6.nd.na.flag.rsv == 0
|
||||
""",
|
||||
PBBR_ETH=PBBR_ETH)
|
||||
|
||||
# PBBR must send unsolicited NA for ROUTER1's DUA
|
||||
pkts.filter_eth_src(PBBR_ETH).filter_LLANMA().filter_icmpv6_nd_na(ROUTER1_DUA).filter(
|
||||
'icmpv6.nd.na.flag.s == 0').must_next().must_verify("""
|
||||
icmpv6.opt.linkaddr == {PBBR_ETH}
|
||||
and icmpv6.nd.na.flag.r == 1
|
||||
and icmpv6.nd.na.flag.o == 1
|
||||
and icmpv6.nd.na.flag.rsv == 0
|
||||
""",
|
||||
PBBR_ETH=PBBR_ETH)
|
||||
|
||||
# HOST should send NS for PBBR's DUA
|
||||
pkts.filter_eth_src(HOST_ETH).filter_icmpv6_nd_ns(PBBR_DUA).must_next()
|
||||
# PBBR must send solicited NA for PBBR's DUA
|
||||
pkts.filter_eth_src(PBBR_ETH).filter_ipv6_dst(HOST_BGUA).filter_icmpv6_nd_na(PBBR_DUA).filter(
|
||||
'icmpv6.nd.na.flag.s == 1').must_next().must_verify("""
|
||||
icmpv6.opt.linkaddr == {PBBR_ETH}
|
||||
and icmpv6.nd.na.flag.r == 1
|
||||
and icmpv6.nd.na.flag.o == 0
|
||||
and icmpv6.nd.na.flag.rsv == 0
|
||||
""",
|
||||
PBBR_ETH=PBBR_ETH)
|
||||
|
||||
# Host should send NS for SBBR's DUA
|
||||
pkts.filter_eth_src(HOST_ETH).filter_icmpv6_nd_ns(SBBR_DUA).must_next()
|
||||
# PBBR must send solicited NA for SBBR's DUA
|
||||
pkts.filter_eth_src(PBBR_ETH).filter_ipv6_dst(HOST_BGUA).filter_icmpv6_nd_na(SBBR_DUA).filter(
|
||||
'icmpv6.nd.na.flag.s == 1').must_next().must_verify("""
|
||||
icmpv6.opt.linkaddr == {PBBR_ETH}
|
||||
and icmpv6.nd.na.flag.r == 1
|
||||
and icmpv6.nd.na.flag.o == 0
|
||||
and icmpv6.nd.na.flag.rsv == 0
|
||||
""",
|
||||
PBBR_ETH=PBBR_ETH)
|
||||
|
||||
# Host should send NS for ROUTER1's DUA
|
||||
pkts.filter_eth_src(HOST_ETH).filter_icmpv6_nd_ns(ROUTER1_DUA).must_next()
|
||||
# PBBR must send solicited NA for ROUTER1's DUA
|
||||
pkts.filter_eth_src(PBBR_ETH).filter_ipv6_dst(HOST_BGUA).filter_icmpv6_nd_na(ROUTER1_DUA).filter(
|
||||
'icmpv6.nd.na.flag.s == 1').must_next().must_verify("""
|
||||
icmpv6.opt.linkaddr == {PBBR_ETH}
|
||||
and icmpv6.nd.na.flag.r == 1
|
||||
and icmpv6.nd.na.flag.o == 0
|
||||
and icmpv6.nd.na.flag.rsv == 0
|
||||
""",
|
||||
PBBR_ETH=PBBR_ETH)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user