diff --git a/tools/otci/otci/otci.py b/tools/otci/otci/otci.py index 08283e72d..01cf44dbb 100644 --- a/tools/otci/otci/otci.py +++ b/tools/otci/otci/otci.py @@ -719,6 +719,397 @@ class OTCI(object): return table + # + # DNS server & client utilities + # + + _IPV6_SERVER_PORT_PATTERN = re.compile(r'\[(.*)\]:(\d+)') + + def dns_get_config(self): + """Get DNS client query config.""" + output = self.execute_command('dns config') + config = {} + for line in output: + k, v = line.split(': ') + if k == 'Server': + ip, port = re.match(OTCI._IPV6_SERVER_PORT_PATTERN, v).groups() + config['server'] = (Ip6Addr(ip), int(port)) + elif k == 'ResponseTimeout': + config['response_timeout'] = int(v[:-3]) + elif k == 'MaxTxAttempts': + config['max_tx_attempts'] = int(v) + elif k == 'RecursionDesired': + config['recursion_desired'] = (v == 'yes') + else: + logging.warning("dns config ignored: %s", line) + + return config + + def dns_set_config(self, + server: Tuple[Union[str, ipaddress.IPv6Address], int], + response_timeout: int = None, + max_tx_attempts: int = None, + recursion_desired: bool = None): + """Set DNS client query config.""" + cmd = f'dns config {str(server[0])} {server[1]}' + if response_timeout is not None: + cmd += f' {response_timeout}' + + assert max_tx_attempts is None or response_timeout is not None, "must specify `response_timeout` if `max_tx_attempts` is specified." + if max_tx_attempts is not None: + cmd += f' {max_tx_attempts}' + + assert recursion_desired is None or max_tx_attempts is not None, 'must specify `max_tx_attempts` if `recursion_desired` is specified.' + if recursion_desired is not None: + cmd += f' {1 if recursion_desired else 0}' + + self.execute_command(cmd) + + def dns_get_compression(self) -> bool: + """Get DNS compression mode.""" + return self.__parse_Enabled_or_Disabled(self.execute_command('dns compression')) + + def dns_enable_compression(self): + """Enable DNS compression mode.""" + self.execute_command('dns compression enable') + + def dns_disable_compression(self): + """Disable DNS compression mode.""" + self.execute_command('dns compression disable') + + def dns_browse(self, service: str) -> List[Dict]: + """Browse DNS service instances.""" + cmd = f'dns browse {service}' + output = '\n'.join(self.execute_command(cmd)) + + result = [] + for ins, port, priority, weight, srv_ttl, hostname, address, aaaa_ttl, txt_data, txt_ttl in re.findall( + r'(.*?)\s+Port:(\d+), Priority:(\d+), Weight:(\d+), TTL:(\d+)\s*Host:(\S+)\s+HostAddress:(\S+) TTL:(\d+)\s+TXT:(\[.*?\]) TTL:(\d+)', + output): + result.append({ + 'instance': ins, + 'service': service, + 'port': int(port), + 'priority': int(priority), + 'weight': int(weight), + 'host': hostname, + 'address': Ip6Addr(address), + 'txt': self.__parse_srp_server_service_txt(txt_data), + 'srv_ttl': int(srv_ttl), + 'txt_ttl': int(txt_ttl), + 'aaaa_ttl': int(aaaa_ttl), + }) + + return result + + def dns_resolve(self, hostname: str) -> List[Dict]: + """Resolve a DNS host name.""" + cmd = f'dns resolve {hostname}' + output = self.execute_command(cmd) + dns_resp = output[0] + addrs = dns_resp.strip().split(' - ')[1].split(' ') + ips = [Ip6Addr(item.strip()) for item in addrs[::2]] + ttls = [int(item.split('TTL:')[1]) for item in addrs[1::2]] + + return [{ + 'address': ip, + 'ttl': ttl, + } for ip, ttl in zip(ips, ttls)] + + def dns_resolve_service(self, instance: str, service: str) -> Dict: + """Resolves aservice instance.""" + cmd = f'dns service {instance} {service}' + output = self.execute_command(cmd) + + m = re.match( + r'.*Port:(\d+), Priority:(\d+), Weight:(\d+), TTL:(\d+)\s+Host:(.*?)\s+HostAddress:(\S+) TTL:(\d+)\s+TXT:(\[.*?\]) TTL:(\d+)', + '\t'.join(output)) + if m: + port, priority, weight, srv_ttl, hostname, address, aaaa_ttl, txt_data, txt_ttl = m.groups() + return { + 'instance': instance, + 'service': service, + 'port': int(port), + 'priority': int(priority), + 'weight': int(weight), + 'host': hostname, + 'address': Ip6Addr(address), + 'txt': self.__parse_srp_server_service_txt(txt_data), + 'srv_ttl': int(srv_ttl), + 'txt_ttl': int(txt_ttl), + 'aaaa_ttl': int(aaaa_ttl), + } + else: + raise CommandError(cmd, output) + + # + # SRP server & client utilities + # + + def srp_server_enable(self): + """Enable SRP server.""" + self.execute_command('srp server enable') + + def srp_server_disable(self): + """Disable SRP server.""" + self.execute_command('srp server disable') + + def srp_server_get_domain(self) -> str: + """Get the SRP server domain.""" + return self.__parse_str(self.execute_command('srp server domain')) + + def srp_server_set_domain(self, domain: str): + """Set the SRP server domain.""" + self.execute_command(f'srp server domain {domain}') + + def srp_server_get_hosts(self) -> List[Dict]: + """Get SRP server registered hosts.""" + return self.__parse_srp_server_hosts(self.execute_command('srp server host')) + + def srp_server_get_services(self) -> List[Dict]: + """Get SRP server registered services.""" + output = self.execute_command('srp server service') + return self.__parse_srp_server_services(output) + + def __parse_srp_server_hosts(self, output: List[str]) -> List[Dict]: + result = [] + info = None + for line in output: + if not line.startswith(' '): + info = {'host': line} + result.append(info) + else: + k, v = line.strip().split(': ') + if k == 'deleted': + if v not in ('true', 'false'): + raise UnexpectedCommandOutput(output) + + info['deleted'] = (v == 'true') + + elif k == 'addresses': + if not v.startswith('[') or not v.endswith(']'): + raise UnexpectedCommandOutput(output) + + v = v[1:-1] + info['addresses'] = list(map(Ip6Addr, v.split(', '))) + else: + raise UnexpectedCommandOutput(output) + + return result + + def __parse_srp_server_services(self, output: List[str]) -> List[Dict]: + result = [] + info = None + for line in output: + if not line.startswith(' '): + info = {'instance': line} + result.append(info) + else: + k, v = line.strip().split(': ') + if k == 'deleted': + if v not in ('true', 'false'): + raise UnexpectedCommandOutput(output) + + info['deleted'] = (v == 'true') + + elif k == 'addresses': + if not v.startswith('[') or not v.endswith(']'): + raise UnexpectedCommandOutput(output) + + v = v[1:-1] + info['addresses'] = list(map(Ip6Addr, v.split(', '))) + elif k in ('port', 'weight', 'priority'): + info[k] = int(v) + elif k in ('host',): + info[k] = v + elif k == 'TXT': + info['txt'] = self.__parse_srp_server_service_txt(v) + else: + raise UnexpectedCommandOutput(output) + + return result + + def __parse_srp_server_service_txt(self, txt: str) -> Dict[str, Union[bytes, bool]]: + # example value: [txt11=76616c3131, txt12=76616c3132] + assert txt.startswith('[') and txt.endswith(']') + txt_dict = {} + for entry in txt[1:-1].split(', '): + if not entry: + continue + + equal_pos = entry.find('=') + + if equal_pos != -1: + k, v = entry[:equal_pos], entry[equal_pos + 1:] + txt_dict[k] = bytes(int(v[i:i + 2], 16) for i in range(0, len(v), 2)) + else: + txt_dict[entry] = True + + return txt_dict + + def srp_server_set_lease(self, min_lease: int, max_lease: int, min_key_lease: int, max_key_lease: int): + """Configure SRP server LEASE & KEY-LEASE range (in seconds).""" + self.execute_command(f'srp server lease {min_lease} {max_lease} {min_key_lease} {max_key_lease}') + + def srp_client_get_state(self) -> bool: + """Get SRP client state.""" + return self.__parse_Enabled_or_Disabled(self.execute_command('srp client state')) + + def srp_client_start(self, server_ip: Union[str, ipaddress.IPv6Address], server_port: int): + """Start SRP client.""" + self.execute_command(f'srp client start {str(server_ip)} {server_port}') + + def srp_client_stop(self): + """Stop SRP client.""" + self.execute_command('srp client stop') + + def srp_client_get_autostart(self) -> bool: + """Get SRP client autostart mode.""" + return self.__parse_Enabled_or_Disabled(self.execute_command('srp client autostart')) + + def srp_client_enable_autostart(self): + """Enable SRP client autostart mode.""" + self.execute_command('srp client autostart enable') + + def srp_client_disable_autostart(self): + """Disable SRP client autostart mode.""" + self.execute_command('srp client autostart disable') + + def srp_client_get_callback(self) -> bool: + """Get SRP client callback mode.""" + return self.__parse_Enabled_or_Disabled(self.execute_command('srp client callback')) + + def srp_client_enable_callback(self): + """Enable SRP client callback mode.""" + self.execute_command('srp client callback enable') + + def srp_client_disable_callback(self): + """Disable SRP client callback mode.""" + self.execute_command('srp client callback disable') + + def srp_client_set_host_name(self, name: str): + """Set SRP client host name.""" + self.execute_command(f'srp client host name {name}') + + def srp_client_get_host(self) -> Dict: + """Get SRP client host.""" + output = self.__parse_str(self.execute_command('srp client host')) + return self.__parse_srp_client_host(output) + + _SRP_CLIENT_HOST_PATTERN = re.compile(r'name:("(.*)"|(\(null\))), state:(\S+), addrs:\[(.*)\]') + + def __parse_srp_client_host(self, line: str) -> Dict: + m = re.match(OTCI._SRP_CLIENT_HOST_PATTERN, line) + if not m: + raise UnexpectedCommandOutput([line]) + + _, host, _, state, addrs = m.groups() + return { + 'host': host or '', + 'state': state, + 'addresses': [Ip6Addr(ip) for ip in addrs.split(', ')] if addrs else [], + } + + def srp_client_get_host_name(self) -> str: + """Get SRP client host name.""" + name = self.__parse_str(self.execute_command('srp client host name')) + return name if name != '(null)' else '' + + def srp_client_get_host_addresses(self) -> List[Ip6Addr]: + """Get SRP client host addresses.""" + return self.__parse_ip6addr_list(self.execute_command('srp client host address')) + + def srp_client_set_host_addresses(self, *addrs: Union[str, ipaddress.IPv6Address]): + """Set SRP client host addresses.""" + self.execute_command(f'srp client host address {" ".join(map(str, addrs))}') + + def srp_client_get_host_state(self): + """Get SRP client host state.""" + return self.__parse_str(self.execute_command('srp client host state')) + + def srp_client_remove_host(self, remove_key_lease=False): + """Remove SRP client host.""" + cmd = 'srp client host remove' + if remove_key_lease: + cmd += ' 1' + + self.execute_command(cmd) + + def srp_client_get_services(self) -> List[Dict]: + """Get SRP client services.""" + output = self.execute_command('srp client service') + return [self.__parse_srp_client_service(line) for line in output] + + _SRP_CLIENT_SERVICE_PATTERN = re.compile( + r'instance:"(.*)", name:"(.*)", state:(\S+), port:(\d+), priority:(\d+), weight:(\d+)') + + def __parse_srp_client_service(self, line: str) -> Dict: + # e.g. instance:"ins2", name:"_meshcop._udp", state:ToAdd, port:2000, priority:2, weight:2 + m = OTCI._SRP_CLIENT_SERVICE_PATTERN.match(line) + if m is None: + raise UnexpectedCommandOutput([line]) + + instance, service, state, port, priority, weight = m.groups() + port, priority, weight = int(port), int(priority), int(weight) + return { + 'instance': instance, + 'service': service, + 'state': state, + 'port': port, + 'priority': priority, + 'weight': weight, + } + + def srp_client_add_service(self, + instance: str, + service: str, + port: int, + priority: int = 0, + weight: int = 0, + txt: Dict[str, Union[str, bytes, bool]] = None): + cmd = f'srp client service add {instance} {service} {port} {priority} {weight}' + if txt: + cmd += f' {self.__txt_to_hex(txt)}' + self.execute_command(cmd) + + def srp_client_remove_service(self, instance: str, service: str): + """Remove a service from SRP client.""" + self.execute_command(f'srp client service remove {instance} {service}') + + def srp_client_get_key_lease_interval(self) -> int: + """Get SRP client key lease interval (in seconds).""" + return self.__parse_int(self.execute_command('srp client keyleaseinterval')) + + def srp_client_set_key_lease_interval(self, interval: int): + """Set SRP client key lease interval (in seconds).""" + self.execute_command(f'srp client keyleaseinterval {interval}') + + def srp_client_get_lease_interval(self) -> int: + """Get SRP client lease interval (in seconds).""" + return self.__parse_int(self.execute_command('srp client leaseinterval')) + + def srp_client_set_lease_interval(self, interval: int): + """Set SRP client lease interval (in seconds).""" + self.execute_command(f'srp client leaseinterval {interval}') + + def srp_client_get_server(self) -> Tuple[Ip6Addr, int]: + """Get the SRP server (IP, port).""" + result = self.__parse_str(self.execute_command('srp client server')) + ip, port = re.match(OTCI._IPV6_SERVER_PORT_PATTERN, result).groups() + return Ip6Addr(ip), int(port) + + def srp_client_get_service_key(self) -> bool: + """Get SRP client "service key record inclusion" mode.""" + return self.__parse_Enabled_or_Disabled(self.execute_command('srp client service key')) + + def srp_client_enable_service_key(self): + """Enable SRP client "service key record inclusion" mode.""" + self.execute_command('srp client service key enable') + + def srp_client_disable_service_key(self): + """Disable SRP client "service key record inclusion" mode.""" + self.execute_command('srp client service key disable') + def __split_table_row(self, row: str) -> List[str]: if not (row.startswith('|') and row.endswith('|')): raise ValueError(row) @@ -1847,6 +2238,9 @@ class OTCI(object): def __parse_ip6addr(self, output: List[str]) -> Ip6Addr: return Ip6Addr(self.__parse_str(output)) + def __parse_ip6addr_list(self, output: List[str]) -> List[Ip6Addr]: + return [Ip6Addr(line) for line in output] + def __parse_int(self, output: List[str], base=10) -> int: if len(output) != 1: raise UnexpectedCommandOutput(output) @@ -1935,6 +2329,26 @@ class OTCI(object): # So we sleep for a while after reset. time.sleep(3) + def __txt_to_hex(self, txt: Dict[str, Union[str, bytes, bool]]) -> str: + txt_bin = b'' + for k, v in txt.items(): + assert '=' not in k, 'TXT key must not contain `=`' + + if isinstance(v, str): + entry = f'{k}={v}'.encode('utf8') + elif isinstance(v, bytes): + entry = f'{k}='.encode('utf8') + v + else: + assert v is True, 'TXT val must be str or bytes or True' + entry = k.encode('utf8') + + assert len(entry) <= 255, 'TXT entry is too long' + + txt_bin += bytes([len(entry)]) + txt_bin += entry + + return ''.join('%02x' % b for b in txt_bin) + def connect_cli_sim(executable: str, nodeid: int, simulator: Optional[Simulator] = None) -> OTCI: cli_handler = connectors.OtCliSim(executable, nodeid, simulator=simulator) diff --git a/tools/otci/tests/test_otci.py b/tools/otci/tests/test_otci.py index 0e0b2161c..aa2f21d7f 100644 --- a/tools/otci/tests/test_otci.py +++ b/tools/otci/tests/test_otci.py @@ -26,6 +26,7 @@ # 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 os @@ -33,9 +34,10 @@ import subprocess import unittest import otci +from otci import OTCI from otci.errors import CommandError -logging.basicConfig(level=logging.INFO) +logging.basicConfig(level=logging.DEBUG) TEST_CHANNEL = 22 TEST_NETWORK_NAME = 'OT CI' @@ -318,6 +320,178 @@ class TestOTCI(unittest.TestCase): logging.info('dataset active -x: %r', leader.get_dataset_bytes('active')) logging.info('dataset pending -x: %r', leader.get_dataset_bytes('pending')) + # Test SRP server & client + self._test_otci_srp(leader, leader) + + # Test DNS client and server + self._test_otci_dns(leader, leader) + + self._test_otci_srp_remove(leader, leader) + + def _test_otci_dns(self, client: OTCI, server: OTCI): + dns_cfg = client.dns_get_config() + self.assertTrue(dns_cfg['server']) + self.assertIn('response_timeout', dns_cfg) + self.assertIn('max_tx_attempts', dns_cfg) + self.assertIn('recursion_desired', dns_cfg) + + client.dns_set_config(server=(server.get_ipaddr_rloc(), 53), + response_timeout=10000, + max_tx_attempts=4, + recursion_desired=False) + self.assertEqual( + { + 'server': (server.get_ipaddr_rloc(), 53), + 'response_timeout': 10000, + 'max_tx_attempts': 4, + 'recursion_desired': False + }, client.dns_get_config()) + + self.assertTrue(client.dns_get_compression()) + client.dns_disable_compression() + self.assertFalse(client.dns_get_compression()) + client.dns_enable_compression() + self.assertTrue(client.dns_get_compression()) + + logging.info('dns browse: %r', client.dns_browse('_ipps._tcp.default.service.arpa.')) + logging.info('dns browse: %r', client.dns_browse('_meshcop._udp.default.service.arpa.')) + logging.info('dns resolve: %r', client.dns_resolve_service('ins1', '_ipps._tcp.default.service.arpa.')) + logging.info('dns resolve: %r', client.dns_resolve('host1.default.service.arpa.')) + + def _test_otci_srp(self, client: OTCI, server: OTCI): + self.assertEqual('default.service.arpa.', server.srp_server_get_domain()) + server.srp_server_set_domain('example1.com') + self.assertEqual('example1.com.', server.srp_server_get_domain()) + server.srp_server_set_domain('example2.com.') + self.assertEqual('example2.com.', server.srp_server_get_domain()) + server.srp_server_set_domain('default.service.arpa.') + self.assertEqual('default.service.arpa.', server.srp_server_get_domain()) + + self.assertFalse(client.srp_client_get_state()) + self.assertEqual('Removed', client.srp_client_get_host_state()) + self.assertEqual(('::', 0), client.srp_client_get_server()) + + self.assertFalse(client.srp_client_get_service_key()) + client.srp_client_enable_service_key() + self.assertTrue(client.srp_client_get_service_key()) + client.srp_client_disable_service_key() + self.assertFalse(client.srp_client_get_service_key()) + + server.srp_server_disable() + client.wait(3) + server.srp_server_enable() + client.wait(3) + self.assertEqual([], server.srp_server_get_hosts()) + + self.assertFalse(client.srp_client_get_autostart()) + client.srp_client_enable_autostart() + self.assertTrue(client.srp_client_get_autostart()) + client.wait(3) + self.assertTrue(client.srp_client_get_state()) + self.assertNotEqual(('::', 0), client.srp_client_get_server()) + + self.assertEqual('', client.srp_client_get_host_name()) + client.srp_client_set_host_name('host1') + self.assertEqual('host1', client.srp_client_get_host_name()) + + self.assertEqual([], client.srp_client_get_host_addresses()) + client.srp_client_set_host_addresses('2001::1') + self.assertEqual(['2001::1'], client.srp_client_get_host_addresses()) + client.srp_client_set_host_addresses('2001::1', '2001::2') + self.assertEqual(['2001::1', '2001::2'], client.srp_client_get_host_addresses()) + srp_client_host = client.srp_client_get_host() + self.assertEqual('host1', srp_client_host['host']) + self.assertEqual('ToAdd', srp_client_host['state']) + self.assertEqual( + {ipaddress.IPv6Address('2001::1'), ipaddress.IPv6Address('2001::2')}, set(srp_client_host['addresses'])) + + self.assertEqual([], client.srp_client_get_services()) + client.srp_client_add_service('ins1', + '_ipps._tcp', + 1000, + 1, + 1, + txt={ + 'txt11': 'val11', + 'txt12': b'val12', + 'txt13': True + }) + client.srp_client_add_service('ins2', + '_meshcop._udp', + 2000, + 2, + 2, + txt={ + 'txt21': 'val21', + 'txt22': b'val22', + 'txt23': True + }) + self.assertEqual(2, len(client.srp_client_get_services())) + self.assertIn( + { + 'instance': 'ins1', + 'service': '_ipps._tcp', + 'state': 'ToAdd', + 'port': 1000, + 'priority': 1, + 'weight': 1, + }, client.srp_client_get_services()) + self.assertIn( + { + 'instance': 'ins2', + 'service': '_meshcop._udp', + 'state': 'ToAdd', + 'port': 2000, + 'priority': 2, + 'weight': 2, + }, client.srp_client_get_services()) + + client.wait(3) + + self.assertEqual('Registered', client.srp_client_get_host()['state']) + + srp_server_hosts = server.srp_server_get_hosts() + logging.info('srp_server_hosts %r', srp_server_hosts) + self.assertEqual(1, len(srp_server_hosts)) + self.assertEqual('host1.default.service.arpa.', srp_server_hosts[0]['host']) + self.assertEqual(False, srp_server_hosts[0]['deleted']) + self.assertEqual( + {ipaddress.IPv6Address('2001::1'), ipaddress.IPv6Address('2001::2')}, + set(srp_server_hosts[0]['addresses'])) + + srp_server_services = server.srp_server_get_services() + logging.info('srp_server_services %r', srp_server_services) + self.assertEqual(2, len(srp_server_services)) + for service in srp_server_services: + if service['instance'] == 'ins1._ipps._tcp.default.service.arpa.': + self.assertEqual(False, service['deleted']) + self.assertEqual(1000, service['port']) + self.assertEqual(1, service['priority']) + self.assertEqual(1, service['weight']) + self.assertEqual('host1.default.service.arpa.', service['host']) + self.assertEqual({ipaddress.IPv6Address('2001::1'), + ipaddress.IPv6Address('2001::2')}, set(service['addresses'])) + self.assertEqual({'txt11': b'val11', 'txt12': b'val12', 'txt13': True}, service['txt']) + elif service['instance'] == 'ins2._meshcop._udp.default.service.arpa.': + self.assertEqual(False, service['deleted']) + self.assertEqual(2000, service['port']) + self.assertEqual(2, service['priority']) + self.assertEqual(2, service['weight']) + self.assertEqual('host1.default.service.arpa.', service['host']) + self.assertEqual({ipaddress.IPv6Address('2001::1'), + ipaddress.IPv6Address('2001::2')}, set(service['addresses'])) + self.assertEqual({'txt21': b'val21', 'txt22': b'val22', 'txt23': True}, service['txt']) + else: + self.fail(service) + + def _test_otci_srp_remove(self, client: OTCI, server: OTCI): + client.srp_client_remove_host(remove_key_lease=True) + client.wait(3) + self.assertEqual([], client.srp_client_get_services()) + self.assertEqual('Removed', client.srp_client_get_host()['state']) + self.assertEqual([], server.srp_server_get_hosts()) + self.assertEqual([], server.srp_server_get_services()) + def _test_otci_example(self, node1, node2): node1.dataset_init_buffer() node1.dataset_set_buffer(network_name='test',