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 the basic MLR feature works.
30#
31import unittest
32
33import config
34import thread_cert
35from pktverify.packet_verifier import PacketVerifier
36
37CH1 = 11
38CH2 = 22
39
40PBBR1 = 1
41ROUTER1 = 2
42PBBR2 = 3
43ROUTER2 = 4
44HOST = 5
45
46MA1 = 'ff05::1234:777a:1'
47MA2 = 'ff05::1234:777a:2'
48
49BBR_REGISTRATION_JITTER = 1
50WAIT_REDUNDANCE = 3
51
52
53class TestMlr(thread_cert.TestCase):
54    USE_MESSAGE_FACTORY = False
55
56    # Topology:
57    #   --------(eth)-----------
58    #       |     |      |
59    #     PBBR  HOST   PBBR2
60    #      |            |
61    #    ROUTER1      ROUTER2
62    #
63    TOPOLOGY = {
64        PBBR1: {
65            'name': 'PBBR1',
66            'allowlist': [ROUTER1],
67            'is_otbr': True,
68            'version': '1.2',
69            'bbr_registration_jitter': BBR_REGISTRATION_JITTER,
70            'channel': CH1,
71        },
72        ROUTER1: {
73            'name': 'ROUTER1',
74            'allowlist': [PBBR1],
75            'version': '1.2',
76            'channel': CH1,
77        },
78        PBBR2: {
79            'name': 'PBBR2',
80            'allowlist': [ROUTER2],
81            'is_otbr': True,
82            'version': '1.2',
83            'bbr_registration_jitter': BBR_REGISTRATION_JITTER,
84            'channel': CH2,
85        },
86        ROUTER2: {
87            'name': 'ROUTER2',
88            'allowlist': [PBBR2],
89            'version': '1.2',
90            'channel': CH2,
91        },
92        HOST: {
93            'name': 'Host',
94            'is_host': True
95        },
96    }
97
98    def test(self):
99        # Bring up Host
100        self.nodes[HOST].start()
101        self.nodes[HOST].add_ipmaddr(MA2, backbone=True)
102
103        # Bring up PBBR1
104        self.nodes[PBBR1].start()
105        self.simulator.go(config.LEADER_STARTUP_DELAY)
106        self.assertEqual('leader', self.nodes[PBBR1].get_state())
107
108        self.nodes[PBBR1].enable_backbone_router()
109        self.simulator.go(10)
110        self.assertTrue(self.nodes[PBBR1].is_primary_backbone_router)
111        self.nodes[PBBR1].add_prefix(config.DOMAIN_PREFIX, "parosD")
112        self.nodes[PBBR1].register_netdata()
113        self.simulator.go(WAIT_REDUNDANCE)
114
115        # Bring up ROUTER1
116        self.nodes[ROUTER1].start()
117        self.simulator.go(5)
118        self.assertEqual('router', self.nodes[ROUTER1].get_state())
119        self.nodes[ROUTER1].add_ipmaddr(MA1)
120        self.simulator.go(WAIT_REDUNDANCE)
121
122        # Bring up PBBR2
123        self.nodes[PBBR2].start()
124        self.simulator.go(config.LEADER_STARTUP_DELAY)
125        self.assertEqual('leader', self.nodes[PBBR2].get_state())
126
127        self.nodes[PBBR2].enable_backbone_router()
128        self.simulator.go(10)
129        self.assertTrue(self.nodes[PBBR2].is_primary_backbone_router)
130
131        self.nodes[PBBR2].add_prefix(config.DOMAIN_PREFIX, "parosD")
132        self.nodes[PBBR2].register_netdata()
133        self.simulator.go(WAIT_REDUNDANCE)
134
135        # Bring up ROUTER2
136        self.nodes[ROUTER2].start()
137        self.simulator.go(5)
138        self.assertEqual('router', self.nodes[ROUTER1].get_state())
139        self.nodes[ROUTER2].add_ipmaddr(MA1)
140        self.nodes[ROUTER2].add_ipmaddr(MA2)
141        self.simulator.go(WAIT_REDUNDANCE)
142
143        self.collect_ipaddrs()
144        self.collect_rloc16s()
145
146        # ping MA1 from Host could generate a reply from R1 and R2
147        self.assertTrue(self.nodes[HOST].ping(MA1, backbone=True, ttl=5))
148        self.simulator.go(WAIT_REDUNDANCE)
149        self.verify_border_routing_counters(self.nodes[PBBR1], {'inbound_multicast': 1, 'outbound_unicast': 1})
150        self.verify_border_routing_counters(self.nodes[PBBR2], {'inbound_multicast': 1, 'outbound_unicast': 1})
151
152        # ping MA2 from R1 could generate a reply from Host and R2
153        self.assertTrue(self.nodes[ROUTER1].ping(MA2))
154        self.simulator.go(WAIT_REDUNDANCE)
155        self.verify_border_routing_counters(self.nodes[PBBR1], {'inbound_unicast': 2, 'outbound_multicast': 1})
156        self.verify_border_routing_counters(self.nodes[PBBR2], {'inbound_multicast': 1, 'outbound_unicast': 1})
157
158        # ping MA2 from R1's MLE-ID shouldn't generate a reply from Host or R2
159        self.assertFalse(self.nodes[ROUTER1].ping(MA2, interface=self.nodes[ROUTER1].get_mleid()))
160        self.simulator.go(WAIT_REDUNDANCE)
161
162        # ping MA2 from R1's LLA shouldn't generate a reply from Host or R2
163        self.assertFalse(self.nodes[ROUTER1].ping(MA2, interface=self.nodes[ROUTER1].get_linklocal()))
164        self.simulator.go(WAIT_REDUNDANCE)
165
166    def verify(self, pv: PacketVerifier):
167        pkts = pv.pkts
168        pv.add_common_vars()
169        pv.summary.show()
170
171        PBBR1 = pv.vars['PBBR1']
172        PBBR1_ETH = pv.vars['PBBR1_ETH']
173        PBBR2 = pv.vars['PBBR2']
174        PBBR2_ETH = pv.vars['PBBR2_ETH']
175        ROUTER1 = pv.vars['ROUTER1']
176        ROUTER2 = pv.vars['ROUTER2']
177        HOST_ETH = pv.vars['Host_ETH']
178        HOST_BGUA = pv.vars['Host_BGUA']
179
180        ROUTER1_DUA = pv.vars['ROUTER1_DUA']
181        ROUTER2_DUA = pv.vars['ROUTER2_DUA']
182
183        ROUTER1_RLOC16 = pv.vars['ROUTER1_RLOC16']
184
185        #
186        # Verify Host ping MA1 to R1 and R2
187        #
188
189        # Host should send ping MA1 in the Backbone link
190        ping_ma1 = pkts.filter_eth_src(HOST_ETH).filter_ipv6_src_dst(HOST_BGUA, MA1).filter_ping_request().must_next()
191
192        with pkts.save_index():
193            # PBBR1 should forward ping reply to the Backbone link
194            pkts.filter_eth_src(PBBR1_ETH).filter_ipv6_src_dst(
195                ROUTER1_DUA, HOST_BGUA).filter_ping_reply(identifier=ping_ma1.icmpv6.echo.identifier).must_next()
196
197        # PBBR2 should forward ping reply to the Backbone link
198        pkts.filter_eth_src(PBBR2_ETH).filter_ipv6_src_dst(
199            ROUTER2_DUA, HOST_BGUA).filter_ping_reply(identifier=ping_ma1.icmpv6.echo.identifier).must_next()
200
201        #
202        # Verify R1 pings MA2 to Host and R2
203        #
204
205        # ROUTER1 should send the multicast ping request
206        ping_ma2 = pkts.filter_wpan_src64(ROUTER1).filter_AMPLFMA(
207            mpl_seed_id=ROUTER1_RLOC16).filter_ping_request().must_next()
208
209        # PBBR1 should forward the multicast ping request to the Backbone link
210        pkts.filter_eth_src(PBBR1_ETH).filter_ipv6_src_dst(
211            ROUTER1_DUA, MA2).filter_ping_request(identifier=ping_ma2.icmpv6.echo.identifier).must_next()
212
213        with pkts.save_index():
214            # Host should send ping reply to Router1
215            pkts.filter_eth_src(HOST_ETH).filter_ipv6_dst(ROUTER1_DUA).filter_ping_reply(
216                identifier=ping_ma2.icmpv6.echo.identifier).must_next()
217
218        # PBBR2 should forward the multicast ping request to Thread network at CH2
219        pkts.filter_wpan_src64(PBBR2).filter_AMPLFMA().filter_ping_request(
220            identifier=ping_ma2.icmpv6.echo.identifier).must_next()
221
222        # ROUTER2 should send ping reply back to ROUTER1
223        pkts.filter_wpan_src64(ROUTER2).filter_ipv6_src_dst(
224            ROUTER2_DUA, ROUTER1_DUA).filter_ping_reply(identifier=ping_ma2.icmpv6.echo.identifier).must_next()
225
226        #
227        # Verify pinging MA2 from R1's MLE-ID will not be forwarded to the Backbone link
228        #
229
230        # ROUTER1 should send the multicast ping request
231        ping_ma2_2 = pkts.filter_wpan_src64(ROUTER1).filter_AMPLFMA(mpl_seed_id=ROUTER1_RLOC16).filter_ping_request(
232            identifier=ping_ma2.icmpv6.echo.identifier + 1).must_next()
233
234        # PBBR1 shouldn't forward the multicast ping request to the Backbone link
235        pkts.filter_eth_src(PBBR1_ETH).filter_ping_request(ping_ma2_2.icmpv6.echo.identifier).must_not_next()
236
237        #
238        # Verify pinging MA2 from R1's Link-Local address will not be forwarded to the Backbone link
239        #
240
241        # ROUTER1 should send the multicast ping request
242        ping_ma2_3 = pkts.filter_wpan_src64(ROUTER1).filter_AMPLFMA(mpl_seed_id=ROUTER1_RLOC16).filter_ping_request(
243            identifier=ping_ma2.icmpv6.echo.identifier + 1).must_next()
244
245        # PBBR1 shouldn't forward the multicast ping request to the Backbone link
246        pkts.filter_eth_src(PBBR1_ETH).filter_ping_request(ping_ma2_3.icmpv6.echo.identifier).must_not_next()
247
248    def verify_border_routing_counters(self, br, expect_delta):
249        delta_counters = br.read_border_routing_counters_delta()
250        self.assertEqual(set(delta_counters.keys()), set(expect_delta.keys()))
251        for key in delta_counters:
252            self.assertGreaterEqual(delta_counters[key][0], expect_delta[key])
253            self.assertGreater(delta_counters[key][1], 0)
254
255
256if __name__ == '__main__':
257    unittest.main()
258