fix(SKY-11917): validate script deploy file paths (#7145)

skyvern/cli/mcp_tools/scripts.py

@@ -15,6 +15,7 @@ from pydantic import Field, ValidationError
 
 from skyvern.client.errors import NotFoundError
 from skyvern.client.types import ScriptFileCreate
+from skyvern.utils.script_file_paths import normalize_script_file_path
 
 from ._common import ErrorCode, Timer, make_error, make_result, raw_http_get
 from ._session import get_skyvern
@@ -247,8 +248,11 @@ async def skyvern_script_deploy(
         parsed_files = json.loads(files)
         if not isinstance(parsed_files, list):
             raise ValueError("files must be a JSON array")
-        typed_files = [ScriptFileCreate(**file_data) for file_data in parsed_files]
-    except (json.JSONDecodeError, TypeError, ValueError, ValidationError) as e:
+        typed_files = [
+            ScriptFileCreate(**{**file_data, "path": normalize_script_file_path(file_data["path"])})
+            for file_data in parsed_files
+        ]
+    except (json.JSONDecodeError, TypeError, ValueError, ValidationError, KeyError) as e:
         return make_result(
             "skyvern_script_deploy",
             ok=False,

skyvern/forge/sdk/artifact/storage/azure.py

@@ -33,6 +33,7 @@ from skyvern.forge.sdk.schemas.ai_suggestions import AISuggestion
 from skyvern.forge.sdk.schemas.files import FileInfo
 from skyvern.forge.sdk.schemas.task_v2 import TaskV2, Thought
 from skyvern.forge.sdk.schemas.workflow_runs import WorkflowRunBlock
+from skyvern.utils.script_file_paths import build_script_file_storage_uri
 
 LOG = structlog.get_logger()
 
@@ -118,7 +119,12 @@ class AzureStorage(BaseStorage):
         self, *, organization_id: str, script_id: str, script_version: int, file_path: str
     ) -> str:
         """Build the Azure URI for a script file."""
-        return f"{self._build_base_uri(organization_id)}/scripts/{script_id}/{script_version}/{file_path}"
+        return build_script_file_storage_uri(
+            self._build_base_uri(organization_id),
+            script_id=script_id,
+            script_version=script_version,
+            file_path=file_path,
+        )
 
     async def store_artifact(self, artifact: Artifact, data: bytes) -> None:
         tier = await self._get_storage_tier_for_org(artifact.organization_id)

skyvern/forge/sdk/artifact/storage/gcs.py

@@ -38,6 +38,7 @@ from skyvern.forge.sdk.schemas.ai_suggestions import AISuggestion
 from skyvern.forge.sdk.schemas.files import FileInfo
 from skyvern.forge.sdk.schemas.task_v2 import TaskV2, Thought
 from skyvern.forge.sdk.schemas.workflow_runs import WorkflowRunBlock
+from skyvern.utils.script_file_paths import build_script_file_storage_uri
 
 LOG = structlog.get_logger()
 
@@ -122,7 +123,12 @@ class GcsStorage(BaseStorage):
         self, *, organization_id: str, script_id: str, script_version: int, file_path: str
     ) -> str:
         """Build the GCS URI for a script file."""
-        return f"{self._build_base_uri(organization_id)}/scripts/{script_id}/{script_version}/{file_path}"
+        return build_script_file_storage_uri(
+            self._build_base_uri(organization_id),
+            script_id=script_id,
+            script_version=script_version,
+            file_path=file_path,
+        )
 
     async def store_artifact(self, artifact: Artifact, data: bytes) -> None:
         storage_class = await self._get_storage_class_for_org(artifact.organization_id)

skyvern/forge/sdk/artifact/storage/local.py

@@ -22,6 +22,7 @@ from skyvern.forge.sdk.schemas.ai_suggestions import AISuggestion
 from skyvern.forge.sdk.schemas.files import FileInfo
 from skyvern.forge.sdk.schemas.task_v2 import TaskV2, Thought
 from skyvern.forge.sdk.schemas.workflow_runs import WorkflowRunBlock
+from skyvern.utils.script_file_paths import build_script_file_storage_uri
 from skyvern.webeye.session_cookies import SESSION_COOKIES_FILENAME
 
 LOG = structlog.get_logger()
@@ -122,7 +123,12 @@ class LocalStorage(BaseStorage):
     def build_script_file_uri(
         self, *, organization_id: str, script_id: str, script_version: int, file_path: str
     ) -> str:
-        return f"file://{self.artifact_path}/{settings.ENV}/{organization_id}/scripts/{script_id}/{script_version}/{file_path}"
+        return build_script_file_storage_uri(
+            f"file://{self.artifact_path}/{settings.ENV}/{organization_id}",
+            script_id=script_id,
+            script_version=script_version,
+            file_path=file_path,
+        )
 
     async def store_artifact(self, artifact: Artifact, data: bytes) -> None:
         file_path = None

skyvern/forge/sdk/artifact/storage/s3.py

@@ -41,6 +41,7 @@ from skyvern.forge.sdk.schemas.ai_suggestions import AISuggestion
 from skyvern.forge.sdk.schemas.files import FileInfo
 from skyvern.forge.sdk.schemas.task_v2 import TaskV2, Thought
 from skyvern.forge.sdk.schemas.workflow_runs import WorkflowRunBlock
+from skyvern.utils.script_file_paths import build_script_file_storage_uri
 from skyvern.webeye.video_utils import prepare_recording_for_upload
 
 LOG = structlog.get_logger()
@@ -147,7 +148,12 @@ class S3Storage(BaseStorage):
         Returns:
             The S3 URI for the script file
         """
-        return f"{self._build_base_uri(organization_id)}/scripts/{script_id}/{script_version}/{file_path}"
+        return build_script_file_storage_uri(
+            self._build_base_uri(organization_id),
+            script_id=script_id,
+            script_version=script_version,
+            file_path=file_path,
+        )
 
     async def store_artifact(self, artifact: Artifact, data: bytes) -> None:
         # We compress HAR files with zstd level 3 to reduce storage size.

skyvern/schemas/scripts.py

@@ -5,7 +5,9 @@ from datetime import datetime
 from enum import StrEnum
 from typing import Literal
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from skyvern.utils.script_file_paths import normalize_script_file_path
 
 
 class FileEncoding(StrEnum):
@@ -50,6 +52,11 @@ class ScriptFileCreate(BaseModel):
     encoding: FileEncoding = Field(default=FileEncoding.BASE64, description="Content encoding")
     mime_type: str | None = Field(default=None, description="MIME type (auto-detected if not provided)")
 
+    @field_validator("path")
+    @classmethod
+    def validate_path(cls, value: str) -> str:
+        return normalize_script_file_path(value)
+
 
 class CreateScriptRequest(BaseModel):
     workflow_id: str | None = Field(default=None, description="Associated workflow ID")

skyvern/services/cached_script_deploy_service.py

@@ -26,6 +26,7 @@ from skyvern.schemas.scripts import (
     WorkflowScript,
 )
 from skyvern.services.workflow_script_service import CacheKeyResolutionError, resolve_cache_key_value
+from skyvern.utils.script_file_paths import SCRIPT_FILE_PATH_ERROR, normalize_script_file_path
 
 _CODE_VERSION_STATIC = 1
 _CODE_VERSION_ADAPTIVE = 2
@@ -87,12 +88,13 @@ def _decode_script_file_bytes(file: ScriptFileCreate) -> bytes:
 
 
 def _validate_script_file_path(file_path: str) -> None:
-    parts = file_path.split("/")
-    if file_path.startswith("/") or "\\" in file_path or any(part in ("", ".", "..") for part in parts):
+    try:
+        normalize_script_file_path(file_path)
+    except ValueError as exc:
         raise HTTPException(
             status_code=400,
-            detail=f"File path {file_path!r} must be a relative POSIX path without empty, '.', or '..' segments",
-        )
+            detail=f"File path {file_path!r} is invalid: {SCRIPT_FILE_PATH_ERROR}",
+        ) from exc
 
 
 def _validate_script_files(files: list[ScriptFileCreate]) -> list[_ValidatedScriptFile]:

skyvern/utils/script_file_paths.py

@@ -0,0 +1,49 @@
+from pathlib import PurePosixPath, PureWindowsPath
+from urllib.parse import unquote
+
+SCRIPT_FILE_PATH_ERROR = (
+    "Script file path must be a relative POSIX path without empty, '.', '..', absolute, "
+    "drive-qualified, backslash, or null-byte segments"
+)
+_MAX_SCRIPT_FILE_PATH_DECODE_ITERATIONS = 16
+
+
+def normalize_script_file_path(file_path: str) -> str:
+    normalized = _decode_storage_path(file_path)
+    if (
+        not normalized
+        or "\x00" in normalized
+        or "\\" in normalized
+        or normalized.startswith(("/", "\\"))
+        or PurePosixPath(normalized).is_absolute()
+        or PureWindowsPath(normalized).drive
+    ):
+        raise ValueError(SCRIPT_FILE_PATH_ERROR)
+
+    parts = normalized.split("/")
+    if any(part in ("", ".", "..") for part in parts):
+        raise ValueError(SCRIPT_FILE_PATH_ERROR)
+    return "/".join(parts)
+
+
+def build_script_file_storage_uri(
+    storage_base_uri: str,
+    *,
+    script_id: str,
+    script_version: int,
+    file_path: str,
+) -> str:
+    normalized_file_path = normalize_script_file_path(file_path)
+    return f"{storage_base_uri.rstrip('/')}/scripts/{script_id}/{script_version}/{normalized_file_path}"
+
+
+def _decode_storage_path(file_path: str) -> str:
+    decoded = file_path
+    for _ in range(_MAX_SCRIPT_FILE_PATH_DECODE_ITERATIONS):
+        next_decoded = unquote(decoded)
+        if next_decoded == decoded:
+            return decoded
+        decoded = next_decoded
+    if unquote(decoded) != decoded:
+        raise ValueError(SCRIPT_FILE_PATH_ERROR)
+    return decoded