1#!/usr/bin/env python3
2#
3#  Copyright (c) 2016, 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
30import unittest
31
32import config
33import thread_cert
34from pktverify.consts import MLE_ADVERTISEMENT, MLE_PARENT_REQUEST, MLE_CHILD_ID_RESPONSE, MLE_ANNOUNCE, CHANNEL_TLV, PAN_ID_TLV, ACTIVE_TIMESTAMP_TLV
35from pktverify.packet_verifier import PacketVerifier
36
37CHANNEL1 = 11
38CHANNEL2 = 18
39CHANNEL_MASK = 1 << 18
40PANID_INIT = 0xface
41
42LEADER1 = 1
43LEADER2 = 2
44ED1 = 3
45
46
47class Cert_9_2_17_Orphan(thread_cert.TestCase):
48    SUPPORT_NCP = False
49
50    TOPOLOGY = {
51        LEADER1: {
52            'name': 'LEADER_1',
53            'active_dataset': {
54                'timestamp': 10,
55                'panid': PANID_INIT,
56                'channel': CHANNEL1,
57                'channel_mask': CHANNEL_MASK
58            },
59            'mode': 'rdn',
60            'allowlist': [ED1]
61        },
62        LEADER2: {
63            'name': 'LEADER_2',
64            'active_dataset': {
65                'timestamp': 20,
66                'panid': PANID_INIT,
67                'channel': CHANNEL2,
68                'channel_mask': CHANNEL_MASK
69            },
70            'mode': 'rdn',
71        },
72        ED1: {
73            'name': 'ED',
74            'channel': CHANNEL1,
75            'is_mtd': True,
76            'mode': 'rn',
77            'panid': PANID_INIT,
78            'timeout': config.DEFAULT_CHILD_TIMEOUT,
79            'allowlist': [LEADER1]
80        },
81    }
82
83    def test(self):
84        self.nodes[LEADER1].start()
85        self.simulator.go(5)
86        self.assertEqual(self.nodes[LEADER1].get_state(), 'leader')
87
88        self.nodes[LEADER2].start()
89        self.nodes[LEADER2].set_state('leader')
90        self.assertEqual(self.nodes[LEADER2].get_state(), 'leader')
91
92        self.nodes[ED1].start()
93        self.simulator.go(5)
94        self.assertEqual(self.nodes[ED1].get_state(), 'child')
95
96        self.nodes[LEADER1].stop()
97        self.nodes[LEADER2].add_allowlist(self.nodes[ED1].get_addr64())
98        self.nodes[ED1].add_allowlist(self.nodes[LEADER2].get_addr64())
99        self.simulator.go(20)
100
101        self.assertEqual(self.nodes[ED1].get_state(), 'child')
102        self.assertEqual(self.nodes[ED1].get_channel(), CHANNEL2)
103
104        self.collect_ipaddrs()
105        ipaddrs = self.nodes[ED1].get_addrs()
106        for ipaddr in ipaddrs:
107            self.assertTrue(self.nodes[LEADER2].ping(ipaddr))
108
109    def verify(self, pv):
110        pkts = pv.pkts
111        pv.summary.show()
112
113        LEADER_1 = pv.vars['LEADER_1']
114        LEADER_2 = pv.vars['LEADER_2']
115        ED = pv.vars['ED']
116
117        # Step 1: Ensure the topology is formed correctly
118        # Verify that Leader_1 & Leader_2 are sending MLE Advertisements on separate channels.
119        pkts.filter_wpan_src64(LEADER_1).filter_mle_cmd(MLE_CHILD_ID_RESPONSE).must_next().must_verify(
120            lambda p: p.wpan.dst64 == ED and p.thread_meshcop.tlv.channel == [CHANNEL1])
121        pkts.filter_wpan_src64(LEADER_1).filter_mle_cmd(MLE_ADVERTISEMENT).must_next()
122        pkts.copy().filter_wpan_src64(LEADER_2).filter_mle_cmd(MLE_ADVERTISEMENT).must_next()
123
124        # Step 4: powers-down Leader_1 and enables connectivity between the ED and Leader_2
125        # ED MUST send a MLE Parent Request
126        _epkts = pkts.filter_wpan_src64(ED)
127        _epkts.filter_mle_cmd(MLE_PARENT_REQUEST).must_next()
128
129        # Step 6: ED MUST send a MLE Announce Message
130        # The Destination PAN ID (0xFFFF) in the IEEE 802.15.4 MAC and MUST be secured using Key ID Mode 2.
131        _epkts.filter_mle_cmd(MLE_ANNOUNCE).must_next().must_verify(
132            lambda p: {CHANNEL_TLV, PAN_ID_TLV, ACTIVE_TIMESTAMP_TLV} == set(
133                p.mle.tlv.type) and p.wpan.dst_pan == 0xffff and p.wpan.aux_sec.key_id_mode == 0x2)
134
135        # Step 8: ED MUST attempt to attach on the Secondary channel,
136        # with the new PAN ID it received in the MLE Announce message from Leader_2
137        pkts.range(_epkts.index).filter_mle_cmd(MLE_ANNOUNCE).filter_wpan_src64(LEADER_2).filter_wpan_dst64(
138            ED).must_next().must_verify(lambda p: p.mle.tlv.channel == CHANNEL2)
139        _epkts.range(pkts.index).filter_mle_cmd(MLE_PARENT_REQUEST).must_next()
140
141        # Step 9: ED MUST respond with an ICMPv6 Echo Reply
142        _epkts.filter('ipv6.dst == {LEADER_2_MLEID} and ipv6.src == {ED_MLEID}',
143                      **pv.vars).filter_ping_reply().must_next()
144
145
146if __name__ == '__main__':
147    unittest.main()
148