1#!/usr/bin/env python3
2#
3#  Copyright (c) 2021, 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#
29import logging
30import unittest
31
32import ipaddress
33import config
34import thread_cert
35
36# Test description:
37#   This test verifies that when external routes are changed in Network Data, a
38#   border router will make according updates to routes at the linux kernel.
39#
40# Topology:
41#    ----------------(eth)----------------------
42#           |                  |          |
43#          BR1 (Leader) ----- BR2       HOST
44#           |                  |
45#         ROUTER1             ROUTER2
46
47BR1 = 1
48BR2 = 2
49ROUTER1 = 3
50ROUTER2 = 4
51HOST = 5
52
53ROUTE1 = '2402:1234:1234:1234::/64'
54
55
56class ExternalRoutes(thread_cert.TestCase):
57    USE_MESSAGE_FACTORY = False
58
59    TOPOLOGY = {
60        BR1: {
61            'name': 'BR_1',
62            'allowlist': [BR2, ROUTER1],
63            'is_otbr': True,
64            'version': '1.2',
65        },
66        BR2: {
67            'name': 'BR_2',
68            'allowlist': [BR1, ROUTER2],
69            'is_otbr': True,
70            'version': '1.2',
71        },
72        ROUTER1: {
73            'name': 'Router_1',
74            'allowlist': [BR1],
75            'version': '1.2',
76        },
77        ROUTER2: {
78            'name': 'Router_2',
79            'allowlist': [BR2],
80            'version': '1.2',
81        },
82        HOST: {
83            'name': 'Host',
84            'is_host': True,
85        }
86    }
87
88    def test(self):
89        br1 = self.nodes[BR1]
90        br2 = self.nodes[BR2]
91        router1 = self.nodes[ROUTER1]
92        router2 = self.nodes[ROUTER2]
93        host = self.nodes[HOST]
94
95        br1.start()
96        self.simulator.go(config.LEADER_STARTUP_DELAY)
97        self.assertEqual('leader', br1.get_state())
98
99        br2.start()
100        self.simulator.go(config.BORDER_ROUTER_STARTUP_DELAY)
101        self.assertEqual('router', br2.get_state())
102
103        router1.start()
104        router2.start()
105        host.start()
106        self.simulator.go(5)
107        self.assertEqual('router', router1.get_state())
108        self.assertEqual('router', router2.get_state())
109
110        # Manually add ROUTE1 at BR1
111        br1.add_route(ROUTE1)
112        br1.register_netdata()
113
114        # BRs has installed all external routes in their kernels respectively
115        netdata = br1.get_netdata()
116        br1_kernel_routes = self.get_routes_from_kernel(br1)
117        for route in self.get_routes_from_netdata(netdata, br1.get_addr16()):
118            self.assertIn(route, br1_kernel_routes)
119        br2_kernel_routes = self.get_routes_from_kernel(br2)
120        for route in self.get_routes_from_netdata(netdata, br2.get_addr16()):
121            self.assertIn(route, br2_kernel_routes)
122
123        # ROUTE1 has been installed in BR2 but not BR1
124        self.assertNotIn(ipaddress.IPv6Network(ROUTE1), br1_kernel_routes)
125        self.assertIn(ipaddress.IPv6Network(ROUTE1), br2_kernel_routes)
126
127        # Remove ROUTE1
128        br1.remove_route(ROUTE1)
129        br1.register_netdata()
130
131        # Verify that external routes are still in kernel routes
132        netdata = br1.get_netdata()
133        br1_kernel_routes = self.get_routes_from_kernel(br1)
134        for route in self.get_routes_from_netdata(netdata, br1.get_addr16()):
135            self.assertIn(route, br1_kernel_routes)
136        br2_kernel_routes = self.get_routes_from_kernel(br2)
137        for route in self.get_routes_from_netdata(netdata, br2.get_addr16()):
138            self.assertIn(route, br2_kernel_routes)
139
140        # ROUTE1 has been removed from kernel routes
141        self.assertNotIn(ipaddress.IPv6Network(ROUTE1), br1_kernel_routes)
142        self.assertNotIn(ipaddress.IPv6Network(ROUTE1), br2_kernel_routes)
143
144        br2.bash('ip link set eth0 down')
145        self.simulator.go(10)
146
147        # HOST pings BR2's OMR, the ping should succeed
148        self.assertTrue(host.ping_ether(br2.get_ip6_address(address_type=config.ADDRESS_TYPE.OMR)[0]))
149
150    def get_routes_from_netdata(self, netdata, exclude_rloc16):
151        routes = []
152        for entry in netdata['Routes']:
153            items = entry.split()
154            prefix = items[0]
155            rloc16 = int(items[-1], 16)
156            if rloc16 != exclude_rloc16:
157                routes.append(ipaddress.IPv6Network(prefix))
158        return routes
159
160    def get_routes_from_kernel(self, br):
161        routes = []
162        for entry in br.bash('ip -6 -d route list dev wpan0'):
163            routes.append(ipaddress.IPv6Network(entry.split()[1]))
164        return routes
165
166
167if __name__ == '__main__':
168    unittest.main()
169