feat: Add universal HTTP/HTTPS proxy configuration support to Python CDK

- Add ProxyConfig class with support for HTTP/HTTPS proxies, authentication, and SSL certificates
- Integrate proxy configuration into HttpClient with session configuration logic
- Update HttpRequester to extract proxy config from connector configuration
- Update HttpStream to accept optional proxy configuration parameter
- Add comprehensive unit tests for proxy functionality
- Maintain full backward compatibility with existing connectors
- Support environment variable fallback for proxy configuration
- Handle secure certificate storage using temporary files with restricted permissions

Co-Authored-By: AJ Steers <[email protected]>

Author: devin-ai-integration[bot]

airbyte_cdk/sources/declarative/requesters/http_requester.py

@@ -26,6 +26,7 @@
 from airbyte_cdk.sources.streams.call_rate import APIBudget
 from airbyte_cdk.sources.streams.http import HttpClient
 from airbyte_cdk.sources.streams.http.error_handlers import ErrorHandler
+from airbyte_cdk.sources.streams.http.proxy_config import ProxyConfig
 from airbyte_cdk.sources.types import Config, EmptyString, StreamSlice, StreamState
 from airbyte_cdk.utils.mapping_helpers import (
     combine_mappings,
@@ -104,6 +105,8 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         else:
             backoff_strategies = None
 
+        proxy_config = ProxyConfig.from_config(self.config)
+
         self._http_client = HttpClient(
             name=self.name,
             logger=self.logger,
@@ -114,6 +117,7 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
             backoff_strategy=backoff_strategies,
             disable_retries=self.disable_retries,
             message_repository=self.message_repository,
+            proxy_config=proxy_config,
         )
 
     @property

airbyte_cdk/sources/streams/http/__init__.py

@@ -6,5 +6,12 @@
 from .exceptions import UserDefinedBackoffException
 from .http import HttpStream, HttpSubStream
 from .http_client import HttpClient
+from .proxy_config import ProxyConfig
 
-__all__ = ["HttpClient", "HttpStream", "HttpSubStream", "UserDefinedBackoffException"]
+__all__ = [
+    "HttpClient",
+    "HttpStream",
+    "HttpSubStream",
+    "ProxyConfig",
+    "UserDefinedBackoffException",
+]

airbyte_cdk/sources/streams/http/http.py

@@ -34,6 +34,7 @@
     ResponseAction,
 )
 from airbyte_cdk.sources.streams.http.http_client import HttpClient
+from airbyte_cdk.sources.streams.http.proxy_config import ProxyConfig
 from airbyte_cdk.sources.types import Record, StreamSlice
 from airbyte_cdk.sources.utils.types import JsonType
 
@@ -52,7 +53,10 @@ class HttpStream(Stream, CheckpointMixin, ABC):
     )
 
     def __init__(
-        self, authenticator: Optional[AuthBase] = None, api_budget: Optional[APIBudget] = None
+        self,
+        authenticator: Optional[AuthBase] = None,
+        api_budget: Optional[APIBudget] = None,
+        proxy_config: Optional[ProxyConfig] = None,
     ):
         self._exit_on_rate_limit: bool = False
         self._http_client = HttpClient(
@@ -64,6 +68,7 @@ def __init__(
             use_cache=self.use_cache,
             backoff_strategy=self.get_backoff_strategy(),
             message_repository=InMemoryMessageRepository(),
+            proxy_config=proxy_config,
         )
 
         # There are three conditions that dictate if RFR should automatically be applied to a stream

airbyte_cdk/sources/streams/http/http_client.py

@@ -40,6 +40,7 @@
     RequestBodyException,
     UserDefinedBackoffException,
 )
+from airbyte_cdk.sources.streams.http.proxy_config import ProxyConfig
 from airbyte_cdk.sources.streams.http.rate_limiting import (
     http_client_default_backoff_handler,
     rate_limit_default_backoff_handler,
@@ -91,9 +92,11 @@ def __init__(
         error_message_parser: Optional[ErrorMessageParser] = None,
         disable_retries: bool = False,
         message_repository: Optional[MessageRepository] = None,
+        proxy_config: Optional[ProxyConfig] = None,
     ):
         self._name = name
         self._api_budget: APIBudget = api_budget or APIBudget(policies=[])
+        self._proxy_config = proxy_config
         if session:
             self._session = session
         else:
@@ -105,6 +108,10 @@ def __init__(
                     pool_connections=MAX_CONNECTION_POOL_SIZE, pool_maxsize=MAX_CONNECTION_POOL_SIZE
                 ),
             )
+
+        if self._proxy_config:
+            self._session = self._proxy_config.configure_session(self._session)
+
         if isinstance(authenticator, AuthBase):
             self._session.auth = authenticator
         self._logger = logger

airbyte_cdk/sources/streams/http/proxy_config.py

@@ -0,0 +1,121 @@
+#
+#
+
+import base64
+import os
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, Optional, Tuple
+
+import requests
+from requests.auth import HTTPProxyAuth
+
+
+@dataclass
+class ProxyConfig:
+    """Configuration for HTTP/HTTPS proxy settings."""
+
+    enabled: bool = False
+    http_proxy: Optional[str] = None
+    https_proxy: Optional[str] = None
+    no_proxy: Optional[str] = None
+    proxy_username: Optional[str] = None
+    proxy_password: Optional[str] = None
+    verify_ssl: bool = True
+    ca_certificate: Optional[str] = None
+    client_certificate: Optional[str] = None
+    client_key: Optional[str] = None
+
+    def get_proxies(self) -> Dict[str, str]:
+        """Get proxy configuration for requests library."""
+        if not self.enabled:
+            return {}
+
+        proxies = {}
+        if self.http_proxy:
+            proxies["http"] = self.http_proxy
+        if self.https_proxy:
+            proxies["https"] = self.https_proxy
+
+        return proxies
+
+    def get_proxy_auth(self) -> Optional[HTTPProxyAuth]:
+        """Get proxy authentication if configured."""
+        if self.proxy_username and self.proxy_password:
+            return HTTPProxyAuth(self.proxy_username, self.proxy_password)
+        return None
+
+    def configure_session(self, session: requests.Session) -> requests.Session:
+        """Configure a requests session with proxy settings."""
+        if not self.enabled:
+            return session
+
+        proxies = self.get_proxies()
+        if proxies:
+            session.proxies.update(proxies)
+
+        proxy_auth = self.get_proxy_auth()
+        if proxy_auth:
+            session.auth = proxy_auth
+
+        if not self.verify_ssl:
+            session.verify = False
+        elif self.ca_certificate:
+            ca_cert_path = self._write_temp_cert(self.ca_certificate, "ca_cert")
+            session.verify = ca_cert_path
+
+        if self.client_certificate and self.client_key:
+            client_cert_path = self._write_temp_cert(self.client_certificate, "client_cert")
+            client_key_path = self._write_temp_cert(self.client_key, "client_key")
+            session.cert = (client_cert_path, client_key_path)
+
+        return session
+
+    def _write_temp_cert(self, cert_content: str, cert_type: str) -> str:
+        """Write certificate content to a secure temporary file."""
+        try:
+            cert_bytes = base64.b64decode(cert_content)
+        except Exception as e:
+            raise ValueError(f"Invalid base64 certificate content for {cert_type}: {e}")
+
+        fd, temp_path = tempfile.mkstemp(suffix=f"_{cert_type}.pem", prefix="airbyte_proxy_")
+        try:
+            os.fchmod(fd, 0o600)  # Restrict to owner only
+            os.write(fd, cert_bytes)
+        finally:
+            os.close(fd)
+
+        return temp_path
+
+    @classmethod
+    def from_config(cls, config: Dict) -> Optional["ProxyConfig"]:
+        """Create ProxyConfig from connector configuration."""
+        proxy_section = config.get("proxy", {})
+
+        if not proxy_section.get("enabled", False):
+            http_proxy = os.getenv("HTTP_PROXY") or os.getenv("http_proxy")
+            https_proxy = os.getenv("HTTPS_PROXY") or os.getenv("https_proxy")
+            no_proxy = os.getenv("NO_PROXY") or os.getenv("no_proxy")
+
+            if http_proxy or https_proxy:
+                return cls(
+                    enabled=True,
+                    http_proxy=http_proxy,
+                    https_proxy=https_proxy,
+                    no_proxy=no_proxy,
+                )
+            return None
+
+        return cls(
+            enabled=proxy_section.get("enabled", False),
+            http_proxy=proxy_section.get("http_proxy"),
+            https_proxy=proxy_section.get("https_proxy"),
+            no_proxy=proxy_section.get("no_proxy"),
+            proxy_username=proxy_section.get("proxy_username"),
+            proxy_password=proxy_section.get("proxy_password"),
+            verify_ssl=proxy_section.get("verify_ssl", True),
+            ca_certificate=proxy_section.get("ca_certificate"),
+            client_certificate=proxy_section.get("client_certificate"),
+            client_key=proxy_section.get("client_key"),
+        )