SDK: file uploading (public url only) (#3867)

skyvern/forge/sdk/api/files.py

@@ -69,6 +69,56 @@ def is_valid_mime_type(file_path: str) -> bool:
     return mime_type is not None
 
 
+def validate_download_url(url: str) -> bool:
+    """Validate if a URL is supported for downloading.
+
+    Security validation for URL downloads to prevent:
+    - File system access outside allowed directories
+    - Access to local file system in non-local environments
+    - Unsupported or dangerous URL schemes
+
+    Args:
+        url: The URL to validate
+
+    Returns:
+        True if valid, False otherwise.
+    """
+    try:
+        parsed_url = urlparse(url)
+        scheme = parsed_url.scheme.lower()
+
+        # Allow http/https URLs (includes Google Drive which uses https)
+        if scheme in ("http", "https"):
+            return True
+
+        # Allow S3 URIs for Skyvern uploads bucket
+        if scheme == "s3":
+            if url.startswith(f"s3://{settings.AWS_S3_BUCKET_UPLOADS}/{settings.ENV}/o_"):
+                return True
+            return False
+
+        # Allow file:// URLs only in local environment
+        if scheme == "file":
+            if settings.ENV != "local":
+                return False
+
+            # Validate the file path is within allowed directories
+            try:
+                file_path = parse_uri_to_path(url)
+                allowed_prefix = f"{REPO_ROOT_DIR}/downloads"
+                if not file_path.startswith(allowed_prefix):
+                    return False
+                return True
+            except ValueError:
+                return False
+
+        # Reject unsupported schemes
+        return False
+
+    except Exception:
+        return False
+
+
 async def download_file(url: str, max_size_mb: int | None = None) -> str:
     try:
         # Check if URL is a Google Drive link

skyvern/forge/sdk/routes/sdk.py

@@ -6,6 +6,7 @@ from fastapi import Depends, HTTPException, status
 from skyvern.core.script_generations.real_skyvern_page_ai import RealSkyvernPageAi
 from skyvern.core.script_generations.script_skyvern_page import ScriptSkyvernPage
 from skyvern.forge import app
+from skyvern.forge.sdk.api.files import validate_download_url
 from skyvern.forge.sdk.core import skyvern_context
 from skyvern.forge.sdk.core.skyvern_context import SkyvernContext
 from skyvern.forge.sdk.routes.routers import base_router
@@ -95,8 +96,8 @@ async def run_sdk_action(
     task = await app.DATABASE.create_task(
         organization_id=organization_id,
         url=action_request.url,
-        navigation_goal=action.intention,
-        navigation_payload=action.data,
+        navigation_goal=action.get_navigation_goal(),
+        navigation_payload=action.get_navigation_payload(),
         data_extraction_goal=None,
         title=f"SDK Action Task: {action_request.action.type}",
         workflow_run_id=workflow_run.workflow_run_id,
@@ -174,6 +175,16 @@ async def run_sdk_action(
                 data=action.data,
                 timeout=action.timeout,
             )
+        elif action.type == "ai_upload_file":
+            if action.file_url and not validate_download_url(action.file_url):
+                raise HTTPException(status_code=400, detail="Unsupported file url")
+            result = await page_ai.ai_upload_file(
+                selector=action.selector,
+                files=action.file_url,
+                intention=action.intention,
+                data=action.data,
+                timeout=action.timeout,
+            )
         elif action.type == "ai_act":
             await page_ai.ai_act(
                 prompt=action.intention,

skyvern/forge/sdk/schemas/sdk_actions.py

@@ -12,6 +12,7 @@ class SdkActionType(str, Enum):
     AI_CLICK = "ai_click"
     AI_INPUT_TEXT = "ai_input_text"
     AI_SELECT_OPTION = "ai_select_option"
+    AI_UPLOAD_FILE = "ai_upload_file"
     AI_ACT = "ai_act"
     EXTRACT = "extract"
 
@@ -22,6 +23,12 @@ class SdkActionBase(BaseModel):
 
     type: str = Field(..., description="The type of action")
 
+    def get_navigation_goal(self) -> str | None:
+        return None
+
+    def get_navigation_payload(self) -> dict[str, Any] | None:
+        return None
+
 
 # Specific action types
 class ClickAction(SdkActionBase):
@@ -33,6 +40,12 @@ class ClickAction(SdkActionBase):
     data: str | dict[str, Any] | None = Field(None, description="Additional context data")
     timeout: float = Field(default=settings.BROWSER_ACTION_TIMEOUT_MS, description="Timeout in milliseconds")
 
+    def get_navigation_goal(self) -> str | None:
+        return self.intention
+
+    def get_navigation_payload(self) -> dict[str, Any] | None:
+        return self.data if isinstance(self.data, dict) else None
+
 
 class InputTextAction(SdkActionBase):
     """Input text action parameters."""
@@ -46,6 +59,12 @@ class InputTextAction(SdkActionBase):
     totp_url: str | None = Field(None, description="TOTP URL for input_text actions")
     timeout: float = Field(default=settings.BROWSER_ACTION_TIMEOUT_MS, description="Timeout in milliseconds")
 
+    def get_navigation_goal(self) -> str | None:
+        return self.intention
+
+    def get_navigation_payload(self) -> dict[str, Any] | None:
+        return self.data if isinstance(self.data, dict) else None
+
 
 class SelectOptionAction(SdkActionBase):
     """Select option action parameters."""
@@ -57,6 +76,35 @@ class SelectOptionAction(SdkActionBase):
     data: str | dict[str, Any] | None = Field(None, description="Additional context data")
     timeout: float = Field(default=settings.BROWSER_ACTION_TIMEOUT_MS, description="Timeout in milliseconds")
 
+    def get_navigation_goal(self) -> str | None:
+        return self.intention
+
+    def get_navigation_payload(self) -> dict[str, Any] | None:
+        return self.data if isinstance(self.data, dict) else None
+
+
+class UploadFileAction(SdkActionBase):
+    """Upload file action parameters."""
+
+    type: Literal["ai_upload_file"] = "ai_upload_file"
+    selector: str | None = Field(default="", description="CSS selector for the element")
+    file_url: str | None = Field(default="", description="File URL for upload")
+    intention: str = Field(default="", description="The intention or goal of the upload")
+    data: str | dict[str, Any] | None = Field(None, description="Additional context data")
+    timeout: float = Field(default=settings.BROWSER_ACTION_TIMEOUT_MS, description="Timeout in milliseconds")
+
+    def get_navigation_goal(self) -> str | None:
+        return self.intention
+
+    def get_navigation_payload(self) -> dict[str, Any] | None:
+        if self.data and not isinstance(self.data, dict):
+            return None
+
+        data = self.data or {}
+        if "files" not in data:
+            data["files"] = self.file_url
+        return data
+
 
 class ActAction(SdkActionBase):
     """AI act action parameters."""
@@ -65,6 +113,12 @@ class ActAction(SdkActionBase):
     intention: str = Field(default="", description="Natural language prompt for the action")
     data: str | dict[str, Any] | None = Field(None, description="Additional context data")
 
+    def get_navigation_goal(self) -> str | None:
+        return self.intention
+
+    def get_navigation_payload(self) -> dict[str, Any] | None:
+        return self.data if isinstance(self.data, dict) else None
+
 
 class ExtractAction(SdkActionBase):
     """Extract data action parameters."""
@@ -76,10 +130,16 @@ class ExtractAction(SdkActionBase):
     intention: str | None = Field(None, description="The intention or goal of the extraction")
     data: str | dict[str, Any] | None = Field(None, description="Additional context data")
 
+    def get_navigation_goal(self) -> str | None:
+        return self.intention
+
+    def get_navigation_payload(self) -> dict[str, Any] | None:
+        return self.data if isinstance(self.data, dict) else None
+
 
 # Discriminated union of all action types
 SdkAction = Annotated[
-    Union[ClickAction, InputTextAction, SelectOptionAction, ActAction, ExtractAction],
+    Union[ClickAction, InputTextAction, SelectOptionAction, UploadFileAction, ActAction, ExtractAction],
     Field(discriminator="type"),
 ]