diff --git a/README.md b/README.md index 73074d3..793c129 100644 --- a/README.md +++ b/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. diff --git a/examples/ssh.py b/examples/ssh.py index 4f34c46..7f368e8 100644 --- a/examples/ssh.py +++ b/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() diff --git a/leap0/__init__.py b/leap0/__init__.py index 504e4d8..1fa700d 100644 --- a/leap0/__init__.py +++ b/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"), diff --git a/leap0/_async/sandbox.py b/leap0/_async/sandbox.py index f9ebb1f..8e35406 100644 --- a/leap0/_async/sandbox.py +++ b/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. diff --git a/leap0/_async/ssh.py b/leap0/_async/ssh.py index eed8f5e..0c7c023 100644 --- a/leap0/_async/ssh.py +++ b/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)) diff --git a/leap0/_schemas/sandbox.py b/leap0/_schemas/sandbox.py index 976c2f7..cf59be4 100644 --- a/leap0/_schemas/sandbox.py +++ b/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 diff --git a/leap0/_sync/sandbox.py b/leap0/_sync/sandbox.py index e1d20ce..11978c3 100644 --- a/leap0/_sync/sandbox.py +++ b/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. diff --git a/leap0/_sync/ssh.py b/leap0/_sync/ssh.py index 53c007c..578bb5e 100644 --- a/leap0/_sync/ssh.py +++ b/leap0/_sync/ssh.py @@ -11,9 +11,8 @@ class SshClient: """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 @@ def create_access(self, sandbox: SandboxRef, http_timeout: float | None = None) return SshAccess.from_dict(cast(dict, data)) @intercept_errors("Failed to delete SSH access: ") - def delete_access(self, sandbox: SandboxRef, http_timeout: float | None = None) -> None: - """Revoke SSH access for a sandbox. The credential is invalidated immediately. + 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. """ - self._transport.request("DELETE", f"/v1/sandbox/{sandbox_id_of(sandbox)}/ssh/access", expected_status=204, timeout=http_timeout) + 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: ") - 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. + 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 @@ def validate_access(self, sandbox: SandboxRef, *, access_id: str, password: str, """ data = 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: ") - 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. + 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: SshAccess: Newly generated SSH credential bundle. """ - data = self._transport.request_json("POST", f"/v1/sandbox/{sandbox_id_of(sandbox)}/ssh/regen", timeout=http_timeout) + data = 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)) diff --git a/leap0/models/sandbox.py b/leap0/models/sandbox.py index 8afcfe6..527eb41 100644 --- a/leap0/models/sandbox.py +++ b/leap0/models/sandbox.py @@ -12,8 +12,10 @@ from .._internal.types import SandboxHandle from .._internal.types import JsonObject from .._schemas.sandbox import ( + CreatePresignedURLRequestDict, ListSandboxesResponseDict, NetworkPolicyDict, + PresignedURLResponseDict, SandboxCreateResponseDict, SandboxListItemResponseDict, SandboxStatusResponseDict, @@ -141,6 +143,29 @@ def to_payload(self) -> JsonObject: CreateSandboxParams.model_rebuild(_types_namespace={"NetworkPolicyMode": NetworkPolicyMode}) + +class CreatePresignedURLParams(BaseModel): + """Validated presigned URL creation parameters.""" + + model_config = ConfigDict(extra="forbid") + + port: int + expires_in: int | None = None + + @model_validator(mode="after") + def _validate_values(self) -> CreatePresignedURLParams: + if not 1 <= self.port <= 65535: + raise ValueError("port must be between 1 and 65535") + if self.expires_in is not None and self.expires_in < 1: + raise ValueError("expires_in must be at least 1") + return self + + def to_payload(self) -> CreatePresignedURLRequestDict: + return self.model_dump(exclude_none=True) + + +CreatePresignedURLParams.model_rebuild() + @dataclass(slots=True) class Sandbox(SandboxHandle): """Sandbox model returned by sandbox creation APIs.""" @@ -248,6 +273,57 @@ def from_dict(cls, data: ListSandboxesResponseDict) -> SandboxListResponse: total_items=int(data.get("total_items", 0)), ) + +@dataclass(slots=True) +class PresignedURL: + """Presigned URL response returned by sandbox sharing APIs.""" + + id: str + token: str + url: str + sandbox_id: str + port: int + expires_at: str + created_at: str + + def __repr__(self) -> str: + return ( + "PresignedURL(" + f"id={self.id!r}, " + "token='', " + "url='', " + f"sandbox_id={self.sandbox_id!r}, " + f"port={self.port!r}, " + f"expires_at={self.expires_at!r}, " + f"created_at={self.created_at!r}" + ")" + ) + + @classmethod + def from_dict(cls, data: PresignedURLResponseDict) -> PresignedURL: + presigned_id = data.get("id") + token = data.get("token") + url = data.get("url") + sandbox_id = data.get("sandbox_id") + if not all(isinstance(value, str) and value.strip() for value in (presigned_id, token, url, sandbox_id)): + raise ValueError("PresignedURL response missing required non-empty string fields") + port = data.get("port") + if not isinstance(port, int) or not 1 <= port <= 65535: + raise ValueError(f"PresignedURL response has invalid 'port', got: {port!r}") + expires_at = data.get("expires_at") + created_at = data.get("created_at") + if not all(isinstance(value, str) and value.strip() for value in (expires_at, created_at)): + raise ValueError("PresignedURL response missing required non-empty timestamp fields") + return cls( + id=presigned_id, + token=token, + url=url, + sandbox_id=sandbox_id, + port=port, + expires_at=expires_at, + created_at=created_at, + ) + SandboxRef: TypeAlias = str | SandboxHandle def _parse_sandbox_state(value: SandboxState | str | None) -> SandboxState | str: diff --git a/tests/_async/test_sandboxes.py b/tests/_async/test_sandboxes.py index 856d93d..eb6677d 100644 --- a/tests/_async/test_sandboxes.py +++ b/tests/_async/test_sandboxes.py @@ -160,6 +160,42 @@ async def run() -> None: asyncio.run(run()) + def test_create_presigned_url(self, async_mock_transport): + async def run() -> None: + async_mock_transport.request_json.return_value = { + "id": "psu-1", + "token": "tok_1", + "url": "https://tok_1.leap0.app", + "sandbox_id": "sbx-1", + "port": 8080, + "expires_at": "2026-01-01T00:15:00Z", + "created_at": "2026-01-01T00:00:00Z", + } + + result = await AsyncSandboxesClient(async_mock_transport, sandbox_domain="s.dev").create_presigned_url( + "sbx-1", + port=8080, + expires_in=900, + ) + + args, kwargs = async_mock_transport.request_json.call_args + assert args == ("POST", "/v1/sandbox/sbx-1/presigned-url") + assert kwargs["json"] == {"port": 8080, "expires_in": 900} + assert kwargs["expected_status"] == 201 + assert result.token == "tok_1" + + asyncio.run(run()) + + def test_delete_presigned_url(self, async_mock_transport): + async def run() -> None: + await AsyncSandboxesClient(async_mock_transport, sandbox_domain="s.dev").delete_presigned_url("sbx-1", "psu-1") + + args, kwargs = async_mock_transport.request.call_args + assert args == ("DELETE", "/v1/sandbox/sbx-1/presigned-url/psu-1") + assert kwargs["expected_status"] == 204 + + asyncio.run(run()) + class TestAsyncSandbox: @@ -210,3 +246,29 @@ async def get_workdir(sandbox: object, http_timeout: float | None = None): assert await sandbox.get_workdir(http_timeout=2.5) == "/home/steve/agent" asyncio.run(run()) + + def test_presigned_url_helpers_delegate_to_sandboxes_client(self): + async def run() -> None: + sandboxes = SimpleNamespace() + fake_client = SimpleNamespace( + _filesystem=SimpleNamespace(), _git=SimpleNamespace(), _process=SimpleNamespace(), _pty=SimpleNamespace(), + _lsp=SimpleNamespace(), _ssh=SimpleNamespace(), _code_interpreter=SimpleNamespace(), _desktop=SimpleNamespace(), + sandboxes=sandboxes, + ) + + async def create_presigned_url(sandbox: object, **kwargs): + assert kwargs == {"port": 8080, "expires_in": 900, "http_timeout": 2.5} + return None + + async def delete_presigned_url(sandbox: object, presigned_url_id: str, http_timeout: float | None = None): + assert presigned_url_id == "psu-1" + assert http_timeout == 3.5 + + sandboxes.create_presigned_url = create_presigned_url + sandboxes.delete_presigned_url = delete_presigned_url + sandbox = AsyncSandbox(fake_client, Sandbox(id="sbx-1", state="running")) + + await sandbox.create_presigned_url(port=8080, expires_in=900, http_timeout=2.5) + await sandbox.delete_presigned_url("psu-1", http_timeout=3.5) + + asyncio.run(run()) diff --git a/tests/_async/test_ssh.py b/tests/_async/test_ssh.py index 53db3f7..0e22ace 100644 --- a/tests/_async/test_ssh.py +++ b/tests/_async/test_ssh.py @@ -20,7 +20,7 @@ async def run() -> None: def test_delete_access(self, async_mock_transport): async def run() -> None: async_mock_transport.request.return_value = MagicMock(status_code=204) - await AsyncSshClient(async_mock_transport).delete_access("sbx-1") - assert async_mock_transport.request.call_args[0][:2] == ("DELETE", "/v1/sandbox/sbx-1/ssh/access") + await AsyncSshClient(async_mock_transport).delete_access("sbx-1", id="ssh-1") + assert async_mock_transport.request.call_args[0][:2] == ("DELETE", "/v1/sandbox/sbx-1/ssh/ssh-1") asyncio.run(run()) diff --git a/tests/_sync/test_sandboxes.py b/tests/_sync/test_sandboxes.py index 7f43a7a..f550ea9 100644 --- a/tests/_sync/test_sandboxes.py +++ b/tests/_sync/test_sandboxes.py @@ -91,6 +91,38 @@ def test_get_workdir(self, mock_transport): assert result == "/home/steve/agent" assert mock_transport.request_json.call_args[0][1] == "/v1/sandbox/sbx-1/system/workdir" + def test_create_presigned_url(self, mock_transport): + mock_transport.request_json.return_value = { + "id": "psu-1", + "token": "tok_1", + "url": "https://tok_1.leap0.app", + "sandbox_id": "sbx-1", + "port": 8080, + "expires_at": "2026-01-01T00:15:00Z", + "created_at": "2026-01-01T00:00:00Z", + } + + result = SandboxesClient(mock_transport, sandbox_domain="s.dev").create_presigned_url( + "sbx-1", + port=8080, + expires_in=900, + ) + + args, kwargs = mock_transport.request_json.call_args + assert args == ("POST", "/v1/sandbox/sbx-1/presigned-url") + assert kwargs["json"] == {"port": 8080, "expires_in": 900} + assert kwargs["expected_status"] == 201 + assert result.token == "tok_1" + + def test_delete_presigned_url(self, mock_transport): + mock_transport.request.return_value = MagicMock(status_code=204) + + SandboxesClient(mock_transport, sandbox_domain="s.dev").delete_presigned_url("sbx-1", "psu-1") + + args, kwargs = mock_transport.request.call_args + assert args == ("DELETE", "/v1/sandbox/sbx-1/presigned-url/psu-1") + assert kwargs["expected_status"] == 204 + def test_accepts_sandbox_object(self, mock_transport): mock_transport.request_json.return_value = { @@ -258,6 +290,32 @@ def test_pause_forwards_http_timeout(self): sandboxes.pause.assert_called_once_with(sandbox, http_timeout=7.5) assert sandbox.state == "paused" + def test_presigned_url_helpers_delegate_to_sandboxes_client(self): + sandboxes = MagicMock() + client = SimpleNamespace( + _filesystem=MagicMock(), + _git=MagicMock(), + _process=MagicMock(), + _pty=MagicMock(), + _lsp=MagicMock(), + _ssh=MagicMock(), + _code_interpreter=MagicMock(), + _desktop=MagicMock(), + sandboxes=sandboxes, + ) + sandbox = RichSandbox(client, Sandbox(id="sbx-1", state="running")) + + sandbox.create_presigned_url(port=8080, expires_in=900, http_timeout=2.5) + sandbox.delete_presigned_url("psu-1", http_timeout=3.5) + + sandboxes.create_presigned_url.assert_called_once_with( + sandbox, + port=8080, + expires_in=900, + http_timeout=2.5, + ) + sandboxes.delete_presigned_url.assert_called_once_with(sandbox, "psu-1", http_timeout=3.5) + def test_runtime_info_helpers_delegate_to_sandboxes_client(self): sandboxes = MagicMock() sandboxes.get_user_home_dir.return_value = "/home/steven" diff --git a/tests/_sync/test_ssh.py b/tests/_sync/test_ssh.py index 05e50be..48797ee 100644 --- a/tests/_sync/test_ssh.py +++ b/tests/_sync/test_ssh.py @@ -15,5 +15,5 @@ def test_create_access(self, mock_transport): def test_delete_access(self, mock_transport): mock_transport.request.return_value = MagicMock(status_code=204) - SshClient(mock_transport).delete_access("sbx-1") - assert mock_transport.request.call_args[0][:2] == ("DELETE", "/v1/sandbox/sbx-1/ssh/access") + SshClient(mock_transport).delete_access("sbx-1", id="ssh-1") + assert mock_transport.request.call_args[0][:2] == ("DELETE", "/v1/sandbox/sbx-1/ssh/ssh-1") diff --git a/tests/models/test_sandbox.py b/tests/models/test_sandbox.py index ef61435..01e4002 100644 --- a/tests/models/test_sandbox.py +++ b/tests/models/test_sandbox.py @@ -1,8 +1,9 @@ from __future__ import annotations import pytest +from pydantic import ValidationError -from leap0.models.sandbox import CreateSandboxParams, Sandbox, SandboxStatus, _validate_network_policy, sandbox_id_of +from leap0.models.sandbox import CreatePresignedURLParams, CreateSandboxParams, PresignedURL, Sandbox, SandboxStatus, _validate_network_policy, sandbox_id_of class TestSandboxIdOf: @@ -73,3 +74,68 @@ def test_rejects_invalid_network_policy(self): with pytest.raises(ValueError, match=r"transforms\[0\] must be a mapping"): _validate_network_policy({"mode": "custom", "transforms": ["bad"]}) + + +class TestCreatePresignedURLParams: + def test_rejects_invalid_values(self): + with pytest.raises(ValidationError, match="port"): + CreatePresignedURLParams(port=0) + + with pytest.raises(ValidationError, match="expires_in"): + CreatePresignedURLParams(port=8080, expires_in=0) + + +class TestPresignedURL: + def test_from_dict(self): + result = PresignedURL.from_dict({ + "id": "psu-1", + "token": "tok_1", + "url": "https://tok_1.leap0.app", + "sandbox_id": "sbx_1", + "port": 8080, + "expires_at": "2026-01-01T00:15:00Z", + "created_at": "2026-01-01T00:00:00Z", + }) + + assert result.id == "psu-1" + assert result.url == "https://tok_1.leap0.app" + assert result.port == 8080 + + def test_from_dict_rejects_missing_timestamps(self): + with pytest.raises(ValueError, match="timestamp"): + PresignedURL.from_dict({ + "id": "psu-1", + "token": "tok_1", + "url": "https://tok_1.leap0.app", + "sandbox_id": "sbx_1", + "port": 8080, + "expires_at": "", + }) + + def test_from_dict_rejects_invalid_port(self): + with pytest.raises(ValueError, match=r"invalid 'port'.*0"): + PresignedURL.from_dict({ + "id": "psu-1", + "token": "tok_1", + "url": "https://tok_1.leap0.app", + "sandbox_id": "sbx_1", + "port": 0, + "expires_at": "2026-01-01T00:15:00Z", + "created_at": "2026-01-01T00:00:00Z", + }) + + def test_repr_redacts_sensitive_fields(self): + result = PresignedURL.from_dict({ + "id": "psu-1", + "token": "tok_1", + "url": "https://tok_1.leap0.app", + "sandbox_id": "sbx_1", + "port": 8080, + "expires_at": "2026-01-01T00:15:00Z", + "created_at": "2026-01-01T00:00:00Z", + }) + + rendered = repr(result) + assert "tok_1" not in rendered + assert "https://tok_1.leap0.app" not in rendered + assert "" in rendered