Add presigned URL APIs

README.md

@@ -168,6 +168,17 @@ ssh = sandbox.ssh.create_access()
 print(ssh.hostname, ssh.port, ssh.username)
 ```
 
+### Presigned URLs
+
+Create a temporary public URL for a sandbox port.
+
+```python
+presigned = sandbox.create_presigned_url(port=8080, expires_in=900)
+print(presigned.url)
+
+sandbox.delete_presigned_url(presigned.id)
+```
+
 ### Desktop Automation
 
 Control a graphical desktop inside the sandbox. Take screenshots, move the pointer, click, type, and record the screen.

examples/ssh.py

@@ -17,10 +17,12 @@ def main() -> None:
         print("ssh command:", access.ssh_command)
 
         validation: SshValidation = sandbox.ssh.validate_access(
-            access_id=access.id,
+            id=access.id,
             password=access.password,
         )
         print("ssh valid:", validation.valid)
+        rotated: SshAccess = sandbox.ssh.regenerate_access(id=access.id)
+        print("rotated ssh command:", rotated.ssh_command)
     finally:
         sandbox.delete()
         client.close()

leap0/__init__.py

@@ -75,8 +75,10 @@
     from .models.process import ProcessResult
     from .models.pty import CreatePtySessionParams, PtyConnection, PtySession
     from .models.sandbox import (
+        CreatePresignedURLParams,
         CreateSandboxParams,
         NetworkPolicyMode,
+        PresignedURL,
         SandboxListItem,
         SandboxListResponse,
         SandboxState,
@@ -144,6 +146,7 @@
     "CodeExecutionResult",
     "CodeInterpreterClient",
     "CreatePtySessionParams",
+    "CreatePresignedURLParams",
     "CreateSandboxParams",
     "CreateSnapshotParams",
     "CreateTemplateParams",
@@ -191,6 +194,7 @@
     "NetworkPolicyMode",
     "ProcessClient",
     "ProcessResult",
+    "PresignedURL",
     "PtyClient",
     "PtyConnection",
     "PtySession",
@@ -266,6 +270,7 @@
     "Leap0TimeoutError": (".models.errors", "Leap0TimeoutError"),
     "Leap0WebSocketError": (".models.errors", "Leap0WebSocketError"),
     "CreateSandboxParams": (".models.sandbox", "CreateSandboxParams"),
+    "CreatePresignedURLParams": (".models.sandbox", "CreatePresignedURLParams"),
     "NetworkPolicyMode": (".models.sandbox", "NetworkPolicyMode"),
     "SandboxListItem": (".models.sandbox", "SandboxListItem"),
     "SandboxListResponse": (".models.sandbox", "SandboxListResponse"),
@@ -288,6 +293,7 @@
     "GitCommitResult": (".models.git", "GitCommitResult"),
     "GitResult": (".models.git", "GitResult"),
     "ProcessResult": (".models.process", "ProcessResult"),
+    "PresignedURL": (".models.sandbox", "PresignedURL"),
     "CreatePtySessionParams": (".models.pty", "CreatePtySessionParams"),
     "PtyConnection": (".models.pty", "PtyConnection"),
     "PtySession": (".models.pty", "PtySession"),

leap0/_async/sandbox.py

@@ -14,14 +14,16 @@
     DEFAULT_VCPU,
 )
 from ..models.sandbox import (
+    CreatePresignedURLParams,
     CreateSandboxParams,
+    PresignedURL,
     Sandbox as SandboxData,
     SandboxListResponse,
     SandboxRef,
     SandboxStatus,
     sandbox_id_of,
 )
-from .._schemas.sandbox import ListSandboxesResponseDict, NetworkPolicyDict, SandboxCreateResponseDict, SandboxStatusResponseDict
+from .._schemas.sandbox import ListSandboxesResponseDict, NetworkPolicyDict, PresignedURLResponseDict, SandboxCreateResponseDict, SandboxStatusResponseDict
 from .._utils.errors import intercept_errors
 from .._utils.url import ensure_leading_slash, sandbox_base_url, websocket_url_from_http
 from ._transport import AsyncTransport
@@ -130,6 +132,25 @@ async def delete(self, http_timeout: float | None = None) -> None:
         """
         await self._client.sandboxes.delete(self, http_timeout=http_timeout)
 
+    async def create_presigned_url(
+        self,
+        *,
+        port: int,
+        expires_in: int | None = None,
+        http_timeout: float | None = None,
+    ) -> PresignedURL:
+        """Create a temporary public URL for a specific sandbox port."""
+        return await self._client.sandboxes.create_presigned_url(
+            self,
+            port=port,
+            expires_in=expires_in,
+            http_timeout=http_timeout,
+        )
+
+    async def delete_presigned_url(self, presigned_url_id: str, http_timeout: float | None = None) -> None:
+        """Delete a previously issued presigned URL."""
+        await self._client.sandboxes.delete_presigned_url(self, presigned_url_id, http_timeout=http_timeout)
+
     def invoke_url(self, path: str = "/", *, port: int | None = None) -> str:
         """Build an HTTPS URL for this sandbox.
 
@@ -401,6 +422,37 @@ async def get_workdir(self, sandbox: SandboxRef, http_timeout: float | None = No
             raise ValueError("Sandbox workdir response missing 'workdir'")
         return value
 
+    @intercept_errors("Failed to create presigned URL: ")
+    async def create_presigned_url(
+        self,
+        sandbox: SandboxRef,
+        *,
+        port: int,
+        expires_in: int | None = None,
+        http_timeout: float | None = None,
+    ) -> PresignedURL:
+        params = CreatePresignedURLParams(port=port, expires_in=expires_in)
+        data: PresignedURLResponseDict = await self._transport.request_json(
+            "POST",
+            f"/v1/sandbox/{sandbox_id_of(sandbox)}/presigned-url",
+            json=params.to_payload(),
+            expected_status=201,
+            timeout=http_timeout,
+        )
+        return PresignedURL.from_dict(data)
+
+    @intercept_errors("Failed to delete presigned URL: ")
+    async def delete_presigned_url(self, sandbox: SandboxRef, presigned_url_id: str, http_timeout: float | None = None) -> None:
+        id_value = presigned_url_id.strip()
+        if not id_value:
+            raise ValueError("presigned_url_id must be a non-empty string")
+        await self._transport.request(
+            "DELETE",
+            f"/v1/sandbox/{sandbox_id_of(sandbox)}/presigned-url/{id_value}",
+            expected_status=204,
+            timeout=http_timeout,
+        )
+
 
     def invoke_url(self, sandbox: SandboxRef, path: str = "/", *, port: int | None = None) -> str:
         """Build an HTTPS URL for this sandbox.

leap0/_async/ssh.py

@@ -11,9 +11,8 @@
 class AsyncSshClient:
     """Manage SSH access credentials for a sandbox.
 
-        Each sandbox supports a single set of SSH credentials at a time. Creating
-        access when credentials already exist returns 409 Conflict. Use
-        :meth:`regenerate_access` to rotate credentials without revoking first.
+        Each sandbox supports up to 10 active SSH credentials at a time. Target a
+        specific credential ID when validating, deleting, or regenerating credentials.
 
         Example:
             ```python
@@ -47,22 +46,23 @@ async def create_access(self, sandbox: SandboxRef, http_timeout: float | None =
         return SshAccess.from_dict(cast(dict, data))
 
     @intercept_errors("Failed to delete SSH access: ")
-    async def delete_access(self, sandbox: SandboxRef, http_timeout: float | None = None) -> None:
-        """Revoke SSH access for a sandbox. The credential is invalidated immediately.
+    async def delete_access(self, sandbox: SandboxRef, *, id: str, http_timeout: float | None = None) -> None:
+        """Revoke a specific SSH access credential. The credential is invalidated immediately.
 
         Args:
             sandbox: Sandbox ID or object.
+            id: SSH credential ID.
             http_timeout: Optional HTTP request timeout in seconds for this SDK call.
         """
-        await self._transport.request("DELETE", f"/v1/sandbox/{sandbox_id_of(sandbox)}/ssh/access", expected_status=204, timeout=http_timeout)
+        await self._transport.request("DELETE", f"/v1/sandbox/{sandbox_id_of(sandbox)}/ssh/{id}", expected_status=204, timeout=http_timeout)
 
     @intercept_errors("Failed to validate SSH access: ")
-    async def validate_access(self, sandbox: SandboxRef, *, access_id: str, password: str, http_timeout: float | None = None) -> SshValidation:
-        """Check whether an SSH access credential is still valid and not expired.
+    async def validate_access(self, sandbox: SandboxRef, *, id: str, password: str, http_timeout: float | None = None) -> SshValidation:
+        """Check whether a specific SSH access credential is still valid and not expired.
 
         Args:
             sandbox: Sandbox ID or object.
-            access_id: SSH access identifier.
+            id: SSH credential ID.
             password: SSH password.
             http_timeout: Optional HTTP request timeout in seconds for this SDK call.
 
@@ -71,22 +71,23 @@ async def validate_access(self, sandbox: SandboxRef, *, access_id: str, password
         """
         data = await self._transport.request_json(
             "POST",
-            f"/v1/sandbox/{sandbox_id_of(sandbox)}/ssh/validate",
-            json={"id": access_id, "password": password},
+            f"/v1/sandbox/{sandbox_id_of(sandbox)}/ssh/{id}/validate",
+            json={"password": password},
             timeout=http_timeout,
         )
         return SshValidation.from_dict(cast(dict, data))
 
     @intercept_errors("Failed to regenerate SSH access: ")
-    async def regenerate_access(self, sandbox: SandboxRef, http_timeout: float | None = None) -> SshAccess:
-        """Invalidate the current credential and generate a new one. The expiry is also reset.
-
-                Args:
-                    sandbox: Sandbox ID or object.
-                    http_timeout: Optional HTTP request timeout in seconds for this SDK call.
-
+    async def regenerate_access(self, sandbox: SandboxRef, *, id: str, http_timeout: float | None = None) -> SshAccess:
+        """Invalidate a specific credential and generate a new one. The expiry is also reset.
+
+        Args:
+            sandbox: Sandbox ID or object.
+            id: SSH credential ID.
+            http_timeout: Optional HTTP request timeout in seconds for this SDK call.
+
         Returns:
-            object: Result returned by this operation.
+            SshAccess: Newly generated SSH credential bundle.
         """
-        data = await self._transport.request_json("POST", f"/v1/sandbox/{sandbox_id_of(sandbox)}/ssh/regen", timeout=http_timeout)
+        data = await self._transport.request_json("POST", f"/v1/sandbox/{sandbox_id_of(sandbox)}/ssh/{id}/regen", timeout=http_timeout)
         return SshAccess.from_dict(cast(dict, data))

leap0/_schemas/sandbox.py

@@ -56,3 +56,22 @@ class ListSandboxesResponseDict(TypedDict):
     """Wire schema for paginated sandbox list responses."""
     items: list[SandboxListItemResponseDict]
     total_items: int
+
+
+class CreatePresignedURLRequestDict(TypedDict, total=False):
+    """Wire schema for presigned URL creation requests."""
+
+    port: Required[int]
+    expires_in: int
+
+
+class PresignedURLResponseDict(TypedDict):
+    """Wire schema for presigned URL responses."""
+
+    id: str
+    token: str
+    url: str
+    sandbox_id: str
+    port: int
+    expires_at: str
+    created_at: str

leap0/_sync/sandbox.py

@@ -13,14 +13,16 @@
     DEFAULT_VCPU,
 )
 from ..models.sandbox import (
+    CreatePresignedURLParams,
     CreateSandboxParams,
+    PresignedURL,
     Sandbox as SandboxData,
     SandboxListResponse,
     SandboxRef,
     SandboxStatus,
     sandbox_id_of,
 )
-from .._schemas.sandbox import ListSandboxesResponseDict, NetworkPolicyDict, SandboxCreateResponseDict, SandboxStatusResponseDict
+from .._schemas.sandbox import ListSandboxesResponseDict, NetworkPolicyDict, PresignedURLResponseDict, SandboxCreateResponseDict, SandboxStatusResponseDict
 from .._utils.errors import intercept_errors
 from .._utils.url import ensure_leading_slash, sandbox_base_url, websocket_url_from_http
 from ._transport import Transport
@@ -127,6 +129,25 @@ def delete(self, http_timeout: float | None = None) -> None:
         """
         self._client.sandboxes.delete(self, http_timeout=http_timeout)
 
+    def create_presigned_url(
+        self,
+        *,
+        port: int,
+        expires_in: int | None = None,
+        http_timeout: float | None = None,
+    ) -> PresignedURL:
+        """Create a temporary public URL for a specific sandbox port."""
+        return self._client.sandboxes.create_presigned_url(
+            self,
+            port=port,
+            expires_in=expires_in,
+            http_timeout=http_timeout,
+        )
+
+    def delete_presigned_url(self, presigned_url_id: str, http_timeout: float | None = None) -> None:
+        """Delete a previously issued presigned URL."""
+        self._client.sandboxes.delete_presigned_url(self, presigned_url_id, http_timeout=http_timeout)
+
     def invoke_url(self, path: str = "/", *, port: int | None = None) -> str:
         """Build an HTTPS URL that routes directly to this sandbox.
 
@@ -400,6 +421,37 @@ def get_workdir(self, sandbox: SandboxRef, http_timeout: float | None = None) ->
             raise ValueError("Sandbox workdir response missing 'workdir'")
         return value
 
+    @intercept_errors("Failed to create presigned URL: ")
+    def create_presigned_url(
+        self,
+        sandbox: SandboxRef,
+        *,
+        port: int,
+        expires_in: int | None = None,
+        http_timeout: float | None = None,
+    ) -> PresignedURL:
+        params = CreatePresignedURLParams(port=port, expires_in=expires_in)
+        data: PresignedURLResponseDict = self._transport.request_json(
+            "POST",
+            f"/v1/sandbox/{sandbox_id_of(sandbox)}/presigned-url",
+            json=params.to_payload(),
+            expected_status=201,
+            timeout=http_timeout,
+        )
+        return PresignedURL.from_dict(data)
+
+    @intercept_errors("Failed to delete presigned URL: ")
+    def delete_presigned_url(self, sandbox: SandboxRef, presigned_url_id: str, http_timeout: float | None = None) -> None:
+        id_value = presigned_url_id.strip()
+        if not id_value:
+            raise ValueError("presigned_url_id must be a non-empty string")
+        self._transport.request(
+            "DELETE",
+            f"/v1/sandbox/{sandbox_id_of(sandbox)}/presigned-url/{id_value}",
+            expected_status=204,
+            timeout=http_timeout,
+        )
+
     def invoke_url(self, sandbox: SandboxRef, path: str = "/", *, port: int | None = None) -> str:
         """Build an HTTPS URL that routes directly to the sandbox.