russell@unturf.com
27a0df3d12
Replace manual HMAC authentication with official Unsandbox Python SDK (un.py). Simplifies code execution proxy endpoints by using SDK methods: execute_async, get_job, and cancel_job. Removes ~40 lines of manual HTTP/auth code. Changes: - Add official Unsandbox Python SDK (un.py) - Refactor app.py proxy endpoints to use SDK methods - Update CLAUDE.md documentation with SDK setup and usage - Remove manual HMAC signing code - Maintain backward compatibility with existing API endpoints
2829 lines
91 KiB
Python
2829 lines
91 KiB
Python
"""
|
|
PUBLIC DOMAIN - NO LICENSE, NO WARRANTY
|
|
|
|
unsandbox.com Python SDK (Synchronous)
|
|
|
|
Library Usage:
|
|
from un import (
|
|
# Execution
|
|
execute_code,
|
|
execute_async,
|
|
get_job,
|
|
wait_for_job,
|
|
cancel_job,
|
|
list_jobs,
|
|
get_languages,
|
|
detect_language,
|
|
# Sessions
|
|
list_sessions,
|
|
get_session,
|
|
create_session,
|
|
delete_session,
|
|
freeze_session,
|
|
unfreeze_session,
|
|
boost_session,
|
|
unboost_session,
|
|
shell_session,
|
|
# Services
|
|
list_services,
|
|
create_service,
|
|
get_service,
|
|
update_service,
|
|
delete_service,
|
|
freeze_service,
|
|
unfreeze_service,
|
|
lock_service,
|
|
unlock_service,
|
|
get_service_logs,
|
|
get_service_env,
|
|
set_service_env,
|
|
delete_service_env,
|
|
export_service_env,
|
|
redeploy_service,
|
|
execute_in_service,
|
|
# Snapshots
|
|
session_snapshot,
|
|
service_snapshot,
|
|
list_snapshots,
|
|
restore_snapshot,
|
|
delete_snapshot,
|
|
lock_snapshot,
|
|
unlock_snapshot,
|
|
clone_snapshot,
|
|
# Key validation
|
|
validate_keys,
|
|
# Image generation
|
|
image,
|
|
)
|
|
|
|
# Execute code synchronously
|
|
result = execute_code("python", 'print("hello")', public_key, secret_key)
|
|
|
|
# Execute asynchronously
|
|
job_id = execute_async("javascript", 'console.log("hello")', public_key, secret_key)
|
|
|
|
# Wait for job completion with exponential backoff
|
|
result = wait_for_job(job_id, public_key, secret_key)
|
|
|
|
# List all jobs
|
|
jobs = list_jobs(public_key, secret_key)
|
|
|
|
# Get supported languages
|
|
languages = get_languages(public_key, secret_key)
|
|
|
|
# Detect language from filename
|
|
lang = detect_language("script.py") # Returns "python"
|
|
|
|
# Snapshot operations
|
|
snapshot_id = session_snapshot(session_id, public_key, secret_key, name="my-snapshot")
|
|
snapshot_id = service_snapshot(service_id, public_key, secret_key, name="svc-snapshot")
|
|
snapshots = list_snapshots(public_key, secret_key)
|
|
result = restore_snapshot(snapshot_id, public_key, secret_key)
|
|
delete_snapshot(snapshot_id, public_key, secret_key)
|
|
|
|
Authentication Priority (4-tier):
|
|
1. Function arguments (public_key, secret_key)
|
|
2. Environment variables (UNSANDBOX_PUBLIC_KEY, UNSANDBOX_SECRET_KEY)
|
|
3. Config file (~/.unsandbox/accounts.csv, line 0 by default)
|
|
4. Local directory (./accounts.csv, line 0 by default)
|
|
|
|
Format: public_key,secret_key (one per line)
|
|
Account selection: UNSANDBOX_ACCOUNT=N env var (0-based index)
|
|
|
|
Request Authentication (HMAC-SHA256):
|
|
Authorization: Bearer <public_key> (identifies account)
|
|
X-Timestamp: <unix_seconds> (replay prevention)
|
|
X-Signature: HMAC-SHA256(secret_key, msg) (proves secret + body integrity)
|
|
|
|
Message format: "timestamp:METHOD:path:body"
|
|
- timestamp: seconds since epoch
|
|
- METHOD: GET, POST, DELETE, etc. (uppercase)
|
|
- path: e.g., "/execute", "/jobs/123"
|
|
- body: JSON payload (empty string for GET/DELETE)
|
|
|
|
Languages Cache:
|
|
- Cached in ~/.unsandbox/languages.json
|
|
- TTL: 1 hour
|
|
- Updated on successful API calls
|
|
"""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import time
|
|
import requests
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, Any, List
|
|
|
|
|
|
API_BASE = "https://api.unsandbox.com"
|
|
POLL_DELAYS_MS = [300, 450, 700, 900, 650, 1600, 2000]
|
|
LANGUAGES_CACHE_TTL = 3600 # 1 hour
|
|
|
|
|
|
class CredentialsError(Exception):
|
|
"""Raised when credentials cannot be found or are invalid."""
|
|
pass
|
|
|
|
|
|
def _get_unsandbox_dir() -> Path:
|
|
"""Get ~/.unsandbox directory path, creating if necessary."""
|
|
home = Path.home()
|
|
unsandbox_dir = home / ".unsandbox"
|
|
unsandbox_dir.mkdir(exist_ok=True, mode=0o700)
|
|
return unsandbox_dir
|
|
|
|
|
|
def _load_credentials_from_csv(csv_path: Path, account_index: int = 0) -> Optional[tuple[str, str]]:
|
|
"""Load credentials from CSV file (public_key,secret_key per line)."""
|
|
if not csv_path.exists():
|
|
return None
|
|
|
|
try:
|
|
with open(csv_path, "r") as f:
|
|
for i, line in enumerate(f):
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if i == account_index:
|
|
parts = line.split(",")
|
|
if len(parts) >= 2:
|
|
return (parts[0].strip(), parts[1].strip())
|
|
return None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _resolve_credentials(
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
account_index: Optional[int] = None,
|
|
) -> tuple[str, str]:
|
|
"""
|
|
Resolve credentials from 4-tier priority system.
|
|
|
|
Priority:
|
|
1. Function arguments
|
|
2. Environment variables
|
|
3. ~/.unsandbox/accounts.csv
|
|
4. ./accounts.csv
|
|
"""
|
|
# Tier 1: Function arguments
|
|
if public_key and secret_key:
|
|
return (public_key, secret_key)
|
|
|
|
# Tier 2: Environment variables
|
|
env_pk = os.environ.get("UNSANDBOX_PUBLIC_KEY")
|
|
env_sk = os.environ.get("UNSANDBOX_SECRET_KEY")
|
|
if env_pk and env_sk:
|
|
return (env_pk, env_sk)
|
|
|
|
# Determine account index
|
|
if account_index is None:
|
|
account_index = int(os.environ.get("UNSANDBOX_ACCOUNT", "0"))
|
|
|
|
# Tier 3: ~/.unsandbox/accounts.csv
|
|
unsandbox_dir = _get_unsandbox_dir()
|
|
creds = _load_credentials_from_csv(unsandbox_dir / "accounts.csv", account_index)
|
|
if creds:
|
|
return creds
|
|
|
|
# Tier 4: ./accounts.csv
|
|
creds = _load_credentials_from_csv(Path("accounts.csv"), account_index)
|
|
if creds:
|
|
return creds
|
|
|
|
raise CredentialsError(
|
|
"No credentials found. Please provide via:\n"
|
|
" 1. Function arguments (public_key, secret_key)\n"
|
|
" 2. Environment variables (UNSANDBOX_PUBLIC_KEY, UNSANDBOX_SECRET_KEY)\n"
|
|
" 3. ~/.unsandbox/accounts.csv\n"
|
|
" 4. ./accounts.csv"
|
|
)
|
|
|
|
|
|
def _sign_request(
|
|
secret_key: str,
|
|
timestamp: int,
|
|
method: str,
|
|
path: str,
|
|
body: Optional[str] = None,
|
|
) -> str:
|
|
"""
|
|
Sign a request using HMAC-SHA256.
|
|
|
|
Message format: "timestamp:METHOD:path:body"
|
|
Returns: 64-character hex string
|
|
"""
|
|
body_str = body or ""
|
|
message = f"{timestamp}:{method}:{path}:{body_str}"
|
|
signature = hmac.new(
|
|
secret_key.encode(),
|
|
message.encode(),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
return signature
|
|
|
|
|
|
def _make_request(
|
|
method: str,
|
|
path: str,
|
|
public_key: str,
|
|
secret_key: str,
|
|
data: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Make an authenticated HTTP request to the API.
|
|
|
|
Raises requests.RequestException on network errors.
|
|
Raises ValueError if response is not valid JSON.
|
|
"""
|
|
url = f"{API_BASE}{path}"
|
|
timestamp = int(time.time())
|
|
body = json.dumps(data) if data else ""
|
|
|
|
signature = _sign_request(secret_key, timestamp, method, path, body if data else None)
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {public_key}",
|
|
"X-Timestamp": str(timestamp),
|
|
"X-Signature": signature,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
if method == "GET":
|
|
response = requests.get(url, headers=headers, timeout=120)
|
|
elif method == "POST":
|
|
response = requests.post(url, headers=headers, json=data, timeout=120)
|
|
elif method == "PATCH":
|
|
response = requests.patch(url, headers=headers, json=data, timeout=120)
|
|
elif method == "DELETE":
|
|
response = requests.delete(url, headers=headers, timeout=120)
|
|
else:
|
|
raise ValueError(f"Unsupported HTTP method: {method}")
|
|
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def _get_languages_cache_path() -> Path:
|
|
"""Get path to languages cache file."""
|
|
return _get_unsandbox_dir() / "languages.json"
|
|
|
|
|
|
def _load_languages_cache() -> Optional[List[str]]:
|
|
"""Load languages from cache if valid (< 1 hour old)."""
|
|
cache_path = _get_languages_cache_path()
|
|
if not cache_path.exists():
|
|
return None
|
|
|
|
try:
|
|
with open(cache_path, "r") as f:
|
|
data = json.load(f)
|
|
|
|
# Check if cache is fresh
|
|
mtime = cache_path.stat().st_mtime
|
|
age_seconds = time.time() - mtime
|
|
if age_seconds < LANGUAGES_CACHE_TTL:
|
|
return data.get("languages")
|
|
except Exception:
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
def _save_languages_cache(languages: List[str]) -> None:
|
|
"""Save languages to cache."""
|
|
try:
|
|
cache_path = _get_languages_cache_path()
|
|
with open(cache_path, "w") as f:
|
|
json.dump({"languages": languages, "timestamp": int(time.time())}, f)
|
|
except Exception:
|
|
pass # Cache failures are non-fatal
|
|
|
|
|
|
def execute_code(
|
|
language: str,
|
|
code: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Execute code synchronously (blocks until completion).
|
|
|
|
Args:
|
|
language: Programming language (e.g., "python", "javascript", "go")
|
|
code: Source code to execute
|
|
public_key: Optional API key (uses credentials resolution if not provided)
|
|
secret_key: Optional API secret (uses credentials resolution if not provided)
|
|
|
|
Returns:
|
|
Response dict containing stdout, stderr, exit code, etc.
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
response = _make_request(
|
|
"POST",
|
|
"/execute",
|
|
public_key,
|
|
secret_key,
|
|
{"language": language, "code": code},
|
|
)
|
|
|
|
# If we got a job_id, poll until completion
|
|
job_id = response.get("job_id")
|
|
status = response.get("status")
|
|
|
|
if job_id and status in ("pending", "running"):
|
|
return wait_for_job(job_id, public_key, secret_key)
|
|
|
|
return response
|
|
|
|
|
|
def execute_async(
|
|
language: str,
|
|
code: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> str:
|
|
"""
|
|
Execute code asynchronously (returns immediately with job_id).
|
|
|
|
Args:
|
|
language: Programming language (e.g., "python", "javascript")
|
|
code: Source code to execute
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Job ID string
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
response = _make_request(
|
|
"POST",
|
|
"/execute",
|
|
public_key,
|
|
secret_key,
|
|
{"language": language, "code": code},
|
|
)
|
|
return response.get("job_id")
|
|
|
|
|
|
def get_job(
|
|
job_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Get current status/result of a job (single poll, no waiting).
|
|
|
|
Args:
|
|
job_id: Job ID from execute_async()
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Job response dict
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("GET", f"/jobs/{job_id}", public_key, secret_key)
|
|
|
|
|
|
def wait_for_job(
|
|
job_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
timeout: Optional[float] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Wait for job completion with exponential backoff polling.
|
|
|
|
Polling delays (ms): [300, 450, 700, 900, 650, 1600, 2000, ...]
|
|
Cumulative: 300, 750, 1450, 2350, 3000, 4600, 6600ms+
|
|
|
|
Args:
|
|
job_id: Job ID from execute_async()
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
timeout: Optional maximum wait time in seconds (None = wait indefinitely)
|
|
|
|
Returns:
|
|
Final job result when status is terminal (completed, failed, timeout, cancelled)
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
TimeoutError: If timeout is exceeded before job completes
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
poll_count = 0
|
|
start_time = time.time()
|
|
|
|
while True:
|
|
# Check timeout
|
|
if timeout is not None:
|
|
elapsed = time.time() - start_time
|
|
if elapsed >= timeout:
|
|
raise TimeoutError(f"Job {job_id} did not complete within {timeout} seconds")
|
|
|
|
# Sleep before polling
|
|
delay_idx = min(poll_count, len(POLL_DELAYS_MS) - 1)
|
|
time.sleep(POLL_DELAYS_MS[delay_idx] / 1000.0)
|
|
poll_count += 1
|
|
|
|
response = get_job(job_id, public_key, secret_key)
|
|
status = response.get("status")
|
|
|
|
if status in ("completed", "failed", "timeout", "cancelled"):
|
|
return response
|
|
|
|
# Still running, continue polling
|
|
|
|
|
|
def cancel_job(
|
|
job_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Cancel a running job.
|
|
|
|
Args:
|
|
job_id: Job ID to cancel
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with cancellation confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("DELETE", f"/jobs/{job_id}", public_key, secret_key)
|
|
|
|
|
|
def list_jobs(
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
List all jobs for the authenticated account.
|
|
|
|
Args:
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
List of job dicts
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
response = _make_request("GET", "/jobs", public_key, secret_key)
|
|
return response.get("jobs", [])
|
|
|
|
|
|
def get_languages(
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> List[str]:
|
|
"""
|
|
Get list of supported programming languages.
|
|
|
|
Results are cached for 1 hour in ~/.unsandbox/languages.json
|
|
|
|
Args:
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
List of language identifiers (e.g., ["python", "javascript", "go", ...])
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
# Try cache first
|
|
cached = _load_languages_cache()
|
|
if cached:
|
|
return cached
|
|
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
response = _make_request("GET", "/languages", public_key, secret_key)
|
|
languages = response.get("languages", [])
|
|
|
|
# Cache the result
|
|
_save_languages_cache(languages)
|
|
return languages
|
|
|
|
|
|
# Language detection mapping (file extension -> language)
|
|
_LANGUAGE_MAP = {
|
|
"py": "python",
|
|
"js": "javascript",
|
|
"ts": "typescript",
|
|
"rb": "ruby",
|
|
"php": "php",
|
|
"pl": "perl",
|
|
"sh": "bash",
|
|
"r": "r",
|
|
"R": "r",
|
|
"lua": "lua",
|
|
"go": "go",
|
|
"rs": "rust",
|
|
"c": "c",
|
|
"cpp": "cpp",
|
|
"cc": "cpp",
|
|
"cxx": "cpp",
|
|
"java": "java",
|
|
"kt": "kotlin",
|
|
"m": "objc",
|
|
"cs": "csharp",
|
|
"fs": "fsharp",
|
|
"hs": "haskell",
|
|
"ml": "ocaml",
|
|
"clj": "clojure",
|
|
"scm": "scheme",
|
|
"ss": "scheme",
|
|
"erl": "erlang",
|
|
"ex": "elixir",
|
|
"exs": "elixir",
|
|
"jl": "julia",
|
|
"d": "d",
|
|
"nim": "nim",
|
|
"zig": "zig",
|
|
"v": "v",
|
|
"cr": "crystal",
|
|
"dart": "dart",
|
|
"groovy": "groovy",
|
|
"f90": "fortran",
|
|
"f95": "fortran",
|
|
"lisp": "commonlisp",
|
|
"lsp": "commonlisp",
|
|
"cob": "cobol",
|
|
"tcl": "tcl",
|
|
"raku": "raku",
|
|
"pro": "prolog",
|
|
"p": "prolog",
|
|
"4th": "forth",
|
|
"forth": "forth",
|
|
"fth": "forth",
|
|
}
|
|
|
|
|
|
def detect_language(filename: str) -> Optional[str]:
|
|
"""
|
|
Detect programming language from filename extension.
|
|
|
|
Args:
|
|
filename: Filename to detect language from (e.g., "script.py")
|
|
|
|
Returns:
|
|
Language identifier (e.g., "python") or None if unknown
|
|
|
|
Examples:
|
|
detect_language("hello.py") # -> "python"
|
|
detect_language("script.js") # -> "javascript"
|
|
detect_language("main.go") # -> "go"
|
|
detect_language("unknown") # -> None
|
|
"""
|
|
if not filename or "." not in filename:
|
|
return None
|
|
|
|
ext = filename.rsplit(".", 1)[-1].lower()
|
|
return _LANGUAGE_MAP.get(ext)
|
|
|
|
|
|
def session_snapshot(
|
|
session_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
name: Optional[str] = None,
|
|
ephemeral: bool = False,
|
|
) -> str:
|
|
"""
|
|
Create a snapshot of a session.
|
|
|
|
Args:
|
|
session_id: Session ID to snapshot
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
name: Optional snapshot name
|
|
ephemeral: If True, snapshot is temporary and may be auto-deleted
|
|
|
|
Returns:
|
|
Snapshot ID
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
data = {"session_id": session_id, "ephemeral": ephemeral}
|
|
if name:
|
|
data["name"] = name
|
|
|
|
response = _make_request("POST", "/snapshots", public_key, secret_key, data)
|
|
return response.get("snapshot_id")
|
|
|
|
|
|
def service_snapshot(
|
|
service_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
name: Optional[str] = None,
|
|
) -> str:
|
|
"""
|
|
Create a snapshot of a service.
|
|
|
|
Args:
|
|
service_id: Service ID to snapshot
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
name: Optional snapshot name
|
|
|
|
Returns:
|
|
Snapshot ID
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
data = {"service_id": service_id}
|
|
if name:
|
|
data["name"] = name
|
|
|
|
response = _make_request("POST", "/snapshots", public_key, secret_key, data)
|
|
return response.get("snapshot_id")
|
|
|
|
|
|
def list_snapshots(
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
List all snapshots (NEW).
|
|
|
|
Args:
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
List of snapshot dicts
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
response = _make_request("GET", "/snapshots", public_key, secret_key)
|
|
return response.get("snapshots", [])
|
|
|
|
|
|
def restore_snapshot(
|
|
snapshot_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Restore a snapshot (NEW).
|
|
|
|
Args:
|
|
snapshot_id: Snapshot ID to restore
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with restored resource info
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/snapshots/{snapshot_id}/restore", public_key, secret_key, {})
|
|
|
|
|
|
def delete_snapshot(
|
|
snapshot_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Delete a snapshot (NEW).
|
|
|
|
Args:
|
|
snapshot_id: Snapshot ID to delete
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with deletion confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("DELETE", f"/snapshots/{snapshot_id}", public_key, secret_key)
|
|
|
|
|
|
# =============================================================================
|
|
# Session Management Functions
|
|
# =============================================================================
|
|
|
|
|
|
def list_sessions(
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
List all sessions for the authenticated account.
|
|
|
|
Args:
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
List of session dicts containing id, container_name, status, etc.
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
response = _make_request("GET", "/sessions", public_key, secret_key)
|
|
return response.get("sessions", [])
|
|
|
|
|
|
def get_session(
|
|
session_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Get details of a specific session.
|
|
|
|
Args:
|
|
session_id: Session ID to get details for
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Session details dict
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("GET", f"/sessions/{session_id}", public_key, secret_key)
|
|
|
|
|
|
def create_session(
|
|
language: Optional[str] = None,
|
|
network_mode: str = "zerotrust",
|
|
ttl: int = 3600,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
shell: Optional[str] = None,
|
|
multiplexer: Optional[str] = None,
|
|
vcpu: int = 1,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create a new interactive session.
|
|
|
|
Args:
|
|
language: Optional programming language for the session
|
|
network_mode: Network mode - "zerotrust" (default, no network) or "semitrusted" (with network)
|
|
ttl: Time to live in seconds (default 3600)
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
shell: Optional shell to use (e.g., "bash", "python3")
|
|
multiplexer: Optional terminal multiplexer ("tmux" or "screen")
|
|
vcpu: Number of vCPUs (1-8, default 1)
|
|
|
|
Returns:
|
|
Response dict containing session_id, container_name, etc.
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
data: Dict[str, Any] = {
|
|
"network_mode": network_mode,
|
|
"ttl": ttl,
|
|
}
|
|
if language:
|
|
data["language"] = language
|
|
if shell:
|
|
data["shell"] = shell
|
|
if multiplexer:
|
|
data["multiplexer"] = multiplexer
|
|
if vcpu > 1:
|
|
data["vcpu"] = vcpu
|
|
|
|
return _make_request("POST", "/sessions", public_key, secret_key, data)
|
|
|
|
|
|
def delete_session(
|
|
session_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Delete/terminate a session.
|
|
|
|
Args:
|
|
session_id: Session ID to delete
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with deletion confirmation and optional artifacts
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("DELETE", f"/sessions/{session_id}", public_key, secret_key)
|
|
|
|
|
|
def freeze_session(
|
|
session_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Freeze a session (pause execution, preserve state).
|
|
|
|
Args:
|
|
session_id: Session ID to freeze
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with freeze confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/sessions/{session_id}/freeze", public_key, secret_key, {})
|
|
|
|
|
|
def unfreeze_session(
|
|
session_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Unfreeze a session (resume execution).
|
|
|
|
Args:
|
|
session_id: Session ID to unfreeze
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with unfreeze confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/sessions/{session_id}/unfreeze", public_key, secret_key, {})
|
|
|
|
|
|
def boost_session(
|
|
session_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Boost a session (increase resources).
|
|
|
|
Args:
|
|
session_id: Session ID to boost
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with boost confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/sessions/{session_id}/boost", public_key, secret_key, {})
|
|
|
|
|
|
def unboost_session(
|
|
session_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Unboost a session (return to normal resources).
|
|
|
|
Args:
|
|
session_id: Session ID to unboost
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with unboost confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/sessions/{session_id}/unboost", public_key, secret_key, {})
|
|
|
|
|
|
def shell_session(
|
|
session_id: str,
|
|
command: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Execute a shell command in a session.
|
|
|
|
Note: This is for one-off commands. For interactive shell access,
|
|
use WebSocket connection to /sessions/{id}/shell.
|
|
|
|
Args:
|
|
session_id: Session ID to execute command in
|
|
command: Shell command to execute
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with command output
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request(
|
|
"POST",
|
|
f"/sessions/{session_id}/shell",
|
|
public_key,
|
|
secret_key,
|
|
{"command": command},
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Service Management Functions
|
|
# =============================================================================
|
|
|
|
|
|
def list_services(
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
List all services for the authenticated account.
|
|
|
|
Args:
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
List of service dicts containing id, name, status, ports, etc.
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
response = _make_request("GET", "/services", public_key, secret_key)
|
|
return response.get("services", [])
|
|
|
|
|
|
def create_service(
|
|
name: str,
|
|
ports: List[int],
|
|
bootstrap: Optional[str] = None,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
network_mode: str = "semitrusted",
|
|
custom_domains: Optional[List[str]] = None,
|
|
vcpu: int = 1,
|
|
service_type: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create a new persistent service.
|
|
|
|
Args:
|
|
name: Service name (used for subdomain: name.on.unsandbox.com)
|
|
ports: List of ports to expose (e.g., [80, 443])
|
|
bootstrap: Bootstrap script content, URL, or inline command
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
network_mode: Network mode (default "semitrusted" for services)
|
|
custom_domains: Optional list of custom domain names
|
|
vcpu: Number of vCPUs (1-8, default 1)
|
|
service_type: Optional service type for SRV records (e.g., "minecraft")
|
|
|
|
Returns:
|
|
Response dict containing service_id, etc.
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
data: Dict[str, Any] = {
|
|
"name": name,
|
|
"ports": ports,
|
|
"network_mode": network_mode,
|
|
}
|
|
if bootstrap:
|
|
# Check if it looks like a URL
|
|
if bootstrap.startswith("http://") or bootstrap.startswith("https://"):
|
|
data["bootstrap"] = bootstrap
|
|
else:
|
|
data["bootstrap_content"] = bootstrap
|
|
if custom_domains:
|
|
data["custom_domains"] = custom_domains
|
|
if vcpu > 1:
|
|
data["vcpu"] = vcpu
|
|
if service_type:
|
|
data["service_type"] = service_type
|
|
|
|
return _make_request("POST", "/services", public_key, secret_key, data)
|
|
|
|
|
|
def get_service(
|
|
service_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Get details of a specific service.
|
|
|
|
Args:
|
|
service_id: Service ID to get details for
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Service details dict
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("GET", f"/services/{service_id}", public_key, secret_key)
|
|
|
|
|
|
def update_service(
|
|
service_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
vcpu: Optional[int] = None,
|
|
**kwargs,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Update a service (e.g., resize vCPU/memory).
|
|
|
|
Args:
|
|
service_id: Service ID to update
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
vcpu: Optional new vCPU count (1-8)
|
|
**kwargs: Additional fields to update
|
|
|
|
Returns:
|
|
Response dict with update confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
data: Dict[str, Any] = {}
|
|
if vcpu is not None:
|
|
data["vcpu"] = vcpu
|
|
data.update(kwargs)
|
|
|
|
return _make_request("PATCH", f"/services/{service_id}", public_key, secret_key, data)
|
|
|
|
|
|
def delete_service(
|
|
service_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Delete/destroy a service.
|
|
|
|
Args:
|
|
service_id: Service ID to delete
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with deletion confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("DELETE", f"/services/{service_id}", public_key, secret_key)
|
|
|
|
|
|
def freeze_service(
|
|
service_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Freeze a service (pause execution, preserve state).
|
|
|
|
Args:
|
|
service_id: Service ID to freeze
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with freeze confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/services/{service_id}/freeze", public_key, secret_key, {})
|
|
|
|
|
|
def unfreeze_service(
|
|
service_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Unfreeze a service (resume execution).
|
|
|
|
Args:
|
|
service_id: Service ID to unfreeze
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with unfreeze confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/services/{service_id}/unfreeze", public_key, secret_key, {})
|
|
|
|
|
|
def lock_service(
|
|
service_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Lock a service to prevent accidental deletion.
|
|
|
|
Args:
|
|
service_id: Service ID to lock
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with lock confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/services/{service_id}/lock", public_key, secret_key, {})
|
|
|
|
|
|
def unlock_service(
|
|
service_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Unlock a service to allow deletion.
|
|
|
|
Args:
|
|
service_id: Service ID to unlock
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with unlock confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/services/{service_id}/unlock", public_key, secret_key, {})
|
|
|
|
|
|
def get_service_logs(
|
|
service_id: str,
|
|
all_logs: bool = False,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Get bootstrap/runtime logs for a service.
|
|
|
|
Args:
|
|
service_id: Service ID to get logs for
|
|
all_logs: If True, get all logs; if False, get last ~9000 lines (tail)
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict containing "log" field with log content
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
path = f"/services/{service_id}/logs"
|
|
if all_logs:
|
|
path += "?all=true"
|
|
return _make_request("GET", path, public_key, secret_key)
|
|
|
|
|
|
def get_service_env(
|
|
service_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Get environment vault status for a service.
|
|
|
|
Returns metadata about the vault (has_vault, count, updated_at)
|
|
but NOT the actual secrets. Use export_service_env to retrieve secrets.
|
|
|
|
Args:
|
|
service_id: Service ID to get env status for
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with has_vault, count, updated_at fields
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("GET", f"/services/{service_id}/env", public_key, secret_key)
|
|
|
|
|
|
def set_service_env(
|
|
service_id: str,
|
|
env_dict: Dict[str, str],
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Set environment variables for a service.
|
|
|
|
Replaces the entire environment vault with the provided variables.
|
|
Variables are encrypted at rest and injected into the container.
|
|
|
|
Args:
|
|
service_id: Service ID to set env for
|
|
env_dict: Dictionary of environment variables (KEY: VALUE)
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with count of variables set
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
# Convert dict to .env format for the API
|
|
env_content = "\n".join(f"{k}={v}" for k, v in env_dict.items())
|
|
|
|
# Note: This endpoint expects text/plain body, but we'll send as JSON
|
|
# and let the API handle conversion
|
|
return _make_request(
|
|
"POST",
|
|
f"/services/{service_id}/env",
|
|
public_key,
|
|
secret_key,
|
|
{"env": env_content},
|
|
)
|
|
|
|
|
|
def delete_service_env(
|
|
service_id: str,
|
|
keys: Optional[List[str]] = None,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Delete environment vault or specific keys from a service.
|
|
|
|
Args:
|
|
service_id: Service ID to delete env from
|
|
keys: Optional list of specific keys to delete; if None, deletes entire vault
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with deletion confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
path = f"/services/{service_id}/env"
|
|
# If specific keys provided, could add as query params (API dependent)
|
|
return _make_request("DELETE", path, public_key, secret_key)
|
|
|
|
|
|
def export_service_env(
|
|
service_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Export environment vault secrets for a service.
|
|
|
|
Requires HMAC authentication to prove ownership.
|
|
Returns the actual secret values in .env format.
|
|
|
|
Args:
|
|
service_id: Service ID to export env from
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict containing "env" field with KEY=VALUE content
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/services/{service_id}/env/export", public_key, secret_key, {})
|
|
|
|
|
|
def redeploy_service(
|
|
service_id: str,
|
|
bootstrap: Optional[str] = None,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Redeploy a service (re-run bootstrap script).
|
|
|
|
Bootstrap scripts should be idempotent for proper upgrade behavior.
|
|
|
|
Args:
|
|
service_id: Service ID to redeploy
|
|
bootstrap: Optional new bootstrap script/URL (uses existing if not provided)
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with redeploy confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
data: Dict[str, Any] = {}
|
|
if bootstrap:
|
|
if bootstrap.startswith("http://") or bootstrap.startswith("https://"):
|
|
data["bootstrap"] = bootstrap
|
|
else:
|
|
data["bootstrap_content"] = bootstrap
|
|
|
|
return _make_request("POST", f"/services/{service_id}/redeploy", public_key, secret_key, data)
|
|
|
|
|
|
def execute_in_service(
|
|
service_id: str,
|
|
command: str,
|
|
timeout: int = 30000,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Execute a command in a running service container.
|
|
|
|
Uses async job polling for long-running commands.
|
|
|
|
Args:
|
|
service_id: Service ID to execute command in
|
|
command: Shell command to execute
|
|
timeout: Command timeout in milliseconds (default 30000)
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with job_id for async polling, or direct result
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request(
|
|
"POST",
|
|
f"/services/{service_id}/execute",
|
|
public_key,
|
|
secret_key,
|
|
{"command": command, "timeout": timeout},
|
|
)
|
|
|
|
|
|
# =============================================================================
|
|
# Additional Snapshot Functions
|
|
# =============================================================================
|
|
|
|
|
|
def lock_snapshot(
|
|
snapshot_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Lock a snapshot to prevent accidental deletion.
|
|
|
|
Args:
|
|
snapshot_id: Snapshot ID to lock
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with lock confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/snapshots/{snapshot_id}/lock", public_key, secret_key, {})
|
|
|
|
|
|
def unlock_snapshot(
|
|
snapshot_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Unlock a snapshot to allow deletion.
|
|
|
|
Args:
|
|
snapshot_id: Snapshot ID to unlock
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with unlock confirmation
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/snapshots/{snapshot_id}/unlock", public_key, secret_key, {})
|
|
|
|
|
|
def clone_snapshot(
|
|
snapshot_id: str,
|
|
clone_type: str = "session",
|
|
name: Optional[str] = None,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
shell: Optional[str] = None,
|
|
ports: Optional[List[int]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Clone a snapshot to create a new session or service.
|
|
|
|
Args:
|
|
snapshot_id: Snapshot ID to clone from
|
|
clone_type: Type of resource to create ("session" or "service")
|
|
name: Optional name for the new resource
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
shell: Optional shell for session clones
|
|
ports: Optional ports list for service clones
|
|
|
|
Returns:
|
|
Response dict containing session_id or service_id
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
data: Dict[str, Any] = {"type": clone_type}
|
|
if name:
|
|
data["name"] = name
|
|
if shell:
|
|
data["shell"] = shell
|
|
if ports:
|
|
data["ports"] = ports
|
|
|
|
return _make_request("POST", f"/snapshots/{snapshot_id}/clone", public_key, secret_key, data)
|
|
|
|
|
|
# =============================================================================
|
|
# Images (LXD Container Images)
|
|
# =============================================================================
|
|
|
|
|
|
def image_publish(
|
|
source_type: str,
|
|
source_id: str,
|
|
name: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Publish a service or snapshot as a portable LXD image.
|
|
|
|
Images are independent of containers and survive service deletion.
|
|
They can be shared with other users or made public.
|
|
|
|
Args:
|
|
source_type: Type of source ("service" or "snapshot")
|
|
source_id: ID of the service or snapshot to publish
|
|
name: Optional friendly name for the image
|
|
description: Optional description
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict containing:
|
|
- id: Image ID (unsb-image-xxxx-xxxx-xxxx-xxxx)
|
|
- name: Image name
|
|
- fingerprint: LXD image fingerprint
|
|
- size_bytes: Image size in bytes
|
|
|
|
Example:
|
|
>>> img = image_publish("service", "unsb-service-xxxx")
|
|
>>> print(img["id"])
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
data: Dict[str, Any] = {"source_type": source_type, "source_id": source_id}
|
|
if name:
|
|
data["name"] = name
|
|
if description:
|
|
data["description"] = description
|
|
return _make_request("POST", "/images", public_key, secret_key, data)
|
|
|
|
|
|
def list_images(
|
|
filter_type: Optional[str] = None,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
List images accessible to this API key.
|
|
|
|
By default lists all accessible images (owned + shared + public).
|
|
Use filter_type to narrow results.
|
|
|
|
Args:
|
|
filter_type: Optional filter - "owned", "shared", or "public"
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict containing:
|
|
- images: List of image objects
|
|
- count: Total number of images
|
|
|
|
Example:
|
|
>>> result = list_images()
|
|
>>> for img in result["images"]:
|
|
... print(f"{img['id']}: {img['name']} ({img['access']})")
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
endpoint = "/images"
|
|
if filter_type:
|
|
endpoint = f"/images/{filter_type}"
|
|
return _make_request("GET", endpoint, public_key, secret_key)
|
|
|
|
|
|
def get_image(
|
|
image_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Get details of a specific image.
|
|
|
|
Args:
|
|
image_id: Image ID
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict containing image details:
|
|
- id, name, description, fingerprint
|
|
- source_type, source_id
|
|
- size_bytes, locked, visibility
|
|
- trusted_keys, created_at
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("GET", f"/images/{image_id}", public_key, secret_key)
|
|
|
|
|
|
def delete_image(
|
|
image_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Delete an image.
|
|
|
|
Cannot delete locked images - unlock first.
|
|
|
|
Args:
|
|
image_id: Image ID to delete
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with deletion confirmation
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("DELETE", f"/images/{image_id}", public_key, secret_key)
|
|
|
|
|
|
def lock_image(
|
|
image_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Lock an image to prevent accidental deletion.
|
|
|
|
Args:
|
|
image_id: Image ID to lock
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with lock status
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/images/{image_id}/lock", public_key, secret_key, {})
|
|
|
|
|
|
def unlock_image(
|
|
image_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Unlock an image to allow deletion.
|
|
|
|
Args:
|
|
image_id: Image ID to unlock
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with unlock confirmation
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/images/{image_id}/unlock", public_key, secret_key, {})
|
|
|
|
|
|
def set_image_visibility(
|
|
image_id: str,
|
|
visibility: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Set image visibility.
|
|
|
|
Args:
|
|
image_id: Image ID
|
|
visibility: One of "private", "unlisted", or "public"
|
|
- private: Only owner can see/use
|
|
- unlisted: Hidden but can be shared via trusted_keys
|
|
- public: Visible to all users (marketplace)
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with updated image info
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/images/{image_id}/visibility", public_key, secret_key, {"visibility": visibility})
|
|
|
|
|
|
def grant_image_access(
|
|
image_id: str,
|
|
trusted_api_key: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Grant access to an image for another API key.
|
|
|
|
Works for private/unlisted images. Public images are already accessible.
|
|
|
|
Args:
|
|
image_id: Image ID
|
|
trusted_api_key: Public key of the user to grant access
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with updated trusted_keys list
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/images/{image_id}/grant", public_key, secret_key, {"trusted_api_key": trusted_api_key})
|
|
|
|
|
|
def revoke_image_access(
|
|
image_id: str,
|
|
trusted_api_key: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Revoke access to an image from another API key.
|
|
|
|
Args:
|
|
image_id: Image ID
|
|
trusted_api_key: Public key of the user to revoke access
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with updated trusted_keys list
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/images/{image_id}/revoke", public_key, secret_key, {"trusted_api_key": trusted_api_key})
|
|
|
|
|
|
def list_image_trusted(
|
|
image_id: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
List all API keys that have access to an image.
|
|
|
|
Args:
|
|
image_id: Image ID
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict containing:
|
|
- trusted_keys: List of API public keys with access
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("GET", f"/images/{image_id}/trusted", public_key, secret_key)
|
|
|
|
|
|
def transfer_image(
|
|
image_id: str,
|
|
to_api_key: str,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Transfer image ownership to another API key.
|
|
|
|
This is a database-only operation - the LXD image stays on the same node.
|
|
|
|
Args:
|
|
image_id: Image ID to transfer
|
|
to_api_key: Public key of the recipient
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with updated owner info
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
return _make_request("POST", f"/images/{image_id}/transfer", public_key, secret_key, {"to_api_key": to_api_key})
|
|
|
|
|
|
def spawn_from_image(
|
|
image_id: str,
|
|
name: Optional[str] = None,
|
|
ports: Optional[List[int]] = None,
|
|
bootstrap: Optional[str] = None,
|
|
network_mode: str = "zerotrust",
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create a new service from an image.
|
|
|
|
Spawns a new container using the image as the base.
|
|
|
|
Args:
|
|
image_id: Image ID to spawn from
|
|
name: Optional service name
|
|
ports: Optional list of ports to expose
|
|
bootstrap: Optional bootstrap script (image may already have app)
|
|
network_mode: "zerotrust" or "semitrusted"
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict containing service info with source_image field
|
|
|
|
Example:
|
|
>>> svc = spawn_from_image("unsb-image-xxxx", name="my-app", ports=[8080])
|
|
>>> print(svc["service_id"])
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
data: Dict[str, Any] = {"network_mode": network_mode}
|
|
if name:
|
|
data["name"] = name
|
|
if ports:
|
|
data["ports"] = ports
|
|
if bootstrap:
|
|
data["bootstrap"] = bootstrap
|
|
return _make_request("POST", f"/images/{image_id}/spawn", public_key, secret_key, data)
|
|
|
|
|
|
def clone_image(
|
|
image_id: str,
|
|
name: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Clone an image to create a copy owned by you.
|
|
|
|
The clone inherits the source image's content but:
|
|
- Gets a new unique ID
|
|
- Is owned by the requesting user
|
|
- Starts as private visibility
|
|
- Has empty trusted_keys
|
|
|
|
Args:
|
|
image_id: Image ID to clone
|
|
name: Optional name for the clone
|
|
description: Optional description
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict containing new image info
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
data: Dict[str, Any] = {}
|
|
if name:
|
|
data["name"] = name
|
|
if description:
|
|
data["description"] = description
|
|
return _make_request("POST", f"/images/{image_id}/clone", public_key, secret_key, data)
|
|
|
|
|
|
# =============================================================================
|
|
# Key Validation
|
|
# =============================================================================
|
|
|
|
|
|
PORTAL_BASE = "https://unsandbox.com"
|
|
|
|
|
|
def validate_keys(
|
|
public_key: Optional[str] = None,
|
|
secret_key: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Validate API keys against the portal.
|
|
|
|
Checks if the keys are valid, not expired, and not suspended.
|
|
|
|
Args:
|
|
public_key: Optional API key
|
|
secret_key: Optional API secret
|
|
|
|
Returns:
|
|
Response dict with validation result:
|
|
- valid: True if keys are valid
|
|
- tier: Account tier level
|
|
- expires_at: Expiration timestamp (if applicable)
|
|
- reason: Reason for invalid status (if applicable)
|
|
|
|
Raises:
|
|
requests.RequestException: Network errors
|
|
ValueError: Invalid response format
|
|
CredentialsError: Missing credentials
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
|
|
url = f"{API_BASE}/keys/validate"
|
|
timestamp = int(time.time())
|
|
body = ""
|
|
|
|
signature = _sign_request(secret_key, timestamp, "POST", "/keys/validate", body)
|
|
|
|
headers = {
|
|
"Authorization": f"Bearer {public_key}",
|
|
"X-Timestamp": str(timestamp),
|
|
"X-Signature": signature,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
response = requests.post(url, headers=headers, data=body, timeout=30)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
# =============================================================================
|
|
# Image Generation
|
|
# =============================================================================
|
|
|
|
|
|
def image(
|
|
prompt: str,
|
|
*,
|
|
model: str = None,
|
|
size: str = "1024x1024",
|
|
quality: str = "standard",
|
|
n: int = 1,
|
|
public_key: str = None,
|
|
secret_key: str = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Generate images from text prompt using AI.
|
|
|
|
Args:
|
|
prompt: Text description of the image to generate
|
|
model: Model to use (optional, uses default)
|
|
size: Image size (e.g., "1024x1024", "512x512")
|
|
quality: "standard" or "hd"
|
|
n: Number of images to generate
|
|
public_key: API public key (optional)
|
|
secret_key: API secret key (optional)
|
|
|
|
Returns:
|
|
dict with keys: images (list of base64 or URLs), created_at
|
|
|
|
Example:
|
|
>>> result = image("A sunset over mountains")
|
|
>>> print(result["images"][0])
|
|
"""
|
|
public_key, secret_key = _resolve_credentials(public_key, secret_key)
|
|
payload = {
|
|
"prompt": prompt,
|
|
"size": size,
|
|
"quality": quality,
|
|
"n": n,
|
|
}
|
|
if model:
|
|
payload["model"] = model
|
|
|
|
return _make_request("POST", "/image", public_key, secret_key, payload)
|
|
|
|
|
|
# =============================================================================
|
|
# CLI Implementation
|
|
# =============================================================================
|
|
|
|
import sys
|
|
import argparse
|
|
|
|
|
|
def _parse_env_file(file_path: str) -> Dict[str, str]:
|
|
"""Parse a .env file into a dictionary."""
|
|
env_dict = {}
|
|
try:
|
|
with open(file_path, "r") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
if "=" in line:
|
|
key, _, value = line.partition("=")
|
|
# Handle quoted values
|
|
value = value.strip()
|
|
if (value.startswith('"') and value.endswith('"')) or \
|
|
(value.startswith("'") and value.endswith("'")):
|
|
value = value[1:-1]
|
|
env_dict[key.strip()] = value
|
|
except Exception as e:
|
|
print(f"Error: Failed to parse env file: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
return env_dict
|
|
|
|
|
|
def _format_list_output(items: List[Dict[str, Any]], resource_type: str) -> str:
|
|
"""Format list output in table format."""
|
|
if not items:
|
|
return f"No {resource_type}s found."
|
|
|
|
# Determine columns based on resource type
|
|
if resource_type == "session":
|
|
headers = ["ID", "STATUS", "SHELL", "CREATED"]
|
|
rows = []
|
|
for item in items:
|
|
rows.append([
|
|
item.get("id", item.get("session_id", ""))[:36],
|
|
item.get("status", "unknown"),
|
|
item.get("shell", "bash"),
|
|
item.get("created_at", "")[:19] if item.get("created_at") else "",
|
|
])
|
|
elif resource_type == "service":
|
|
headers = ["ID", "NAME", "STATUS", "PORTS", "CREATED"]
|
|
rows = []
|
|
for item in items:
|
|
ports = item.get("ports", [])
|
|
ports_str = ",".join(str(p) for p in ports) if ports else ""
|
|
rows.append([
|
|
item.get("id", item.get("service_id", ""))[:36],
|
|
item.get("name", "")[:20],
|
|
item.get("status", "unknown"),
|
|
ports_str[:15],
|
|
item.get("created_at", "")[:19] if item.get("created_at") else "",
|
|
])
|
|
elif resource_type == "snapshot":
|
|
headers = ["ID", "NAME", "TYPE", "SIZE", "CREATED"]
|
|
rows = []
|
|
for item in items:
|
|
rows.append([
|
|
item.get("id", item.get("snapshot_id", ""))[:36],
|
|
item.get("name", "")[:20],
|
|
item.get("source_type", "unknown"),
|
|
item.get("size", ""),
|
|
item.get("created_at", "")[:19] if item.get("created_at") else "",
|
|
])
|
|
elif resource_type == "image":
|
|
headers = ["ID", "NAME", "VISIBILITY", "SOURCE", "CREATED"]
|
|
rows = []
|
|
for item in items:
|
|
rows.append([
|
|
item.get("id", item.get("image_id", ""))[:36],
|
|
item.get("name", "")[:20],
|
|
item.get("visibility", "private"),
|
|
item.get("source_type", "")[:10],
|
|
item.get("created_at", "")[:19] if item.get("created_at") else "",
|
|
])
|
|
else:
|
|
headers = ["ID", "STATUS"]
|
|
rows = [[str(item.get("id", "")), str(item.get("status", ""))] for item in items]
|
|
|
|
# Calculate column widths
|
|
widths = [len(h) for h in headers]
|
|
for row in rows:
|
|
for i, cell in enumerate(row):
|
|
widths[i] = max(widths[i], len(str(cell)))
|
|
|
|
# Build output
|
|
lines = []
|
|
header_line = " ".join(h.ljust(widths[i]) for i, h in enumerate(headers))
|
|
lines.append(header_line)
|
|
for row in rows:
|
|
line = " ".join(str(cell).ljust(widths[i]) for i, cell in enumerate(row))
|
|
lines.append(line)
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _build_parser() -> argparse.ArgumentParser:
|
|
"""Build the argument parser for the CLI."""
|
|
parser = argparse.ArgumentParser(
|
|
prog="un.py",
|
|
description="Unsandbox CLI - Execute code in secure containers",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
python un.py script.py Execute Python script
|
|
python un.py -s bash 'echo hello' Inline bash command
|
|
python un.py session --list List active sessions
|
|
python un.py service --list List all services
|
|
python un.py snapshot --list List all snapshots
|
|
python un.py key Check API key
|
|
python un.py languages List available languages
|
|
python un.py languages --json List languages as JSON
|
|
""",
|
|
)
|
|
|
|
# Global options
|
|
parser.add_argument("-s", "--shell", metavar="LANG",
|
|
help="Language for inline code execution")
|
|
parser.add_argument("-e", "--env", action="append", metavar="KEY=VAL",
|
|
help="Set environment variable (can be used multiple times)")
|
|
parser.add_argument("-f", "--file", action="append", metavar="FILE",
|
|
help="Add input file to /tmp/ (can be used multiple times)")
|
|
parser.add_argument("-F", "--file-path", action="append", metavar="FILE",
|
|
help="Add input file with path preserved")
|
|
parser.add_argument("-a", "--artifacts", action="store_true",
|
|
help="Return compiled artifacts")
|
|
parser.add_argument("-o", "--output", metavar="DIR",
|
|
help="Output directory for artifacts")
|
|
parser.add_argument("-p", "--public-key", metavar="KEY",
|
|
help="API public key")
|
|
parser.add_argument("-k", "--secret-key", metavar="KEY",
|
|
help="API secret key")
|
|
parser.add_argument("-n", "--network", choices=["zerotrust", "semitrusted"],
|
|
default="zerotrust", help="Network mode (default: zerotrust)")
|
|
parser.add_argument("-v", "--vcpu", type=int, default=1, choices=range(1, 9),
|
|
metavar="N", help="vCPU count (1-8, default: 1)")
|
|
parser.add_argument("-y", "--yes", action="store_true",
|
|
help="Skip confirmation prompts")
|
|
|
|
# Subcommands
|
|
subparsers = parser.add_subparsers(dest="command", help="Commands")
|
|
|
|
# Session subcommand
|
|
session_parser = subparsers.add_parser("session", help="Manage interactive sessions")
|
|
session_group = session_parser.add_mutually_exclusive_group()
|
|
session_group.add_argument("-l", "--list", action="store_true",
|
|
help="List active sessions")
|
|
session_group.add_argument("--attach", metavar="ID",
|
|
help="Reconnect to existing session")
|
|
session_group.add_argument("--kill", metavar="ID",
|
|
help="Terminate a session")
|
|
session_group.add_argument("--freeze", metavar="ID",
|
|
help="Pause session")
|
|
session_group.add_argument("--unfreeze", metavar="ID",
|
|
help="Resume session")
|
|
session_group.add_argument("--boost", metavar="ID",
|
|
help="Add resources to session")
|
|
session_group.add_argument("--unboost", metavar="ID",
|
|
help="Remove boost from session")
|
|
session_group.add_argument("--snapshot", metavar="ID",
|
|
help="Create snapshot of session")
|
|
session_parser.add_argument("--shell", metavar="SHELL",
|
|
help="Shell/REPL to use (default: bash)")
|
|
session_parser.add_argument("--tmux", action="store_true",
|
|
help="Enable persistence with tmux")
|
|
session_parser.add_argument("--screen", action="store_true",
|
|
help="Enable persistence with screen")
|
|
session_parser.add_argument("--snapshot-name", metavar="NAME",
|
|
help="Name for snapshot")
|
|
session_parser.add_argument("--hot", action="store_true",
|
|
help="Live snapshot (no freeze)")
|
|
session_parser.add_argument("--audit", action="store_true",
|
|
help="Record session")
|
|
|
|
# Service subcommand
|
|
service_parser = subparsers.add_parser("service", help="Manage persistent services")
|
|
service_group = service_parser.add_mutually_exclusive_group()
|
|
service_group.add_argument("-l", "--list", action="store_true",
|
|
help="List all services")
|
|
service_group.add_argument("--info", metavar="ID",
|
|
help="Get service details")
|
|
service_group.add_argument("--logs", metavar="ID",
|
|
help="Get all logs")
|
|
service_group.add_argument("--tail", metavar="ID",
|
|
help="Get last 9000 lines of logs")
|
|
service_group.add_argument("--freeze", metavar="ID",
|
|
help="Pause service")
|
|
service_group.add_argument("--unfreeze", metavar="ID",
|
|
help="Resume service")
|
|
service_group.add_argument("--destroy", metavar="ID",
|
|
help="Delete service")
|
|
service_group.add_argument("--lock", metavar="ID",
|
|
help="Prevent deletion")
|
|
service_group.add_argument("--unlock", metavar="ID",
|
|
help="Allow deletion")
|
|
service_group.add_argument("--resize", metavar="ID",
|
|
help="Resize service (with --vcpu)")
|
|
service_group.add_argument("--redeploy", metavar="ID",
|
|
help="Re-run bootstrap")
|
|
service_group.add_argument("--execute", nargs=2, metavar=("ID", "CMD"),
|
|
help="Run command in service")
|
|
service_group.add_argument("--snapshot", metavar="ID",
|
|
help="Create snapshot of service")
|
|
service_parser.add_argument("--name", metavar="NAME",
|
|
help="Service name (creates new)")
|
|
service_parser.add_argument("--ports", metavar="PORTS",
|
|
help="Comma-separated ports")
|
|
service_parser.add_argument("--domains", metavar="DOMAINS",
|
|
help="Custom domains")
|
|
service_parser.add_argument("--type", metavar="TYPE", dest="service_type",
|
|
help="Service type (minecraft, tcp, udp)")
|
|
service_parser.add_argument("--bootstrap", metavar="CMD",
|
|
help="Bootstrap command")
|
|
service_parser.add_argument("--bootstrap-file", metavar="FILE",
|
|
help="Bootstrap from file")
|
|
service_parser.add_argument("--env-file", metavar="FILE",
|
|
help="Load env from .env file")
|
|
service_parser.add_argument("--snapshot-name", metavar="NAME",
|
|
help="Name for snapshot")
|
|
service_parser.add_argument("--hot", action="store_true",
|
|
help="Live snapshot (no freeze)")
|
|
|
|
# Service env subcommand
|
|
service_env_parser = subparsers.add_parser("service-env",
|
|
help="Manage service environment vault")
|
|
service_env_parser.add_argument("action", choices=["status", "set", "export", "delete"],
|
|
help="Environment action")
|
|
service_env_parser.add_argument("service_id", metavar="ID",
|
|
help="Service ID")
|
|
service_env_parser.add_argument("--env-file", metavar="FILE",
|
|
help="Load env from .env file (for set)")
|
|
|
|
# Snapshot subcommand
|
|
snapshot_parser = subparsers.add_parser("snapshot", help="Manage snapshots")
|
|
snapshot_group = snapshot_parser.add_mutually_exclusive_group()
|
|
snapshot_group.add_argument("-l", "--list", action="store_true",
|
|
help="List all snapshots")
|
|
snapshot_group.add_argument("--info", metavar="ID",
|
|
help="Get snapshot details")
|
|
snapshot_group.add_argument("--delete", metavar="ID",
|
|
help="Delete snapshot")
|
|
snapshot_group.add_argument("--lock", metavar="ID",
|
|
help="Prevent deletion")
|
|
snapshot_group.add_argument("--unlock", metavar="ID",
|
|
help="Allow deletion")
|
|
snapshot_group.add_argument("--clone", metavar="ID",
|
|
help="Clone snapshot")
|
|
snapshot_parser.add_argument("--type", choices=["session", "service"],
|
|
dest="clone_type", help="Clone type")
|
|
snapshot_parser.add_argument("--name", metavar="NAME",
|
|
help="Name for cloned resource")
|
|
snapshot_parser.add_argument("--shell", metavar="SHELL",
|
|
help="Shell for cloned session")
|
|
snapshot_parser.add_argument("--ports", metavar="PORTS",
|
|
help="Ports for cloned service")
|
|
|
|
# Image subcommand
|
|
image_parser = subparsers.add_parser("image", help="Manage images")
|
|
image_group = image_parser.add_mutually_exclusive_group()
|
|
image_group.add_argument("-l", "--list", action="store_true",
|
|
help="List all images")
|
|
image_group.add_argument("--info", metavar="ID",
|
|
help="Get image details")
|
|
image_group.add_argument("--delete", metavar="ID",
|
|
help="Delete image")
|
|
image_group.add_argument("--lock", metavar="ID",
|
|
help="Prevent deletion")
|
|
image_group.add_argument("--unlock", metavar="ID",
|
|
help="Allow deletion")
|
|
image_group.add_argument("--publish", metavar="ID",
|
|
help="Publish image from service/snapshot (requires --source-type)")
|
|
image_group.add_argument("--visibility", nargs=2, metavar=("ID", "MODE"),
|
|
help="Set visibility (private, unlisted, public)")
|
|
image_group.add_argument("--spawn", metavar="ID",
|
|
help="Spawn new service from image")
|
|
image_group.add_argument("--clone", metavar="ID",
|
|
help="Clone an image")
|
|
image_parser.add_argument("--source-type", metavar="TYPE",
|
|
choices=["service", "snapshot"],
|
|
help="Source type for publish (service or snapshot)")
|
|
image_parser.add_argument("--name", metavar="NAME",
|
|
help="Name for spawned service or cloned image")
|
|
image_parser.add_argument("--ports", metavar="PORTS",
|
|
help="Ports for spawned service (comma-separated)")
|
|
|
|
# Key subcommand
|
|
subparsers.add_parser("key", help="Check API key validity")
|
|
|
|
# Languages subcommand
|
|
languages_parser = subparsers.add_parser("languages", help="List available languages")
|
|
languages_parser.add_argument("--json", action="store_true",
|
|
help="Output as JSON array")
|
|
|
|
# Positional argument for source file or inline code
|
|
parser.add_argument("source", nargs="?",
|
|
help="Source file or inline code (with -s)")
|
|
|
|
return parser
|
|
|
|
|
|
def cli_main():
|
|
"""Main entry point for CLI."""
|
|
parser = _build_parser()
|
|
args = parser.parse_args()
|
|
|
|
# Resolve credentials
|
|
try:
|
|
public_key, secret_key = _resolve_credentials(
|
|
args.public_key, args.secret_key
|
|
)
|
|
except CredentialsError as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(3)
|
|
|
|
try:
|
|
# Handle subcommands
|
|
if args.command == "session":
|
|
_handle_session_command(args, public_key, secret_key)
|
|
elif args.command == "service":
|
|
_handle_service_command(args, public_key, secret_key)
|
|
elif args.command == "service-env":
|
|
_handle_service_env_command(args, public_key, secret_key)
|
|
elif args.command == "snapshot":
|
|
_handle_snapshot_command(args, public_key, secret_key)
|
|
elif args.command == "image":
|
|
_handle_image_command(args, public_key, secret_key)
|
|
elif args.command == "key":
|
|
_handle_key_command(public_key, secret_key)
|
|
elif args.command == "languages":
|
|
_handle_languages_command(args, public_key, secret_key)
|
|
elif args.source or args.shell:
|
|
_handle_execute_command(args, public_key, secret_key)
|
|
else:
|
|
parser.print_help()
|
|
sys.exit(2)
|
|
except CredentialsError as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(3)
|
|
except requests.exceptions.HTTPError as e:
|
|
if e.response is not None and e.response.status_code == 401:
|
|
print("Error: Authentication failed", file=sys.stderr)
|
|
sys.exit(3)
|
|
print(f"Error: API error - {e}", file=sys.stderr)
|
|
sys.exit(4)
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"Error: Network error - {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def _handle_execute_command(args, public_key: str, secret_key: str):
|
|
"""Handle code execution command."""
|
|
# Determine language and code
|
|
if args.shell:
|
|
# Inline code mode
|
|
if not args.source:
|
|
print("Error: Code required with -s/--shell", file=sys.stderr)
|
|
sys.exit(2)
|
|
language = args.shell
|
|
code = args.source
|
|
else:
|
|
# File mode
|
|
if not args.source:
|
|
print("Error: Source file required", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
# Detect language from filename
|
|
language = detect_language(args.source)
|
|
if not language:
|
|
print(f"Error: Cannot detect language from '{args.source}'", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
# Read source file
|
|
try:
|
|
with open(args.source, "r") as f:
|
|
code = f.read()
|
|
except FileNotFoundError:
|
|
print(f"Error: File not found: {args.source}", file=sys.stderr)
|
|
sys.exit(1)
|
|
except Exception as e:
|
|
print(f"Error: Failed to read file: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Execute code
|
|
result = execute_code(language, code, public_key, secret_key)
|
|
|
|
# Output result
|
|
stdout = result.get("stdout", "")
|
|
stderr = result.get("stderr", "")
|
|
exit_code = result.get("exit_code", 0)
|
|
execution_time = result.get("execution_time_ms", 0)
|
|
|
|
if stdout:
|
|
print(stdout, end="")
|
|
if not stdout.endswith("\n"):
|
|
print()
|
|
|
|
if stderr:
|
|
print(stderr, end="", file=sys.stderr)
|
|
if not stderr.endswith("\n"):
|
|
print(file=sys.stderr)
|
|
|
|
print("---")
|
|
print(f"Exit code: {exit_code}")
|
|
print(f"Execution time: {execution_time}ms")
|
|
|
|
sys.exit(exit_code if exit_code else 0)
|
|
|
|
|
|
def _handle_session_command(args, public_key: str, secret_key: str):
|
|
"""Handle session subcommand."""
|
|
if args.list:
|
|
sessions = list_sessions(public_key, secret_key)
|
|
print(_format_list_output(sessions, "session"))
|
|
elif args.attach:
|
|
# Get session info for attach
|
|
session = get_session(args.attach, public_key, secret_key)
|
|
print(f"Session ID: {session.get('id', session.get('session_id', ''))}")
|
|
print(f"Status: {session.get('status', 'unknown')}")
|
|
print(f"WebSocket URL: wss://api.unsandbox.com/sessions/{args.attach}/shell")
|
|
print("\nUse a WebSocket client to connect interactively.")
|
|
elif args.kill:
|
|
result = delete_session(args.kill, public_key, secret_key)
|
|
print(f"Session {args.kill} terminated")
|
|
elif args.freeze:
|
|
result = freeze_session(args.freeze, public_key, secret_key)
|
|
print(f"Session {args.freeze} frozen")
|
|
elif args.unfreeze:
|
|
result = unfreeze_session(args.unfreeze, public_key, secret_key)
|
|
print(f"Session {args.unfreeze} unfrozen")
|
|
elif args.boost:
|
|
result = boost_session(args.boost, public_key, secret_key)
|
|
print(f"Session {args.boost} boosted")
|
|
elif args.unboost:
|
|
result = unboost_session(args.unboost, public_key, secret_key)
|
|
print(f"Session {args.unboost} unboosted")
|
|
elif args.snapshot:
|
|
snapshot_id = session_snapshot(
|
|
args.snapshot, public_key, secret_key,
|
|
name=args.snapshot_name,
|
|
ephemeral=not args.hot
|
|
)
|
|
print(f"Snapshot created: {snapshot_id}")
|
|
else:
|
|
# Create new session
|
|
multiplexer = None
|
|
if args.tmux:
|
|
multiplexer = "tmux"
|
|
elif args.screen:
|
|
multiplexer = "screen"
|
|
|
|
result = create_session(
|
|
shell=args.shell,
|
|
network_mode="semitrusted" if hasattr(args, 'network') and args.network == "semitrusted" else "zerotrust",
|
|
public_key=public_key,
|
|
secret_key=secret_key,
|
|
multiplexer=multiplexer,
|
|
)
|
|
|
|
session_id = result.get("session_id", result.get("id", ""))
|
|
print(f"Session created: {session_id}")
|
|
print(f"WebSocket URL: wss://api.unsandbox.com/sessions/{session_id}/shell")
|
|
|
|
|
|
def _handle_service_command(args, public_key: str, secret_key: str):
|
|
"""Handle service subcommand."""
|
|
if args.list:
|
|
services = list_services(public_key, secret_key)
|
|
print(_format_list_output(services, "service"))
|
|
elif args.info:
|
|
service = get_service(args.info, public_key, secret_key)
|
|
print(json.dumps(service, indent=2))
|
|
elif args.logs:
|
|
result = get_service_logs(args.logs, all_logs=True, public_key=public_key, secret_key=secret_key)
|
|
print(result.get("log", ""))
|
|
elif args.tail:
|
|
result = get_service_logs(args.tail, all_logs=False, public_key=public_key, secret_key=secret_key)
|
|
print(result.get("log", ""))
|
|
elif args.freeze:
|
|
result = freeze_service(args.freeze, public_key, secret_key)
|
|
print(f"Service {args.freeze} frozen")
|
|
elif args.unfreeze:
|
|
result = unfreeze_service(args.unfreeze, public_key, secret_key)
|
|
print(f"Service {args.unfreeze} unfrozen")
|
|
elif args.destroy:
|
|
result = delete_service(args.destroy, public_key, secret_key)
|
|
print(f"Service {args.destroy} destroyed")
|
|
elif args.lock:
|
|
result = lock_service(args.lock, public_key, secret_key)
|
|
print(f"Service {args.lock} locked")
|
|
elif args.unlock:
|
|
result = unlock_service(args.unlock, public_key, secret_key)
|
|
print(f"Service {args.unlock} unlocked")
|
|
elif args.resize:
|
|
vcpu = getattr(args, 'vcpu', 1) or 1
|
|
result = update_service(args.resize, public_key, secret_key, vcpu=vcpu)
|
|
print(f"Service {args.resize} resized to {vcpu} vCPU(s)")
|
|
elif args.redeploy:
|
|
bootstrap = None
|
|
if args.bootstrap_file:
|
|
with open(args.bootstrap_file, "r") as f:
|
|
bootstrap = f.read()
|
|
elif args.bootstrap:
|
|
bootstrap = args.bootstrap
|
|
result = redeploy_service(args.redeploy, bootstrap=bootstrap, public_key=public_key, secret_key=secret_key)
|
|
print(f"Service {args.redeploy} redeployed")
|
|
elif args.execute:
|
|
service_id, command = args.execute
|
|
result = execute_in_service(service_id, command, public_key=public_key, secret_key=secret_key)
|
|
# Handle async result
|
|
if result.get("job_id"):
|
|
job_result = wait_for_job(result["job_id"], public_key, secret_key)
|
|
stdout = job_result.get("stdout", "")
|
|
stderr = job_result.get("stderr", "")
|
|
if stdout:
|
|
print(stdout, end="")
|
|
if stderr:
|
|
print(stderr, end="", file=sys.stderr)
|
|
else:
|
|
stdout = result.get("stdout", "")
|
|
stderr = result.get("stderr", "")
|
|
if stdout:
|
|
print(stdout, end="")
|
|
if stderr:
|
|
print(stderr, end="", file=sys.stderr)
|
|
elif args.snapshot:
|
|
snapshot_id = service_snapshot(
|
|
args.snapshot, public_key, secret_key,
|
|
name=getattr(args, 'snapshot_name', None)
|
|
)
|
|
print(f"Snapshot created: {snapshot_id}")
|
|
elif args.name:
|
|
# Create new service
|
|
if not args.ports:
|
|
print("Error: --ports required when creating service", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
ports = [int(p.strip()) for p in args.ports.split(",")]
|
|
|
|
bootstrap = None
|
|
if args.bootstrap_file:
|
|
with open(args.bootstrap_file, "r") as f:
|
|
bootstrap = f.read()
|
|
elif args.bootstrap:
|
|
bootstrap = args.bootstrap
|
|
|
|
custom_domains = None
|
|
if args.domains:
|
|
custom_domains = [d.strip() for d in args.domains.split(",")]
|
|
|
|
result = create_service(
|
|
name=args.name,
|
|
ports=ports,
|
|
bootstrap=bootstrap,
|
|
public_key=public_key,
|
|
secret_key=secret_key,
|
|
custom_domains=custom_domains,
|
|
vcpu=getattr(args, 'vcpu', 1) or 1,
|
|
service_type=args.service_type,
|
|
)
|
|
|
|
service_id = result.get("service_id", result.get("id", ""))
|
|
print(f"Service created: {service_id}")
|
|
print(f"URL: https://{args.name}.on.unsandbox.com")
|
|
else:
|
|
print("Error: No action specified for service command", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
|
|
def _handle_service_env_command(args, public_key: str, secret_key: str):
|
|
"""Handle service env subcommand."""
|
|
if args.action == "status":
|
|
result = get_service_env(args.service_id, public_key, secret_key)
|
|
print(f"Has vault: {result.get('has_vault', False)}")
|
|
print(f"Variable count: {result.get('count', 0)}")
|
|
if result.get('updated_at'):
|
|
print(f"Updated at: {result.get('updated_at')}")
|
|
elif args.action == "set":
|
|
# Read env from file or stdin
|
|
if args.env_file:
|
|
env_dict = _parse_env_file(args.env_file)
|
|
else:
|
|
# Read from stdin
|
|
print("Enter environment variables (KEY=VALUE), one per line. Ctrl+D to finish:", file=sys.stderr)
|
|
env_dict = {}
|
|
for line in sys.stdin:
|
|
line = line.strip()
|
|
if line and "=" in line:
|
|
key, _, value = line.partition("=")
|
|
env_dict[key.strip()] = value.strip()
|
|
|
|
result = set_service_env(args.service_id, env_dict, public_key, secret_key)
|
|
print(f"Environment set: {result.get('count', len(env_dict))} variables")
|
|
elif args.action == "export":
|
|
result = export_service_env(args.service_id, public_key, secret_key)
|
|
env_content = result.get("env", "")
|
|
print(env_content)
|
|
elif args.action == "delete":
|
|
result = delete_service_env(args.service_id, public_key=public_key, secret_key=secret_key)
|
|
print(f"Environment vault deleted for service {args.service_id}")
|
|
|
|
|
|
def _handle_snapshot_command(args, public_key: str, secret_key: str):
|
|
"""Handle snapshot subcommand."""
|
|
if args.list:
|
|
snapshots = list_snapshots(public_key, secret_key)
|
|
print(_format_list_output(snapshots, "snapshot"))
|
|
elif args.info:
|
|
# Get snapshot info via listing and filtering
|
|
snapshots = list_snapshots(public_key, secret_key)
|
|
snapshot = next((s for s in snapshots if s.get("id") == args.info or s.get("snapshot_id") == args.info), None)
|
|
if snapshot:
|
|
print(json.dumps(snapshot, indent=2))
|
|
else:
|
|
print(f"Error: Snapshot {args.info} not found", file=sys.stderr)
|
|
sys.exit(1)
|
|
elif args.delete:
|
|
result = delete_snapshot(args.delete, public_key, secret_key)
|
|
print(f"Snapshot {args.delete} deleted")
|
|
elif args.lock:
|
|
result = lock_snapshot(args.lock, public_key, secret_key)
|
|
print(f"Snapshot {args.lock} locked")
|
|
elif args.unlock:
|
|
result = unlock_snapshot(args.unlock, public_key, secret_key)
|
|
print(f"Snapshot {args.unlock} unlocked")
|
|
elif args.clone:
|
|
clone_type = args.clone_type or "session"
|
|
ports = None
|
|
if args.ports:
|
|
ports = [int(p.strip()) for p in args.ports.split(",")]
|
|
|
|
result = clone_snapshot(
|
|
args.clone,
|
|
clone_type=clone_type,
|
|
name=args.name,
|
|
public_key=public_key,
|
|
secret_key=secret_key,
|
|
shell=args.shell,
|
|
ports=ports,
|
|
)
|
|
|
|
if clone_type == "session":
|
|
print(f"Session created: {result.get('session_id', result.get('id', ''))}")
|
|
else:
|
|
print(f"Service created: {result.get('service_id', result.get('id', ''))}")
|
|
else:
|
|
print("Error: No action specified for snapshot command", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
|
|
def _handle_image_command(args, public_key: str, secret_key: str):
|
|
"""Handle image subcommand."""
|
|
if args.list:
|
|
images = list_images(public_key=public_key, secret_key=secret_key)
|
|
print(_format_list_output(images, "image"))
|
|
elif args.info:
|
|
image = get_image(args.info, public_key=public_key, secret_key=secret_key)
|
|
print(json.dumps(image, indent=2))
|
|
elif args.delete:
|
|
result = delete_image(args.delete, public_key=public_key, secret_key=secret_key)
|
|
print(f"Image {args.delete} deleted")
|
|
elif args.lock:
|
|
result = lock_image(args.lock, public_key=public_key, secret_key=secret_key)
|
|
print(f"Image {args.lock} locked")
|
|
elif args.unlock:
|
|
result = unlock_image(args.unlock, public_key=public_key, secret_key=secret_key)
|
|
print(f"Image {args.unlock} unlocked")
|
|
elif args.publish:
|
|
if not args.source_type:
|
|
print("Error: --source-type required for --publish", file=sys.stderr)
|
|
sys.exit(2)
|
|
result = image_publish(
|
|
source_type=args.source_type,
|
|
source_id=args.publish,
|
|
name=args.name,
|
|
public_key=public_key,
|
|
secret_key=secret_key,
|
|
)
|
|
image_id = result.get("image_id", result.get("id", ""))
|
|
print(f"Image published: {image_id}")
|
|
elif args.visibility:
|
|
image_id, mode = args.visibility
|
|
if mode not in ("private", "unlisted", "public"):
|
|
print(f"Error: Invalid visibility mode '{mode}'. Must be private, unlisted, or public", file=sys.stderr)
|
|
sys.exit(2)
|
|
result = set_image_visibility(image_id, mode, public_key=public_key, secret_key=secret_key)
|
|
print(f"Image {image_id} visibility set to {mode}")
|
|
elif args.spawn:
|
|
if not args.name:
|
|
print("Error: --name required for --spawn", file=sys.stderr)
|
|
sys.exit(2)
|
|
ports = None
|
|
if args.ports:
|
|
ports = [int(p.strip()) for p in args.ports.split(",")]
|
|
result = spawn_from_image(
|
|
args.spawn,
|
|
name=args.name,
|
|
ports=ports,
|
|
public_key=public_key,
|
|
secret_key=secret_key,
|
|
)
|
|
service_id = result.get("service_id", result.get("id", ""))
|
|
print(f"Service spawned: {service_id}")
|
|
elif args.clone:
|
|
result = clone_image(
|
|
args.clone,
|
|
name=args.name,
|
|
public_key=public_key,
|
|
secret_key=secret_key,
|
|
)
|
|
image_id = result.get("image_id", result.get("id", ""))
|
|
print(f"Image cloned: {image_id}")
|
|
else:
|
|
print("Error: No action specified for image command", file=sys.stderr)
|
|
sys.exit(2)
|
|
|
|
|
|
def _handle_key_command(public_key: str, secret_key: str):
|
|
"""Handle key validation command."""
|
|
result = validate_keys(public_key, secret_key)
|
|
|
|
print(f"Public key: {public_key}")
|
|
print(f"Valid: {result.get('valid', False)}")
|
|
if result.get('tier'):
|
|
print(f"Tier: {result.get('tier')}")
|
|
if result.get('expires_at'):
|
|
print(f"Expires: {result.get('expires_at')}")
|
|
if result.get('reason'):
|
|
print(f"Reason: {result.get('reason')}")
|
|
|
|
|
|
def _handle_languages_command(args, public_key: str, secret_key: str):
|
|
"""Handle languages list command."""
|
|
languages = get_languages(public_key, secret_key)
|
|
|
|
if args.json:
|
|
# Output as JSON array
|
|
print(json.dumps(languages))
|
|
else:
|
|
# Output one language per line (pipe-friendly)
|
|
for lang in languages:
|
|
print(lang)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli_main()
|