Merge pull request #818 from airweave-ai/feat/pipedream-proxy-auth

feat(pipedream): add proxy auth

Author: orhanrauf

.cursor/rules/auth-providers.mdc

@@ -90,3 +90,35 @@ token_manager = TokenManager(
 - **auth_provider_connections**: User's configured providers (encrypted)
 - **source_connections**: Links to auth provider via `auth_provider_connection_id`
 # Airweave Auth Providers
+
+
+### Pipedream Proxy Authentication
+
+Pipedream supports two authentication modes depending on the OAuth client type:
+
+#### Direct Mode (Custom OAuth Clients)
+- Used when accounts are connected via custom OAuth clients
+- Credentials are retrievable and used directly by sources
+- Standard authentication flow
+
+#### Proxy Mode (Default OAuth Clients)
+- Required when using Pipedream's default OAuth client
+- Credentials are not exposed for security
+- All API requests route through Pipedream's proxy endpoint
+- Pipedream injects authentication credentials server-side
+
+The system automatically detects which mode to use:
+1. Attempts to retrieve credentials
+2. If unavailable (default OAuth), switches to proxy mode
+3. Sources use `PipedreamProxyClient` transparently
+
+**Benefits:**
+- Supports both custom and default OAuth clients
+- Maintains security for default OAuth credentials
+- Transparent to source implementations (same API)
+- No code changes needed in individual sources
+
+**Technical Details:**
+- Proxy URL: `https://api.pipedream.com/v1/connect/{project_id}/proxy/{encoded_url}`
+- Auth headers stripped and injected by Pipedream
+- Non-auth headers forwarded with `x-pd-proxy-` prefix

.github/workflows/test-public-api.yml

@@ -53,6 +53,11 @@ jobs:
           TEST_COMPOSIO_API_KEY: ${{ secrets.TEST_COMPOSIO_API_KEY }}
           TEST_COMPOSIO_AUTH_CONFIG_ID: ${{ secrets.TEST_COMPOSIO_AUTH_CONFIG_ID }}
           TEST_COMPOSIO_ACCOUNT_ID: ${{ secrets.TEST_COMPOSIO_ACCOUNT_ID }}
+          TEST_PIPEDREAM_CLIENT_ID: ${{ secrets.TEST_PIPEDREAM_CLIENT_ID }}
+          TEST_PIPEDREAM_CLIENT_SECRET: ${{ secrets.TEST_PIPEDREAM_CLIENT_SECRET }}
+          TEST_PIPEDREAM_PROJECT_ID: ${{ secrets.TEST_PIPEDREAM_PROJECT_ID }}
+          TEST_PIPEDREAM_ACCOUNT_ID: ${{ secrets.TEST_PIPEDREAM_ACCOUNT_ID }}
+          TEST_PIPEDREAM_EXTERNAL_USER_ID: ${{ secrets.TEST_PIPEDREAM_EXTERNAL_USER_ID }}
           # Override images to use locally built test images
           BACKEND_IMAGE: test-backend:latest
           FRONTEND_IMAGE: test-frontend:latest

backend/airweave/platform/auth_providers/_base.py

@@ -4,6 +4,7 @@
 from typing import Any, Dict, List, Optional
 
 from airweave.core.logging import logger
+from airweave.platform.auth_providers.auth_result import AuthResult
 
 
 class BaseAuthProvider(ABC):
@@ -65,3 +66,22 @@ async def validate(self) -> bool:
             HTTPException: If validation fails with detailed error message
         """
         pass
+
+    async def get_auth_result(
+        self, source_short_name: str, source_auth_config_fields: List[str]
+    ) -> AuthResult:
+        """Get auth result with explicit mode (direct vs proxy).
+
+        Default implementation calls get_creds_for_source and returns direct mode.
+        Subclasses can override to return proxy mode when needed.
+
+        Args:
+            source_short_name: The short name of the source
+            source_auth_config_fields: The fields required for the source auth config
+
+        Returns:
+            AuthResult with explicit mode and credentials/config
+        """
+        # Default: try to get credentials directly
+        credentials = await self.get_creds_for_source(source_short_name, source_auth_config_fields)
+        return AuthResult.direct(credentials)

backend/airweave/platform/auth_providers/auth_result.py

@@ -0,0 +1,39 @@
+"""Auth result types for explicit auth mode communication."""
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any, Dict, Optional
+
+
+class AuthProviderMode(Enum):
+    """Explicit auth mode enumeration."""
+
+    DIRECT = "direct"  # Use credentials directly
+    PROXY = "proxy"  # Route through proxy (e.g., Pipedream)
+
+
+@dataclass
+class AuthResult:
+    """Result of auth provider credential fetch.
+
+    This makes explicit whether to use credentials directly or via proxy.
+    """
+
+    mode: AuthProviderMode
+    credentials: Optional[Any] = None  # Actual credentials (if DIRECT mode)
+    proxy_config: Optional[Dict[str, Any]] = None  # Config for proxy (if PROXY mode)
+
+    @classmethod
+    def direct(cls, credentials: Any) -> "AuthResult":
+        """Create a direct auth result with credentials."""
+        return cls(mode=AuthProviderMode.DIRECT, credentials=credentials)
+
+    @classmethod
+    def proxy(cls, config: Optional[Dict[str, Any]] = None) -> "AuthResult":
+        """Create a proxy auth result."""
+        return cls(mode=AuthProviderMode.PROXY, proxy_config=config or {})
+
+    @property
+    def requires_proxy(self) -> bool:
+        """Check if proxy is required."""
+        return self.mode == AuthProviderMode.PROXY

backend/airweave/platform/auth_providers/pipedream.py

@@ -8,9 +8,30 @@
 
 from airweave.core.credential_sanitizer import safe_log_credentials
 from airweave.platform.auth_providers._base import BaseAuthProvider
+from airweave.platform.auth_providers.auth_result import AuthResult
 from airweave.platform.decorators import auth_provider
 
 
+class PipedreamDefaultOAuthException(Exception):
+    """Raised when trying to access credentials for a Pipedream default OAuth client.
+
+    This happens when the connected account uses Pipedream's built-in OAuth client
+    rather than a custom OAuth client. In this case, credentials cannot be retrieved
+    directly and the proxy must be used.
+    """
+
+    def __init__(self, source_short_name: str, message: str = None):
+        """Initialize the exception."""
+        self.source_short_name = source_short_name
+        if message is None:
+            message = (
+                f"Cannot retrieve credentials for {source_short_name}. "
+                "This account uses Pipedream's default OAuth client. "
+                "Proxy mode must be used for this connection."
+            )
+        super().__init__(message)
+
+
 @auth_provider(
     name="Pipedream",
     short_name="pipedream",
@@ -98,10 +119,10 @@ async def create(
         instance = cls()
         instance.client_id = credentials["client_id"]
         instance.client_secret = credentials["client_secret"]
-        instance.project_id = config.get("project_id")
-        instance.account_id = config.get("account_id")
+        instance.project_id = config["project_id"]
+        instance.account_id = config["account_id"]
+        instance.external_user_id = config["external_user_id"]
         instance.environment = config.get("environment", "production")
-        instance.external_user_id = config.get("external_user_id")
 
         # Initialize token management
         instance._access_token = None
@@ -369,11 +390,7 @@ async def _get_account_with_credentials(
                     "❌ [Pipedream] No credentials in response. This usually means the account "
                     "was created with Pipedream's default OAuth client, not a custom one."
                 )
-                raise HTTPException(
-                    status_code=422,
-                    detail="Credentials not available. Pipedream only exposes credentials for "
-                    "accounts created with custom OAuth clients, not default Pipedream OAuth.",
-                )
+                raise PipedreamDefaultOAuthException(source_short_name)
 
             self.logger.info(
                 f"✅ [Pipedream] Found account '{account_data.get('name')}' "
@@ -464,3 +481,43 @@ def _extract_and_map_credentials(
         )
 
         return found_credentials
+
+    async def get_auth_result(
+        self, source_short_name: str, source_auth_config_fields: List[str]
+    ) -> AuthResult:
+        """Get auth result with explicit mode for Pipedream.
+
+        Determines whether to use direct credentials or proxy based on OAuth client type.
+        """
+        # Check if source is in blocked list (must use proxy)
+        if source_short_name in self.BLOCKED_SOURCES:
+            self.logger.info(f"Source {source_short_name} is in blocked list - using proxy mode")
+            return AuthResult.proxy(
+                {
+                    "reason": "blocked_source",
+                    "source": source_short_name,
+                }
+            )
+
+        # Try to get credentials to determine OAuth client type
+        try:
+            credentials = await self.get_creds_for_source(
+                source_short_name, source_auth_config_fields
+            )
+            # Custom OAuth client - can use direct access
+            self.logger.info(
+                f"Custom OAuth client detected for {source_short_name} - using direct mode"
+            )
+            return AuthResult.direct(credentials)
+
+        except PipedreamDefaultOAuthException:
+            # Default OAuth client - must use proxy
+            self.logger.info(
+                f"Default OAuth client detected for {source_short_name} - using proxy mode"
+            )
+            return AuthResult.proxy(
+                {
+                    "reason": "default_oauth",
+                    "source": source_short_name,
+                }
+            )

backend/airweave/platform/configs/config.py

@@ -1,7 +1,5 @@
 """Configuration classes for platform components."""
 
-from typing import Optional
-
 from pydantic import Field, validator
 
 from airweave.platform.configs._base import BaseConfig
@@ -304,13 +302,12 @@ class PipedreamConfig(AuthProviderConfig):
         title="Account ID",
         description="Pipedream account ID (e.g., apn_gyha5Ky)",
     )
+    external_user_id: str = Field(
+        title="External User ID",
+        description="External user ID associated with the account",
+    )
     environment: str = Field(
         default="production",
         title="Environment",
         description="Pipedream environment (production or development)",
     )
-    external_user_id: Optional[str] = Field(
-        default=None,
-        title="External User ID",
-        description="External user ID associated with the account",
-    )

backend/airweave/platform/http_client/__init__.py

@@ -0,0 +1,5 @@
+"""HTTP client implementations for Airweave platform."""
+
+from .pipedream_proxy import PipedreamProxyClient
+
+__all__ = ["PipedreamProxyClient"]