Repository: GoogleCloudPlatform/cloud-sql-python-connector

Issue:
Add domain name validation to Python Connector
Add regex to validate domain name used is in proper format.

Similar to how we validate an instance connection name:

https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/blob/fb8c21c89af524bdd00c51eeb0bcb4a6bf4dbbd3/google/cloud/sql/connector/connection_name.py#L46-L54

Below is what we used in Go Connector:

```go
// The domain name pattern in accordance with RFC 1035, RFC 1123 and RFC 2181.
domainNameRegex = regexp.MustCompile(`^(?:[_a-z0-9](?:[_a-z0-9-]{0,61}[a-z0-9])?\.)+(?:[a-z](?:[a-z0-9-]{0,61}[a-z0-9])?)?$`)
```

Can use Go as ref: https://github.com/GoogleCloudPlatform/cloud-sql-go-connector/pull/925

Hints from issue comments:

Solution patch:
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/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)