chore: Merge branch 'main' into mdx

Author: jackwotherspoon

.ci/cloudbuild.yaml

@@ -0,0 +1,75 @@
+# Copyright 2025 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+steps:
+  - id: run integration tests
+    name: python:${_VERSION}
+    entrypoint: bash
+    env: 
+      - "IP_TYPE=${_IP_TYPE}"
+    secretEnv: ["MYSQL_CONNECTION_NAME", "MYSQL_USER", "MYSQL_IAM_USER", "MYSQL_PASS", "MYSQL_DB", "POSTGRES_CONNECTION_NAME", "POSTGRES_USER", "POSTGRES_IAM_USER", "POSTGRES_PASS", "POSTGRES_DB",  "POSTGRES_CAS_CONNECTION_NAME", "POSTGRES_CAS_PASS", "POSTGRES_CUSTOMER_CAS_CONNECTION_NAME", "POSTGRES_CUSTOMER_CAS_PASS", "POSTGRES_CUSTOMER_CAS_PASS_VALID_DOMAIN_NAME","SQLSERVER_CONNECTION_NAME", "SQLSERVER_USER", "SQLSERVER_PASS", "SQLSERVER_DB"]
+    args:
+      - "-c"
+      - |
+        pip install nox
+        nox -s system-${_VERSION}
+availableSecrets:
+  secretManager:
+  - versionName: 'projects/$PROJECT_ID/secrets/MYSQL_CONNECTION_NAME/versions/latest'
+    env: 'MYSQL_CONNECTION_NAME'
+  - versionName: 'projects/$PROJECT_ID/secrets/MYSQL_USER/versions/latest'
+    env: 'MYSQL_USER'
+  - versionName: 'projects/$PROJECT_ID/secrets/CLOUD_BUILD_MYSQL_IAM_USER/versions/latest'
+    env: 'MYSQL_IAM_USER'
+  - versionName: 'projects/$PROJECT_ID/secrets/MYSQL_PASS/versions/latest'
+    env: 'MYSQL_PASS'
+  - versionName: 'projects/$PROJECT_ID/secrets/MYSQL_DB/versions/latest'
+    env: 'MYSQL_DB'
+  - versionName: 'projects/$PROJECT_ID/secrets/POSTGRES_CONNECTION_NAME/versions/latest'
+    env: 'POSTGRES_CONNECTION_NAME'
+  - versionName: 'projects/$PROJECT_ID/secrets/POSTGRES_USER/versions/latest'
+    env: 'POSTGRES_USER'
+  - versionName: 'projects/$PROJECT_ID/secrets/CLOUD_BUILD_POSTGRES_IAM_USER/versions/latest'
+    env: 'POSTGRES_IAM_USER'
+  - versionName: 'projects/$PROJECT_ID/secrets/POSTGRES_PASS/versions/latest'
+    env: 'POSTGRES_PASS'
+  - versionName: 'projects/$PROJECT_ID/secrets/POSTGRES_DB/versions/latest'
+    env: 'POSTGRES_DB'
+  - versionName: 'projects/$PROJECT_ID/secrets/POSTGRES_CAS_CONNECTION_NAME/versions/latest'
+    env: 'POSTGRES_CAS_CONNECTION_NAME'
+  - versionName: 'projects/$PROJECT_ID/secrets/POSTGRES_CAS_PASS/versions/latest'
+    env: 'POSTGRES_CAS_PASS'
+  - versionName: 'projects/$PROJECT_ID/secrets/POSTGRES_CUSTOMER_CAS_CONNECTION_NAME/versions/latest'
+    env: 'POSTGRES_CUSTOMER_CAS_CONNECTION_NAME'
+  - versionName: 'projects/$PROJECT_ID/secrets/POSTGRES_CUSTOMER_CAS_PASS/versions/latest'
+    env: 'POSTGRES_CUSTOMER_CAS_PASS'
+  - versionName: 'projects/$PROJECT_ID/secrets/POSTGRES_CUSTOMER_CAS_PASS_VALID_DOMAIN_NAME/versions/latest'
+    env: 'POSTGRES_CUSTOMER_CAS_PASS_VALID_DOMAIN_NAME'
+  - versionName: 'projects/$PROJECT_ID/secrets/SQLSERVER_CONNECTION_NAME/versions/latest'
+    env: 'SQLSERVER_CONNECTION_NAME'
+  - versionName: 'projects/$PROJECT_ID/secrets/SQLSERVER_USER/versions/latest'
+    env: 'SQLSERVER_USER'
+  - versionName: 'projects/$PROJECT_ID/secrets/SQLSERVER_PASS/versions/latest'
+    env: 'SQLSERVER_PASS'
+  - versionName: 'projects/$PROJECT_ID/secrets/SQLSERVER_DB/versions/latest'
+    env: 'SQLSERVER_DB'
+substitutions:
+  _VERSION: ${_VERSION}
+  _IP_TYPE: ${_IP_TYPE}
+  
+options:
+  dynamicSubstitutions: true
+  pool:
+    name: ${_POOL_NAME}
+  logging: CLOUD_LOGGING_ONLY

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/system/test_asyncpg_connection.py

@@ -32,6 +32,7 @@ async def create_sqlalchemy_engine(
     user: str,
     password: str,
     db: str,
+    ip_type: str = "public",
     refresh_strategy: str = "background",
     resolver: Union[type[DefaultResolver], type[DnsResolver]] = DefaultResolver,
 ) -> tuple[sqlalchemy.ext.asyncio.engine.AsyncEngine, Connector]:
@@ -63,6 +64,9 @@ async def create_sqlalchemy_engine(
             The database user's password, e.g., secret-password
         db (str):
             The name of the database, e.g., mydb
+        ip_type (str):
+            The IP type of the Cloud SQL instance to connect to. Can be one
+            of "public", "private", or "psc".
         refresh_strategy (Optional[str]):
             Refresh strategy for the Cloud SQL Connector. Can be one of "lazy"
             or "background". For serverless environments use "lazy" to avoid
@@ -87,7 +91,7 @@ async def create_sqlalchemy_engine(
             user=user,
             password=password,
             db=db,
-            ip_type="public",  # can also be "private" or "psc"
+            ip_type=ip_type,  # can be "public", "private" or "psc"
         ),
         execution_options={"isolation_level": "AUTOCOMMIT"},
     )
@@ -99,6 +103,7 @@ async def create_asyncpg_pool(
     user: str,
     password: str,
     db: str,
+    ip_type: str = "public",
     refresh_strategy: str = "background",
 ) -> tuple[asyncpg.Pool, Connector]:
     """Creates a native asyncpg connection pool for a Cloud SQL instance and
@@ -128,6 +133,9 @@ async def create_asyncpg_pool(
             The database user's password, e.g., secret-password
         db (str):
             The name of the database, e.g., mydb
+        ip_type (str):
+            The IP type of the Cloud SQL instance to connect to. Can be one
+            of "public", "private", or "psc".
         refresh_strategy (Optional[str]):
             Refresh strategy for the Cloud SQL Connector. Can be one of "lazy"
             or "background". For serverless environments use "lazy" to avoid
@@ -145,7 +153,7 @@ async def getconn(
             user=user,
             password=password,
             db=db,
-            ip_type="public",  # can also be "private" or "psc",
+            ip_type=ip_type,  # can be "public", "private" or "psc"
             **kwargs,
         )
         return conn
@@ -161,8 +169,11 @@ async def test_sqlalchemy_connection_with_asyncpg() -> None:
     user = os.environ["POSTGRES_USER"]
     password = os.environ["POSTGRES_PASS"]
     db = os.environ["POSTGRES_DB"]
+    ip_type = os.environ.get("IP_TYPE", "public")
 
-    pool, connector = await create_sqlalchemy_engine(inst_conn_name, user, password, db)
+    pool, connector = await create_sqlalchemy_engine(
+        inst_conn_name, user, password, db, ip_type
+    )
 
     async with pool.connect() as conn:
         res = (await conn.execute(sqlalchemy.text("SELECT 1"))).fetchone()
@@ -177,9 +188,10 @@ async def test_lazy_sqlalchemy_connection_with_asyncpg() -> None:
     user = os.environ["POSTGRES_USER"]
     password = os.environ["POSTGRES_PASS"]
     db = os.environ["POSTGRES_DB"]
+    ip_type = os.environ.get("IP_TYPE", "public")
 
     pool, connector = await create_sqlalchemy_engine(
-        inst_conn_name, user, password, db, "lazy"
+        inst_conn_name, user, password, db, ip_type, "lazy"
     )
 
     async with pool.connect() as conn:
@@ -195,9 +207,10 @@ async def test_custom_SAN_with_dns_sqlalchemy_connection_with_asyncpg() -> None:
     user = os.environ["POSTGRES_USER"]
     password = os.environ["POSTGRES_CUSTOMER_CAS_PASS"]
     db = os.environ["POSTGRES_DB"]
+    ip_type = os.environ.get("IP_TYPE", "public")
 
     pool, connector = await create_sqlalchemy_engine(
-        inst_conn_name, user, password, db, resolver=DnsResolver
+        inst_conn_name, user, password, db, ip_type, resolver=DnsResolver
     )
 
     async with pool.connect() as conn:
@@ -213,8 +226,11 @@ async def test_connection_with_asyncpg() -> None:
     user = os.environ["POSTGRES_USER"]
     password = os.environ["POSTGRES_PASS"]
     db = os.environ["POSTGRES_DB"]
+    ip_type = os.environ.get("IP_TYPE", "public")
 
-    pool, connector = await create_asyncpg_pool(inst_conn_name, user, password, db)
+    pool, connector = await create_asyncpg_pool(
+        inst_conn_name, user, password, db, ip_type
+    )
 
     async with pool.acquire() as conn:
         res = await conn.fetch("SELECT 1")
@@ -229,9 +245,10 @@ async def test_lazy_connection_with_asyncpg() -> None:
     user = os.environ["POSTGRES_USER"]
     password = os.environ["POSTGRES_PASS"]
     db = os.environ["POSTGRES_DB"]
+    ip_type = os.environ.get("IP_TYPE", "public")
 
     pool, connector = await create_asyncpg_pool(
-        inst_conn_name, user, password, db, "lazy"
+        inst_conn_name, user, password, db, ip_type, "lazy"
     )
 
     async with pool.acquire() as conn:

tests/system/test_asyncpg_iam_auth.py

@@ -27,6 +27,7 @@ async def create_sqlalchemy_engine(
     instance_connection_name: str,
     user: str,
     db: str,
+    ip_type: str = "public",
     refresh_strategy: str = "background",
 ) -> tuple[sqlalchemy.ext.asyncio.engine.AsyncEngine, Connector]:
     """Creates a connection pool for a Cloud SQL instance and returns the pool
@@ -55,6 +56,9 @@ async def create_sqlalchemy_engine(
             e.g., [email protected], [email protected]
         db (str):
             The name of the database, e.g., mydb
+        ip_type (str):
+            The IP type of the Cloud SQL instance to connect to. Can be one
+            of "public", "private", or "psc".
         refresh_strategy (Optional[str]):
             Refresh strategy for the Cloud SQL Connector. Can be one of "lazy"
             or "background". For serverless environments use "lazy" to avoid
@@ -71,7 +75,7 @@ async def create_sqlalchemy_engine(
             "asyncpg",
             user=user,
             db=db,
-            ip_type="public",  # can also be "private" or "psc"
+            ip_type=ip_type,  # can be "public", "private" or "psc"
             enable_iam_auth=True,
         ),
         execution_options={"isolation_level": "AUTOCOMMIT"},
@@ -84,8 +88,9 @@ async def test_iam_authn_connection_with_asyncpg() -> None:
     inst_conn_name = os.environ["POSTGRES_CONNECTION_NAME"]
     user = os.environ["POSTGRES_IAM_USER"]
     db = os.environ["POSTGRES_DB"]
+    ip_type = os.environ.get("IP_TYPE", "public")
 
-    pool, connector = await create_sqlalchemy_engine(inst_conn_name, user, db)
+    pool, connector = await create_sqlalchemy_engine(inst_conn_name, user, db, ip_type)
 
     async with pool.connect() as conn:
         res = (await conn.execute(sqlalchemy.text("SELECT 1"))).fetchone()
@@ -99,8 +104,11 @@ async def test_lazy_iam_authn_connection_with_asyncpg() -> None:
     inst_conn_name = os.environ["POSTGRES_CONNECTION_NAME"]
     user = os.environ["POSTGRES_IAM_USER"]
     db = os.environ["POSTGRES_DB"]
+    ip_type = os.environ.get("IP_TYPE", "public")
 
-    pool, connector = await create_sqlalchemy_engine(inst_conn_name, user, db, "lazy")
+    pool, connector = await create_sqlalchemy_engine(
+        inst_conn_name, user, db, ip_type, "lazy"
+    )
 
     async with pool.connect() as conn:
         res = (await conn.execute(sqlalchemy.text("SELECT 1"))).fetchone()

tests/system/test_connector_object.py

@@ -38,6 +38,7 @@ def getconn() -> pymysql.connections.Connection:
             user=os.environ["MYSQL_USER"],
             password=os.environ["MYSQL_PASS"],
             db=os.environ["MYSQL_DB"],
+            ip_type=os.environ.get("IP_TYPE", "public"),
         )
         return conn