fix: re-raise DNS failures as ConnectionError in filtered_send for universal retryability

The SSRF protection in filtered_send caught socket.gaierror from
_is_private_url and re-raised it as requests.exceptions.InvalidURL.
While CDK streams handle InvalidURL as retryable (via PR #384), external
SDKs (e.g. facebook-business) do not, causing transient DNS blips to
produce unrecoverable sync failures.

Change the exception type to requests.ConnectionError, which is
universally retried by both CDK streams and external SDK error
handlers.

Co-Authored-By: bot_apk <apk@cognition.ai>

Author: devin-ai-integration[bot]

airbyte_cdk/entrypoint.py

@@ -430,10 +430,13 @@ def filtered_send(self: Any, request: PreparedRequest, **kwargs: Any) -> Respons
                     message="Invalid URL endpoint: The endpoint that data is being requested from belongs to a private network. Source connectors only support requesting data from public API endpoints.",
                 )
         except socket.gaierror as exception:
-            # This is a special case where the developer specifies an IP address string that is not formatted correctly like trailing
-            # whitespace which will fail the socket IP lookup. This only happens when using IP addresses and not text hostnames.
-            # Knowing that this is a request using the requests library, we will mock the exception without calling the lib
-            raise requests.exceptions.InvalidURL(f"Invalid URL {parsed_url}: {exception}")
+            # socket.gaierror fires on transient DNS failures (EAI_AGAIN / errno -3)
+            # for valid hostnames, not just malformed IPs.  Re-raise as ConnectionError
+            # so every caller (CDK streams *and* external SDKs) treats it as retryable.
+            hostname = parsed_url.hostname or ""
+            raise requests.ConnectionError(
+                f"DNS resolution failed for {str(hostname)}: {str(exception)}"
+            )
 
         return wrapped_fn(self, request, **kwargs)
 

unit_tests/test_entrypoint.py

@@ -3,6 +3,7 @@
 #
 
 import os
+import socket
 from argparse import Namespace
 from collections import defaultdict
 from copy import deepcopy
@@ -559,7 +560,7 @@ def test_invalid_command(entrypoint: AirbyteEntrypoint, config_mock):
         pytest.param(
             "CLOUD",
             "https://192.168.27.30    ",
-            ValueError,
+            requests.ConnectionError,
             id="test_cloud_incorrect_ip_format_is_rejected",
         ),
         pytest.param(
@@ -944,3 +945,55 @@ def test_memory_failfast_flushes_queued_state_before_raising(mocker):
     with pytest.raises(AirbyteTracedException) as exc_info:
         next(gen)
     assert exc_info.value is fail_fast_exc
+
+
+@pytest.mark.parametrize(
+    "gaierror_errno,gaierror_msg",
+    [
+        pytest.param(
+            socket.EAI_AGAIN, "Temporary failure in name resolution", id="transient_dns_EAI_AGAIN"
+        ),
+        pytest.param(socket.EAI_NONAME, "Name or service not known", id="unknown_host_EAI_NONAME"),
+        pytest.param(
+            socket.EAI_FAIL,
+            "Non-recoverable failure in name resolution",
+            id="permanent_dns_EAI_FAIL",
+        ),
+    ],
+)
+def test_filtered_send_raises_connection_error_on_dns_failure(gaierror_errno, gaierror_msg):
+    """filtered_send must re-raise socket.gaierror as ConnectionError, not InvalidURL."""
+    entrypoint_module._init_internal_request_filter()
+
+    prepared = requests.PreparedRequest()
+    prepared.prepare_url("https://graph.facebook.com/v25.0/", None)
+
+    dns_error = socket.gaierror(gaierror_errno, gaierror_msg)
+
+    with patch.object(entrypoint_module, "_is_private_url", side_effect=dns_error):
+        with pytest.raises(
+            requests.ConnectionError, match="DNS resolution failed for graph.facebook.com"
+        ):
+            requests.Session().send(prepared)
+
+
+def test_filtered_send_does_not_raise_invalid_url_on_dns_failure():
+    """Verify the old behavior (InvalidURL) no longer occurs for DNS errors."""
+    entrypoint_module._init_internal_request_filter()
+
+    prepared = requests.PreparedRequest()
+    prepared.prepare_url("https://graph.facebook.com/v25.0/", None)
+
+    dns_error = socket.gaierror(socket.EAI_AGAIN, "Temporary failure in name resolution")
+
+    with patch.object(entrypoint_module, "_is_private_url", side_effect=dns_error):
+        with pytest.raises(requests.ConnectionError):
+            requests.Session().send(prepared)
+        # Confirm it is NOT InvalidURL
+        try:
+            with patch.object(entrypoint_module, "_is_private_url", side_effect=dns_error):
+                requests.Session().send(prepared)
+        except requests.ConnectionError:
+            pass
+        except requests.exceptions.InvalidURL:
+            pytest.fail("DNS failure should raise ConnectionError, not InvalidURL")