PYTHON-5288: SRV hostname validation fails when resolver and resolved hostnames are identical with three domain levels (#2272)

Author: aclark4life

doc/changelog.rst

@@ -1,6 +1,22 @@
 Changelog
 =========
 
+Changes in Version 4.12.1 (XXXX/XX/XX)
+--------------------------------------
+
+Version 4.12.1 is a bug fix release.
+
+- Fixed a bug causing SRV hostname validation to fail when resolver and resolved hostnames are identical with three domain levels.
+
+Issues Resolved
+...............
+
+See the `PyMongo 4.12 release notes in JIRA`_ for the list of resolved issues
+in this release.
+
+.. _PyMongo 4.12 release notes in JIRA: https://jira.mongodb.org/secure/ReleaseNote.jspa?projectId=10004&version=41916
+.. _PYTHON-5288: https://jira.mongodb.org/browse/PYTHON-5288
+
 Changes in Version 4.12.0 (2025/04/08)
 --------------------------------------
 

pymongo/asynchronous/srv_resolver.py

@@ -96,6 +96,7 @@ def __init__(
         except Exception:
             raise ConfigurationError(_INVALID_HOST_MSG % (fqdn,)) from None
         self.__slen = len(self.__plist)
+        self.nparts = len(split_fqdn)
 
     async def get_options(self) -> Optional[str]:
         from dns import resolver
@@ -137,12 +138,13 @@ async def _get_srv_response_and_hosts(
 
         # Validate hosts
         for node in nodes:
-            if self.__fqdn == node[0].lower():
+            srv_host = node[0].lower()
+            if self.__fqdn == srv_host and self.nparts < 3:
                 raise ConfigurationError(
                     "Invalid SRV host: return address is identical to SRV hostname"
                 )
             try:
-                nlist = node[0].lower().split(".")[1:][-self.__slen :]
+                nlist = srv_host.split(".")[1:][-self.__slen :]
             except Exception:
                 raise ConfigurationError(f"Invalid SRV host: {node[0]}") from None
             if self.__plist != nlist:

pymongo/synchronous/srv_resolver.py

@@ -96,6 +96,7 @@ def __init__(
         except Exception:
             raise ConfigurationError(_INVALID_HOST_MSG % (fqdn,)) from None
         self.__slen = len(self.__plist)
+        self.nparts = len(split_fqdn)
 
     def get_options(self) -> Optional[str]:
         from dns import resolver
@@ -137,12 +138,13 @@ def _get_srv_response_and_hosts(
 
         # Validate hosts
         for node in nodes:
-            if self.__fqdn == node[0].lower():
+            srv_host = node[0].lower()
+            if self.__fqdn == srv_host and self.nparts < 3:
                 raise ConfigurationError(
                     "Invalid SRV host: return address is identical to SRV hostname"
                 )
             try:
-                nlist = node[0].lower().split(".")[1:][-self.__slen :]
+                nlist = srv_host.split(".")[1:][-self.__slen :]
             except Exception:
                 raise ConfigurationError(f"Invalid SRV host: {node[0]}") from None
             if self.__plist != nlist:

test/asynchronous/test_dns.py

@@ -220,12 +220,15 @@ async def mock_resolve(query, record_type, *args, **kwargs):
                 mock_resolver.side_effect = mock_resolve
                 domain = case["query"].split("._tcp.")[1]
                 connection_string = f"mongodb+srv://{domain}"
-                try:
+                if "expected_error" not in case:
                     await parse_uri(connection_string)
-                except ConfigurationError as e:
-                    self.assertIn(case["expected_error"], str(e))
                 else:
-                    self.fail(f"ConfigurationError was not raised for query: {case['query']}")
+                    try:
+                        await parse_uri(connection_string)
+                    except ConfigurationError as e:
+                        self.assertIn(case["expected_error"], str(e))
+                    else:
+                        self.fail(f"ConfigurationError was not raised for query: {case['query']}")
 
     async def test_1_allow_srv_hosts_with_fewer_than_three_dot_separated_parts(self):
         with patch("dns.asyncresolver.resolve"):
@@ -289,6 +292,17 @@ async def test_4_throw_when_return_address_does_not_contain_dot_separating_share
         ]
         await self.run_initial_dns_seedlist_discovery_prose_tests(test_cases)
 
+    async def test_5_when_srv_hostname_has_two_dot_separated_parts_it_is_valid_for_the_returned_hostname_to_be_identical(
+        self
+    ):
+        test_cases = [
+            {
+                "query": "_mongodb._tcp.blogs.mongodb.com",
+                "mock_target": "blogs.mongodb.com",
+            },
+        ]
+        await self.run_initial_dns_seedlist_discovery_prose_tests(test_cases)
+
 
 if __name__ == "__main__":
     unittest.main()

test/test_dns.py

@@ -218,12 +218,15 @@ def mock_resolve(query, record_type, *args, **kwargs):
                 mock_resolver.side_effect = mock_resolve
                 domain = case["query"].split("._tcp.")[1]
                 connection_string = f"mongodb+srv://{domain}"
-                try:
+                if "expected_error" not in case:
                     parse_uri(connection_string)
-                except ConfigurationError as e:
-                    self.assertIn(case["expected_error"], str(e))
                 else:
-                    self.fail(f"ConfigurationError was not raised for query: {case['query']}")
+                    try:
+                        parse_uri(connection_string)
+                    except ConfigurationError as e:
+                        self.assertIn(case["expected_error"], str(e))
+                    else:
+                        self.fail(f"ConfigurationError was not raised for query: {case['query']}")
 
     def test_1_allow_srv_hosts_with_fewer_than_three_dot_separated_parts(self):
         with patch("dns.resolver.resolve"):
@@ -287,6 +290,17 @@ def test_4_throw_when_return_address_does_not_contain_dot_separating_shared_part
         ]
         self.run_initial_dns_seedlist_discovery_prose_tests(test_cases)
 
+    def test_5_when_srv_hostname_has_two_dot_separated_parts_it_is_valid_for_the_returned_hostname_to_be_identical(
+        self
+    ):
+        test_cases = [
+            {
+                "query": "_mongodb._tcp.blogs.mongodb.com",
+                "mock_target": "blogs.mongodb.com",
+            },
+        ]
+        self.run_initial_dns_seedlist_discovery_prose_tests(test_cases)
+
 
 if __name__ == "__main__":
     unittest.main()