test: update server used for tests to use SSL/TLS (GoogleCloudPlatform#1262)

This PR improves the local server used for unit tests.

The previous server fixture was just a plain socket unable to perform SSL/TLS.
This meant in certain tests we were skipping the SSL/TLS handshake because it
would fail.

The new proxy_server fixture more closely behaves to that of the real Proxy server.
It allows SSL/TLS by loading in the server CA used for testing, the client can then
use the new context fixture to properly use SSL to connect to the server.

This will provide a foundation for testing the metadata exchange which will be
introduced in the near future.

Author: jackwotherspoon

tests/conftest.py

@@ -17,18 +17,23 @@
 import asyncio
 import os
 import socket
+import ssl
 from threading import Thread
-from typing import Any, AsyncGenerator, Generator
+from typing import Any, AsyncGenerator
 
+from aiofiles.tempfile import TemporaryDirectory
 from aiohttp import web
+from cryptography.hazmat.primitives import serialization
 import pytest  # noqa F401 Needed to run the tests
+from unit.mocks import create_ssl_context  # type: ignore
 from unit.mocks import FakeCredentials  # type: ignore
 from unit.mocks import FakeCSQLInstance  # type: ignore
 
 from google.cloud.sql.connector.client import CloudSQLClient
 from google.cloud.sql.connector.connection_name import ConnectionName
 from google.cloud.sql.connector.instance import RefreshAheadCache
 from google.cloud.sql.connector.utils import generate_keys
+from google.cloud.sql.connector.utils import write_to_file
 
 SCOPES = ["https://www.googleapis.com/auth/sqlservice.admin"]
 
@@ -79,25 +84,60 @@ def fake_credentials() -> FakeCredentials:
     return FakeCredentials()
 
 
-def mock_server(server_sock: socket.socket) -> None:
-    """Create mock server listening on specified ip_address and port."""
+async def start_proxy_server(instance: FakeCSQLInstance) -> None:
+    """Run local proxy server capable of performing mTLS"""
     ip_address = "127.0.0.1"
     port = 3307
-    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-    server_sock.bind((ip_address, port))
-    server_sock.listen(0)
-    server_sock.accept()
+    # create socket
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        # create SSL/TLS context
+        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        context.minimum_version = ssl.TLSVersion.TLSv1_3
+        # tmpdir and its contents are automatically deleted after the CA cert
+        # and cert chain are loaded into the SSLcontext. The values
+        # need to be written to files in order to be loaded by the SSLContext
+        server_key_bytes = instance.server_key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.TraditionalOpenSSL,
+            encryption_algorithm=serialization.NoEncryption(),
+        )
+        async with TemporaryDirectory() as tmpdir:
+            server_filename, _, key_filename = await write_to_file(
+                tmpdir, instance.server_cert_pem, "", server_key_bytes
+            )
+            context.load_cert_chain(server_filename, key_filename)
+        # allow socket to be re-used
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        # bind socket to Cloud SQL proxy server port on localhost
+        sock.bind((ip_address, port))
+        # listen for incoming connections
+        sock.listen(5)
+
+        with context.wrap_socket(sock, server_side=True) as ssock:
+            while True:
+                conn, _ = ssock.accept()
+                conn.close()
+
+
+@pytest.fixture(scope="session")
+def proxy_server(fake_instance: FakeCSQLInstance) -> None:
+    """Run local proxy server capable of performing mTLS"""
+    thread = Thread(
+        target=asyncio.run,
+        args=(
+            start_proxy_server(
+                fake_instance,
+            ),
+        ),
+        daemon=True,
+    )
+    thread.start()
+    thread.join(1.0)  # add a delay to allow the proxy server to start
 
 
 @pytest.fixture
-def server() -> Generator:
-    """Create thread with server listening on proper port"""
-    server_sock = socket.socket()
-    thread = Thread(target=mock_server, args=(server_sock,), daemon=True)
-    thread.start()
-    yield thread
-    server_sock.close()
-    thread.join()
+async def context(fake_instance: FakeCSQLInstance) -> ssl.SSLContext:
+    return await create_ssl_context(fake_instance)
 
 
 @pytest.fixture
@@ -107,7 +147,7 @@ def kwargs() -> Any:
     return kwargs
 
 
-@pytest.fixture
+@pytest.fixture(scope="session")
 def fake_instance() -> FakeCSQLInstance:
     return FakeCSQLInstance()
 

tests/unit/mocks.py

@@ -1,4 +1,4 @@
-""""
+"""
 Copyright 2022 Google LLC
 
 Licensed under the Apache License, Version 2.0 (the "License");
@@ -16,6 +16,8 @@
 
 # file containing all mocks used for Cloud SQL Python Connector unit tests
 
+from __future__ import annotations
+
 import datetime
 import json
 import ssl
@@ -184,28 +186,28 @@ def client_key_signed_cert(
         .not_valid_after(cert_expiration)  # type: ignore
     )
     return (
-        cert.sign(priv_key, hashes.SHA256(), default_backend())
+        cert.sign(priv_key, hashes.SHA256())
         .public_bytes(encoding=serialization.Encoding.PEM)
         .decode("UTF-8")
     )
 
 
-async def create_ssl_context() -> ssl.SSLContext:
+async def create_ssl_context(instance: FakeCSQLInstance) -> ssl.SSLContext:
     """Helper method to build an ssl.SSLContext for tests"""
-    # generate keys and certs for test
-    cert, private_key = generate_cert("my-project", "my-instance")
-    server_ca_cert = self_signed_cert(cert, private_key)
     client_private, client_bytes = await generate_keys()
     client_key: rsa.RSAPublicKey = serialization.load_pem_public_key(
-        client_bytes.encode("UTF-8"), default_backend()
+        client_bytes.encode("UTF-8"),
     )  # type: ignore
-    ephemeral_cert = client_key_signed_cert(cert, private_key, client_key)
-    # build default ssl.SSLContext
-    context = ssl.create_default_context()
+    ephemeral_cert = client_key_signed_cert(
+        instance.server_ca, instance.server_key, client_key
+    )
+    # create SSL/TLS context
+    context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    context.check_hostname = False
     # load ssl.SSLContext with certs
     async with TemporaryDirectory() as tmpdir:
         ca_filename, cert_filename, key_filename = await write_to_file(
-            tmpdir, server_ca_cert, ephemeral_cert, client_private
+            tmpdir, instance.server_cert_pem, ephemeral_cert, client_private
         )
         context.load_cert_chain(cert_filename, keyfile=key_filename)
         context.load_verify_locations(cafile=ca_filename)
@@ -279,8 +281,8 @@ async def generate_ephemeral(self, request: Any) -> web.Response:
         body = await request.json()
         pub_key = body["public_key"]
         client_key: rsa.RSAPublicKey = serialization.load_pem_public_key(
-            pub_key.encode("UTF-8"), default_backend()
-        )  # type: ignore
+            pub_key.encode("UTF-8"),
+        )
         ephemeral_cert = client_key_signed_cert(
             self.server_ca,
             self.server_key,

tests/unit/test_instance.py

@@ -186,17 +186,18 @@ async def test_RefreshAheadCache_close(cache: RefreshAheadCache) -> None:
 @pytest.mark.asyncio
 async def test_perform_refresh(
     cache: RefreshAheadCache,
-    fake_instance: mocks.FakeCSQLInstance,
 ) -> None:
     """
     Test that _perform_refresh returns valid ConnectionInfo object.
     """
     instance_metadata = await cache._perform_refresh()
-
     # verify instance metadata object is returned
     assert isinstance(instance_metadata, ConnectionInfo)
     # verify instance metadata expiration
-    assert fake_instance.server_cert.not_valid_after_utc == instance_metadata.expiration
+    assert (
+        cache._client.instance.cert_expiration.replace(microsecond=0)
+        == instance_metadata.expiration
+    )
 
 
 @pytest.mark.asyncio

tests/unit/test_monitored_cache.py

@@ -14,13 +14,13 @@
 
 import asyncio
 import socket
+import ssl
 
 import dns.message
 import dns.rdataclass
 import dns.rdatatype
 import dns.resolver
 from mock import patch
-from mocks import create_ssl_context
 import pytest
 
 from google.cloud.sql.connector.client import CloudSQLClient
@@ -149,8 +149,10 @@ async def test_MonitoredCache_with_disabled_failover(
     assert monitored_cache.closed is True
 
 
-@pytest.mark.usefixtures("server")
-async def test_MonitoredCache_check_domain_name(fake_client: CloudSQLClient) -> None:
+@pytest.mark.usefixtures("proxy_server")
+async def test_MonitoredCache_check_domain_name(
+    context: ssl.SSLContext, fake_client: CloudSQLClient
+) -> None:
     """
     Test that MonitoredCache is closed when _check_domain_name has domain change.
     """
@@ -177,11 +179,9 @@ async def test_MonitoredCache_check_domain_name(fake_client: CloudSQLClient) ->
 
         # configure a local socket
         ip_addr = "127.0.0.1"
-        context = await create_ssl_context()
         sock = context.wrap_socket(
             socket.create_connection((ip_addr, 3307)),
             server_hostname=ip_addr,
-            do_handshake_on_connect=False,
         )
         # verify socket is open
         assert sock.fileno() != -1
@@ -198,8 +198,10 @@ async def test_MonitoredCache_check_domain_name(fake_client: CloudSQLClient) ->
         assert sock.fileno() == -1
 
 
-@pytest.mark.usefixtures("server")
-async def test_MonitoredCache_purge_closed_sockets(fake_client: CloudSQLClient) -> None:
+@pytest.mark.usefixtures("proxy_server")
+async def test_MonitoredCache_purge_closed_sockets(
+    context: ssl.SSLContext, fake_client: CloudSQLClient
+) -> None:
     """
     Test that MonitoredCache._purge_closed_sockets removes closed sockets from
     cache.
@@ -215,11 +217,9 @@ async def test_MonitoredCache_purge_closed_sockets(fake_client: CloudSQLClient)
     )
     # configure a local socket
     ip_addr = "127.0.0.1"
-    context = await create_ssl_context()
     sock = context.wrap_socket(
         socket.create_connection((ip_addr, 3307)),
         server_hostname=ip_addr,
-        do_handshake_on_connect=False,
     )
 
     # set failover to 0 to disable polling

tests/unit/test_pg8000.py

@@ -15,26 +15,22 @@
 """
 
 import socket
+import ssl
 from typing import Any
 
 from mock import patch
-from mocks import create_ssl_context
 import pytest
 
 from google.cloud.sql.connector.pg8000 import connect
 
 
-@pytest.mark.usefixtures("server")
-@pytest.mark.asyncio
-async def test_pg8000(kwargs: Any) -> None:
+@pytest.mark.usefixtures("proxy_server")
+async def test_pg8000(context: ssl.SSLContext, kwargs: Any) -> None:
     """Test to verify that pg8000 gets to proper connection call."""
     ip_addr = "127.0.0.1"
-    # build ssl.SSLContext
-    context = await create_ssl_context()
     sock = context.wrap_socket(
         socket.create_connection((ip_addr, 3307)),
         server_hostname=ip_addr,
-        do_handshake_on_connect=False,
     )
     with patch("pg8000.dbapi.connect") as mock_connect:
         mock_connect.return_value = True

tests/unit/test_pymysql.py

@@ -19,7 +19,6 @@
 from typing import Any
 
 from mock import patch
-from mocks import create_ssl_context
 import pytest
 
 from google.cloud.sql.connector.pymysql import connect as pymysql_connect
@@ -33,17 +32,14 @@ def connect(sock: ssl.SSLSocket) -> None:  # type: ignore
         assert isinstance(sock, ssl.SSLSocket)
 
 
-@pytest.mark.usefixtures("server")
+@pytest.mark.usefixtures("proxy_server")
 @pytest.mark.asyncio
-async def test_pymysql(kwargs: Any) -> None:
+async def test_pymysql(context: ssl.SSLContext, kwargs: Any) -> None:
     """Test to verify that pymysql gets to proper connection call."""
     ip_addr = "127.0.0.1"
-    # build ssl.SSLContext
-    context = await create_ssl_context()
     sock = context.wrap_socket(
         socket.create_connection((ip_addr, 3307)),
         server_hostname=ip_addr,
-        do_handshake_on_connect=False,
     )
     kwargs["timeout"] = 30
     with patch("pymysql.Connection") as mock_connect: