espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-07-10 22:30:21 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
5b2bcee2716cd28bea2a76fda00a880fef5b9568
openthread/tests/scripts/thread-cert/border_router/test_dnssd_server.py
T
Abtin Keshavarzian 20aefc2215 [dnssd] support generic record queries via discovery proxy and SRP (#11357) 
This commit enhances the OpenThread DNSSD name server/resolver and its
native Discovery Proxy to support queries for arbitrary record
types.

To enable this, a new set of `otPlatDnssd` APIs are introduced for
generic `RecordQuerier`. These APIs mirror the existing APIs in the
OpenThread native mDNS module, allowing direct use of the native mDNS
implementation.

The discovery proxy implementation is updated to start and stop the
mDNS `RecordQuerier` when receiving a query for an arbitrary record
type, passing the first response record back to the client.

The unit tests `test_dnssd_discovery_proxy` and `test_dns_client`
are updated to cover all the newly added behaviors in discovery proxy.
2025-04-23 12:14:05 -07:00

345 lines
14 KiB
Python
Executable File
Raw Blame History

#!/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 works on a BR and is accessible from a Host.
#
# Topology:
# ----------------(eth)--------------------
# | |
# BR1 (Leader, Server) HOST
# / \
# CLIENT1 CLIENT2
SERVER = BR1 = 1
CLIENT1, CLIENT2 = 2, 3
HOST = 4
DOMAIN = 'default.service.arpa.'
SERVICE = '_testsrv._udp'
SERVICE_FULL_NAME = f'{SERVICE}.{DOMAIN}'
VALID_SERVICE_NAMES = [
'_abc._udp.default.service.arpa.',
'_abc._tcp.default.service.arpa.',
]
WRONG_SERVICE_NAMES = [
'_testsrv._udp.default.service.xxxx.',
'_testsrv._txp,default.service.arpa.',
]
class TestDnssdServerOnBr(thread_cert.TestCase):
USE_MESSAGE_FACTORY = False
TOPOLOGY = {
BR1: {
'name': 'SERVER',
'is_otbr': True,
'version': '1.2',
},
CLIENT1: {
'name': 'CLIENT1',
},
CLIENT2: {
'name': 'CLIENT2',
},
HOST: {
'name': 'Host',
'is_host': True
},
}
def test(self):
server = br1 = self.nodes[BR1]
client1 = self.nodes[CLIENT1]
client2 = self.nodes[CLIENT2]
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)
server.dns_upstream_query_state = False
client1.start()
self.simulator.go(config.ROUTER_STARTUP_DELAY)
self.assertEqual('router', client1.get_state())
client2.start()
self.simulator.go(config.ROUTER_STARTUP_DELAY)
self.assertEqual('router', client2.get_state())
self.simulator.go(10)
server_addr = server.get_ip6_address(config.ADDRESS_TYPE.OMR)[0]
# Router1 can ping to/from the Host on infra link.
self.assertTrue(br1.ping(host.get_ip6_address(config.ADDRESS_TYPE.ONLINK_ULA)[0], backbone=True))
self.assertTrue(host.ping(br1.get_ip6_address(config.ADDRESS_TYPE.OMR)[0], backbone=True))
client1_addrs = [client1.get_mleid(), client1.get_ip6_address(config.ADDRESS_TYPE.OMR)[0]]
self._config_srp_client_services(client1, 'ins1', 'host1', 11111, 1, 1, client1_addrs)
client2_addrs = [client2.get_mleid(), client2.get_ip6_address(config.ADDRESS_TYPE.OMR)[0]]
self._config_srp_client_services(client2, 'ins2', 'host2', 22222, 2, 2, client2_addrs)
ins1_full_name = f'ins1.{SERVICE_FULL_NAME}'
ins2_full_name = f'ins2.{SERVICE_FULL_NAME}'
host1_full_name = f'host1.{DOMAIN}'
host2_full_name = f'host2.{DOMAIN}'
EMPTY_TXT = {}
# 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'ins1.{SERVICE_FULL_NAME}'),
(SERVICE_FULL_NAME, 'IN', 'PTR', f'ins2.{SERVICE_FULL_NAME}')],
})
# check if SRV query works
dig_result = digger.dns_dig(server_addr, ins1_full_name, 'SRV')
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(ins1_full_name, 'IN', 'SRV')],
'ANSWER': [(ins1_full_name, 'IN', 'SRV', 1, 1, 11111, host1_full_name),],
'ADDITIONAL': [
(host1_full_name, 'IN', 'AAAA', client1_addrs[0]),
(host1_full_name, 'IN', 'AAAA', client1_addrs[1]),
],
})
dig_result = digger.dns_dig(server_addr, ins2_full_name, 'SRV')
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(ins2_full_name, 'IN', 'SRV')],
'ANSWER': [(ins2_full_name, 'IN', 'SRV', 2, 2, 22222, host2_full_name),],
'ADDITIONAL': [
(host2_full_name, 'IN', 'AAAA', client2_addrs[0]),
(host2_full_name, 'IN', 'AAAA', client2_addrs[1]),
],
})
# check if TXT query works
dig_result = digger.dns_dig(server_addr, ins1_full_name, 'TXT')
self._assert_dig_result_matches(dig_result, {
'QUESTION': [(ins1_full_name, 'IN', 'TXT')],
'ANSWER': [(ins1_full_name, 'IN', 'TXT', EMPTY_TXT),],
})
dig_result = digger.dns_dig(server_addr, ins2_full_name, 'TXT')
self._assert_dig_result_matches(dig_result, {
'QUESTION': [(ins2_full_name, 'IN', 'TXT')],
'ANSWER': [(ins2_full_name, 'IN', 'TXT', EMPTY_TXT),],
})
# check if AAAA query works
dig_result = digger.dns_dig(server_addr, host1_full_name, 'AAAA')
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(host1_full_name, 'IN', 'AAAA'),],
'ANSWER': [
(host1_full_name, 'IN', 'AAAA', client1_addrs[0]),
(host1_full_name, 'IN', 'AAAA', client1_addrs[1]),
],
})
dig_result = digger.dns_dig(server_addr, host2_full_name, 'AAAA')
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(host2_full_name, 'IN', 'AAAA'),],
'ANSWER': [
(host2_full_name, 'IN', 'AAAA', client2_addrs[0]),
(host2_full_name, 'IN', 'AAAA', client2_addrs[1]),
],
})
# check some invalid queries
for service_name in WRONG_SERVICE_NAMES:
dig_result = digger.dns_dig(server_addr, service_name, 'PTR')
self._assert_dig_result_matches(dig_result, {
'status': 'NXDOMAIN',
})
# verify Discovery Proxy works for _meshcop._udp
self._verify_discovery_proxy_meshcop(server_addr, server.get_network_name(), digger)
def _verify_discovery_proxy_meshcop(self, server_addr, network_name, digger):
dp_service_name = '_meshcop._udp.default.service.arpa.'
dp_hostname = lambda x: x.endswith('.default.service.arpa.')
def check_border_agent_port(port):
return 0 < port <= 65535
dig_result = digger.dns_dig(server_addr, dp_service_name, 'PTR')
for answer in dig_result['ANSWER']:
if len(answer) >= 2 and answer[-2] == 'PTR':
dp_instance_name = answer[-1]
break
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(dp_service_name, 'IN', 'PTR'),],
'ANSWER': [(dp_service_name, 'IN', 'PTR', dp_instance_name),],
'ADDITIONAL': [
(dp_instance_name, 'IN', 'SRV', 0, 0, check_border_agent_port, dp_hostname),
(dp_instance_name, 'IN', 'TXT', lambda txt: (isinstance(txt, dict) and txt.get(
'nn') == network_name and 'xp' in txt and 'tv' in txt and 'xa' in txt)),
],
})
# Find the actual host name and IPv6 address
dp_ip6_address = None
for rr in dig_result['ADDITIONAL']:
if rr[3] == 'SRV':
dp_hostname = rr[7]
elif rr[3] == 'AAAA':
dp_ip6_address = rr[4]
assert isinstance(dp_hostname, str), dig_result
dig_result = digger.dns_dig(server_addr, dp_instance_name, 'SRV')
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(dp_instance_name, 'IN', 'SRV'),],
'ANSWER': [(dp_instance_name, 'IN', 'SRV', 0, 0, check_border_agent_port, dp_hostname),],
'ADDITIONAL': [(dp_instance_name, 'IN', 'TXT', lambda txt: (isinstance(txt, dict) and txt.get(
'nn') == network_name and 'xp' in txt and 'tv' in txt and 'xa' in txt)),],
})
dig_result = digger.dns_dig(server_addr, dp_instance_name, 'TXT')
self._assert_dig_result_matches(
dig_result, {
'QUESTION': [(dp_instance_name, 'IN', 'TXT'),],
'ANSWER': [(dp_instance_name, 'IN', 'TXT', lambda txt: (isinstance(txt, dict) and txt.get(
'nn') == network_name and 'xp' in txt and 'tv' in txt and 'xa' in txt)),],
'ADDITIONAL': [(dp_instance_name, 'IN', 'SRV', 0, 0, check_border_agent_port, dp_hostname),],
})
if dp_ip6_address is not None:
dig_result = digger.dns_dig(server_addr, dp_hostname, 'AAAA')
self._assert_dig_result_matches(dig_result, {
'QUESTION': [(dp_hostname, 'IN', 'AAAA'),],
'ANSWER': [(dp_hostname, 'IN', 'AAAA', dp_ip6_address),],
},
allow_extra_answer=True)
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, allow_extra_answer=False):
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:
if allow_extra_answer:
self.assertGreaterEqual(len(dig_result['ANSWER']), len(expected_result['ANSWER']), dig_result)
else:
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))
if __name__ == '__main__':
unittest.main()
Reference in New Issue View Git Blame Copy Permalink