# SPDX-FileCopyrightText: 2022 Espressif Systems (Shanghai) CO LTD # SPDX-License-Identifier: Unlicense OR CC0-1.0 import http.server import logging import multiprocessing import os import socket import ssl from typing import Callable import pexpect import pytest from pytest_embedded import Dut from RangeHTTPServer import RangeRequestHandler def get_my_ip() -> str: s1 = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) s1.connect(('8.8.8.8', 80)) my_ip = '' my_ip = s1.getsockname()[0] s1.close() return my_ip def get_server_status(host_ip: str, port: int) -> bool: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_status = sock.connect_ex((host_ip, port)) sock.close() if server_status == 0: return True return False def https_request_handler() -> Callable[...,http.server.BaseHTTPRequestHandler]: """ Returns a request handler class that handles broken pipe exception """ class RequestHandler(RangeRequestHandler): protocol_version = 'HTTP/1.1' def finish(self) -> None: try: if not self.wfile.closed: self.wfile.flush() self.wfile.close() except socket.error: pass self.rfile.close() def handle(self) -> None: try: RangeRequestHandler.handle(self) except socket.error: pass def do_GET(self) -> None: self.close_connection = True self.send_response(200) self.end_headers() return RequestHandler def start_https_server(server_file: str, key_file: str, server_ip: str, server_port: int) -> None: requestHandler = https_request_handler() httpd = http.server.HTTPServer((server_ip, server_port), requestHandler) httpd.socket = ssl.wrap_socket(httpd.socket, keyfile=key_file, certfile=server_file, server_side=True) httpd.serve_forever() @pytest.mark.esp32 @pytest.mark.esp32c3 @pytest.mark.esp32s2 @pytest.mark.esp32s3 @pytest.mark.ethernet @pytest.mark.parametrize('config', ['cli_ses_tkt',], indirect=True) def test_examples_protocol_https_request_cli_session_tickets(dut: Dut) -> None: logging.info("Testing for \"esp_tls client session tickets\"") # check and log bin size binary_file = os.path.join(dut.app.binary_path, 'https_request.bin') bin_size = os.path.getsize(binary_file) logging.info('https_request_bin_size : {}KB'.format(bin_size // 1024)) # start test host_ip = get_my_ip() server_port = 8070 server_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_cert.pem') key_file = os.path.join(os.path.dirname(__file__), 'main', 'local_server_key.pem') if (get_server_status(host_ip, server_port) is False): thread1 = multiprocessing.Process(target=start_https_server, args=(server_file, key_file, host_ip, server_port)) thread1.daemon = True thread1.start() logging.info('The server started on {}:{}'.format(host_ip, server_port)) dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r' (sta|eth) ip: (\d+\.\d+\.\d+\.\d+)', timeout=60)[2].decode() print('Connected to AP with IP: {}'.format(ip_address)) except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') dut.expect('Start https_request example', timeout=30) print('writing to device: {}'.format('https://' + host_ip + ':' + str(server_port))) dut.write('https://' + host_ip + ':' + str(server_port)) logging.info("Testing for \"https_request using saved session\"") # Check for connection using already saved client session try: dut.expect('https_request to local server', timeout=30) dut.expect(['Connection established...', 'Reading HTTP response...', 'HTTP/1.1 200 OK', 'connection closed'], expect_all=True) except Exception: logging.info("Failed to connect to local https server\"") raise try: dut.expect('https_request using saved client session', timeout=20) dut.expect(['Connection established...', 'Reading HTTP response...', 'HTTP/1.1 200 OK', 'connection closed'], expect_all=True) except Exception: logging.info("Failed the test for \"https_request using saved client session\"") raise logging.info("Passed the test for \"https_request using saved client session\"") thread1.terminate() @pytest.mark.esp32 @pytest.mark.esp32c3 @pytest.mark.esp32s2 @pytest.mark.esp32s3 @pytest.mark.ethernet @pytest.mark.parametrize('config', ['ssldyn',], indirect=True) def test_examples_protocol_https_request_dynamic_buffers(dut: Dut) -> None: # Check for connection using crt bundle with mbedtls dynamic resource enabled # check and log bin size binary_file = os.path.join(dut.app.binary_path, 'https_request.bin') bin_size = os.path.getsize(binary_file) logging.info('https_request_bin_size : {}KB'.format(bin_size // 1024)) dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r' (sta|eth) ip: (\d+\.\d+\.\d+\.\d+)', timeout=60)[2].decode() print('Connected to AP with IP: {}'.format(ip_address)) except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') # only check if one connection is established logging.info("Testing for \"https_request using crt bundle\" with mbedtls dynamic resource enabled") try: dut.expect('https_request using crt bundle', timeout=30) dut.expect(['Connection established...', 'Reading HTTP response...', 'HTTP/1.1 200 OK', 'connection closed'], expect_all=True) except Exception: logging.info("Failed the test for \"https_request using crt bundle\" when mbedtls dynamic resource was enabled") raise logging.info("Passed the test for \"https_request using crt bundle\" when mbedtls dynamic resource was enabled") @pytest.mark.supported_targets @pytest.mark.ethernet def test_examples_protocol_https_request(dut: Dut) -> None: """ steps: | 1. join AP 2. establish TLS connection to www.howsmyssl.com:443 with multiple certificate verification options 3. send http request """ # check and log bin size binary_file = os.path.join(dut.app.binary_path, 'https_request.bin') bin_size = os.path.getsize(binary_file) logging.info('https_request_bin_size : {}KB'.format(bin_size // 1024)) logging.info('Starting https_request simple test app') dut.expect('Loaded app from partition at offset', timeout=30) try: ip_address = dut.expect(r' (sta|eth) ip: (\d+\.\d+\.\d+\.\d+)', timeout=60)[2].decode() print('Connected to AP with IP: {}'.format(ip_address)) except pexpect.exceptions.TIMEOUT: raise ValueError('ENV_TEST_FAILURE: Cannot connect to AP') # Check for connection using crt bundle logging.info("Testing for \"https_request using crt bundle\"") try: dut.expect('https_request using crt bundle', timeout=30) dut.expect(['Certificate validated', 'Connection established...', 'Reading HTTP response...', 'HTTP/1.1 200 OK', 'connection closed'], expect_all=True) except Exception: logging.info("Failed the test for \"https_request using crt bundle\"") raise logging.info("Passed the test for \"https_request using crt bundle\"") # Check for connection using cacert_buf logging.info("Testing for \"https_request using cacert_buf\"") try: dut.expect('https_request using cacert_buf', timeout=20) dut.expect(['Connection established...', 'Reading HTTP response...', 'HTTP/1.1 200 OK', 'connection closed'], expect_all=True) except Exception: logging.info("Passed the test for \"https_request using cacert_buf\"") raise logging.info("Passed the test for \"https_request using cacert_buf\"") # Check for connection using global ca_store logging.info("Testing for \"https_request using global ca_store\"") try: dut.expect('https_request using global ca_store', timeout=20) dut.expect(['Connection established...', 'Reading HTTP response...', 'HTTP/1.1 200 OK', 'connection closed'], expect_all=True) except Exception: logging.info("Failed the test for \"https_request using global ca_store\"") raise logging.info("Passed the test for \"https_request using global ca_store\"")