hanzo-sandbox: a client for the sandboxes cloud actually serves

The package is named for the cloud app it speaks to — cloud apps/sandbox is hanzo-sandbox here — so there is one name per concept across the two languages. Built against the routes rather than guessed: GET|POST /v1/sandboxes list, create GET|DELETE /v1/sandboxes/:id get, delete POST /v1/sandboxes/:id/exec argv | command, stdin, dir, timeoutSec GET|POST /v1/sandboxes/:id/fs read, write The API key carries the org, and the client sends nothing else. cloud resolves it through principal.OrgOf(user, org), which returns nothing without a validated principal — "no validated principal, so the org claim is untrusted" — and the gateway strips a client-supplied X-Org-Id before a handler sees it. A first cut took `org` as a required argument and sent the header on every call; that would have been ignored at best. Two things the tests found, both real: * with no key configured the client still sent `Authorization: Bearer `, an empty credential that reads as a present one and hides "no key" behind whatever the server says about a bad one. It sends no header now. * cloud spells the size field `class`, which python cannot, so Sandbox maps it to `cls` at the boundary instead of leaving callers to index a dict. This is the REMOTE half. hanzo_sandbox.sandbox stays the local one — unshare, namespaces, a boundary on this machine. Same word, unrelated concerns, separate modules. Served by a real HTTP server in the tests, since what breaks in a client of this shape is the request: path, method, credential. Co-authored-by: Hanzo Dev <[email protected]>
2026-10-23 20:23:00 +00:00 · 2026-08-09 19:36:57 -07:00
parent 9cd2d37717
commit 1fe73de0c3
3 changed files with 319 additions and 0 deletions
@@ -14,6 +14,10 @@ from hanzo_sandbox.sandbox import (
 )

 __all__ = [
+    "Exec",
+    "SandboxError",
+    "Sandbox",
+    "Sandboxes",
    "ContainerEnvironment",
    "FilesystemIsolationMode",
    "LinuxSandboxCommand",
@@ -27,3 +31,6 @@ __all__ = [
    "resolve_sandbox_status",
    "resolve_sandbox_status_for_request",
 ]
+
+# The remote half: cloud's /v1/sandboxes. The names above are the local one.
+from .cloud import Exec, Sandbox, SandboxError, Sandboxes  # noqa: E402
@@ -0,0 +1,181 @@
+"""Client for the sandboxes cloud serves at /v1/sandboxes.
+
+The package is named for the cloud app it speaks to — cloud `apps/sandbox` is
+`hanzo-sandbox` here — so the two move together and there is one name per
+concept across the two languages.
+
+This is the REMOTE half. `hanzo_sandbox.sandbox` is the local one: unshare,
+namespaces, a boundary drawn on this machine. They share a word and nothing
+else, so they stay separate modules rather than one class that means both.
+
+The API key carries the org. cloud resolves it with principal.OrgOf(user, org),
+which returns nothing unless a VALIDATED principal came with the request — "no
+validated principal, so the org claim is untrusted" — and the gateway strips
+client-supplied X-Org-Id before a handler ever sees it. So a caller sends its
+key and nothing else; an org passed by hand would be ignored at best.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass, field
+from typing import Any
+
+DEFAULT_BASE_URL = "https://api.hanzo.ai"
+
+
+class SandboxError(RuntimeError):
+    """A sandbox call that cloud refused, carrying the status it refused with."""
+
+    def __init__(self, status: int, message: str) -> None:
+        super().__init__(f"{status}: {message}")
+        self.status = status
+        self.message = message
+
+
+@dataclass(frozen=True)
+class Sandbox:
+    """One sandbox, as cloud reports it.
+
+    `runtime` is the isolation boundary the sandbox GOT, which is not always the
+    one it asked for — cloud answers with what it could actually give. Empty
+    means the node's default, which is an answer and not a gap.
+    """
+
+    id: str
+    org: str = ""
+    kind: str = ""
+    cls: str = ""
+    project: str = ""
+    status: str = ""
+    image: str = ""
+    runtime: str = ""
+    volume: str = ""
+    error: str = ""
+    created_at: int = 0
+    last_used_at: int = 0
+    expires_at: int = 0
+
+    @classmethod
+    def of(cls, d: dict[str, Any]) -> Sandbox:
+        return cls(
+            id=d.get("id", ""),
+            org=d.get("org", ""),
+            kind=d.get("kind", ""),
+            cls=d.get("class", ""),
+            project=d.get("project", ""),
+            status=d.get("status", ""),
+            image=d.get("image", ""),
+            runtime=d.get("runtime", ""),
+            volume=d.get("volume", ""),
+            error=d.get("error", ""),
+            created_at=d.get("createdAt", 0),
+            last_used_at=d.get("lastUsedAt", 0),
+            expires_at=d.get("expiresAt", 0),
+        )
+
+    @property
+    def running(self) -> bool:
+        return self.status == "running"
+
+
+@dataclass(frozen=True)
+class Exec:
+    """What a command left behind."""
+
+    exit_code: int = 0
+    stdout: str = ""
+    stderr: str = ""
+
+    @property
+    def ok(self) -> bool:
+        return self.exit_code == 0
+
+
+@dataclass
+class Sandboxes:
+    """Cloud's sandboxes, over HTTP.
+
+        boxes = Sandboxes(api_key=os.environ["HANZO_API_KEY"])   # pk-… or sk-…
+        box = boxes.create(image="ghcr.io/hanzoai/base:latest")
+        boxes.exec(box.id, argv=["python", "-c", "print(1)"])
+        boxes.delete(box.id)
+    """
+
+    api_key: str = field(default_factory=lambda: os.environ.get("HANZO_API_KEY", ""))
+    base_url: str = field(default_factory=lambda: os.environ.get("HANZO_API_URL", DEFAULT_BASE_URL))
+    timeout: float = 30.0
+
+    def list(self) -> list[Sandbox]:
+        got = self._call("GET", "")
+        rows = got.get("sandboxes", got) if isinstance(got, dict) else got
+        return [Sandbox.of(r) for r in (rows or [])]
+
+    def get(self, id: str) -> Sandbox:
+        return Sandbox.of(self._call("GET", f"/{id}"))
+
+    def create(self, *, image: str = "", kind: str = "", cls: str = "", project: str = "") -> Sandbox:
+        body = {"image": image, "kind": kind, "class": cls, "project": project}
+        return Sandbox.of(self._call("POST", "", {k: v for k, v in body.items() if v}))
+
+    def delete(self, id: str) -> None:
+        self._call("DELETE", f"/{id}")
+
+    def exec(
+        self,
+        id: str,
+        *,
+        argv: list[str] | None = None,
+        command: str = "",
+        stdin: str = "",
+        dir: str = "",
+        timeout_sec: int = 0,
+    ) -> Exec:
+        """Run something. `argv` runs it directly; `command` runs it through a shell."""
+        if not argv and not command:
+            raise ValueError("exec needs argv or command")
+        body: dict[str, Any] = {}
+        if argv:
+            body["argv"] = argv
+        if command:
+            body["command"] = command
+        if stdin:
+            body["stdin"] = stdin
+        if dir:
+            body["dir"] = dir
+        if timeout_sec:
+            body["timeoutSec"] = timeout_sec
+        got = self._call("POST", f"/{id}/exec", body)
+        return Exec(
+            exit_code=got.get("exitCode", got.get("exit_code", 0)),
+            stdout=got.get("stdout", ""),
+            stderr=got.get("stderr", ""),
+        )
+
+    def read(self, id: str, path: str) -> str:
+        return self._call("GET", f"/{id}/fs?path={urllib.parse.quote(path)}").get("content", "")
+
+    def write(self, id: str, path: str, content: str) -> None:
+        self._call("POST", f"/{id}/fs", {"path": path, "content": content})
+
+    def _call(self, method: str, path: str, body: dict[str, Any] | None = None) -> Any:
+        url = f"{self.base_url.rstrip('/')}/v1/sandboxes{path}"
+        data = json.dumps(body).encode() if body is not None else None
+        req = urllib.request.Request(url, data=data, method=method)
+        req.add_header("Content-Type", "application/json")
+        # An empty bearer is worse than none: it reads as a credential to anything
+        # counting headers, and hides "no key configured" behind whatever the
+        # server says about a bad one.
+        if self.api_key:
+            req.add_header("Authorization", f"Bearer {self.api_key}")
+        try:
+            with urllib.request.urlopen(req, timeout=self.timeout) as r:
+                raw = r.read()
+        except urllib.error.HTTPError as e:
+            detail = e.read().decode(errors="replace")[:400]
+            raise SandboxError(e.code, detail or e.reason) from None
+        return json.loads(raw) if raw else {}
@@ -0,0 +1,131 @@
+"""The cloud sandboxes client, checked against the contract cloud actually serves.
+
+Served by a real HTTP server rather than a mocked urlopen, because most of what
+can go wrong here is in the request: the path, the method, and the credential.
+
+The org is NOT sent. cloud derives it from the validated principal the API key
+carries, and the gateway strips a client-supplied X-Org-Id anyway — so a test
+that asserted the header would be pinning a thing the server ignores.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+from hanzo_sandbox import Exec, Sandbox, SandboxError, Sandboxes
+
+SEEN: list[dict] = []
+
+
+class Handler(BaseHTTPRequestHandler):
+    def log_message(self, *a):  # keep the test output quiet
+        pass
+
+    def _record(self) -> dict:
+        n = int(self.headers.get("Content-Length") or 0)
+        body = json.loads(self.rfile.read(n)) if n else None
+        row = {"method": self.command, "path": self.path, "org": self.headers.get("X-Org-Id"),
+               "auth": self.headers.get("Authorization"), "body": body}  # org captured to prove it is absent
+        SEEN.append(row)
+        return row
+
+    def _send(self, code: int, payload) -> None:
+        raw = json.dumps(payload).encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(raw)))
+        self.end_headers()
+        self.wfile.write(raw)
+
+    def do_GET(self):
+        row = self._record()
+        # cloud refuses an unauthenticated caller; the org rides inside the key
+        if not row["auth"]:
+            return self._send(403, {"error": "X-Org-Id required"})
+        if row["path"] == "/v1/sandboxes":
+            return self._send(200, [{"id": "sb_1", "class": "small", "status": "running"}])
+        if row["path"].endswith("/fs?path=/etc/hosts"):
+            return self._send(200, {"content": "127.0.0.1 localhost"})
+        return self._send(200, {"id": "sb_1", "class": "small", "status": "running", "createdAt": 7})
+
+    def do_POST(self):
+        row = self._record()
+        if row["path"].endswith("/exec"):
+            return self._send(200, {"exitCode": 0, "stdout": "hi\n", "stderr": ""})
+        if row["path"].endswith("/fs"):
+            return self._send(200, {})
+        return self._send(200, {"id": "sb_new", "status": "pending", "image": row["body"].get("image", "")})
+
+    def do_DELETE(self):
+        self._record()
+        self._send(200, {})
+
+
+class CloudSandboxesTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.srv = HTTPServer(("127.0.0.1", 0), Handler)
+        threading.Thread(target=cls.srv.serve_forever, daemon=True).start()
+        cls.base = f"http://127.0.0.1:{cls.srv.server_port}"
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.srv.shutdown()
+
+    def setUp(self):
+        SEEN.clear()
+        self.boxes = Sandboxes(api_key="sk-test", base_url=self.base)
+
+    def test_list_hits_the_collection_and_parses_class(self):
+        got = self.boxes.list()
+        self.assertEqual(SEEN[0]["path"], "/v1/sandboxes")
+        self.assertEqual(SEEN[0]["method"], "GET")
+        # cloud spells it `class`, which python cannot; the client maps it to cls
+        self.assertEqual(got[0].cls, "small")
+        self.assertTrue(got[0].running)
+
+    def test_every_call_carries_the_key_and_never_the_org(self):
+        self.boxes.list()
+        self.boxes.get("sb_1")
+        self.boxes.delete("sb_1")
+        self.assertTrue(all(r["auth"] == "Bearer sk-test" for r in SEEN), SEEN)
+        # the org is the key's to state, not the caller's
+        self.assertTrue(all(r["org"] is None for r in SEEN), SEEN)
+
+    def test_create_posts_to_the_collection_and_drops_empty_fields(self):
+        got = self.boxes.create(image="ghcr.io/hanzoai/base:latest")
+        self.assertEqual(SEEN[0]["method"], "POST")
+        self.assertEqual(SEEN[0]["path"], "/v1/sandboxes")
+        self.assertEqual(SEEN[0]["body"], {"image": "ghcr.io/hanzoai/base:latest"})
+        self.assertEqual(got.id, "sb_new")
+
+    def test_exec_sends_argv_and_reads_exitcode(self):
+        got = self.boxes.exec("sb_1", argv=["echo", "hi"], timeout_sec=5)
+        self.assertEqual(SEEN[0]["path"], "/v1/sandboxes/sb_1/exec")
+        self.assertEqual(SEEN[0]["body"], {"argv": ["echo", "hi"], "timeoutSec": 5})
+        self.assertTrue(got.ok)
+        self.assertEqual(got.stdout, "hi\n")
+
+    def test_exec_refuses_to_send_nothing(self):
+        with self.assertRaises(ValueError):
+            self.boxes.exec("sb_1")
+        self.assertEqual(SEEN, [], "a call with no command must not reach the network")
+
+    def test_fs_round_trip(self):
+        self.assertEqual(self.boxes.read("sb_1", "/etc/hosts"), "127.0.0.1 localhost")
+        self.boxes.write("sb_1", "/tmp/x", "body")
+        self.assertEqual(SEEN[-1]["body"], {"path": "/tmp/x", "content": "body"})
+
+    def test_a_refusal_arrives_as_SandboxError_with_its_status(self):
+        anon = Sandboxes(api_key="", base_url=self.base)
+        with self.assertRaises(SandboxError) as e:
+            anon.list()
+        self.assertEqual(e.exception.status, 403)
+        self.assertIn("X-Org-Id", e.exception.message)
+
+
+if __name__ == "__main__":
+    unittest.main()