fix: harden JSON execution boundaries

README.md

@@ -84,6 +84,18 @@ make setup
 make test
 ```
 
+## Remote values
+
+Queue and task calls use finite JSON-compatible arguments, return values,
+and stream chunks. JSON conversion turns tuples into lists and supported
+dictionary keys into strings, including in same-worker remote calls.
+Bytes, custom objects, circular references, and non-finite numbers produce
+explicit errors. `.local()` remains a direct Python call.
+
+Successful function responses contain `json_result`, including for a null
+return value. SDK and worker runtimes share the JSON protocol validators;
+use compatible package releases when building worker images.
+
 ## ⚡ | Serverless Worker (SDK)
 
 This python package can also be used to create a serverless worker that can be deployed to Runpod as a custom endpoint API.

requirements.txt

@@ -5,7 +5,6 @@ backoff >= 2.2.1
 boto3 >= 1.43.118
 certifi >= 2026.10.19
 click >= 8.5.0
-cloudpickle >= 3.1.2
 colorama >= 0.4.6, < 0.4.7
 cryptography >= 50.0.2
 fastapi[all] >= 0.162.0

runpod/_serialization.py

@@ -0,0 +1,21 @@
+"""finite json validation and wire-value conversion for callers and workers."""
+
+import json
+from typing import Any
+
+
+def _json_text(value: Any, label: str) -> str:
+    try:
+        return json.dumps(value, allow_nan=False)
+    except (TypeError, ValueError, RecursionError) as exc:
+        raise TypeError(f"{label} must be json-serializable: {exc}") from exc
+
+
+def validate_json(value: Any, label: str = "value") -> None:
+    """require values encodable as finite json."""
+    _json_text(value, label)
+
+
+def normalize_json(value: Any, label: str = "value") -> Any:
+    """apply json's container and key conversions without a network roundtrip."""
+    return json.loads(_json_text(value, label))

runpod/apps/build.py

@@ -50,7 +50,6 @@
     "aiohttp[speedups]",
     "aiohttp-retry",
     "backoff",
-    "cloudpickle",
     "py-cpuinfo",
     "requests",
     "tomli",

runpod/apps/handles.py

@@ -20,7 +20,6 @@ def load_model(): ...                    # worker startup hook
 
 import asyncio
 import inspect
-import json
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -32,14 +31,14 @@ def load_model(): ...                    # worker startup hook
     Type,
 )
 
+from .._serialization import normalize_json
 from .context import Context, block, current_context
 from .discovery_state import DiscoveryInvocationError, in_discovery
 from .errors import InvalidResourceError
 from .invoker import Invoker, StreamInvoker
 from .job import Job
 from .markers import INIT_ATTR, is_init, route_of
 from .schedule import SCHEDULE_ATTR
-from .serialization import require_json_value
 from .spec import ResourceKind, ResourceSpec, RouteSpec
 
 if TYPE_CHECKING:
@@ -126,10 +125,12 @@ def _call_local_sync(self, *args: Any, **kwargs: Any) -> Any:
 
     @staticmethod
     def _json_value(value: Any) -> Any:
-        require_json_value(
-            value, "remote arguments and results must be JSON-serializable"
-        )
-        return json.loads(json.dumps(value, allow_nan=False))
+        try:
+            return normalize_json(value)
+        except TypeError:
+            raise ValueError(
+                "remote arguments and results must be JSON-serializable"
+            ) from None
 
     @staticmethod
     async def _await_local_result(result: Any) -> Any:

runpod/apps/images.py

@@ -38,11 +38,9 @@ def runtime_tag() -> str:
 def local_python_version() -> str:
     """the local interpreter's version, which the worker must match.
 
-    dev sessions and tasks ship cloudpickle payloads produced on the
-    client interpreter; those are not reliably portable across python
-    versions, so a silent fallback would surface as cryptic
-    deserialization failures on the worker. mismatches fail here,
-    loudly, instead.
+    live execution ships python source from the client. matching the
+    interpreter keeps syntax and language behavior consistent between
+    local development and the worker.
     """
     version = f"{sys.version_info.major}.{sys.version_info.minor}"
     if version not in SUPPORTED_PYTHON_VERSIONS:

runpod/apps/protocol.py

@@ -1,21 +1,20 @@
 """wire protocol for remote function execution.
 
 this is the contract between the sdk and the worker runtime images.
-the worker repo imports these shapes from the runpod package so client
-and worker can never drift.
+the worker runtime and client share these shapes and validation rules.
 
 live mode ships function source with every request; deployed mode omits
 the code (the worker unpacked the build and resolves the function by
-name). args/kwargs are base64 cloudpickle unless serialization_format
-is "json".
+name). arguments, results, and stream chunks are json values.
 """
 
-from dataclasses import asdict, dataclass, field
+from dataclasses import asdict, dataclass, field, fields
 from typing import Any, Dict, List, Optional
 
+from .._serialization import normalize_json, validate_json
+
 EXECUTION_FUNCTION = "function"
 
-FORMAT_CLOUDPICKLE = "cloudpickle"
 FORMAT_JSON = "json"
 
 
@@ -31,10 +30,30 @@ class FunctionRequest:
     system_dependencies: Optional[List[str]] = None
     execution_type: str = EXECUTION_FUNCTION
     accelerate_downloads: bool = True
-    serialization_format: str = FORMAT_CLOUDPICKLE
+    serialization_format: str = FORMAT_JSON
+
+    def validate(self) -> None:
+        if self.serialization_format != FORMAT_JSON:
+            raise ValueError("serialization_format must be 'json'")
+        if not isinstance(self.args, list) or not isinstance(self.kwargs, dict):
+            raise TypeError("args must be a list and kwargs must be a dictionary")
+        validate_json(
+            {f.name: getattr(self, f.name) for f in fields(self)}, "function request"
+        )
+
+    @classmethod
+    def from_input(cls, data: Dict[str, Any]) -> "FunctionRequest":
+        if not isinstance(data, dict):
+            raise TypeError("function request must be a dictionary")
+        request = cls(**{f.name: data[f.name] for f in fields(cls) if f.name in data})
+        request.validate()
+        request.args = normalize_json(request.args, "arguments")
+        request.kwargs = normalize_json(request.kwargs, "keyword arguments")
+        return request
 
     def to_input(self) -> Dict[str, Any]:
         """serialize as the job input dict."""
+        self.validate()
         data = asdict(self)
         return {k: v for k, v in data.items() if v is not None}
 
@@ -44,17 +63,24 @@ class FunctionResponse:
     """the worker's result for one function execution."""
 
     success: bool
-    result: Optional[str] = None
     json_result: Any = None
     error: Optional[str] = None
     stdout: Optional[str] = None
 
     @classmethod
     def from_output(cls, output: Dict[str, Any]) -> "FunctionResponse":
+        if not isinstance(output, dict) or not isinstance(output.get("success"), bool):
+            raise TypeError("function response must contain a boolean success field")
+        if output.get("serialization_format", FORMAT_JSON) != FORMAT_JSON:
+            raise ValueError("serialization_format must be 'json'")
+        if output.get("result") is not None:
+            raise ValueError("function responses must use json_result")
+        if output["success"] and "json_result" not in output:
+            raise ValueError("successful function response is missing json_result")
+        validate_json(output, "function response")
         return cls(
-            success=output.get("success", False),
-            result=output.get("result"),
-            json_result=output.get("json_result"),
+            success=output["success"],
+            json_result=normalize_json(output.get("json_result"), "return value"),
             error=output.get("error"),
             stdout=output.get("stdout"),
         )

runpod/apps/serialization.py

@@ -1,52 +1,37 @@
 """argument and result serialization for remote execution.
 
 remote calls use finite json-compatible arguments and results. this
-module validates that contract, provides cloudpickle codecs for explicitly
-encoded values, and extracts function source for live workers.
+module validates that contract, decodes json response envelopes, and extracts
+function source for live workers.
 """
 
 import ast
-import base64
 import inspect
 import os
 import textwrap
-from typing import Any, Callable, Dict, List
+from typing import Any, Callable
 
-import cloudpickle
+from .._serialization import validate_json
+from .errors import RemoteExecutionError
+from .protocol import FunctionResponse
 
 
-def serialize_arg(arg: Any) -> str:
-    return base64.b64encode(cloudpickle.dumps(arg)).decode("utf-8")
-
-
-def serialize_args(args: tuple) -> List[str]:
-    return [serialize_arg(a) for a in args]
-
-
-def serialize_kwargs(kwargs: dict) -> Dict[str, str]:
-    return {k: serialize_arg(v) for k, v in kwargs.items()}
-
-
-def deserialize_result(result_b64: str) -> Any:
+def unwrap_function_response(output: dict) -> Any:
     try:
-        return cloudpickle.loads(base64.b64decode(result_b64))
-    except ModuleNotFoundError as exc:
-        from .errors import RemoteExecutionError
-
+        response = FunctionResponse.from_output(output)
+    except (TypeError, ValueError) as exc:
+        raise RemoteExecutionError(f"invalid remote response: {exc}") from exc
+    if not response.success:
         raise RemoteExecutionError(
-            f"the remote result contains an object from the "
-            f"'{exc.name}' package, which is not installed locally. "
-            f"return plain python types (e.g. str(...)) or install "
-            f"'{exc.name}' on this machine."
-        ) from exc
+            f"remote execution failed: {response.error or 'unknown'}"
+        )
+    return response.json_result
 
 
 def require_json_value(value: Any, message: str) -> None:
     try:
-        import json
-
-        json.dumps(value, allow_nan=False)
-    except (TypeError, ValueError):
+        validate_json(value)
+    except TypeError:
         raise ValueError(message) from None
 
 

runpod/apps/targets.py

@@ -15,7 +15,7 @@
 
 from ..user_agent import USER_AGENT
 from .errors import EndpointNotFound, RemoteExecutionError
-from .protocol import FORMAT_JSON, FunctionRequest, FunctionResponse
+from .protocol import FORMAT_JSON, FunctionRequest
 from .serialization import require_json_value
 from .spec import ResourceSpec
 from .utils.events import emit
@@ -696,16 +696,9 @@ def unwrap(self, data: Dict[str, Any]) -> Any:
     def _unwrap_response(self, output: Any) -> Any:
         require_json_value(output, "queue results must be JSON-serializable")
         if isinstance(output, dict) and "success" in output:
-            response = FunctionResponse.from_output(output)
-            if not response.success:
-                raise RemoteExecutionError(
-                    f"remote execution failed: {response.error or 'unknown'}"
-                )
-            if response.result is not None:
-                from .serialization import deserialize_result
+            from .serialization import unwrap_function_response
 
-                return deserialize_result(response.result)
-            return response.json_result
+            return unwrap_function_response(output)
         return output
 
     def _monitor(self):

runpod/apps/tasks.py

@@ -21,7 +21,6 @@
 
 from .api import AppsApiClient, is_capacity_error
 from .context import CLEANUP_TIMEOUT
-from .errors import RemoteExecutionError
 from .spec import ResourceSpec
 
 log = logging.getLogger(__name__)
@@ -264,8 +263,11 @@ async def set_timeout(self, timeout: float) -> None:
         await self._post("/timeout", {"timeout": timeout})
 
     async def _post(self, path: str, request: Dict[str, Any]) -> None:
+        from .serialization import require_json_value
+
         timeout = request.get("timeout")
         _validate_timeout(timeout)
+        require_json_value(request, "task request body must be JSON-serializable")
         url = f"{_proxy_url(self.pod_id)}{path}"
         attempts = 6
         async with aiohttp.ClientSession() as session:
@@ -473,19 +475,10 @@ async def _terminate_pod(self):
 
 def unwrap_task_response(response: Dict[str, Any]) -> Any:
     """extract the function result from a runner response."""
-    if not response.get("success"):
-        raise RemoteExecutionError(
-            f"task execution failed: {response.get('error', 'unknown')}"
-        )
-    if response.get("result") is not None:
-        from .serialization import deserialize_result
-
-        return deserialize_result(response["result"])
-    from .serialization import require_json_value
+    from .serialization import require_json_value, unwrap_function_response
 
-    result = response.get("json_result")
-    require_json_value(result, "task results must be JSON-serializable")
-    return result
+    require_json_value(response, "task results must be JSON-serializable")
+    return unwrap_function_response(response)
 
 
 class TaskJob: