# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Apache-2.0 import re import select import socket import struct import time from threading import Event, Thread import dpkt import dpkt.dns import pytest UDP_PORT = 5353 MCAST_GRP = '224.0.0.251' # This service is created from esp board startup SERVICE_NAME = u'ESP32-WebServer._http._tcp.local' SUB_SERVICE_NAME = u'_server._sub._http._tcp.local' # This host name answer sent by host, when there is query from board HOST_NAME = u'tinytester.local' # This servce answer sent by host, when there is query from board MDNS_HOST_SERVICE = u'ESP32._http._tcp.local' # Number of retries to receive mdns answear RETRY_COUNT = 10 stop_mdns_listener = Event() start_mdns_listener = Event() esp_service_answered = Event() esp_sub_service_answered = Event() esp_host_answered = Event() esp_delegated_host_answered = Event() @pytest.mark.skip # Get query of ESP32-WebServer._http._tcp.local service def get_mdns_service_query(service): # type:(str) -> dpkt.dns.Msg dns = dpkt.dns.DNS() dns.op = dpkt.dns.DNS_QR | dpkt.dns.DNS_AA dns.rcode = dpkt.dns.DNS_RCODE_NOERR arr = dpkt.dns.DNS.RR() arr.cls = dpkt.dns.DNS_IN arr.type = dpkt.dns.DNS_SRV arr.name = service arr.target = socket.inet_aton('127.0.0.1') arr.srvname = service dns.qd.append(arr) print('Created mdns service query: {} '.format(dns.__repr__())) return dns.pack() @pytest.mark.skip # Get query of _server_.sub._http._tcp.local sub service def get_mdns_sub_service_query(sub_service): # type:(str) -> dpkt.dns.Msg dns = dpkt.dns.DNS() dns.op = dpkt.dns.DNS_QR | dpkt.dns.DNS_AA dns.rcode = dpkt.dns.DNS_RCODE_NOERR arr = dpkt.dns.DNS.RR() arr.cls = dpkt.dns.DNS_IN arr.type = dpkt.dns.DNS_PTR arr.name = sub_service arr.target = socket.inet_aton('127.0.0.1') arr.ptrname = sub_service dns.qd.append(arr) print('Created mdns subtype service query: {} '.format(dns.__repr__())) return dns.pack() @pytest.mark.skip # Get query for host resolution def get_dns_query_for_esp(esp_host): # type:(str) -> dpkt.dns.Msg dns = dpkt.dns.DNS() arr = dpkt.dns.DNS.RR() arr.cls = dpkt.dns.DNS_IN arr.name = esp_host + u'.local' dns.qd.append(arr) print('Created query for esp host: {} '.format(dns.__repr__())) return dns.pack() @pytest.mark.skip # Get mdns answer for host resoloution def get_dns_answer_to_mdns(tester_host): # type:(str) -> dpkt.dns.Msg dns = dpkt.dns.DNS() dns.op = dpkt.dns.DNS_QR | dpkt.dns.DNS_AA dns.rcode = dpkt.dns.DNS_RCODE_NOERR arr = dpkt.dns.DNS.RR() arr.cls = dpkt.dns.DNS_IN arr.type = dpkt.dns.DNS_A arr.name = tester_host arr.ip = socket.inet_aton('127.0.0.1') dns.an.append(arr) print('Created answer to mdns query: {} '.format(dns.__repr__())) return dns.pack() @pytest.mark.skip # Get mdns answer for service query def get_dns_answer_to_service_query( mdns_service): # type:(str) -> dpkt.dns.Msg dns = dpkt.dns.DNS() dns.op = dpkt.dns.DNS_QR | dpkt.dns.DNS_AA dns.rcode = dpkt.dns.DNS_RCODE_NOERR arr = dpkt.dns.DNS.RR() arr.name = mdns_service arr.cls = dpkt.dns.DNS_IN arr.type = dpkt.dns.DNS_SRV arr.priority = 0 arr.weight = 0 arr.port = 100 arr.srvname = mdns_service arr.ip = socket.inet_aton('127.0.0.1') dns.an.append(arr) print('Created answer to mdns query: {} '.format(dns.__repr__())) return dns.pack() @pytest.mark.skip def mdns_listener(esp_host): # type:(str) -> None print('mdns_listener thread started') UDP_IP = '0.0.0.0' sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) sock.setblocking(False) sock.bind((UDP_IP, UDP_PORT)) mreq = struct.pack('4sl', socket.inet_aton(MCAST_GRP), socket.INADDR_ANY) sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) last_query_timepoint = time.time() QUERY_TIMEOUT = 0.2 while not stop_mdns_listener.is_set(): try: start_mdns_listener.set() current_time = time.time() if current_time - last_query_timepoint > QUERY_TIMEOUT: last_query_timepoint = current_time timeout = max( 0, QUERY_TIMEOUT - (current_time - last_query_timepoint)) read_socks, _, _ = select.select([sock], [], [], timeout) if not read_socks: continue data, _ = sock.recvfrom(1024) dns = dpkt.dns.DNS(data) # Receives queries from esp board and sends answers if len(dns.qd) > 0: if dns.qd[0].name == HOST_NAME: print('Received query: {} '.format(dns.__repr__())) sock.sendto(get_dns_answer_to_mdns(HOST_NAME), (MCAST_GRP, UDP_PORT)) if dns.qd[0].name == HOST_NAME: print('Received query: {} '.format(dns.__repr__())) sock.sendto(get_dns_answer_to_mdns(HOST_NAME), (MCAST_GRP, UDP_PORT)) if dns.qd[0].name == MDNS_HOST_SERVICE: print('Received query: {} '.format(dns.__repr__())) sock.sendto( get_dns_answer_to_service_query(MDNS_HOST_SERVICE), (MCAST_GRP, UDP_PORT)) # Receives answers from esp board and sets event flags for python test cases if len(dns.an) == 1: if dns.an[0].name.startswith(SERVICE_NAME): print('Received answer to service query: {}'.format( dns.__repr__())) esp_service_answered.set() if len(dns.an) > 1: if dns.an[1].name.startswith(SUB_SERVICE_NAME): print('Received answer for sub service query: {}'.format( dns.__repr__())) esp_sub_service_answered.set() if len(dns.an) > 0 and dns.an[0].type == dpkt.dns.DNS_A: if dns.an[0].name == esp_host + u'.local': print('Received answer to esp32-mdns query: {}'.format( dns.__repr__())) esp_host_answered.set() if dns.an[0].name == esp_host + u'-delegated.local': print('Received answer to esp32-mdns-delegate query: {}'. format(dns.__repr__())) esp_delegated_host_answered.set() except socket.timeout: break except dpkt.UnpackError: continue @pytest.mark.skip def create_socket(): # type:() -> socket.socket UDP_IP = '0.0.0.0' sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) sock.setblocking(False) sock.bind((UDP_IP, UDP_PORT)) mreq = struct.pack('4sl', socket.inet_aton(MCAST_GRP), socket.INADDR_ANY) sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) return sock @pytest.mark.skip def test_query_dns_http_service(service): # type: (str) -> None print('SRV: Query {}'.format(service)) sock = create_socket() packet = get_mdns_service_query(service) for _ in range(RETRY_COUNT): if esp_service_answered.wait(timeout=25): break sock.sendto(packet, (MCAST_GRP, UDP_PORT)) else: raise RuntimeError( 'Test has failed: did not receive mdns answer within timeout') @pytest.mark.skip def test_query_dns_sub_service(sub_service): # type: (str) -> None print('PTR: Query {}'.format(sub_service)) sock = create_socket() packet = get_mdns_sub_service_query(sub_service) for _ in range(RETRY_COUNT): if esp_sub_service_answered.wait(timeout=25): break sock.sendto(packet, (MCAST_GRP, UDP_PORT)) else: raise RuntimeError( 'Test has failed: did not receive mdns answer within timeout') @pytest.mark.skip def test_query_dns_host(esp_host): # type: (str) -> None print('A: {}'.format(esp_host)) sock = create_socket() packet = get_dns_query_for_esp(esp_host) for _ in range(RETRY_COUNT): if esp_host_answered.wait(timeout=25): break sock.sendto(packet, (MCAST_GRP, UDP_PORT)) else: raise RuntimeError( 'Test has failed: did not receive mdns answer within timeout') @pytest.mark.skip def test_query_dns_host_delegated(esp_host): # type: (str) -> None print('A: {}'.format(esp_host)) sock = create_socket() packet = get_dns_query_for_esp(esp_host + '-delegated') for _ in range(RETRY_COUNT): if esp_delegated_host_answered.wait(timeout=25): break sock.sendto(packet, (MCAST_GRP, UDP_PORT)) else: raise RuntimeError( 'Test has failed: did not receive mdns answer within timeout') @pytest.mark.esp32 @pytest.mark.esp32s2 @pytest.mark.esp32c3 @pytest.mark.generic def test_app_esp_mdns(dut): # 1. get the dut host name (and IP address) specific_host = dut.expect( re.compile(b'mdns hostname set to: \[([^\]]+)\]'), # noqa: W605 timeout=30).group(1).decode() esp_ip = dut.expect( re.compile( b' IPv4 address: ([0-9]+\.[0-9]+\.[0-9]+\.[0-9]+)'), # noqa: W605 timeout=30).group(1).decode() print('Got IP={}'.format(esp_ip)) mdns_responder = Thread(target=mdns_listener, args=(str(specific_host), )) def start_case(case, desc, result): # type: (str, str, str) -> None print('Starting {}: {}'.format(case, desc)) dut.write(case) res = bytes(result, encoding='utf8') dut.expect(re.compile(res), timeout=30) try: # start dns listener thread mdns_responder.start() # wait untill mdns listener thred started if not start_mdns_listener.wait(timeout=5): raise ValueError( 'Test has failed: mdns listener thread did not start') # query dns service from host, answer should be received from esp board test_query_dns_http_service(SERVICE_NAME) # query dns sub-service from host, answer should be received from esp board test_query_dns_sub_service(SUB_SERVICE_NAME) # query dns host name, answer should be received from esp board test_query_dns_host(specific_host) # query dns host name delegated, answer should be received from esp board test_query_dns_host_delegated(specific_host) # query service from esp board, answer should be received from host start_case('CONFIG_TEST_QUERY_SERVICE', 'Query SRV ESP32._http._tcp.local', 'SRV:ESP32') # query dns-host from esp board, answer should be received from host start_case('CONFIG_TEST_QUERY_HOST', 'Query tinytester.local', 'tinytester.local resolved to: 127.0.0.1') # query dns-host aynchrounusely from esp board, answer should be received from host start_case('CONFIG_TEST_QUERY_HOST_ASYNC', 'Query tinytester.local async', 'Async query resolved to A:127.0.0.1') finally: stop_mdns_listener.set() mdns_responder.join()