1#!/usr/bin/env python3
2#
3#  Copyright (c) 2023, 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 unittest
30
31import config
32import thread_cert
33
34import ipaddress
35import shlex
36
37# Test description:
38#   This test verifies forwarding DNS queries sent by 'Router' by using
39# a record resolved by BIND9 server.
40#
41# Topology:
42#    ----------------(eth)--------------------
43#           |                 |
44#          BR (Leader)      HOST
45#           |
46#        ROUTER
47#
48
49BR = 1
50ROUTER = 2
51HOST = 3
52
53TEST_DOMAIN = 'test.domain'
54TEST_DOMAIN_IP6_ADDRESSES = {'2001:db8::1'}
55
56TEST_DOMAIN_BIND_CONF = f'''
57zone "{TEST_DOMAIN}" {{ type master; file "/etc/bind/db.test.domain"; }};
58'''
59
60TEST_DOMAIN_BIND_ZONE = f'''
61$TTL 24h
62@ IN SOA {TEST_DOMAIN} test.{TEST_DOMAIN}. ( 20230330 86400 300 604800 3600 )
63@ IN NS {TEST_DOMAIN}.
64''' + '\n'.join(f'@ IN AAAA {addr}' for addr in TEST_DOMAIN_IP6_ADDRESSES)
65
66
67class UpstreamDns(thread_cert.TestCase):
68    USE_MESSAGE_FACTORY = False
69
70    TOPOLOGY = {
71        BR: {
72            'name': 'BR',
73            'allowlist': [ROUTER],
74            'is_otbr': True,
75            'version': '1.3',
76        },
77        ROUTER: {
78            'name': 'Router',
79            'allowlist': [BR],
80            'version': '1.3',
81        },
82        HOST: {
83            'name': 'Host',
84            'is_host': True
85        },
86    }
87
88    def test(self):
89        br = self.nodes[BR]
90        router = self.nodes[ROUTER]
91        host = self.nodes[HOST]
92
93        host.start(start_radvd=False)
94        self.simulator.go(5)
95
96        br.start()
97        # When feature flag is enabled, NAT64 might be disabled by default. So
98        # ensure NAT64 is enabled here.
99        self.simulator.go(config.LEADER_STARTUP_DELAY)
100        self.assertEqual('leader', br.get_state())
101
102        br.nat64_set_enabled(True)
103        br.srp_server_set_enabled(True)
104
105        br.bash('service bind9 stop')
106
107        br.bash(shlex.join(['echo', TEST_DOMAIN_BIND_CONF]) + ' >> /etc/bind/named.conf.local')
108        br.bash(shlex.join(['echo', TEST_DOMAIN_BIND_ZONE]) + ' >> /etc/bind/db.test.domain')
109
110        br.bash('service bind9 start')
111
112        router.start()
113        self.simulator.go(config.ROUTER_STARTUP_DELAY)
114        self.assertEqual('router', router.get_state())
115
116        self.simulator.go(10)
117        router.srp_client_enable_auto_start_mode()
118
119        # verify the server can forward the DNS query to upstream server.
120        self._verify_upstream_dns(br, router)
121
122    def _verify_upstream_dns(self, br, ed):
123        upstream_dns_enabled = br.dns_upstream_query_state
124        if not upstream_dns_enabled:
125            br.dns_upstream_query_state = True
126        self.assertTrue(br.dns_upstream_query_state)
127
128        resolved_names = ed.dns_resolve(TEST_DOMAIN)
129        self.assertEqual(len(resolved_names), len(TEST_DOMAIN_IP6_ADDRESSES))
130        for record in resolved_names:
131            self.assertIn(ipaddress.IPv6Address(record[0]).compressed, TEST_DOMAIN_IP6_ADDRESSES)
132
133
134if __name__ == '__main__':
135    unittest.main()
136