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 ipaddress
30import logging
31import unittest
32
33import config
34import thread_cert
35
36# Test description:
37#   This test verifies the basic functionality of advertising proxy.
38#
39# Topology:
40#    ----------------(eth)--------------------
41#           |                   |
42#          BR (Leader)    HOST (mDNS Browser)
43#           |
44#        ROUTER
45#
46
47BR = 1
48ROUTER = 2
49HOST = 3
50LEASE = 10  # Seconds
51KEY_LEASE = 20  # Seconds
52
53
54class SingleHostAndService(thread_cert.TestCase):
55    USE_MESSAGE_FACTORY = False
56
57    TOPOLOGY = {
58        BR: {
59            'name': 'BR_1',
60            'allowlist': [ROUTER],
61            'is_otbr': True,
62            'version': '1.2',
63        },
64        ROUTER: {
65            'name': 'Router_1',
66            'allowlist': [BR],
67            'version': '1.2',
68        },
69        HOST: {
70            'name': 'Host',
71            'is_host': True
72        },
73    }
74
75    def test(self):
76        host = self.nodes[HOST]
77        server = self.nodes[BR]
78        client = self.nodes[ROUTER]
79
80        server.srp_server_set_enabled(False)
81        host.start(start_radvd=False)
82        self.simulator.go(5)
83
84        self.assertEqual(server.srp_server_get_state(), 'disabled')
85        server.srp_server_set_enabled(True)
86        server.srp_server_set_lease_range(LEASE, LEASE, KEY_LEASE, KEY_LEASE)
87        server.start()
88        self.simulator.go(10)
89        self.assertEqual('leader', server.get_state())
90        self.assertEqual(server.srp_server_get_state(), 'running')
91
92        client.start()
93        self.simulator.go(5)
94        self.assertEqual('router', client.get_state())
95
96        #
97        # 1. Register a single service.
98        #
99
100        client.srp_client_set_host_name('my-host')
101        client.srp_client_set_host_address('2001::1')
102        client.srp_client_add_service('my-service', '_ipps._tcp', 12345)
103        client.srp_client_add_service('my-service-1', '_ipps._tcp', 12345)
104        client.srp_client_enable_auto_start_mode()
105        self.simulator.go(2)
106
107        self.check_host_and_service(server, client, '2001::1', 'my-service')
108        self.check_host_and_service(server, client, '2001::1', 'my-service-1')
109
110        #
111        # 2. Discover the service by the HOST on the ethernet. This makes sure
112        #    the Advertising Proxy multicasts the same service on ethernet.
113        #
114
115        self.host_check_mdns_service(host, '2001::1', 'my-service')
116        self.host_check_mdns_service(host, '2001::1', 'my-service-1')
117
118        #
119        # 3. Check if the Advertising Proxy removes the service from ethernet
120        #    when the SRP client removes it.
121        #
122
123        client.srp_client_remove_host()
124        self.simulator.go(2)
125
126        self.assertIsNone(host.discover_mdns_service('my-service', '_ipps._tcp', 'my-host'))
127        self.assertIsNone(host.discover_mdns_service('my-service-1', '_ipps._tcp', 'my-host'))
128
129        #
130        # 4. Check if we can discover the mDNS service again when re-registering the
131        #    service from the SRP client.
132        #
133
134        client.srp_client_set_host_name('my-host')
135        client.srp_client_set_host_address('2001::1')
136        client.srp_client_add_service('my-service', '_ipps._tcp', 12345)
137        client.srp_client_add_service('my-service-1', '_ipps._tcp', 12345)
138        self.simulator.go(2)
139
140        self.check_host_and_service(server, client, '2001::1', 'my-service')
141        self.check_host_and_service(server, client, '2001::1', 'my-service-1')
142        self.host_check_mdns_service(host, '2001::1', 'my-service')
143        self.host_check_mdns_service(host, '2001::1', 'my-service-1')
144
145        #
146        # 5. Update the SRP host address and make sure the Advertising Proxy
147        #    will follow it.
148        #
149
150        client.srp_client_set_host_address('2001::2')
151        self.simulator.go(8)
152
153        self.check_host_and_service(server, client, '2001::2', 'my-service')
154        self.check_host_and_service(server, client, '2001::2', 'my-service-1')
155        self.host_check_mdns_service(host, '2001::2', 'my-service')
156        self.host_check_mdns_service(host, '2001::2', 'my-service-1')
157
158        #
159        # 6. Check if the service is removed by the Advertising Proxy when the SRP server is stopped.
160        #
161
162        server.srp_server_set_enabled(False)
163        self.simulator.go(5)
164
165        self.assertEqual(len(server.srp_server_get_hosts()), 0)
166        self.assertEqual(len(server.srp_server_get_services()), 0)
167        self.assertIsNone(host.discover_mdns_service('my-service', '_ipps._tcp', 'my-host'))
168        self.assertIsNone(host.discover_mdns_service('my-service-1', '_ipps._tcp', 'my-host'))
169
170        server.srp_server_set_enabled(True)
171        self.simulator.go(LEASE)
172
173        self.check_host_and_service(server, client, '2001::2', 'my-service')
174        self.check_host_and_service(server, client, '2001::2', 'my-service-1')
175        self.host_check_mdns_service(host, '2001::2', 'my-service')
176        self.host_check_mdns_service(host, '2001::2', 'my-service-1')
177
178        #
179        # 7. Check if the expired service is removed by the Advertising Proxy.
180        #
181
182        client.srp_client_stop()
183        self.simulator.go(LEASE + 2)
184
185        self.assertIsNone(host.discover_mdns_service('my-service', '_ipps._tcp', 'my-host'))
186        self.assertIsNone(host.discover_mdns_service('my-service-1', '_ipps._tcp', 'my-host'))
187
188    def host_check_mdns_service(self, host, host_addr, service_instance):
189        service = host.discover_mdns_service(service_instance, '_ipps._tcp', 'my-host')
190        self.assertIsNotNone(service)
191        self.assertEqual(service['instance'], service_instance)
192        self.assertEqual(service['name'], '_ipps._tcp')
193        self.assertEqual(service['port'], 12345)
194        self.assertEqual(service['priority'], 0)
195        self.assertEqual(service['weight'], 0)
196        self.assertEqual(service['host'], 'my-host')
197        self.assertEqual(ipaddress.ip_address(service['addresses'][0]), ipaddress.ip_address(host_addr))
198        self.assertEqual(len(service['addresses']), 1)
199
200    def check_host_and_service(self, server, client, host_addr, service_instance):
201        """Check that we have properly registered host and service instance.
202        """
203
204        client_services = client.srp_client_get_services()
205        print(client_services)
206        client_services = [service for service in client_services if service['instance'] == service_instance]
207        self.assertEqual(len(client_services), 1)
208        client_service = client_services[0]
209
210        # Verify that the client possesses correct service resources.
211        self.assertEqual(client_service['instance'], service_instance)
212        self.assertEqual(client_service['name'], '_ipps._tcp')
213        self.assertEqual(int(client_service['port']), 12345)
214        self.assertEqual(int(client_service['priority']), 0)
215        self.assertEqual(int(client_service['weight']), 0)
216
217        # Verify that the client received a SUCCESS response for the server.
218        self.assertEqual(client_service['state'], 'Registered')
219
220        server_services = server.srp_server_get_services()
221        print(server_services)
222        server_services = [service for service in server_services if service['instance'] == service_instance]
223        self.assertEqual(len(server_services), 1)
224        server_service = server_services[0]
225
226        # Verify that the server accepted the SRP registration and stores
227        # the same service resources.
228        self.assertEqual(server_service['deleted'], 'false')
229        self.assertEqual(server_service['instance'], client_service['instance'])
230        self.assertEqual(server_service['name'], client_service['name'])
231        self.assertEqual(int(server_service['port']), int(client_service['port']))
232        self.assertEqual(int(server_service['priority']), int(client_service['priority']))
233        self.assertEqual(int(server_service['weight']), int(client_service['weight']))
234        self.assertEqual(server_service['host'], 'my-host')
235
236        server_hosts = server.srp_server_get_hosts()
237        print(server_hosts)
238        self.assertEqual(len(server_hosts), 1)
239        server_host = server_hosts[0]
240
241        self.assertEqual(server_host['deleted'], 'false')
242        self.assertEqual(server_host['fullname'], server_service['host_fullname'])
243        self.assertEqual(len(server_host['addresses']), 1)
244        self.assertEqual(ipaddress.ip_address(server_host['addresses'][0]), ipaddress.ip_address(host_addr))
245
246
247if __name__ == '__main__':
248    unittest.main()
249