block urls pointing to internal addresses (#1012)

skyvern/config.py

@@ -34,6 +34,7 @@ class Settings(BaseSettings):
     LOG_LEVEL: str = "INFO"
     PORT: int = 8000
     ALLOWED_ORIGINS: list[str] = ["*"]
+    BLOCKED_HOSTS: list[str] = ["localhost"]
 
     # Secret key for JWT. Please generate your own secret key in production
     SECRET_KEY: str = "PLACEHOLDER"

skyvern/exceptions.py

@@ -502,3 +502,11 @@ class InvalidUrl(SkyvernHTTPException):
         super().__init__(
             f"Invalid URL: {url}. Skyvern supports HTTP and HTTPS urls.", status_code=status.HTTP_400_BAD_REQUEST
         )
+
+
+class BlockedHost(SkyvernHTTPException):
+    def __init__(self, host: str) -> None:
+        super().__init__(
+            f"The host in your url is blocked: {host}",
+            status_code=status.HTTP_400_BAD_REQUEST,
+        )

skyvern/forge/sdk/core/validators.py

@@ -1,5 +1,8 @@
+import ipaddress
+
 from pydantic import HttpUrl, ValidationError, parse_obj_as
 
+from skyvern.config import settings
 from skyvern.exceptions import InvalidUrl
 
 
@@ -11,3 +14,18 @@ def validate_url(url: str) -> str:
     except ValidationError:
         # Handle the validation error
         raise InvalidUrl(url=url)
+
+
+def is_blocked_host(host: str) -> bool:
+    try:
+        ip = ipaddress.ip_address(host)
+        # Check if the IP is private, link-local, loopback, or reserved
+        return ip.is_private or ip.is_link_local or ip.is_loopback or ip.is_reserved
+    except ValueError:
+        # If the host is not a valid IP address (e.g., it's a domain name like localhost), handle it here
+        for blocked_host in settings.BLOCKED_HOSTS:
+            if blocked_host == host:
+                return True
+        return False
+    except Exception:
+        return False

skyvern/forge/sdk/schemas/tasks.py

@@ -4,9 +4,10 @@ from datetime import datetime
 from enum import StrEnum
 from typing import Any
 
-from pydantic import BaseModel, Field, HttpUrl
+from pydantic import BaseModel, Field, HttpUrl, field_validator
 
-from skyvern.exceptions import InvalidTaskStatusTransition, TaskAlreadyCanceled
+from skyvern.exceptions import BlockedHost, InvalidTaskStatusTransition, TaskAlreadyCanceled
+from skyvern.forge.sdk.core.validators import is_blocked_host
 
 
 class ProxyLocation(StrEnum):
@@ -89,6 +90,17 @@ class TaskRequest(TaskBase):
     )
     totp_verification_url: HttpUrl | None = None
 
+    @field_validator("url", "webhook_callback_url", "totp_verification_url")
+    @classmethod
+    def validate_urls(cls, v: HttpUrl | None) -> HttpUrl | None:
+        if not v or not v.host:
+            return None
+        host = v.host
+        blocked = is_blocked_host(host)
+        if blocked:
+            raise BlockedHost(host=host)
+        return v
+
 
 class TaskStatus(StrEnum):
     created = "created"