1#!/usr/bin/env python3
2#
3#  Copyright (c) 2022, 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 ipaddress
30import unittest
31
32import config
33import thread_cert
34from pktverify.consts import ICMPV6_RA_OPT_TYPE_RIO
35from pktverify.packet_verifier import PacketVerifier
36
37# Test description:
38# The purpose of this test case is to verify that BR can handle manually added OMR prefixes properly.
39#
40# Topology:
41#     BR_1 (Leader)
42#       |
43#       |
44#     BR_2
45#
46
47BR_1 = 1
48BR_2 = 2
49
50
51class ManualOmrsPrefix(thread_cert.TestCase):
52    USE_MESSAGE_FACTORY = False
53
54    TOPOLOGY = {
55        BR_1: {
56            'name': 'BR_1',
57            'is_otbr': True,
58            'version': '1.2',
59        },
60        BR_2: {
61            'name': 'BR_2',
62            'is_otbr': True,
63            'version': '1.2',
64        },
65    }
66
67    def test(self):
68        br1 = self.nodes[BR_1]
69        br2 = self.nodes[BR_2]
70
71        br1.start()
72        self.simulator.go(config.BORDER_ROUTER_STARTUP_DELAY)
73        self.assertEqual('leader', br1.get_state())
74
75        self.simulator.go(10)
76        self.assertEqual(br1.get_netdata_omr_prefixes(), [br1.get_br_omr_prefix()])
77
78        br2.disable_br()
79        br2.start()
80        self.simulator.go(config.BORDER_ROUTER_STARTUP_DELAY)
81        self.assertEqual('router', br2.get_state())
82        self.assertEqual(br1.get_netdata_omr_prefixes(), [br1.get_br_omr_prefix()])
83
84        # Add a smaller OMR prefix. Verify BR_1 withdraws its OMR prefix.
85        self._test_manual_omr_prefix('2001::/64', 'paros', expect_withdraw=True, prf='low')
86        # Add a bigger OMR prefix. Verify BR_1 does not withdraw its OMR prefix.
87        self._test_manual_omr_prefix('fdff:ffff:1::/64', 'paros', expect_withdraw=False, prf='low')
88        # Add a high preference bigger OMR prefix. Verify BR_1 withdraws its OMR prefix.
89        self._test_manual_omr_prefix('fdff:ffff:2::/64', 'paros', expect_withdraw=True, prf='med')
90        # Add a smaller but invalid OMR prefix (P_on_mesh = 0). Verify BR_1 does not withdraw its OMR prefix.
91        self._test_manual_omr_prefix('2002::/64', 'pars', expect_withdraw=False)
92        # Add a smaller but invalid OMR prefix (P_stable = 0). Verify BR_1 does not withdraw its OMR prefix.
93        self._test_manual_omr_prefix('2003::/64', 'paro', expect_withdraw=False)
94        # Add a deprecating OMR prefix (P_preferred = 0). Verify BR_1 does not withdraw its OMR prefix.
95        self._test_manual_omr_prefix('2004::/64', 'aros', expect_withdraw=False)
96
97        self.collect_ipaddrs()
98        self.collect_extra_vars(BR_1_OMR_PREFIX=br1.get_br_omr_prefix())
99
100    def verify(self, pv: PacketVerifier):
101        pkts = pv.pkts
102        pv.summary.show()
103
104        BR_1_ETH = pv.vars['BR_1_ETH']
105
106        BR_1_OMR_PREFIX = pv.vars['BR_1_OMR_PREFIX']
107        assert BR_1_OMR_PREFIX.endswith('::/64')
108        BR_1_OMR_PREFIX = BR_1_OMR_PREFIX[:-3]
109
110        # Filter all ICMPv6 RA messages advertised by BR_1
111        ra_pkts = pkts.filter_eth_src(BR_1_ETH).filter_icmpv6_nd_ra()
112
113        # Verify BR_1 sends RA RIO for both BR OMR prefix and deprecating OMR prefix (i.e. 2004::/64)
114        ra_pkts.filter(lambda p: ICMPV6_RA_OPT_TYPE_RIO in p.icmpv6.opt.type and '2004::' in p.icmpv6.opt.prefix and
115                       BR_1_OMR_PREFIX in p.icmpv6.opt.prefix).must_next()
116
117    def _test_manual_omr_prefix(self, prefix, flags, expect_withdraw, prf='med'):
118        br1 = self.nodes[BR_1]
119        br2 = self.nodes[BR_2]
120
121        br2.add_prefix(prefix, flags, prf=prf)
122        br2.register_netdata()
123        self.simulator.go(10)
124
125        is_omr_prefix = ('a' in flags and 'o' in flags and 's' in flags and 'D' not in flags)
126
127        if expect_withdraw:
128            self._assert_prefixes_equal(br1.get_netdata_omr_prefixes(), [prefix])
129        elif is_omr_prefix:
130            self._assert_prefixes_equal(br1.get_netdata_omr_prefixes(), [br1.get_br_omr_prefix(), prefix])
131        else:
132            self._assert_prefixes_equal(br1.get_netdata_omr_prefixes(), [br1.get_br_omr_prefix()])
133
134        br2.remove_prefix(prefix)
135        br2.register_netdata()
136        self.simulator.go(10)
137        self._assert_prefixes_equal(br1.get_netdata_omr_prefixes(), [br1.get_br_omr_prefix()])
138        self.simulator.go(3)
139
140    def _assert_prefixes_equal(self, prefixes1, prefixes2):
141        prefixes1 = sorted([ipaddress.IPv6Network(p) for p in prefixes1])
142        prefixes2 = sorted([ipaddress.IPv6Network(p) for p in prefixes2])
143        self.assertEqual(prefixes1, prefixes2)
144
145
146if __name__ == '__main__':
147    unittest.main()
148