| #!/usr/bin/env python3 |
| # |
| # Copyright (c) 2021, The OpenThread Authors. |
| # All rights reserved. |
| # |
| # Redistribution and use in source and binary forms, with or without |
| # modification, are permitted provided that the following conditions are met: |
| # 1. Redistributions of source code must retain the above copyright |
| # notice, this list of conditions and the following disclaimer. |
| # 2. Redistributions in binary form must reproduce the above copyright |
| # notice, this list of conditions and the following disclaimer in the |
| # documentation and/or other materials provided with the distribution. |
| # 3. Neither the name of the copyright holder nor the |
| # names of its contributors may be used to endorse or promote products |
| # derived from this software without specific prior written permission. |
| # |
| # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 'AS IS' |
| # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE |
| # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE |
| # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE |
| # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR |
| # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF |
| # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS |
| # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN |
| # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) |
| # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE |
| # POSSIBILITY OF SUCH DAMAGE. |
| # |
| import ipaddress |
| import json |
| import logging |
| import unittest |
| |
| import config |
| import thread_cert |
| |
| # Test description: |
| # This test verifies DNS-SD server and Discovery Proxy can handle Instance names with whitespaces. |
| # |
| # Topology: |
| # ----------------(eth)-------------------- |
| # | | | |
| # BR1 (Server)------BR2 HOST |
| # / |
| # CLIENT1 |
| |
| SERVER = BR1 = 1 |
| CLIENT = 2 |
| HOST = 3 |
| BR2 = 4 |
| |
| DOMAIN = 'default.service.arpa.' |
| SERVICE = '_testsrv._udp' |
| SERVICE_FULL_NAME = f'{SERVICE}.{DOMAIN}' |
| INSTANCE_NAME = 'O T B R' |
| HOST_NAME = 'host1' |
| HOST_FULL_NAME = f'{HOST_NAME}.{DOMAIN}' |
| |
| |
| class TestDnssdInstanceNameWithSpace(thread_cert.TestCase): |
| USE_MESSAGE_FACTORY = False |
| |
| TOPOLOGY = { |
| BR1: { |
| 'name': 'SERVER', |
| 'is_otbr': True, |
| 'version': '1.2', |
| }, |
| BR2: { |
| 'name': 'BR2', |
| 'is_otbr': True, |
| 'version': '1.2', |
| }, |
| CLIENT: { |
| 'name': 'CLIENT1', |
| }, |
| HOST: { |
| 'name': 'Host', |
| 'is_host': True |
| }, |
| } |
| |
| def test(self): |
| server = br1 = self.nodes[BR1] |
| br2 = self.nodes[BR2] |
| client = self.nodes[CLIENT] |
| digger = host = self.nodes[HOST] |
| |
| host.start(start_radvd=False) |
| self.simulator.go(5) |
| |
| br1.start() |
| self.simulator.go(config.LEADER_STARTUP_DELAY) |
| self.assertEqual('leader', br1.get_state()) |
| server.srp_server_set_enabled(True) |
| br1.dns_upstream_query_state = False |
| |
| br2.start() |
| self.simulator.go(config.BORDER_ROUTER_STARTUP_DELAY) |
| self.assertEqual('router', br2.get_state()) |
| |
| client.start() |
| |
| self.simulator.go(config.ROUTER_STARTUP_DELAY) |
| self.assertEqual('router', client.get_state()) |
| |
| self.simulator.go(10) |
| |
| server_addr = server.get_ip6_address(config.ADDRESS_TYPE.OMR)[0] |
| |
| client1_addrs = [client.get_mleid(), client.get_ip6_address(config.ADDRESS_TYPE.OMR)[0]] |
| self._config_srp_client_services(client, INSTANCE_NAME, HOST_NAME, 11111, 0, 0, client1_addrs) |
| |
| full_instance_name = f'{INSTANCE_NAME}.{SERVICE_FULL_NAME}' |
| EMPTY_TXT = {} |
| |
| # In all cases, there is one match, so server should include |
| # service info in additional section of PTR query response. |
| self._verify_service_browse_result(client.dns_browse(SERVICE_FULL_NAME, server=br1.get_rloc())) |
| self._verify_service_browse_result(client.dns_browse(SERVICE_FULL_NAME, server=br2.get_rloc())) |
| self._verify_service_browse_result(client.dns_browse(SERVICE_FULL_NAME.lower(), server=br2.get_rloc())) |
| |
| # check if PTR query works |
| dig_result = digger.dns_dig(server_addr, SERVICE_FULL_NAME, 'PTR') |
| |
| self._assert_dig_result_matches( |
| dig_result, { |
| 'QUESTION': [(SERVICE_FULL_NAME, 'IN', 'PTR')], |
| 'ANSWER': [(SERVICE_FULL_NAME, 'IN', 'PTR', f'{INSTANCE_NAME}.{SERVICE_FULL_NAME}')], |
| 'ADDITIONAL': [ |
| (full_instance_name, 'IN', 'SRV', 0, 0, 11111, HOST_FULL_NAME), |
| (full_instance_name, 'IN', 'TXT', EMPTY_TXT), |
| ], |
| }) |
| |
| # check if SRV query works |
| dig_result = digger.dns_dig(server_addr, full_instance_name, 'SRV') |
| self._assert_dig_result_matches( |
| dig_result, { |
| 'QUESTION': [(full_instance_name, 'IN', 'SRV')], |
| 'ANSWER': [(full_instance_name, 'IN', 'SRV', 0, 0, 11111, HOST_FULL_NAME),], |
| 'ADDITIONAL': [], |
| }) |
| |
| def _config_srp_client_services(self, client, instancename, hostname, port, priority, weight, addrs): |
| client.srp_client_enable_auto_start_mode() |
| client.srp_client_set_host_name(hostname) |
| client.srp_client_set_host_address(*addrs) |
| client.srp_client_add_service(instancename, SERVICE, port, priority, weight) |
| |
| self.simulator.go(5) |
| self.assertEqual(client.srp_client_get_host_state(), 'Registered') |
| |
| def _assert_have_question(self, dig_result, question): |
| for dig_question in dig_result['QUESTION']: |
| if self._match_record(dig_question, question): |
| return |
| |
| self.fail((dig_result, question)) |
| |
| def _assert_have_answer(self, dig_result, record, additional=False): |
| for dig_answer in dig_result['ANSWER' if not additional else 'ADDITIONAL']: |
| dig_answer = list(dig_answer) |
| dig_answer[1:2] = [] # remove TTL from answer |
| |
| record = list(record) |
| |
| # convert IPv6 addresses to `ipaddress.IPv6Address` before matching |
| if dig_answer[2] == 'AAAA': |
| dig_answer[3] = ipaddress.IPv6Address(dig_answer[3]) |
| |
| if record[2] == 'AAAA': |
| record[3] = ipaddress.IPv6Address(record[3]) |
| |
| if self._match_record(dig_answer, record): |
| return |
| |
| print('not match: ', dig_answer, record, |
| list(a == b or (callable(b) and b(a)) for a, b in zip(dig_answer, record))) |
| |
| self.fail((record, dig_result)) |
| |
| def _match_record(self, record, match): |
| assert not any(callable(elem) for elem in record), record |
| |
| if record == match: |
| return True |
| |
| return all(a == b or (callable(b) and b(a)) for a, b in zip(record, match)) |
| |
| def _assert_dig_result_matches(self, dig_result, expected_result): |
| self.assertEqual(dig_result['opcode'], expected_result.get('opcode', 'QUERY'), dig_result) |
| self.assertEqual(dig_result['status'], expected_result.get('status', 'NOERROR'), dig_result) |
| |
| if 'QUESTION' in expected_result: |
| self.assertEqual(len(dig_result['QUESTION']), len(expected_result['QUESTION']), dig_result) |
| |
| for question in expected_result['QUESTION']: |
| self._assert_have_question(dig_result, question) |
| |
| if 'ANSWER' in expected_result: |
| self.assertEqual(len(dig_result['ANSWER']), len(expected_result['ANSWER']), dig_result) |
| |
| for record in expected_result['ANSWER']: |
| self._assert_have_answer(dig_result, record, additional=False) |
| |
| if 'ADDITIONAL' in expected_result: |
| self.assertGreaterEqual(len(dig_result['ADDITIONAL']), len(expected_result['ADDITIONAL']), dig_result) |
| |
| for record in expected_result['ADDITIONAL']: |
| self._assert_have_answer(dig_result, record, additional=True) |
| |
| logging.info("dig result matches:\r%s", json.dumps(dig_result, indent=True)) |
| |
| def _verify_service_browse_result(self, services): |
| assert len(services) == 1, services |
| assert INSTANCE_NAME in services, services |
| service = services[INSTANCE_NAME] |
| logging.info("Service Browse Result: %r", service) |
| self.assertEqual(service['host'], HOST_FULL_NAME) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
