Repository: GoogleCloudPlatform/cloud-sql-python-connector
Pull request #1242 (2025-03-12 03:53:15)

refactor: Use new ConnectSettings.DnsNames field to validate the server TLS certificate. 
When the connector is configured with a DNS name, or if the Cloud SQL Instance reports that it has a DNS Name,
the connector will use standard TLS hostname validation when checking the server certificate. Now, the server's
TLS certificate must contain a SAN record with the instance's DNS name.

The ConnectSettings API added a field dns_names which contains all of the valid DNS names for
an instance.

See also: https://github.com/GoogleCloudPlatform/cloud-sql-go-connector/pull/954


Patch:
diff --git a/.gitignore b/.gitignore
index 9ef6a90..9f449ce 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,3 +5,6 @@ venv
 .python-version
 cloud_sql_python_connector.egg-info/
 dist/
+.idea
+.coverage
+sponge_log.xml
diff --git a/google/cloud/sql/connector/client.py b/google/cloud/sql/connector/client.py
index 8a31eb9..556a01b 100644
--- a/google/cloud/sql/connector/client.py
+++ b/google/cloud/sql/connector/client.py
@@ -156,10 +156,23 @@ class CloudSQLClient:
         # resolve dnsName into IP address for PSC
         # Note that we have to check for PSC enablement also because CAS
         # instances also set the dnsName field.
-        # Remove trailing period from DNS name. Required for SSL in Python
-        dns_name = ret_dict.get("dnsName", "").rstrip(".")
-        if dns_name and ret_dict.get("pscEnabled"):
-            ip_addresses["PSC"] = dns_name
+        if ret_dict.get("pscEnabled"):
+            # Find PSC instance DNS name in the dns_names field
+            psc_dns_names = [
+                d["name"]
+                for d in ret_dict.get("dnsNames", [])
+                if d["connectionType"] == "PRIVATE_SERVICE_CONNECT"
+                and d["dnsScope"] == "INSTANCE"
+            ]
+            dns_name = psc_dns_names[0] if psc_dns_names else None
+
+            # Fall back do dns_name field if dns_names is not set
+            if dns_name is None:
+                dns_name = ret_dict.get("dnsName", None)
+
+            # Remove trailing period from DNS name. Required for SSL in Python
+            if dns_name:
+                ip_addresses["PSC"] = dns_name.rstrip(".")
 
         return {
             "ip_addresses": ip_addresses,
diff --git a/google/cloud/sql/connector/connection_name.py b/google/cloud/sql/connector/connection_name.py
index 1bf711a..437fd66 100644
--- a/google/cloud/sql/connector/connection_name.py
+++ b/google/cloud/sql/connector/connection_name.py
@@ -19,6 +19,10 @@ import re
 # Additionally, we have to support legacy "domain-scoped" projects
 # (e.g. "google.com:PROJECT")
 CONN_NAME_REGEX = re.compile(("([^:]+(:[^:]+)?):([^:]+):([^:]+)"))
+# The domain name pattern in accordance with RFC 1035, RFC 1123 and RFC 2181.
+DOMAIN_NAME_REGEX = re.compile(
+    r"^(?:[_a-z0-9](?:[_a-z0-9-]{0,61}[a-z0-9])?\.)+(?:[a-z](?:[a-z0-9-]{0,61}[a-z0-9])?)?$"
+)
 
 
 @dataclass
@@ -39,6 +43,12 @@ class ConnectionName:
         return f"{self.project}:{self.region}:{self.instance_name}"
 
 
+def _is_valid_domain(domain_name: str) -> bool:
+    if DOMAIN_NAME_REGEX.fullmatch(domain_name) is None:
+        return False
+    return True
+
+
 def _parse_connection_name(connection_name: str) -> ConnectionName:
     return _parse_connection_name_with_domain_name(connection_name, "")
 
diff --git a/google/cloud/sql/connector/resolver.py b/google/cloud/sql/connector/resolver.py
index 39efd04..7d717ca 100644
--- a/google/cloud/sql/connector/resolver.py
+++ b/google/cloud/sql/connector/resolver.py
@@ -17,6 +17,7 @@ import dns.asyncresolver
 from google.cloud.sql.connector.connection_name import (
     _parse_connection_name_with_domain_name,
 )
+from google.cloud.sql.connector.connection_name import _is_valid_domain
 from google.cloud.sql.connector.connection_name import _parse_connection_name
 from google.cloud.sql.connector.connection_name import ConnectionName
 from google.cloud.sql.connector.exceptions import DnsResolutionError
@@ -40,8 +41,16 @@ class DnsResolver(dns.asyncresolver.Resolver):
             conn_name = _parse_connection_name(dns)
         except ValueError:
             # The connection name was not project:region:instance format.
-            # Attempt to query a TXT record to get connection name.
-            conn_name = await self.query_dns(dns)
+            # Check if connection name is a valid DNS domain name
+            if _is_valid_domain(dns):
+                # Attempt to query a TXT record to get connection name.
+                conn_name = await self.query_dns(dns)
+            else:
+                raise ValueError(
+                    "Arg `instance_connection_string` must have "
+                    "format: PROJECT:REGION:INSTANCE or be a valid DNS domain "
+                    f"name, got {dns}."
+                )
         return conn_name
 
     async def query_dns(self, dns: str) -> ConnectionName:


Test patch:
diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml
index e710138..b8e6eb5 100644
--- a/.github/workflows/tests.yml
+++ b/.github/workflows/tests.yml
@@ -81,6 +81,7 @@ jobs:
             POSTGRES_CAS_PASS:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_CAS_PASS
             POSTGRES_CUSTOMER_CAS_CONNECTION_NAME:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_CUSTOMER_CAS_CONNECTION_NAME
             POSTGRES_CUSTOMER_CAS_PASS:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_CUSTOMER_CAS_PASS
+            POSTGRES_CUSTOMER_CAS_PASS_VALID_DOMAIN_NAME:${{ vars.GOOGLE_CLOUD_PROJECT }}/POSTGRES_CUSTOMER_CAS_PASS_VALID_DOMAIN_NAME
             SQLSERVER_CONNECTION_NAME:${{ vars.GOOGLE_CLOUD_PROJECT }}/SQLSERVER_CONNECTION_NAME
             SQLSERVER_USER:${{ vars.GOOGLE_CLOUD_PROJECT }}/SQLSERVER_USER
             SQLSERVER_PASS:${{ vars.GOOGLE_CLOUD_PROJECT }}/SQLSERVER_PASS
@@ -102,6 +103,7 @@ jobs:
           POSTGRES_CAS_PASS: "${{ steps.secrets.outputs.POSTGRES_CAS_PASS }}"
           POSTGRES_CUSTOMER_CAS_CONNECTION_NAME: "${{ steps.secrets.outputs.POSTGRES_CUSTOMER_CAS_CONNECTION_NAME }}"
           POSTGRES_CUSTOMER_CAS_PASS: "${{ steps.secrets.outputs.POSTGRES_CUSTOMER_CAS_PASS }}"
+          POSTGRES_CUSTOMER_CAS_PASS_VALID_DOMAIN_NAME: "${{ steps.secrets.outputs.POSTGRES_CUSTOMER_CAS_PASS_VALID_DOMAIN_NAME }}"
           SQLSERVER_CONNECTION_NAME: "${{ steps.secrets.outputs.SQLSERVER_CONNECTION_NAME }}"
           SQLSERVER_USER: "${{ steps.secrets.outputs.SQLSERVER_USER }}"
           SQLSERVER_PASS: "${{ steps.secrets.outputs.SQLSERVER_PASS }}"
diff --git a/tests/system/test_asyncpg_connection.py b/tests/system/test_asyncpg_connection.py
index 8de14d5..dfcc394 100644
--- a/tests/system/test_asyncpg_connection.py
+++ b/tests/system/test_asyncpg_connection.py
@@ -16,13 +16,15 @@ limitations under the License.
 
 import asyncio
 import os
-from typing import Any
+from typing import Any, Union
 
 import asyncpg
 import sqlalchemy
 import sqlalchemy.ext.asyncio
 
 from google.cloud.sql.connector import Connector
+from google.cloud.sql.connector import DefaultResolver
+from google.cloud.sql.connector import DnsResolver
 
 
 async def create_sqlalchemy_engine(
@@ -31,6 +33,7 @@ async def create_sqlalchemy_engine(
     password: str,
     db: str,
     refresh_strategy: str = "background",
+    resolver: Union[type[DefaultResolver], type[DnsResolver]] = DefaultResolver,
 ) -> tuple[sqlalchemy.ext.asyncio.engine.AsyncEngine, Connector]:
     """Creates a connection pool for a Cloud SQL instance and returns the pool
     and the connector. Callers are responsible for closing the pool and the
@@ -64,9 +67,16 @@ async def create_sqlalchemy_engine(
             Refresh strategy for the Cloud SQL Connector. Can be one of "lazy"
             or "background". For serverless environments use "lazy" to avoid
             errors resulting from CPU being throttled.
+        resolver (Optional[google.cloud.sql.connector.DefaultResolver]):
+            Resolver class for resolving instance connection name. Use
+            google.cloud.sql.connector.DnsResolver when resolving DNS domain
+            names or google.cloud.sql.connector.DefaultResolver for regular
+            instance connection names ("my-project:my-region:my-instance").
     """
     loop = asyncio.get_running_loop()
-    connector = Connector(loop=loop, refresh_strategy=refresh_strategy)
+    connector = Connector(
+        loop=loop, refresh_strategy=refresh_strategy, resolver=resolver
+    )
 
     async def getconn() -> asyncpg.Connection:
         conn: asyncpg.Connection = await connector.connect_async(
@@ -183,6 +193,24 @@ async def test_lazy_sqlalchemy_connection_with_asyncpg() -> None:
     await connector.close_async()
 
 
+async def test_custom_SAN_with_dns_sqlalchemy_connection_with_asyncpg() -> None:
+    """Basic test to get time from database."""
+    inst_conn_name = os.environ["POSTGRES_CUSTOMER_CAS_PASS_VALID_DOMAIN_NAME"]
+    user = os.environ["POSTGRES_USER"]
+    password = os.environ["POSTGRES_CUSTOMER_CAS_PASS"]
+    db = os.environ["POSTGRES_DB"]
+
+    pool, connector = await create_sqlalchemy_engine(
+        inst_conn_name, user, password, db, resolver=DnsResolver
+    )
+
+    async with pool.connect() as conn:
+        res = (await conn.execute(sqlalchemy.text("SELECT 1"))).fetchone()
+        assert res[0] == 1
+
+    await connector.close_async()
+
+
 async def test_connection_with_asyncpg() -> None:
     """Basic test to get time from database."""
     inst_conn_name = os.environ["POSTGRES_CONNECTION_NAME"]
diff --git a/tests/system/test_pg8000_connection.py b/tests/system/test_pg8000_connection.py
index b56a8e8..c47b860 100644
--- a/tests/system/test_pg8000_connection.py
+++ b/tests/system/test_pg8000_connection.py
@@ -18,10 +18,14 @@ from datetime import datetime
 import os
 
 # [START cloud_sql_connector_postgres_pg8000]
+from typing import Union
+
 import pg8000
 import sqlalchemy
 
 from google.cloud.sql.connector import Connector
+from google.cloud.sql.connector import DefaultResolver
+from google.cloud.sql.connector import DnsResolver
 
 
 def create_sqlalchemy_engine(
@@ -30,6 +34,7 @@ def create_sqlalchemy_engine(
     password: str,
     db: str,
     refresh_strategy: str = "background",
+    resolver: Union[type[DefaultResolver], type[DnsResolver]] = DefaultResolver,
 ) -> tuple[sqlalchemy.engine.Engine, Connector]:
     """Creates a connection pool for a Cloud SQL instance and returns the pool
     and the connector. Callers are responsible for closing the pool and the
@@ -64,8 +69,13 @@ def create_sqlalchemy_engine(
             Refresh strategy for the Cloud SQL Connector. Can be one of "lazy"
             or "background". For serverless environments use "lazy" to avoid
             errors resulting from CPU being throttled.
+        resolver (Optional[google.cloud.sql.connector.DefaultResolver]):
+            Resolver class for resolving instance connection name. Use
+            google.cloud.sql.connector.DnsResolver when resolving DNS domain
+            names or google.cloud.sql.connector.DefaultResolver for regular
+            instance connection names ("my-project:my-region:my-instance").
     """
-    connector = Connector(refresh_strategy=refresh_strategy)
+    connector = Connector(refresh_strategy=refresh_strategy, resolver=resolver)
 
     def getconn() -> pg8000.dbapi.Connection:
         conn: pg8000.dbapi.Connection = connector.connect(
@@ -153,3 +163,21 @@ def test_customer_managed_CAS_pg8000_connection() -> None:
         curr_time = time[0]
         assert type(curr_time) is datetime
     connector.close()
+
+
+def test_custom_SAN_with_dns_pg8000_connection() -> None:
+    """Basic test to get time from database."""
+    inst_conn_name = os.environ["POSTGRES_CUSTOMER_CAS_PASS_VALID_DOMAIN_NAME"]
+    user = os.environ["POSTGRES_USER"]
+    password = os.environ["POSTGRES_CUSTOMER_CAS_PASS"]
+    db = os.environ["POSTGRES_DB"]
+
+    engine, connector = create_sqlalchemy_engine(
+        inst_conn_name, user, password, db, resolver=DnsResolver
+    )
+    with engine.connect() as conn:
+        time = conn.execute(sqlalchemy.text("SELECT NOW()")).fetchone()
+        conn.commit()
+        curr_time = time[0]
+        assert type(curr_time) is datetime
+    connector.close()
diff --git a/tests/unit/mocks.py b/tests/unit/mocks.py
index 5d86367..cd3299b 100644
--- a/tests/unit/mocks.py
+++ b/tests/unit/mocks.py
@@ -225,6 +225,7 @@ class FakeCSQLInstance:
             "PRIMARY": "127.0.0.1",
             "PRIVATE": "10.0.0.1",
         },
+        legacy_dns_name: bool = False,
         cert_before: datetime = datetime.datetime.now(datetime.timezone.utc),
         cert_expiration: datetime = datetime.datetime.now(datetime.timezone.utc)
         + datetime.timedelta(hours=1),
@@ -237,6 +238,7 @@ class FakeCSQLInstance:
         self.psc_enabled = False
         self.cert_before = cert_before
         self.cert_expiration = cert_expiration
+        self.legacy_dns_name = legacy_dns_name
         # create self signed CA cert
         self.server_ca, self.server_key = generate_cert(
             self.project, self.name, cert_before, cert_expiration
@@ -255,12 +257,22 @@ class FakeCSQLInstance:
                 "instance": self.name,
                 "expirationTime": str(self.cert_expiration),
             },
-            "dnsName": "abcde.12345.us-central1.sql.goog",
             "pscEnabled": self.psc_enabled,
             "ipAddresses": ip_addrs,
             "region": self.region,
             "databaseVersion": self.db_version,
         }
+        if self.legacy_dns_name:
+            response["dnsName"] = "abcde.12345.us-central1.sql.goog"
+        else:
+            response["dnsNames"] = [
+                {
+                    "name": "abcde.12345.us-central1.sql.goog",
+                    "connectionType": "PRIVATE_SERVICE_CONNECT",
+                    "dnsScope": "INSTANCE",
+                }
+            ]
+
         return web.Response(content_type="application/json", body=json.dumps(response))
 
     async def generate_ephemeral(self, request: Any) -> web.Response:
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
index af42af0..cfe5094 100644
--- a/tests/unit/test_client.py
+++ b/tests/unit/test_client.py
@@ -65,6 +65,28 @@ async def test_get_metadata_with_psc(fake_client: CloudSQLClient) -> None:
     assert isinstance(resp["server_ca_cert"], str)
 
 
+@pytest.mark.asyncio
+async def test_get_metadata_legacy_dns_with_psc(fake_client: CloudSQLClient) -> None:
+    """
+    Test _get_metadata returns successfully with PSC IP type.
+    """
+    # set PSC to enabled on test instance
+    fake_client.instance.psc_enabled = True
+    fake_client.instance.legacy_dns_name = True
+    resp = await fake_client._get_metadata(
+        "test-project",
+        "test-region",
+        "test-instance",
+    )
+    assert resp["database_version"] == "POSTGRES_15"
+    assert resp["ip_addresses"] == {
+        "PRIMARY": "127.0.0.1",
+        "PRIVATE": "10.0.0.1",
+        "PSC": "abcde.12345.us-central1.sql.goog",
+    }
+    assert isinstance(resp["server_ca_cert"], str)
+
+
 @pytest.mark.asyncio
 async def test_get_ephemeral(fake_client: CloudSQLClient) -> None:
     """
diff --git a/tests/unit/test_connection_name.py b/tests/unit/test_connection_name.py
index 783e14f..218034d 100644
--- a/tests/unit/test_connection_name.py
+++ b/tests/unit/test_connection_name.py
@@ -17,6 +17,7 @@ import pytest  # noqa F401 Needed to run the tests
 from google.cloud.sql.connector.connection_name import (
     _parse_connection_name_with_domain_name,
 )
+from google.cloud.sql.connector.connection_name import _is_valid_domain
 from google.cloud.sql.connector.connection_name import _parse_connection_name
 from google.cloud.sql.connector.connection_name import ConnectionName
 
@@ -96,3 +97,40 @@ def test_parse_connection_name_with_domain_name(
     assert expected == _parse_connection_name_with_domain_name(
         connection_name, domain_name
     )
+
+
+@pytest.mark.parametrize(
+    "domain_name, expected",
+    [
+        (
+            "prod-db.mycompany.example.com",
+            True,
+        ),
+        (
+            "example.com.",  # trailing dot
+            True,
+        ),
+        (
+            "-example.com.",  # leading hyphen
+            False,
+        ),
+        (
+            "example",  # missing TLD
+            False,
+        ),
+        (
+            "127.0.0.1",  # IPv4 address
+            False,
+        ),
+        (
+            "0:0:0:0:0:0:0:1",  # IPv6 address
+            False,
+        ),
+    ],
+)
+def test_is_valid_domain(domain_name: str, expected: bool) -> None:
+    """
+    Test that _is_valid_domain works correctly for
+    parsing domain names.
+    """
+    assert expected == _is_valid_domain(domain_name)