1#!/usr/bin/env python3 2# 3# Copyright (c) 2020, The OpenThread Authors. 4# All rights reserved. 5# 6# Redistribution and use in source and binary forms, with or without 7# modification, are permitted provided that the following conditions are met: 8# 1. Redistributions of source code must retain the above copyright 9# notice, this list of conditions and the following disclaimer. 10# 2. Redistributions in binary form must reproduce the above copyright 11# notice, this list of conditions and the following disclaimer in the 12# documentation and/or other materials provided with the distribution. 13# 3. Neither the name of the copyright holder nor the 14# names of its contributors may be used to endorse or promote products 15# derived from this software without specific prior written permission. 16# 17# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS' 18# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 19# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 20# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE 21# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 22# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 23# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 24# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 25# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 26# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 27# POSSIBILITY OF SUCH DAMAGE. 28# 29# This test verifies that PBBR sets DUA routes correctly. 30# 31import ipaddress 32import unittest 33 34import config 35import thread_cert 36from pktverify.packet_verifier import PacketVerifier 37 38# Use two channels 39CH1 = 11 40CH2 = 22 41 42PBBR = 1 43SBBR = 2 44ROUTER1 = 3 45HOST = 4 46PBBR2 = 5 47ROUTER2 = 6 48 49REREG_DELAY = 5 # Seconds 50MLR_TIMEOUT = 300 # Seconds 51WAIT_REDUNDANCE = 3 52 53 54class TestNdProxy(thread_cert.TestCase): 55 USE_MESSAGE_FACTORY = False 56 57 # Topology: 58 # ------(eth)---------------------- 59 # | | | | 60 # PBBR----SBBR HOST PBBR2 61 # \ CH1 / | CH2 62 # ROUTER1 ROUTER2 63 # 64 # PBBR2 is in the secondary channel 65 # 66 TOPOLOGY = { 67 PBBR: { 68 'name': 'PBBR', 69 'allowlist': [SBBR, ROUTER1], 70 'is_otbr': True, 71 'version': '1.2', 72 'channel': CH1, 73 'router_id_range': [0, 30], 74 }, 75 SBBR: { 76 'name': 'SBBR', 77 'allowlist': [PBBR, ROUTER1], 78 'is_otbr': True, 79 'version': '1.2', 80 'channel': CH1, 81 }, 82 ROUTER1: { 83 'name': 'ROUTER1', 84 'allowlist': [PBBR, SBBR], 85 'version': '1.2', 86 'channel': CH1, 87 }, 88 HOST: { 89 'name': 'HOST', 90 'is_host': True 91 }, 92 PBBR2: { 93 'name': 'PBBR2', 94 'is_otbr': True, 95 'version': '1.2', 96 'channel': CH2, 97 'router_id_range': [31, 60], 98 }, 99 ROUTER2: { 100 'name': 'ROUTER2', 101 'version': '1.2', 102 'channel': CH2, 103 }, 104 } 105 106 def _bootstrap(self): 107 # Bring up HOST 108 self.nodes[HOST].start() 109 110 # Bring up PBBR 111 self.nodes[PBBR].start() 112 self.simulator.go(config.LEADER_STARTUP_DELAY) 113 self.assertEqual('leader', self.nodes[PBBR].get_state()) 114 self.wait_node_state(PBBR, 'leader', 10) 115 116 self.nodes[PBBR].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT) 117 self.nodes[PBBR].enable_backbone_router() 118 self.nodes[PBBR].set_domain_prefix(config.DOMAIN_PREFIX, 'prosD') 119 self.simulator.go(5) 120 self.assertTrue(self.nodes[PBBR].is_primary_backbone_router) 121 self.assertIsNotNone(self.nodes[PBBR].get_ip6_address(config.ADDRESS_TYPE.DUA)) 122 123 # Bring up SBBR 124 self.nodes[SBBR].start() 125 self.simulator.go(5) 126 self.assertEqual('router', self.nodes[SBBR].get_state()) 127 128 self.nodes[SBBR].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT) 129 self.nodes[SBBR].enable_backbone_router() 130 self.simulator.go(5) 131 self.assertFalse(self.nodes[SBBR].is_primary_backbone_router) 132 self.assertIsNotNone(self.nodes[SBBR].get_ip6_address(config.ADDRESS_TYPE.DUA)) 133 134 # Bring up ROUTER1 135 self.nodes[ROUTER1].start() 136 self.simulator.go(5) 137 self.assertEqual('router', self.nodes[ROUTER1].get_state()) 138 self.assertIsNotNone(self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA)) 139 140 # Bring up PBBR2 141 self.nodes[PBBR2].start() 142 self.simulator.go(config.LEADER_STARTUP_DELAY) 143 self.assertEqual('leader', self.nodes[PBBR2].get_state()) 144 self.wait_node_state(PBBR2, 'leader', 10) 145 146 self.nodes[PBBR2].set_backbone_router(reg_delay=REREG_DELAY, mlr_timeout=MLR_TIMEOUT) 147 self.nodes[PBBR2].enable_backbone_router() 148 self.nodes[PBBR2].set_domain_prefix(config.DOMAIN_PREFIX, 'prosD') 149 self.simulator.go(5) 150 self.assertTrue(self.nodes[PBBR2].is_primary_backbone_router) 151 self.assertIsNotNone(self.nodes[PBBR2].get_ip6_address(config.ADDRESS_TYPE.DUA)) 152 153 # Bring up ROUTER2 154 self.nodes[ROUTER2].start() 155 self.simulator.go(5) 156 self.assertEqual('router', self.nodes[ROUTER2].get_state()) 157 self.assertIsNotNone(self.nodes[ROUTER2].get_ip6_address(config.ADDRESS_TYPE.DUA)) 158 159 def test(self): 160 self._bootstrap() 161 162 self.collect_ipaddrs() 163 self.collect_rloc16s() 164 165 self.simulator.go(WAIT_REDUNDANCE) 166 167 # Step 1: Host pings PBBR's DUA 168 self.assertTrue(self.nodes[HOST].ping(self.nodes[PBBR].get_ip6_address(config.ADDRESS_TYPE.DUA), 169 backbone=True)) 170 171 self.simulator.go(WAIT_REDUNDANCE) 172 173 # Step 2: Host pings SBBR's DUA 174 self.assertTrue(self.nodes[HOST].ping(self.nodes[SBBR].get_ip6_address(config.ADDRESS_TYPE.DUA), 175 backbone=True)) 176 177 self.simulator.go(WAIT_REDUNDANCE) 178 179 # Step 3: Host pings ROUTER1's DUA 180 self.assertTrue(self.nodes[HOST].ping(self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA), 181 backbone=True)) 182 183 self.simulator.go(WAIT_REDUNDANCE) 184 185 # Step 4: ROUTER2 pings ROUTER1 which should succeed 186 ROUTER1_DUA = self.nodes[ROUTER1].get_ip6_address(config.ADDRESS_TYPE.DUA) 187 self.assertTrue(self.nodes[ROUTER2].ping(ROUTER1_DUA)) 188 189 self.simulator.go(WAIT_REDUNDANCE) 190 191 def _get_mliid(self, nodeid: int) -> str: 192 mleid = self.nodes[nodeid].get_ip6_address(config.ADDRESS_TYPE.ML_EID) 193 mliid = ipaddress.IPv6Address(mleid).packed[-8:] 194 return ''.join(['%02x' % b for b in mliid]) 195 196 def verify(self, pv: PacketVerifier): 197 pkts = pv.pkts 198 pv.add_common_vars() 199 pv.summary.show() 200 201 PBBR = pv.vars['PBBR'] 202 PBBR2 = pv.vars['PBBR2'] 203 ROUTER1 = pv.vars['ROUTER1'] 204 ROUTER2 = pv.vars['ROUTER2'] 205 SBBR = pv.vars['SBBR'] 206 PBBR_DUA = pv.vars['PBBR_DUA'] 207 SBBR_DUA = pv.vars['SBBR_DUA'] 208 ROUTER1_DUA = pv.vars['ROUTER1_DUA'] 209 ROUTER2_DUA = pv.vars['ROUTER2_DUA'] 210 HOST_BGUA = pv.vars['HOST_BGUA'] 211 PBBR_ETH = pv.vars['PBBR_ETH'] 212 PBBR2_ETH = pv.vars['PBBR2_ETH'] 213 214 # Verify that SBBR should not foward any Ping Request to the Thread network. 215 # Use `ipv6.hlim == 63` to avoid false fails because SBBR might still forward Ping Request from PBBR to ROUTER1 216 pkts.filter_wpan_src64(SBBR).filter_ping_request().filter('ipv6.hlim == 63').must_not_next() 217 218 # Step 1: Host pings PBBR's DUA 219 # PBBR should not forward the Ping Request to Thread network 220 ping1_pkts = pkts.filter_ipv6_src_dst(HOST_BGUA, PBBR_DUA).filter_ping_request() 221 ping1_pkts.filter_eth().must_next() 222 ping1_pkts.filter_wpan().must_not_next() 223 224 # Step 2: Host pings SBBR's DUA 225 # PBBR should forward the Ping Request to Thread network according to DUA routes 226 ping2_pkts = pkts.filter_ipv6_src_dst(HOST_BGUA, SBBR_DUA).filter_ping_request() 227 ping2_pkts.filter_eth().must_next() 228 ping2_pkts.filter_wpan().must_next() 229 230 # Step 3: Host pings ROUTER1's DUA 231 # PBBR should forward the Ping Request to Thread network according to DUA routes 232 ping3_pkts = pkts.filter_ipv6_src_dst(HOST_BGUA, ROUTER1_DUA).filter_ping_request() 233 ping3_pkts.filter_eth().must_next() 234 ping3_pkts.filter_wpan().must_next() 235 236 # Step 4: ROUTER2 pings ROUTER1 237 # Verify Ping Request: ROUTER2 -> PBBR2 -> PBBR -> ROUTER1 238 ping4_request_pkts = pkts.filter_ipv6_src_dst(ROUTER2_DUA, ROUTER1_DUA) 239 ping4_id = ping4_request_pkts.filter_wpan_src64( 240 ROUTER2).filter_ping_request().must_next().icmpv6.echo.identifier 241 ping4_request_pkts.filter_eth_src(PBBR2_ETH).filter_ping_request(identifier=ping4_id).must_next() 242 ping4_request_pkts.filter_wpan_src64(PBBR).filter_ping_request(identifier=ping4_id).must_next() 243 # Verify Ping Reply: ROUTER1 -> PBBR -> PBBR2 -> ROUTER2 244 ping4_reply_pkts = pkts.filter_ipv6_src_dst(ROUTER1_DUA, ROUTER2_DUA).filter_ping_reply(identifier=ping4_id) 245 ping4_reply_pkts.filter_eth_src(PBBR_ETH).must_next() 246 247 248if __name__ == '__main__': 249 unittest.main() 250