refactor: move SSL/TLS context creation to ConnectionInfo (#1079)

Author: jackwotherspoon

google/cloud/sql/connector/connector.py

@@ -330,10 +330,17 @@ async def connect_async(
 
             # async drivers are unblocking and can be awaited directly
             if driver in ASYNC_DRIVERS:
-                return await connector(ip_address, instance_data.context, **kwargs)
+                return await connector(
+                    ip_address,
+                    instance_data.create_ssl_context(enable_iam_auth),
+                    **kwargs,
+                )
             # synchronous drivers are blocking and run using executor
             connect_partial = partial(
-                connector, ip_address, instance_data.context, **kwargs
+                connector,
+                ip_address,
+                instance_data.create_ssl_context(enable_iam_auth),
+                **kwargs,
             )
             return await self._loop.run_in_executor(None, connect_partial)
 

google/cloud/sql/connector/exceptions.py

@@ -62,3 +62,11 @@ class DnsNameResolutionError(Exception):
     Exception to be raised when the DnsName of a PSC connection to a
     Cloud SQL instance can not be resolved to a proper IP address.
     """
+
+
+class RefreshNotValidError(Exception):
+    """
+    Exception to be raised when the task returned from refresh is not valid.
+    """
+
+    pass

google/cloud/sql/connector/instance.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import asyncio
+from dataclasses import dataclass
 from enum import Enum
 import logging
 import re
@@ -29,6 +30,7 @@
 from google.cloud.sql.connector.client import CloudSQLClient
 from google.cloud.sql.connector.exceptions import AutoIAMAuthNotSupported
 from google.cloud.sql.connector.exceptions import CloudSQLIPTypeError
+from google.cloud.sql.connector.exceptions import RefreshNotValidError
 from google.cloud.sql.connector.exceptions import TLSVersionError
 from google.cloud.sql.connector.rate_limiter import AsyncRateLimiter
 from google.cloud.sql.connector.refresh_utils import _is_valid
@@ -79,33 +81,39 @@ def _from_str(cls, ip_type_str: str) -> IPTypes:
         return cls(ip_type_str.upper())
 
 
+@dataclass
 class ConnectionInfo:
+    """Contains all necessary information to connect securely to the
+    server-side Proxy running on a Cloud SQL instance."""
+
+    client_cert: str
+    server_ca_cert: str
+    private_key: bytes
     ip_addrs: Dict[str, Any]
-    context: ssl.SSLContext
     database_version: str
     expiration: datetime.datetime
+    context: ssl.SSLContext | None = None
 
-    def __init__(
-        self,
-        ephemeral_cert: str,
-        database_version: str,
-        ip_addrs: Dict[str, Any],
-        private_key: bytes,
-        server_ca_cert: str,
-        expiration: datetime.datetime,
-        enable_iam_auth: bool,
-    ) -> None:
-        self.ip_addrs = ip_addrs
-        self.database_version = database_version
-        self.context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+    def create_ssl_context(self, enable_iam_auth: bool = False) -> ssl.SSLContext:
+        """Constructs a SSL/TLS context for the given connection info.
+
+        Cache the SSL context to ensure we don't read from disk repeatedly when
+        configuring a secure connection.
+        """
+        # if SSL context is cached, use it
+        if self.context is not None:
+            return self.context
+        context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
 
         # update ssl.PROTOCOL_TLS_CLIENT default
-        self.context.check_hostname = False
+        context.check_hostname = False
 
+        # TODO: remove if/else when Python 3.10 is min version. PEP 644 has been
+        # implemented. The ssl module requires OpenSSL 1.1.1 or newer.
         # verify OpenSSL version supports TLSv1.3
         if ssl.HAS_TLSv1_3:
             # force TLSv1.3 if supported by client
-            self.context.minimum_version = ssl.TLSVersion.TLSv1_3
+            context.minimum_version = ssl.TLSVersion.TLSv1_3
         # fallback to TLSv1.2 for older versions of OpenSSL
         else:
             if enable_iam_auth:
@@ -119,18 +127,20 @@ def __init__(
                 f"({ssl.OPENSSL_VERSION}), falling back to TLSv1.2\n"
                 "Upgrade your OpenSSL version to 1.1.1 for TLSv1.3 support."
             )
-            self.context.minimum_version = ssl.TLSVersion.TLSv1_2
-        self.expiration = expiration
+            context.minimum_version = ssl.TLSVersion.TLSv1_2
 
         # tmpdir and its contents are automatically deleted after the CA cert
         # and ephemeral cert are loaded into the SSLcontext. The values
         # need to be written to files in order to be loaded by the SSLContext
         with TemporaryDirectory() as tmpdir:
             ca_filename, cert_filename, key_filename = write_to_file(
-                tmpdir, server_ca_cert, ephemeral_cert, private_key
+                tmpdir, self.server_ca_cert, self.client_cert, self.private_key
             )
-            self.context.load_cert_chain(cert_filename, keyfile=key_filename)
-            self.context.load_verify_locations(cafile=ca_filename)
+            context.load_cert_chain(cert_filename, keyfile=key_filename)
+            context.load_verify_locations(cafile=ca_filename)
+        # set class attribute to cache context for subsequent calls
+        self.context = context
+        return context
 
     def get_preferred_ip(self, ip_type: IPTypes) -> str:
         """Returns the first IP address for the instance, according to the preference
@@ -272,12 +282,11 @@ async def _perform_refresh(self) -> ConnectionInfo:
 
         return ConnectionInfo(
             ephemeral_cert,
-            metadata["database_version"],
-            metadata["ip_addresses"],
-            priv_key,
             metadata["server_ca_cert"],
+            priv_key,
+            metadata["ip_addresses"],
+            metadata["database_version"],
             expiration,
-            self._enable_iam_auth,
         )
 
     def _schedule_refresh(self, delay: int) -> asyncio.Task:
@@ -303,6 +312,11 @@ async def _refresh_task(self: RefreshAheadCache, delay: int) -> ConnectionInfo:
                     await asyncio.sleep(delay)
                 refresh_task = asyncio.create_task(self._perform_refresh())
                 refresh_data = await refresh_task
+                # check that refresh is valid
+                if not await _is_valid(refresh_task):
+                    raise RefreshNotValidError(
+                        f"['{self._instance_connection_string}']: Invalid refresh operation. Certficate appears to be expired."
+                    )
             except asyncio.CancelledError:
                 logger.debug(
                     f"['{self._instance_connection_string}']: Schedule refresh task cancelled."

tests/conftest.py

@@ -105,7 +105,7 @@ def kwargs() -> Any:
     return kwargs
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture
 def fake_instance() -> FakeCSQLInstance:
     return FakeCSQLInstance()
 
@@ -133,7 +133,10 @@ async def fake_client(
         sqlserver_client_cert_uri, sqlserver_instance.generate_ephemeral
     )
     client_session = await aiohttp_client(app)
-    return CloudSQLClient("", "", fake_credentials, client=client_session)
+    client = CloudSQLClient("", "", fake_credentials, client=client_session)
+    # add instance to client to control cert expiration etc.
+    client.instance = fake_instance
+    return client
 
 
 @pytest.fixture

tests/unit/mocks.py

@@ -31,7 +31,6 @@
 from google.auth.credentials import Credentials
 
 from google.cloud.sql.connector.connector import _DEFAULT_UNIVERSE_DOMAIN
-from google.cloud.sql.connector.instance import ConnectionInfo
 from google.cloud.sql.connector.utils import generate_keys
 from google.cloud.sql.connector.utils import write_to_file
 
@@ -85,36 +84,6 @@ def valid(self) -> bool:
         return self.token is not None and not self.expired
 
 
-class BadRefresh(Exception):
-    pass
-
-
-class MockMetadata(ConnectionInfo):
-    """Mock class for ConnectionInfo"""
-
-    def __init__(
-        self, expiration: datetime.datetime, ip_addrs: Dict = {"PRIMARY": "0.0.0.0"}
-    ) -> None:
-        self.expiration = expiration
-        self.ip_addrs = ip_addrs
-
-
-async def instance_metadata_success(*args: Any, **kwargs: Any) -> MockMetadata:
-    return MockMetadata(
-        datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(minutes=10)
-    )
-
-
-async def instance_metadata_expired(*args: Any, **kwargs: Any) -> MockMetadata:
-    return MockMetadata(
-        datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(minutes=10)
-    )
-
-
-async def instance_metadata_error(*args: Any, **kwargs: Any) -> None:
-    raise BadRefresh("something went wrong...")
-
-
 def generate_cert(
     project: str,
     name: str,
@@ -166,6 +135,9 @@ def client_key_signed_cert(
     cert: x509.CertificateBuilder,
     priv_key: rsa.RSAPrivateKey,
     client_key: rsa.RSAPublicKey,
+    cert_before: datetime.datetime = datetime.datetime.now(datetime.timezone.utc),
+    cert_expiration: datetime.datetime = datetime.datetime.now(datetime.timezone.utc)
+    + datetime.timedelta(hours=1),
 ) -> str:
     """
     Create a PEM encoded certificate that is signed by given public key.
@@ -185,8 +157,8 @@ def client_key_signed_cert(
         .issuer_name(issuer)
         .public_key(client_key)
         .serial_number(x509.random_serial_number())
-        .not_valid_before(cert._not_valid_before)
-        .not_valid_after(cert._not_valid_after)  # type: ignore
+        .not_valid_before(cert_before)
+        .not_valid_after(cert_expiration)  # type: ignore
     )
     return (
         cert.sign(priv_key, hashes.SHA256(), default_backend())
@@ -231,7 +203,7 @@ def __init__(
             "PRIVATE": "10.0.0.1",
         },
         cert_before: datetime = datetime.datetime.now(datetime.timezone.utc),
-        cert_expiry: datetime = datetime.datetime.now(datetime.timezone.utc)
+        cert_expiration: datetime = datetime.datetime.now(datetime.timezone.utc)
         + datetime.timedelta(hours=1),
     ) -> None:
         self.project = project
@@ -240,10 +212,10 @@ def __init__(
         self.db_version = db_version
         self.ip_addrs = ip_addrs
         self.cert_before = cert_before
-        self.cert_expiry = cert_expiry
+        self.cert_expiration = cert_expiration
         # create self signed CA cert
         self.server_ca, self.server_key = generate_cert(
-            self.project, self.name, cert_before, cert_expiry
+            self.project, self.name, cert_before, cert_expiration
         )
         self.server_cert = self.server_ca.sign(self.server_key, hashes.SHA256())
         self.server_cert_pem = self.server_cert.public_bytes(
@@ -257,7 +229,7 @@ async def connect_settings(self, request: Any) -> web.Response:
             "serverCaCert": {
                 "cert": self.server_cert_pem,
                 "instance": self.name,
-                "expirationTime": str(self.server_cert.not_valid_after_utc),
+                "expirationTime": str(self.cert_expiration),
             },
             "dnsName": "abcde.12345.us-central1.sql.goog",
             "ipAddresses": ip_addrs,
@@ -273,13 +245,17 @@ async def generate_ephemeral(self, request: Any) -> web.Response:
             pub_key.encode("UTF-8"), default_backend()
         )  # type: ignore
         ephemeral_cert = client_key_signed_cert(
-            self.server_ca, self.server_key, client_key
+            self.server_ca,
+            self.server_key,
+            client_key,
+            self.cert_before,
+            self.cert_expiration,
         )
         response = {
             "ephemeralCert": {
                 "kind": "sql#sslCert",
                 "cert": ephemeral_cert,
-                "expirationTime": str(self.server_cert.not_valid_after_utc),
+                "expirationTime": str(self.cert_expiration),
             }
         }
         return web.Response(content_type="application/json", body=json.dumps(response))