Add test for minimum TLS version

Author: sfodagain

.builder/actions/tls_server_setup.py

@@ -1,14 +1,13 @@
 """
-Setup local TLS server for tests
+Setup local TLS servers for tests
 """
 
 import Builder
 
-import os
+from pathlib import Path
 import sys
 import subprocess
 import atexit
-import time
 
 
 class TlsServerSetup(Builder.Action):
@@ -18,26 +17,47 @@ class TlsServerSetup(Builder.Action):
     This action should be run in the 'pre_build_steps' or 'build_steps' stage.
     """
 
+    @staticmethod
+    def cleanup_tls_server(tls_server_process):
+        tls_server_process.terminate()
+        out, err = tls_server_process.communicate()
+        print("TLS server stdout:")
+        for line in out.splitlines():
+            print(f"  = {line.decode('utf-8')}")
+        print("TLS server stderr:")
+        for line in err.splitlines():
+            print(f"  = {line.decode('utf-8')}")
+
     def run(self, env):
         if not env.project.needs_tests(env):
             print("Skipping TLS server setup because tests disabled for project")
             return
 
         self.env = env
 
-        base_dir = os.path.dirname(os.path.realpath(__file__))
-        dir = os.path.join(base_dir, "..", "..", "tests", "tls_server")
+        root_dir = Path(__file__).resolve().parent / '..' / '..'
+        tls_server_dir = root_dir / 'tests' / 'tls_server'
+        resource_dir = root_dir / 'tests' / 'resources'
 
-        print("Running openssl TLS server")
+        print("Running TLS servers")
 
         python_path = sys.executable
-        p = subprocess.Popen([python_path, "tls_server.py",
-                              ], cwd=dir, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+
+        tls12_server_process = subprocess.Popen(
+            [python_path, tls_server_dir / 'tls_server.py', '--port', '58443', '--resource-dir', resource_dir,
+             '--min-tls', '1.2',
+             '--max-tls', '1.2'],
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+
+        tls13_server_process = subprocess.Popen(
+            [python_path, tls_server_dir / 'tls_server.py', '--port', '59443', '--resource-dir', resource_dir,
+             '--min-tls', '1.3',
+             '--max-tls', '1.3'],
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 
         @atexit.register
-        def close_tls_server():
-            print("Terminating openssl TLS server")
-            p.terminate()
-            out, err = p.communicate()
-            print("TLS server stdout:\n{}".format(out))
-            print("TLS server stderr:\n{}".format(err))
+        def close_tls_servers():
+            print('Terminating TLS 1.2 server')
+            TlsServerSetup.cleanup_tls_server(tls12_server_process)
+            print('Terminating TLS 1.3 server')
+            TlsServerSetup.cleanup_tls_server(tls13_server_process)

tests/CMakeLists.txt

@@ -230,6 +230,7 @@ if(NOT BYO_CRYPTO)
     add_net_test_case(tls_client_channel_negotiation_error_broken_crypto_dh480)
     add_net_test_case(tls_client_channel_negotiation_error_broken_crypto_dh512)
     add_net_test_case(tls_client_channel_negotiation_error_broken_crypto_null)
+    add_net_test_case(tls_client_channel_negotiation_error_tls1_3_to_tls1_2_server)
 
     # Badssl - Legacy crypto suite, includes both negative and positive tests, with override checks where appropriate
     # Our current baseline/default is platform-specific, whereas badssl expects a baseline of 1.2

tests/tls_handler_test.c

@@ -1225,6 +1225,10 @@ static void s_raise_tls_version_to_12(struct aws_tls_ctx_options *options) {
     aws_tls_ctx_options_set_minimum_tls_version(options, AWS_IO_TLSv1_2);
 }
 
+static void s_raise_tls_version_to_13(struct aws_tls_ctx_options *options) {
+    aws_tls_ctx_options_set_minimum_tls_version(options, AWS_IO_TLSv1_3);
+}
+
 static int s_tls_client_channel_negotiation_error_override_legacy_crypto_tls11_fn(
     struct aws_allocator *allocator,
     void *ctx) {
@@ -1334,6 +1338,21 @@ AWS_TEST_CASE(
     tls_client_channel_negotiation_error_socket_closed,
     s_tls_client_channel_negotiation_error_socket_closed_fn);
 
+AWS_STATIC_STRING_FROM_LITERAL(s_aws_local_tls_server_host_name, "127.0.0.1");
+
+static int s_tls_client_channel_negotiation_error_tls1_3_to_tls1_2_server_fn(
+    struct aws_allocator *allocator,
+    void *ctx) {
+    (void)ctx;
+    uint32_t server_tls1_2_port = 58443;
+    return s_verify_negotiation_fails(
+        allocator, s_aws_local_tls_server_host_name, server_tls1_2_port, &s_raise_tls_version_to_13);
+}
+
+AWS_TEST_CASE(
+    tls_client_channel_negotiation_error_tls1_3_to_tls1_2_server,
+    s_tls_client_channel_negotiation_error_tls1_3_to_tls1_2_server_fn)
+
 static int s_verify_good_host(
     struct aws_allocator *allocator,
     const struct aws_string *host_name,
@@ -1623,14 +1642,12 @@ AWS_TEST_CASE(
     s_tls_client_channel_negotiation_success_ecc384_SCHANNEL_CREDS_fn)
 #    endif
 
-static void s_raise_tls_version_to_13(struct aws_tls_ctx_options *options) {
-    aws_tls_ctx_options_set_minimum_tls_version(options, AWS_IO_TLSv1_3);
-}
-
 AWS_STATIC_STRING_FROM_LITERAL(s_aws_ecc384_host_name, "127.0.0.1");
 static int s_tls_client_channel_negotiation_success_mtls_tls1_3_fn(struct aws_allocator *allocator, void *ctx) {
     (void)ctx;
-    return s_verify_good_host_mqtt_connect(allocator, s_aws_ecc384_host_name, 59443, s_raise_tls_version_to_13);
+    uint32_t server_tls1_3_port = 59443;
+    return s_verify_good_host_mqtt_connect(
+        allocator, s_aws_ecc384_host_name, server_tls1_3_port, s_raise_tls_version_to_13);
 }
 
 AWS_TEST_CASE(

tests/tls_server/tls_server.py

@@ -1,23 +1,71 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0.
 
+import argparse
+import pathlib
+import signal
 import socket
 import ssl
+import sys
+
+
+def parse_tls(tls_str):
+    if tls_str == '1.1':
+        return ssl.TLSVersion.TLSv1_1
+    elif tls_str == '1.2':
+        return ssl.TLSVersion.TLSv1_2
+    elif tls_str == '1.3':
+        return ssl.TLSVersion.TLSv1_3
+    raise ValueError('Unknown TLS version')
+
+
+print(f"Starting TLS server")
+
+parser = argparse.ArgumentParser(
+    description="TLS test server",
+    formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+)
+
+optional = parser.add_argument_group("optional arguments")
+
+optional.add_argument("--host", dest="host", default="127.0.0.1", help="Listening host")
+optional.add_argument("--port", type=int, dest="port", default=59443, help="Listening port")
+optional.add_argument("--min-tls", choices=['1.1', '1.2', '1.3'], dest="min_tls", default='1.2',
+                      help="Minimum acceptable TLS version")
+optional.add_argument("--max-tls", choices=['1.1', '1.2', '1.3'], dest="max_tls", default='1.3',
+                      help="Maximum acceptable TLS version")
+optional.add_argument("--resource-dir", type=pathlib.Path, dest="resource_dir", default='./tests/resources/',
+                      help="Path to keys and certificates")
+
+args = parser.parse_args()
 
 context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
-context.minimum_version = ssl.TLSVersion.TLSv1_3
-context.maximum_version = ssl.TLSVersion.TLSv1_3
-context.load_cert_chain('../resources/tls13_server.pem.crt', '../resources/tls13_server.key')
-context.load_verify_locations('../resources/tls13_device_root_ca.pem.crt')
+context.minimum_version = parse_tls(args.min_tls)
+context.maximum_version = parse_tls(args.max_tls)
+context.load_cert_chain(args.resource_dir / 'tls13_server.pem.crt', args.resource_dir / 'tls13_server.key')
+context.load_verify_locations(args.resource_dir / 'tls13_device_root_ca.pem.crt')
 context.verify_mode = ssl.CERT_REQUIRED
 
+
+def signal_handler(signum, frame):
+    sys.stdout.flush()
+    sys.exit(0)
+
+
+signal.signal(signal.SIGTERM, signal_handler)
+
+print(f"Running TLS server on {args.host}:{args.port}")
+print(f"Minimum TLS version: {context.minimum_version.name}")
+print(f"Maximum TLS version: {context.maximum_version.name}")
+
 with socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) as sock:
-    sock.bind(('127.0.0.1', 59443))
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    sock.bind((args.host, args.port))
     sock.listen(1)
     with context.wrap_socket(sock, server_side=True) as ssock:
         while True:
             try:
                 conn, addr = ssock.accept()
-                print("accepted new connection: {}".format(addr))
+                print("Accepted new connection: {}".format(addr))
             except Exception as e:
-                print("accept failed: {}".format(e))
+                print(f"Accept failed: {e}")