diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml index 484f416c..3e53c5c2 100644 --- a/.github/workflows/python.yml +++ b/.github/workflows/python.yml @@ -28,12 +28,14 @@ jobs: with: python-version: "3.12" - name: Install tooling - run: pip install ruff==0.15.20 pytest==8.4.2 && pip install -e sdk/python -e sdk/openai -e sdk/langchain + run: pip install ruff==0.15.20 pytest==8.4.2 mypy==1.19.1 && pip install -e sdk/python -e sdk/openai -e sdk/langchain - name: Lint run: ruff format --check . && ruff check . - name: Test cocoonsandbox - run: cd sdk/python && pytest -q + run: cd sdk/python && python -m pytest -q + - name: Type check cocoonsandbox + run: cd sdk/python && python -m mypy - name: Test openai adapter - run: cd sdk/openai && pytest -q + run: cd sdk/openai && python -m pytest -q - name: Test langchain adapter - run: cd sdk/langchain && pytest -q + run: cd sdk/langchain && python -m pytest -q diff --git a/.github/workflows/sandboxd.yml b/.github/workflows/sandboxd.yml index 5389f8d3..cb05e862 100644 --- a/.github/workflows/sandboxd.yml +++ b/.github/workflows/sandboxd.yml @@ -9,6 +9,7 @@ on: paths: - "sandboxd/**" - "sdk/go/**" + - "sdk/python/**" - "mcp/**" - "e2e/**" - "protocol/**" @@ -17,6 +18,7 @@ on: paths: - "sandboxd/**" - "sdk/go/**" + - "sdk/python/**" - "mcp/**" - "e2e/**" - "protocol/**" @@ -48,3 +50,18 @@ jobs: - name: Test run: make go-test + + - name: Set up Python + uses: actions/setup-python@v7.0.0 + with: + python-version: "3.12" + + - name: Test TLS edge with Caddy + working-directory: e2e + env: + CADDY_BIN: ${{ runner.temp }}/caddy + run: | + curl -fsSL --retry 3 https://github.com/caddyserver/caddy/releases/download/v2.11.4/caddy_2.11.4_linux_amd64.tar.gz -o "$RUNNER_TEMP/caddy.tar.gz" + printf '%s %s\n' '8220d1f013b6f27510247b2360c9e0ca9f018feebd82515f07635318b34ff9777ccc8fd0b6e6f2486ce3a33fe389fbb7db12d05baa474f4587509fb4f5ebf1c9' "$RUNNER_TEMP/caddy.tar.gz" | sha512sum -c - + tar -xzf "$RUNNER_TEMP/caddy.tar.gz" -C "$RUNNER_TEMP" caddy + GOWORK=off go test -race -count=1 -run TestCaddyTLSCluster -timeout 60s -v . diff --git a/docs/cluster.md b/docs/cluster.md index 2b935e7e..8ded4304 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -54,6 +54,10 @@ A claim always enters at whatever node the client dialed: The data plane is never proxied between nodes: the claim response carries `owner_addr` and all sandbox traffic dials the owner directly. +Behind a TLS edge, set each node's `client_advertise` to its own public HTTPS +origin. Owner, redirect, and peer responses use these origins; gossip, peer +probes, and preview forwarding retain the internal `advertise_addr`. See +[TLS termination](deploy.md#tls-for-sdk-clients) for the edge setup. Node death is honest: a dead node's sandboxes die with it (memory state is node-local by design). SWIM detects the death and peers stop redirecting to diff --git a/docs/deploy.md b/docs/deploy.md index 8be9e063..6ce509c8 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -99,7 +99,8 @@ sandboxd reads one JSON file (`-config`, default | `restore_mode` | unset | clone and wake-restore memory mode: `copy`, `ondemand`, or `mmap`; use `mmap` for dense pools | | `no_direct_io` | false | use buffered writable disks for Cloud Hypervisor cold boots and clones; recommended for dense ephemeral pools to avoid direct-I/O CoW journal contention | | `no_balloon` | false | boot pool and template VMs without the virtio-balloon (cocoon otherwise returns 25% of guest memory to the host); clones inherit it from the golden. A guest that thrashes before deflate-on-OOM fires — a 16G build tier running a large typecheck — needs its whole memory | -| `advertise_addr` | = `listen` | the host:port clients reach this node at; returned as a claim's owner address and gossiped to peers. Must be routable when `listen` is a wildcard; a node with `mesh` set refuses to load while it names an unspecified host | +| `advertise_addr` | = `listen` | internal HTTP host:port for peer traffic and preview forwarding; also used by clients when `client_advertise` is unset. Must be routable when `listen` is a wildcard; a node with `mesh` set refuses to load while it names an unspecified host | +| `client_advertise` | unset | client-facing HTTP(S) origin for this node, e.g. `https://node-a.sandbox.example.com`; published in owner, redirect, and peer responses. No path, query, fragment, or userinfo | | `bridges` / `networks` | unset | egress-lane attachment: a list of host bridge devices, or a list of CNI conflist names. Mutually exclusive; with neither set the node serves only the no-network lane. A Linux bridge holds at most 1024 ports (kernel `BR_MAX_PORTS`), so an N-entry list raises the node's egress ceiling to N×1024 — VMs spread over the list by a stable hash of the VM name, so size it with headroom (the spread is statistical, not exact). `bridges` keeps the raw TAP-on-bridge attachment (taps in the root netns, no per-VM network namespace or CNI plugin execution); `networks` runs the CNI chain per VM. [Guarded egress](egress.md) on the egress lane (an egress-lane pool policy or any tenant policy) needs `bridges` and rejects a CNI network at load; none-lane pool policies ride the proxy on either | | `volumes` | unset | node-local catalog of operator-managed dataset images: `[ {"name":"imagenet","path":"/srv/datasets/imagenet.img","directio":"off","tenants":["acme"]}, {"name":"scratch-db","path":"/srv/datasets/scratch.img","writable":true} ]`. Names match `^[a-z][a-z0-9_-]{0,19}$` and cannot start with `cocoon-`; paths are absolute; `directio` is `on`, `off`, or `auto` and defaults to `off` for both read-only and writable entries. `tenants` is an optional access list: empty means every authenticated scope, while every listed name must exist in the node's `tenants` config; root always has access. `writable` (default `false`) lets a claim request `mode: "rw"` on that entry — see [Dataset volumes](#dataset-volumes). The catalog is intentionally not part of the cluster digest | | `secrets` | unset | node-side credentials the egress proxy injects by name: `[{"name": "gh", "header": "Authorization", "value_env": "GH_TOKEN"}]`. A pool or tenant rule references the name; the value comes from the environment, never this file. See [egress](egress.md) | @@ -458,6 +459,84 @@ Add `VOLUME_RW_IMAGE=/srv/datasets/scratch.img` (a second, writable image) to also run the writable leg: a durable write across release, second-writer exclusion, and a clean read-only claim afterward. +## TLS for SDK clients + +sandboxd serves plain HTTP behind a TLS-terminating proxy. Configure a stable +client origin on **every node** reachable through that proxy: + +```json +{ + "listen": ":7777", + "advertise_addr": "node-a.internal:7777", + "client_advertise": "https://node-a.sandbox.example.com" +} +``` + +The mesh gossips both addresses. Client-facing owner replies, claim/template/ +checkpoint redirects, and peer discovery use `client_advertise`. Checkpoint +probing, healing, deletion broadcasts, and preview forwarding continue to use +`advertise_addr` over internal HTTP. When `client_advertise` is unset, direct +HTTP deployments keep their existing address behavior. Configure all cluster +members before using external clients; a member without a client origin still +advertises its internal address to clients. Internal SDK users must also be +able to reach the configured client origins. + +One proxy can serve the whole cluster, but each owner origin must route to +one particular node. An entry load balancer may choose any node for the initial +claim; a shared random-balancing owner origin cannot route later agent and +release requests to the owning node. Unlike Preview URLs, the SDK API does not +forward arbitrary sandbox requests between nodes. + +Caddy reference configuration (node B uses the corresponding hostname and +internal upstream): + +```caddyfile +node-a.sandbox.example.com { + reverse_proxy node-a.internal:7777 { + transport http { + versions 1.1 + } + } +} + +node-b.sandbox.example.com { + reverse_proxy node-b.internal:7777 { + transport http { + versions 1.1 + } + } +} +``` + +For a private development CA, add `tls internal` to each site and provide +Caddy's root certificate to both SDKs using the +[TLS client settings](sdk.md#https-endpoints). For public DNS names, Caddy can +manage the certificates. Keep the upstream listeners and mesh private. + +The edge must pass HTTP/1.1 `Connection: Upgrade`, `Upgrade: silkd`, and the +101 response, then relay both byte streams without response buffering. A +WebSocket-only upgrade allowlist is insufficient. Agent connections must not +negotiate HTTP/2. Configure stream/idle timeouts to exceed the longest relay; +Caddy's default stream timeout is unlimited. Configuration reloads may close +active streams; set `stream_close_delay` when reloads need a drain window. +The Upgrade tunnel does not need `flush_interval -1`. + +The pinned Caddy integration runs both SDKs against two real sandboxd HTTP +handlers and relays (with a fake VM/guest), with an unreachable internal owner +address. It checks redirects, lookup, exec, port forwarding, half-close, +release, and certificate rejection: + +```bash +cd e2e +CADDY_BIN=/path/to/caddy GOWORK=off go test -race -run TestCaddyTLSCluster -v . +``` + +For hardware acceptance, run the same SDK sequence from outside the node +network, including a guest HTTP server through `proxy_port`, and confirm an +A-to-B claim never dials B's internal address. A client-side TLS bridge alone +does not translate returned owners or redirects; use it only when every +returned endpoint is deliberately mapped through a local bridge. + ## Preview URLs `preview_listen` starts a second HTTP server that serves a sandbox's guest diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index 859c23f7..533271e9 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -550,8 +550,9 @@ silkd error; 504 when the command outlives `timeout_seconds` or the request. A h ## GET /v1/sandboxes/{id}/owner Auth: the sandbox's own token. Answers `{"owner_addr": "host:port"}` when -this node owns the sandbox, 404 otherwise. Used by the SDK's `Lookup` -scatter. +this node owns the sandbox, 404 otherwise. With `client_advertise` configured, +`owner_addr` is that node's full HTTP(S) origin instead; the same contract applies +to claim and fork responses. Used by the SDK's `Lookup` scatter. ## GET /v1/info diff --git a/docs/sdk-python.md b/docs/sdk-python.md index 40a85940..7ab7dae9 100644 --- a/docs/sdk-python.md +++ b/docs/sdk-python.md @@ -462,7 +462,8 @@ zero. cwd that is not a directory, no exec bit) / `not_found` / `unimplemented` / `internal` - `ExitError(code, stderr, stdout)` — non-zero exit from `exec` -- `ProtocolError` — broken stream (EOF, oversized or undecodable frame) +- `ProtocolError` — a failed relay dial or TLS handshake, or a broken stream + (EOF, oversized or undecodable frame) ```python try: diff --git a/docs/sdk.md b/docs/sdk.md index 7125ed50..5abb0a32 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -109,7 +109,59 @@ client, err := sandbox.Connect("10.0.0.5:7777", `Info` answer it 403). On a cluster every node shares the same root token and the same tenants set. - `WithHTTPClient(client)` — replace the control-plane HTTP client when the - caller needs a custom transport, proxy, or timeout. + caller needs a custom transport, proxy, or timeout. An `http.Transport`'s + TLS configuration also supplies the agent relay's certificate settings; + HTTP proxies and custom dialers apply only to control requests. +- `WithTLSConfig(config)` — set certificate verification for both HTTPS + requests and agent relays. With `WithHTTPClient`, its transport must be an + `*http.Transport`; the SDK clones it rather than changing the caller's client. + +### HTTPS endpoints + +Both SDKs accept `host:port` (plain HTTP), `http://host[:port]`, or +`https://host[:port]`. The default ports are 80 and 443. IPv6 hosts use brackets. +Endpoints are origins: no userinfo, path prefix, query, or fragment. + +For a public certificate, only the address changes: + +```go +client, err := sandbox.Connect("https://node-a.sandbox.example.com", + sandbox.WithAPIToken(os.Getenv("SANDBOXD_TOKEN"))) +``` + +For a private CA, load it into an `x509.CertPool` and pass +`sandbox.WithTLSConfig(&tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12})`. +The same trust configuration covers control requests and every agent relay. +Certificate and hostname verification are enabled by default. + +Python uses a standard `ssl.SSLContext`: + +```python +import os +import ssl +from cocoonsandbox import Client + +client = Client( + "https://node-a.sandbox.example.com", + api_token=os.environ["SANDBOXD_TOKEN"], + ssl_context=ssl.create_default_context(cafile="edge-ca.pem"), +) +with client.new("rt:24.04") as sb: + assert sb.run(["true"]) == 0 +``` + +Omit `ssl_context` for system trust. The SDK configures that context for +HTTP/1.1; use a dedicated context if another caller requires a different ALPN. +TLS handshakes share the existing dial timeout/cancellation budget. The agent +connection then uses HTTP/1.1 `Upgrade: silkd` and remains a bidirectional +stream; the guest protocol and port-forwarding frames do not change. + +An explicit scheme in an owner, redirect, peer, or `Attach` address wins; +a bare address inherits the entry client's scheme. Trust settings are shared +across these connections. The SDK does not translate private addresses to +public names: configure each node's `client_advertise` as described in +[TLS deployment](deploy.md#tls-for-sdk-clients). Persist the complete owner URL +with the sandbox ID and token when using `Attach` in another process. ### Connecting to clusters diff --git a/e2e/tls_client.py b/e2e/tls_client.py new file mode 100644 index 00000000..0fb3e5c4 --- /dev/null +++ b/e2e/tls_client.py @@ -0,0 +1,52 @@ +"""Exercise both SDK planes through the TLS cluster fixture.""" + +import socket +import ssl +import sys + +from cocoonsandbox import APIError, Client, ProtocolError + + +def main() -> None: + entry, owner, ca = sys.argv[1:] + context = ssl.create_default_context(cafile=ca) + client = Client(entry, api_token="node-token", ssl_context=context) + with client.new("rt:24.04") as sb: + assert sb.owner == owner, sb.owner + assert sb.exec("echo", "tls") == "tls\n" + assert client.lookup(sb.id, sb.token).owner == owner + assert client.attach(owner.removeprefix("https://"), sb.id, sb.token).exec("echo", "bare") == "bare\n" + with sb.dial_port(5000) as port: + port.send(b"tail") + port.close_write() + output = b"" + while chunk := port.recv(): + output += chunk + assert output == b"tail", output + with ( + sb.proxy_port("127.0.0.1:0", 5000) as proxy, + socket.create_connection(proxy.getsockname(), timeout=5) as conn, + ): + conn.sendall(b"proxy") + conn.shutdown(socket.SHUT_WR) + output = b"" + while chunk := conn.recv(4096): + output += chunk + assert output == b"proxy", output + untrusted = Client(entry, api_token="node-token") + try: + untrusted.info() + except APIError: + pass + else: + raise AssertionError("control request accepted an untrusted certificate") + try: + untrusted.attach(owner, sb.id, sb.token).exec("echo", "untrusted") + except ProtocolError: + pass + else: + raise AssertionError("relay accepted an untrusted certificate") + + +if __name__ == "__main__": + main() diff --git a/e2e/tls_test.go b/e2e/tls_test.go new file mode 100644 index 00000000..02d4281a --- /dev/null +++ b/e2e/tls_test.go @@ -0,0 +1,210 @@ +package e2e + +import ( + "bytes" + "cmp" + "crypto/tls" + "crypto/x509" + "encoding/json/v2" + "encoding/pem" + "io" + "net" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "testing" + "time" + + "github.com/cocoonstack/sandbox/sandboxd/server" + sandbox "github.com/cocoonstack/sandbox/sdk/go" +) + +func TestCaddyTLSCluster(t *testing.T) { + bin := os.Getenv("CADDY_BIN") + if bin == "" { + t.Skip("set CADDY_BIN to run the TLS edge integration") + } + dir := t.TempDir() + cert, key, roots := edgeCertificate(t, dir) + la, lb := reserveEdgePort(t), reserveEdgePort(t) + a, b := la.Addr().String(), lb.Addr().String() + owner := "https://" + b + entry := "https://" + a + st := startStack(t, "node-token") + ownerServer := server.New(st.token, nil, owner, st.mgr, st.eng.real, nil, nil, nil, nil) + backend := httptest.NewServer(ownerServer.Handler()) + t.Cleanup(func() { backend.Close(); ownerServer.CloseRelays() }) + origin := startStack(t, "node-token") + placer := &edgePlacer{internal: "owner.invalid:7777", public: owner} + entryServer := server.New(origin.token, nil, entry, origin.mgr, origin.eng.real, placer, nil, nil, nil) + front := httptest.NewServer(entryServer.Handler()) + t.Cleanup(func() { front.Close(); entryServer.CloseRelays() }) + servers := map[string]any{} + for listen, upstream := range map[string]string{a: front.Listener.Addr().String(), b: backend.Listener.Addr().String()} { + servers[listen] = map[string]any{ + "listen": []string{listen}, "tls_connection_policies": []any{map[string]any{}}, + "automatic_https": map[string]bool{"disable": true}, + "routes": []any{map[string]any{"handle": []any{map[string]any{ + "handler": "reverse_proxy", "upstreams": []any{map[string]string{"dial": upstream}}, + "transport": map[string]any{"protocol": "http", "versions": []string{"1.1"}}, + }}}}, + } + } + config, err := json.Marshal(map[string]any{ + "admin": map[string]bool{"disabled": true}, + "apps": map[string]any{ + "tls": map[string]any{"certificates": map[string]any{"load_files": []any{map[string]string{"certificate": cert, "key": key}}}}, + "http": map[string]any{"servers": servers}, + }, + }) + if err != nil { + t.Fatal(err) + } + configPath := filepath.Join(dir, "caddy.json") + if err = os.WriteFile(configPath, config, 0o600); err != nil { + t.Fatal(err) + } + var logs bytes.Buffer + cmd := exec.CommandContext(t.Context(), bin, "run", "--config", configPath) + cmd.Env = append(os.Environ(), "XDG_DATA_HOME="+dir, "XDG_CONFIG_HOME="+dir) + cmd.Stdout, cmd.Stderr = &logs, &logs + _, _ = la.Close(), lb.Close() + if err = cmd.Start(); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = cmd.Process.Kill() + _ = cmd.Wait() + if t.Failed() { + t.Log(logs.String()) + } + }) + client, err := sandbox.Connect(entry, sandbox.WithAPIToken(st.token), sandbox.WithTLSConfig(&tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12})) + if err != nil { + t.Fatal(err) + } + waitFor(t, func() bool { _, err := client.Info(t.Context()); return err == nil }) + t.Run("Go", func(t *testing.T) { checkTLSClient(t, client, owner) }) + t.Run("Python", func(t *testing.T) { + python := cmp.Or(os.Getenv("PYTHON_BIN"), "python3") + cmd := exec.CommandContext(t.Context(), python, "tls_client.py", entry, owner, cert) + cmd.Env = append(os.Environ(), "PYTHONPATH=../sdk/python") + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("Python TLS flow: %v\n%s", err, out) + } + }) +} + +func checkTLSClient(t *testing.T, client *sandbox.Client, owner string) { + t.Helper() + sb, err := client.New(t.Context(), "rt:24.04") + if err != nil { + t.Fatal(err) + } + defer sb.Close() + if sb.Owner() != owner { + t.Fatalf("owner %q, want %q", sb.Owner(), owner) + } + if out, execErr := sb.Exec(t.Context(), "echo", "tls"); execErr != nil || out != "tls\n" { + t.Fatalf("exec = %q, %v", out, execErr) + } + recovered, err := client.Lookup(t.Context(), sb.ID, sb.Token()) + if err != nil || recovered.Owner() != owner { + t.Fatalf("lookup = %v, %v", recovered, err) + } + pc, err := sb.DialPort(t.Context(), 5000) + if err != nil { + t.Fatal(err) + } + defer pc.Close() + if _, err = pc.Write([]byte("tail")); err != nil { + t.Fatal(err) + } + if err = pc.CloseWrite(); err != nil { + t.Fatal(err) + } + if out, readErr := io.ReadAll(pc); readErr != nil || string(out) != "tail" { + t.Fatalf("half-close tail = %q, %v", out, readErr) + } + l, err := sb.ProxyPort(t.Context(), "127.0.0.1:0", 5000) + if err != nil { + t.Fatal(err) + } + defer l.Close() + conn, err := net.DialTimeout("tcp", l.Addr().String(), 5*time.Second) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(5 * time.Second)) + _, _ = io.WriteString(conn, "proxy") + _ = conn.(*net.TCPConn).CloseWrite() + if out, readErr := io.ReadAll(conn); readErr != nil || string(out) != "proxy" { + t.Fatalf("proxy = %q, %v", out, readErr) + } + children, err := sb.Fork(t.Context(), 1, time.Minute) + if err != nil { + t.Fatal(err) + } + defer children[0].Close() + if children[0].Owner() != owner { + t.Fatalf("fork owner = %q", children[0].Owner()) + } + if err := sb.Close(); err != nil { + t.Fatal(err) + } +} + +func edgeCertificate(t *testing.T, dir string) (string, string, *x509.CertPool) { + t.Helper() + fixture := httptest.NewTLSServer(nil) + defer fixture.Close() + cert := fixture.TLS.Certificates[0] + key, err := x509.MarshalPKCS8PrivateKey(cert.PrivateKey) + if err != nil { + t.Fatal(err) + } + certPath, keyPath := filepath.Join(dir, "cert.pem"), filepath.Join(dir, "key.pem") + for path, block := range map[string]*pem.Block{ + certPath: {Type: "CERTIFICATE", Bytes: cert.Certificate[0]}, + keyPath: {Type: "PRIVATE KEY", Bytes: key}, + } { + if err := os.WriteFile(path, pem.EncodeToMemory(block), 0o600); err != nil { + t.Fatal(err) + } + } + roots := x509.NewCertPool() + roots.AddCert(fixture.Certificate()) + return certPath, keyPath, roots +} + +func reserveEdgePort(t *testing.T) net.Listener { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + return l +} + +type edgePlacer struct { + internal string + public string +} + +func (p *edgePlacer) ClientAddr(addr string) string { + if addr == p.internal { + return p.public + } + return addr +} + +func (p *edgePlacer) Candidates(string) []string { return []string{p.internal} } +func (p *edgePlacer) VolumeCandidates(string, []string) []string { return nil } +func (p *edgePlacer) TemplateOwners(string) []string { return nil } +func (p *edgePlacer) VolumeOwners([]string) []string { return nil } +func (p *edgePlacer) TemplateVolumeOwners(string, []string) []string { return nil } +func (p *edgePlacer) VolumeHolders() map[string]int { return nil } +func (p *edgePlacer) PeerAddrs() []string { return []string{p.internal} } +func (p *edgePlacer) ConfigMismatches() int { return 0 } diff --git a/sandboxd/config/client_endpoint_test.go b/sandboxd/config/client_endpoint_test.go new file mode 100644 index 00000000..c407949c --- /dev/null +++ b/sandboxd/config/client_endpoint_test.go @@ -0,0 +1,22 @@ +package config + +import "testing" + +func TestClientAdvertise(t *testing.T) { + for _, addr := range []string{"", "https://node.example", "https://node.example:8443/", "http://[::1]:7777"} { + t.Run(addr, func(t *testing.T) { + cfg := Config{ClientAdvertise: addr} + if err := cfg.validateClientAdvertise(); err != nil { + t.Fatal(err) + } + }) + } + for _, addr := range []string{"node:7777", "https://0.0.0.0", "https://[::]", "https://", "ftp://node", "https://u:p@node", "https://node/path", "https://node?x=1", "https://node#", "https://node#x", "https://node:0", "https://node:65536"} { + t.Run(addr, func(t *testing.T) { + cfg := Config{ClientAdvertise: addr} + if err := cfg.validateClientAdvertise(); err == nil { + t.Fatalf("accepted %q", addr) + } + }) + } +} diff --git a/sandboxd/config/config.go b/sandboxd/config/config.go index 0d4c6824..df0e2b4f 100644 --- a/sandboxd/config/config.go +++ b/sandboxd/config/config.go @@ -13,6 +13,7 @@ import ( "fmt" "net" "net/netip" + "net/url" "os" "path/filepath" "runtime" @@ -41,22 +42,16 @@ type PoolSpec struct { types.PoolKey Warm int `json:"warm"` - // WarmMax, when >0, lets the warm target rise from Warm toward it under demand. WarmMax int `json:"warm_max,omitzero"` - // Egress is this pool's allow-list, intersected with the tenant's; nil denies all egress. Egress *egress.Policy `json:"egress,omitempty"` - // Warmup runs in the golden VM before its snapshot and again in each clone after restore. Warmup []string `json:"warmup,omitempty"` - // IdleHibernateSeconds, when >0, hibernates idle claims after that many seconds. IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitzero"` - // ArchiveAfterSeconds, when >0, archives a hibernated claim; must exceed IdleHibernateSeconds. ArchiveAfterSeconds int `json:"archive_after_seconds,omitzero"` - // ArchiveDeleteAfterSeconds, when >0, purges the checkpoint that long after archiving. ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitzero"` warmSet bool @@ -124,7 +119,6 @@ type TenantSpec struct { Token string `json:"token"` //nolint:gosec // config field, not a hardcoded credential MaxClaims int `json:"max_claims,omitzero"` - // Egress is the tenant's allow-list (see PoolSpec.Egress). Egress *egress.Policy `json:"egress,omitempty"` } @@ -184,82 +178,58 @@ type Config struct { DataDir string `json:"data_dir"` CocoonBin string `json:"cocoon_bin"` - // AdvertiseAddr is the host:port the data plane reaches this node at; defaults to Listen. - AdvertiseAddr string `json:"advertise_addr,omitempty"` + AdvertiseAddr string `json:"advertise_addr,omitempty"` + ClientAdvertise string `json:"client_advertise,omitempty"` - // Bridges shards egress-lane VMs over host bridges; their taps stay lockable in the root netns. Bridges []string `json:"bridges,omitempty"` - // Networks shards egress-lane VMs over CNI conflists; a bridge holds at most 1024 ports. Networks []string `json:"networks,omitempty"` - // RestoreMode is cocoon's --restore-mode; an older CH silently eager-copies an mmap restore. RestoreMode types.RestoreMode `json:"restore_mode,omitempty"` - // NoDirectIO enables buffered writable disks for cold boots and clones. NoDirectIO bool `json:"no_direct_io,omitzero"` - // NoBalloon boots VMs without the virtio-balloon, so a guest keeps its whole memory. NoBalloon bool `json:"no_balloon,omitzero"` - // APIToken, when set, guards claim and info. APIToken string `json:"api_token,omitempty"` //nolint:gosec // config field, not a hardcoded credential - // Tenants adds per-tenant bearer tokens next to APIToken. Tenants []TenantSpec `json:"tenants,omitempty"` - // Secrets registers node-side credentials; values come from value_env, never this file. Secrets []egress.SecretSpec `json:"secrets,omitempty"` - // IdleHibernateSeconds is the idle policy for unpooled claims; per-pool settings override. IdleHibernateSeconds int `json:"idle_hibernate_seconds,omitzero"` - // ArchiveAfterSeconds and ArchiveDeleteAfterSeconds are the archive policy for unpooled keys. ArchiveAfterSeconds int `json:"archive_after_seconds,omitzero"` ArchiveDeleteAfterSeconds int `json:"archive_delete_after_seconds,omitzero"` - PreviewListen string `json:"preview_listen,omitempty"` - PreviewSecret string `json:"preview_secret,omitempty"` //nolint:gosec // config field, not a hardcoded credential - // PreviewAdvertise is the browser-facing base, shareable fleet-wide behind one proxy. + PreviewListen string `json:"preview_listen,omitempty"` + PreviewSecret string `json:"preview_secret,omitempty"` //nolint:gosec // config field, not a hardcoded credential PreviewAdvertise string `json:"preview_advertise,omitempty"` - // CheckpointDir is where checkpoints live; defaults to /checkpoints. CheckpointDir string `json:"checkpoint_dir,omitempty"` - // CheckpointStore selects the checkpoint backend; absent means the dir backend. CheckpointStore *StoreConfig `json:"checkpoint_store,omitempty"` - // CheckpointPeerHeal lets a node pull a checkpoint it lacks from a peer; off by default. CheckpointPeerHeal bool `json:"checkpoint_peer_heal,omitzero"` - // EgressInternalAllow re-admits CIDRs through the proxy's SSRF guard, node-wide. EgressInternalAllow []string `json:"egress_internal_allow,omitempty"` - // CheckpointTTLHours ages out checkpoints; 0 keeps them forever. CheckpointTTLHours int `json:"checkpoint_ttl_hours,omitzero"` - // MaxClaims caps live claims node-wide; 0 means unlimited. MaxClaims int `json:"max_claims,omitzero"` - // AuditLog, when true, appends relayed request ops, never payloads, to audit.jsonl. AuditLog bool `json:"audit_log,omitzero"` - // MaxForkCount caps children per fork call; each child is a full-RAM VM. MaxForkCount int `json:"max_fork_count,omitzero"` - // Volumes is the node-local catalog of operator-managed dataset images. Volumes []VolumeSpec `json:"volumes,omitempty"` - // RefillConcurrency caps concurrent VM provisioning node-wide; 0 auto-scales with CPUs. RefillConcurrency int `json:"refill_concurrency,omitzero"` - // ReleaseDelaySeconds, when >0, parks a released VM in the removal queue for that long instead of removing it inline. ReleaseDelaySeconds int `json:"release_delay_seconds,omitzero"` - // Mesh, when set, joins this node to a memberlist cluster; nil is a mesh of one. Mesh *MeshConfig `json:"mesh,omitempty"` - // EgressCA provisions HTTPS interception; required for any intercept rule. EgressCA *EgressCAConfig `json:"egress_ca,omitempty"` Pools []PoolSpec `json:"pools"` @@ -323,6 +293,9 @@ func (c *Config) applyDefaults() { } func (c *Config) validate() error { + if err := c.validateClientAdvertise(); err != nil { + return err + } for _, field := range []struct { name string value int @@ -423,7 +396,31 @@ func (c *Config) validateVolumes() error { return nil } -// validateMesh fails at load what would otherwise only surface at startMesh. +func (c *Config) validateClientAdvertise() error { + if c.ClientAdvertise == "" { + return nil + } + u, err := url.Parse(c.ClientAdvertise) + if err != nil { + return fmt.Errorf("client_advertise: %w", err) + } + if (u.Scheme != "http" && u.Scheme != "https") || u.Hostname() == "" || u.User != nil || + (u.Path != "" && u.Path != "/") || strings.ContainsAny(c.ClientAdvertise, "?#") { + return fmt.Errorf("client_advertise must be an http or https origin") + } + if ip, _ := netip.ParseAddr(u.Hostname()); ip.IsUnspecified() { + return fmt.Errorf("client_advertise must name a routable host") + } + if port := u.Port(); port != "" { + n, err := strconv.Atoi(port) + if err != nil || n < 1 || n > 65535 { + return fmt.Errorf("client_advertise port must be between 1 and 65535") + } + } + c.ClientAdvertise = strings.TrimSuffix(c.ClientAdvertise, "/") + return nil +} + func (c *Config) validateMesh() error { if c.Mesh == nil { return nil @@ -487,7 +484,6 @@ func (c *Config) validateSecrets() (map[string]struct{}, error) { return names, nil } -// validateAttachment checks the egress-lane attachment; a repeated shard fills one bridge. func (c *Config) validateAttachment() error { if len(c.Bridges) > 0 && len(c.Networks) > 0 { return fmt.Errorf("bridges and networks are mutually exclusive") @@ -498,7 +494,6 @@ func (c *Config) validateAttachment() error { return validateShards(c.Networks, "networks", "conflist") } -// validateEgressAllow rejects a malformed prefix at load, not at a later refused dial. func (c *Config) validateEgressAllow() error { for _, cidr := range c.EgressInternalAllow { if _, err := netip.ParsePrefix(cidr); err != nil { @@ -546,7 +541,6 @@ func Load(path string) (*Config, error) { return nil, fmt.Errorf("read config: %w", err) } cfg := &Config{} - // hand-edited file: a typo must fail load, not silently change policy. if err := utils.DecodeStrictJSON(raw, cfg); err != nil { return nil, fmt.Errorf("parse config: %w", err) } @@ -557,7 +551,6 @@ func Load(path string) (*Config, error) { return cfg, nil } -// validatePolicy checks the rules and secret refs; a nil policy is deny-all. func validatePolicy(p *egress.Policy, secrets map[string]struct{}) error { if p == nil { return nil diff --git a/sandboxd/main.go b/sandboxd/main.go index 8bb9ef41..d77f8d41 100644 --- a/sandboxd/main.go +++ b/sandboxd/main.go @@ -1,6 +1,4 @@ -// sandboxd is the per-node sandbox control plane: it keeps warm pools of -// claim-ready microVMs, serves claims over HTTP, and relays the silkd data -// plane between clients and guests. +// sandboxd is the per-node sandbox control plane: it keeps warm pools of claim-ready microVMs, serves claims over HTTP, and relays the silkd data plane between clients and guests. package main import ( @@ -32,13 +30,11 @@ import ( ) const ( - shutdownGrace = 5 * time.Second - gossipInterval = time.Second - // Slowloris protection; ReadTimeout/WriteTimeout must stay zero for streaming relays. + shutdownGrace = 5 * time.Second + gossipInterval = time.Second readHeaderTimeout = 5 * time.Second ) -// Stamped via -ldflags at release; devel builds fall back to the VCS revision. var version = "devel" func main() { @@ -92,7 +88,6 @@ func main() { ctx, stop := signal.NotifyContext(ctx, os.Interrupt, syscall.SIGTERM) defer stop() - // a node that cannot reconcile cannot trust its view of local VMs. if err := mgr.Reconcile(ctx); err != nil { logger.Fatalf(ctx, err, "reconcile") } @@ -133,7 +128,7 @@ func main() { if cfg.PreviewListen != "" { preview = server.NewPreviewServer(cfg.PreviewSecret, cfg.PreviewAdvertise, cfg.AdvertiseAddr, mgr) } - srv := server.New(cfg.APIToken, cfg.Tenants, cfg.AdvertiseAddr, mgr, eng, placer, prober, probeKey, preview) + srv := server.New(cfg.APIToken, cfg.Tenants, cmp.Or(cfg.ClientAdvertise, cfg.AdvertiseAddr), mgr, eng, placer, prober, probeKey, preview) httpSrv := &http.Server{ Addr: cfg.Listen, Handler: srv.Handler(), @@ -157,7 +152,6 @@ func main() { drained := make(chan struct{}) context.AfterFunc(ctx, func() { defer close(drained) - // must outlive the canceled signal ctx to bound the drain. sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), shutdownGrace) defer cancel() _ = httpSrv.Shutdown(sctx) @@ -169,7 +163,6 @@ func main() { logger.Fatalf(ctx, err, "serve") } <-drained - // a detached recommit may not have converged; leave disk matching memory. if err := mgr.FlushClaims(); err != nil { logger.Error(ctx, err, "flush claims") } @@ -197,7 +190,7 @@ func startMesh(ctx context.Context, cfg *config.Config, mgr *pool.Manager) (*mes if err != nil { return nil, err } - // publish the config digest before Join, so the first gossip carries it. + msh.SetSelfClientAddr(cfg.ClientAdvertise) msh.SetSelfDigest(cfg.ClusterDigest(mgr.EgressCAFingerprint())) msh.UpdateSelf(ctx, mgr.WarmCounts(), mgr.TemplateHashes(), mgr.VolumeNames()) if err := msh.Join(mc.Join); err != nil { diff --git a/sandboxd/mesh/client_endpoint_test.go b/sandboxd/mesh/client_endpoint_test.go new file mode 100644 index 00000000..94d01860 --- /dev/null +++ b/sandboxd/mesh/client_endpoint_test.go @@ -0,0 +1,34 @@ +package mesh + +import ( + "slices" + "testing" +) + +func TestClientAddressesDoNotReplaceInternalAddresses(t *testing.T) { + m := newBoundMesh(t, t.TempDir()) + m.SetSelfClientAddr("https://self.example") + if m.Members()[0].ClientAddr != "https://self.example" { + t.Fatal("self client address was not published") + } + peer := NodeState{NodeID: "peer", Addr: "peer:7777", ClientAddr: "https://peer.example", Epoch: 1, Pools: map[string]int{"key": 1}} + mergeStates(t, m, []NodeState{peer}) + if got := m.ClientAddr(peer.Addr); got != peer.ClientAddr { + t.Fatalf("client address %q, want %q", got, peer.ClientAddr) + } + if got := m.Candidates("key"); !slices.Equal(got, []string{peer.Addr}) { + t.Fatalf("placement addresses %v, want internal address", got) + } + if got := m.PeerAddrs(); !slices.Equal(got, []string{peer.Addr}) { + t.Fatalf("peer addresses %v, want internal address", got) + } + if got := m.ClientAddr("unknown:7777"); got != "unknown:7777" { + t.Fatalf("fallback address = %q", got) + } + peer.Epoch++ + peer.ClientAddr = "" + mergeStates(t, m, []NodeState{peer}) + if got := m.ClientAddr(peer.Addr); got != peer.Addr { + t.Fatalf("direct address = %q", got) + } +} diff --git a/sandboxd/mesh/mesh.go b/sandboxd/mesh/mesh.go index 31138dc3..47a98f75 100644 --- a/sandboxd/mesh/mesh.go +++ b/sandboxd/mesh/mesh.go @@ -1,11 +1,8 @@ -// Package mesh gossips per-node warm counts, promoted templates, and available -// volume names over a hashicorp/memberlist SWIM cluster. Gossip carries only -// placement hints — per-sandbox state stays node-local — so a stale view costs -// at most one failed redirect, never correctness. A single node with no seeds -// is a valid mesh of one. +// Package mesh gossips per-node warm counts, promoted templates, and available volume names over a hashicorp/memberlist SWIM cluster. package mesh import ( + "cmp" "context" "encoding/json" "fmt" @@ -21,22 +18,21 @@ import ( ) const ( - // leaveTimeout bounds the graceful-leave broadcast so a wedged network cannot hang the exit path. leaveTimeout = time.Second - // epochLease keeps the durable floor ahead of the counter, off the gossip tick. epochLease = 1024 ) // NodeState is one node's gossiped placement view; the higher Epoch wins a merge. type NodeState struct { - NodeID string `json:"node_id"` - Addr string `json:"addr"` // data-plane advertise address - Epoch uint64 `json:"epoch"` - Pools map[string]int `json:"pools"` // PoolKey hash → warm count - Templates []string `json:"templates,omitempty"` // promoted-template key hashes on disk - Volumes []string `json:"volumes,omitempty"` // locally available dataset names - Digest string `json:"digest,omitempty"` // cluster-invariant config digest + NodeID string `json:"node_id"` + Addr string `json:"addr"` // data-plane advertise address + ClientAddr string `json:"client_addr,omitempty"` + Epoch uint64 `json:"epoch"` + Pools map[string]int `json:"pools"` // PoolKey hash → warm count + Templates []string `json:"templates,omitempty"` // promoted-template key hashes on disk + Volumes []string `json:"volumes,omitempty"` // locally available dataset names + Digest string `json:"digest,omitempty"` // cluster-invariant config digest } type nodeMatch func(NodeState) bool @@ -45,10 +41,8 @@ type nodeMatch func(NodeState) bool type Mesh struct { ml *memberlist.Memberlist epochPath string - // ctx is the daemon's, for logging inside memberlist callbacks. - ctx context.Context + ctx context.Context - // updateMu serializes UpdateSelf and guards leased; one shared epoch read drops a payload. updateMu sync.Mutex leased uint64 @@ -61,7 +55,6 @@ type Mesh struct { // New starts a mesh member listening per cfg. func New(ctx context.Context, cfg *memberlist.Config, nodeID, selfAddr string, secretKey []byte, dataDir string) (*Mesh, error) { epochPath := filepath.Join(dataDir, "mesh-epoch") - // seed strictly above the persisted floor: merge's `>` rejects a tie with a stale copy. epoch := max(uint64(time.Now().UnixNano()), loadEpoch(epochPath)+1) //nolint:gosec // UnixNano is positive for current times m := &Mesh{ ctx: ctx, @@ -117,7 +110,6 @@ func (m *Mesh) UpdateSelf(ctx context.Context, pools map[string]int, templates, } epoch := m.self.Epoch + 1 m.mu.Unlock() - // the durable floor must stay above anything gossiped: memberlist publishes self at once. if epoch > m.leased { leased := epoch + epochLease if err := m.persistEpoch(leased); err != nil { @@ -143,6 +135,26 @@ func (m *Mesh) SetSelfDigest(digest string) { m.view[m.self.NodeID] = m.self } +// SetSelfClientAddr sets the client origin before Join publishes this node. +func (m *Mesh) SetSelfClientAddr(addr string) { + m.mu.Lock() + defer m.mu.Unlock() + m.self.ClientAddr = addr + m.view[m.self.NodeID] = m.self +} + +// ClientAddr resolves an internal address to its advertised client origin. +func (m *Mesh) ClientAddr(addr string) string { + m.mu.Lock() + defer m.mu.Unlock() + for _, st := range m.view { + if st.Addr == addr { + return cmp.Or(st.ClientAddr, addr) + } + } + return addr +} + // ConfigMismatches counts peers whose config digest differs from this node's. func (m *Mesh) ConfigMismatches() int { m.mu.Lock() @@ -282,7 +294,6 @@ func (m *Mesh) admit(nodeID string) { m.live[nodeID] = struct{}{} } -// forget drops a departed node from the placement view so redirects stop targeting it. func (m *Mesh) forget(nodeID string) { m.mu.Lock() defer m.mu.Unlock() @@ -292,7 +303,6 @@ func (m *Mesh) forget(nodeID string) { } } -// merge absorbs a peer's view, keeping the higher epoch per node; self is never overwritten. func (m *Mesh) merge(states []NodeState) { m.mu.Lock() defer m.mu.Unlock() @@ -307,7 +317,6 @@ func (m *Mesh) merge(states []NodeState) { if ok && st.Epoch <= cur.Epoch { continue } - // warn-only: refusing a divergent digest would partition a rolling credential rotation. if m.self.Digest != "" && st.Digest != "" && st.Digest != m.self.Digest && (!ok || cur.Digest != st.Digest) { log.WithFunc("mesh.merge").Warnf(m.ctx, "peer %s config digest %s differs from this node's %s: cluster-invariant config diverges (redirects may 401, interception may fail)", @@ -329,7 +338,6 @@ func containsAll(have, need []string) bool { var _ memberlist.Delegate = (*delegate)(nil) -// delegate carries this node's full view on each memberlist push/pull sync. type delegate Mesh func (d *delegate) NodeMeta(int) []byte { return nil } @@ -350,7 +358,6 @@ func (d *delegate) MergeRemoteState(buf []byte, _ bool) { var _ memberlist.EventDelegate = (*eventDelegate)(nil) -// eventDelegate tracks SWIM membership: admit on join, prune the view on leave. type eventDelegate Mesh func (e *eventDelegate) NotifyJoin(n *memberlist.Node) { (*Mesh)(e).admit(n.Name) } diff --git a/sandboxd/server/client_endpoint_test.go b/sandboxd/server/client_endpoint_test.go new file mode 100644 index 00000000..65aea71c --- /dev/null +++ b/sandboxd/server/client_endpoint_test.go @@ -0,0 +1,56 @@ +package server + +import ( + "encoding/json" + "net/http/httptest" + "strings" + "testing" + + "github.com/cocoonstack/sandbox/sandboxd/types" +) + +func TestClientAddressesInResponses(t *testing.T) { + for _, tt := range []struct{ method, path, body, field string }{ + {"POST", "/v1/claim", `{"template":"rt"}`, "redirect"}, + {"POST", "/v1/checkpoints/ck_1/claim", `{}`, "redirect"}, + {"DELETE", "/v1/templates?template=rt", "", "redirect"}, + {"GET", "/v1/peers", "", "peers"}, + {"GET", "/v1/info", "", "peers"}, + } { + t.Run(tt.path, func(t *testing.T) { + p := &clientPlacer{addrs: []string{"private:7777"}, owners: []string{"private:7777"}} + s := New("", nil, "https://self.example", &fakeManager{}, &fakeDialer{}, p, &fakeProber{owners: []string{"private:7777"}}, nil, nil) + r := httptest.NewRequest(tt.method, tt.path, strings.NewReader(tt.body)) + w := httptest.NewRecorder() + s.Handler().ServeHTTP(w, r) + var body map[string]json.RawMessage + if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if w.Code != 200 || string(body[tt.field]) != `["https://peer.example"]` { + t.Fatalf("response %d: %s", w.Code, w.Body.String()) + } + if p.addrs[0] != "private:7777" { + t.Fatal("rewrote the placement source slice") + } + }) + } +} + +func TestClientOwner(t *testing.T) { + s := New("", nil, "https://self.example", &fakeManager{}, &fakeDialer{}, nil, nil, nil, nil) + if got := s.claimResponse(&types.Sandbox{ID: "sb_1"}).OwnerAddr; got != "https://self.example" { + t.Fatalf("owner = %q", got) + } + r := httptest.NewRequest("GET", "/v1/sandboxes/sb_1/owner", nil) + r.Header.Set("Authorization", "Bearer tok") + w := httptest.NewRecorder() + s.Handler().ServeHTTP(w, r) + if !strings.Contains(w.Body.String(), `"owner_addr":"https://self.example"`) { + t.Fatalf("owner response: %s", w.Body.String()) + } +} + +type clientPlacer struct{ fakePlacer } + +func (p *clientPlacer) ClientAddr(string) string { return "https://peer.example" } diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index 11dc0275..37133f79 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -1,10 +1,4 @@ -// Package server exposes the v0 control plane: claim/release/info over -// HTTP/JSON, plus the data plane — an HTTP Upgrade relayed byte-for-byte -// between the client and the guest's silkd vsock port. -// -// The owning http.Server must keep ReadTimeout and WriteTimeout at zero: a -// cold-key claim legitimately blocks for the cold probe timeout and relays -// stream indefinitely. Use ReadHeaderTimeout for slowloris protection. +// Package server exposes the v0 control plane: claim/release/info over HTTP/JSON, plus the data plane — an HTTP Upgrade relayed byte-for-byte between the client and the guest's silkd vsock port. package server import ( @@ -34,7 +28,6 @@ const ( previewTTL = time.Hour ) -// an empty msg surfaces err.Error(); a fixed msg avoids echoing internals var poolErrHTTP = []struct { err error code int @@ -106,6 +99,7 @@ type Dialer interface { // Placer names peers for redirect placement and lists the mesh; nil on a single-node deployment. type Placer interface { + ClientAddr(addr string) string Candidates(keyHash string) []string VolumeCandidates(keyHash string, names []string) []string TemplateOwners(keyHash string) []string @@ -124,15 +118,14 @@ type CheckpointProber interface { // InfoResponse is the wire reply of GET /v1/info. type InfoResponse struct { - Pools []pool.PoolInfo `json:"pools"` - Claimed int `json:"claimed"` - Hibernated int `json:"hibernated"` - Archived int `json:"archived"` - Draining bool `json:"draining,omitzero"` - Peers []string `json:"peers,omitempty"` - // AtCapacity marks refill parked because the node refused another VM. - AtCapacity bool `json:"at_capacity,omitzero"` - AtCapacityReason string `json:"at_capacity_reason,omitempty"` + Pools []pool.PoolInfo `json:"pools"` + Claimed int `json:"claimed"` + Hibernated int `json:"hibernated"` + Archived int `json:"archived"` + Draining bool `json:"draining,omitzero"` + Peers []string `json:"peers,omitempty"` + AtCapacity bool `json:"at_capacity,omitzero"` + AtCapacityReason string `json:"at_capacity_reason,omitempty"` } // PoolUpdateRequest is the wire body of PUT /v1/pools; omitted pools are drained. @@ -160,7 +153,6 @@ type Server struct { // New returns a Server; an empty apiToken with no tenants leaves node-level endpoints open. func New(apiToken string, tenants []config.TenantSpec, advertise string, mgr Manager, dialer Dialer, placer Placer, prober CheckpointProber, probeKey []byte, preview *PreviewServer) *Server { - // an unspecified host names nothing a remote client can dial; an empty owner makes the SDK reuse the address it reached if host, _, err := net.SplitHostPort(advertise); err == nil { if ip, _ := netip.ParseAddr(host); host == "" || ip.IsUnspecified() { advertise = "" @@ -190,14 +182,12 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /v1/sandboxes/{id}/wake", s.handleSandboxVerb("wake", s.mgr.Wake)) mux.HandleFunc("GET /v1/sandboxes/{id}", s.requireRoot(s.handleSandbox)) mux.HandleFunc("GET /v1/sandboxes/{id}/stats", s.requireRoot(s.handleSandboxStats)) - // the source sandbox token rides in the body as the ownership proof mux.HandleFunc("POST /v1/sandboxes/{id}/fork", s.requireToken(s.handleFork)) mux.HandleFunc("POST /v1/sandboxes/{id}/promote", s.requireToken(s.handlePromote)) mux.HandleFunc("POST /v1/sandboxes/{id}/preview", s.requireToken(s.handlePreview)) mux.HandleFunc("POST /v1/sandboxes/{id}/checkpoint", s.requireToken(s.handleCheckpoint)) mux.HandleFunc("POST /v1/checkpoints/{id}/claim", s.requireToken(s.handleClaimCheckpoint)) mux.HandleFunc("GET /v1/checkpoints", s.requireToken(s.handleListCheckpoints)) - // GET streams a whole checkpoint with no tenant scoping; HEAD is the unauthenticated probe mux.HandleFunc("GET /v1/checkpoints/{id}/blob", s.requireRoot(s.handleCheckpointBlob)) mux.HandleFunc("HEAD /v1/checkpoints/{id}/blob", s.handleCheckpointProbe) mux.HandleFunc("DELETE /v1/checkpoints/{id}", s.requireToken(s.handleDeleteCheckpoint)) @@ -231,7 +221,6 @@ func (s *Server) handleClaim(w http.ResponseWriter, r *http.Request) { return } - // the data plane must be direct, so a warm peer gets the claim by redirect, not proxy sb, err := s.mgr.ClaimWarm(r.Context(), key, req.TTL(), tenant, req.ClaimRef, nil) if errors.Is(err, pool.ErrNoWarm) { if s.redirectClaim(r.Context(), w, req, key, key.Hash(), tenant) { @@ -239,9 +228,8 @@ func (s *Server) handleClaim(w http.ResponseWriter, r *http.Request) { } sb, err = s.mgr.ClaimProvision(r.Context(), key, req.TTL(), tenant, req.ClaimRef, nil) } - // quota is per node, so a full node bounces the claim to a peer before answering 429 if errors.Is(err, pool.ErrQuota) && s.placer != nil && !req.NoRedirect && - writeRedirect(w, s.placer.Candidates(key.Hash())) { + s.writeRedirect(w, s.placer.Candidates(key.Hash())) { return } writeResult(w, r, "claim", key.Template, "provisioning failed", err, func() { @@ -270,7 +258,7 @@ func (s *Server) handleVolumeClaim(w http.ResponseWriter, r *http.Request, req t } else { sb, err = s.mgr.ClaimWarm(r.Context(), key, req.TTL(), tenant, req.ClaimRef, req.Volumes) if errors.Is(err, pool.ErrNoWarm) { - if s.placer != nil && !req.NoRedirect && writeRedirect(w, s.placer.VolumeCandidates(hash, types.VolumeNames(req.Volumes))) { + if s.placer != nil && !req.NoRedirect && s.writeRedirect(w, s.placer.VolumeCandidates(hash, types.VolumeNames(req.Volumes))) { return } sb, err = s.mgr.ClaimProvision(r.Context(), key, req.TTL(), tenant, req.ClaimRef, req.Volumes) @@ -299,7 +287,6 @@ func (s *Server) redirectVolumeClaim(ctx context.Context, w http.ResponseWriter, } localTemplate := s.mgr.HasPromotedTemplate(ctx, key, tenant) - // a peer's promoted template must not escalate volume claims off the pool golden pooled := s.mgr.HasPoolGolden(key) var templateOwners []string if s.placer != nil && !pooled { @@ -316,7 +303,6 @@ func (s *Server) redirectVolumeClaim(ctx context.Context, w http.ResponseWriter, var owners []string if promoted { - // a shared template store lets a volume holder resolve a hash it has not advertised yet owners = s.templateOwners(func(probe string) []string { return s.placer.TemplateVolumeOwners(probe, names) }, hash, tenant) @@ -329,7 +315,7 @@ func (s *Server) redirectVolumeClaim(ctx context.Context, w http.ResponseWriter, if len(owners) == 0 { return false, pool.ErrVolumeUnavailable } - writeJSON(w, http.StatusOK, types.ClaimResponse{Redirect: owners, RequirePromoted: promoted}) + writeJSON(w, http.StatusOK, types.ClaimResponse{Redirect: s.clientAddrs(owners), RequirePromoted: promoted}) return true, nil } @@ -353,17 +339,15 @@ func (s *Server) templateOwners(query func(string) []string, hash, tenant string return owners } -// redirectClaim bounces a warm-miss to a warm holder, or the template owner when we lack a golden. func (s *Server) redirectClaim(ctx context.Context, w http.ResponseWriter, req types.ClaimRequest, key types.PoolKey, hash, tenant string) bool { if s.placer == nil || req.NoRedirect { return false } - if writeRedirect(w, s.placer.Candidates(hash)) { + if s.writeRedirect(w, s.placer.Candidates(hash)) { return true } - // TemplateOwners is in-memory; HasGolden can be a store round-trip. owners := s.templateOwners(s.placer.TemplateOwners, hash, tenant) - return len(owners) > 0 && !s.mgr.HasGolden(ctx, key, tenant) && writeRedirect(w, owners) + return len(owners) > 0 && !s.mgr.HasGolden(ctx, key, tenant) && s.writeRedirect(w, owners) } func (s *Server) handleSandbox(w http.ResponseWriter, r *http.Request) { @@ -416,7 +400,6 @@ func (s *Server) handleFork(w http.ResponseWriter, r *http.Request) { }) } -// handlePromote publishes a claimed sandbox as a node-local template. func (s *Server) handlePromote(w http.ResponseWriter, r *http.Request) { req, ok := decodeBody[types.PromoteRequest](w, r) if !ok { @@ -441,7 +424,6 @@ func (s *Server) handleCheckpoint(w http.ResponseWriter, r *http.Request) { }) } -// handleClaimCheckpoint claims a sandbox from a checkpoint; a probed owner is not authoritative. func (s *Server) handleClaimCheckpoint(w http.ResponseWriter, r *http.Request) { req, ok := decodeBody[types.CheckpointClaimRequest](w, r) if !ok { @@ -450,7 +432,7 @@ func (s *Server) handleClaimCheckpoint(w http.ResponseWriter, r *http.Request) { ckptID := r.PathValue("id") sb, err := s.mgr.ClaimCheckpoint(r.Context(), ckptID, req.TTL(), tenantFrom(r.Context())) if errors.Is(err, pool.ErrUnknownCheckpoint) { - if !req.NoRedirect && s.prober != nil && writeRedirect(w, s.prober.Owners(r.Context(), ckptID)) { + if !req.NoRedirect && s.prober != nil && s.writeRedirect(w, s.prober.Owners(r.Context(), ckptID)) { return } sb, err = s.mgr.ClaimCheckpointHeal(r.Context(), ckptID, req.TTL(), tenantFrom(r.Context())) @@ -460,14 +442,12 @@ func (s *Server) handleClaimCheckpoint(w http.ResponseWriter, r *http.Request) { }) } -// handleCheckpointBlob streams a checkpoint record as a tar for a peer's pull. func (s *Server) handleCheckpointBlob(w http.ResponseWriter, r *http.Request) { ckptID := r.PathValue("id") dir, meta, release, err := s.mgr.FetchCheckpoint(r.Context(), ckptID) writeResult(w, r, "fetch checkpoint", ckptID, "fetch checkpoint failed", err, func() { defer release() w.Header().Set("Content-Type", "application/x-tar") - // the tar completion marker, not the status, tells the reader the record arrived whole w.WriteHeader(http.StatusOK) if err := peer.TarRecord(dir, meta, w); err != nil { log.WithFunc("server.handleCheckpointBlob").Error(r.Context(), err, "stream checkpoint") @@ -475,7 +455,6 @@ func (s *Server) handleCheckpointBlob(w http.ResponseWriter, r *http.Request) { }) } -// handleCheckpointProbe answers a HEAD probe; without a probeKey the id is the only capability. func (s *Server) handleCheckpointProbe(w http.ResponseWriter, r *http.Request) { id := r.PathValue("id") if len(s.probeKey) > 0 && !peer.VerifyProbeMAC(s.probeKey, id, r.Header.Get(peer.ProbeHeader)) { @@ -489,7 +468,6 @@ func (s *Server) handleCheckpointProbe(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNotFound) } -// handleListCheckpoints lists this node's checkpoints, newest first. func (s *Server) handleListCheckpoints(w http.ResponseWriter, r *http.Request) { ckpts, err := s.mgr.Checkpoints(r.Context(), tenantFrom(r.Context())) if err != nil { @@ -500,7 +478,6 @@ func (s *Server) handleListCheckpoints(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, types.CheckpointListResponse{Checkpoints: ckpts}) } -// handleDeleteCheckpoint removes a checkpoint; no_forward stops a broadcast delete from looping. func (s *Server) handleDeleteCheckpoint(w http.ResponseWriter, r *http.Request) { scope := pool.DeleteFleet if r.URL.Query().Get("no_forward") != "" { @@ -508,7 +485,6 @@ func (s *Server) handleDeleteCheckpoint(w http.ResponseWriter, r *http.Request) } id := r.PathValue("id") err := s.mgr.DeleteCheckpoint(r.Context(), id, tenantFrom(r.Context()), scope) - // evict after the delete: a racing probe would otherwise re-cache the still-present record if s.prober != nil { s.prober.Forget(id) } @@ -517,7 +493,6 @@ func (s *Server) handleDeleteCheckpoint(w http.ResponseWriter, r *http.Request) }) } -// handleDeleteTemplate removes a promoted template; the key axes ride as query parameters. func (s *Server) handleDeleteTemplate(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() req := types.ClaimRequest{ @@ -527,9 +502,8 @@ func (s *Server) handleDeleteTemplate(w http.ResponseWriter, r *http.Request) { } key := req.Key() err := s.mgr.DeleteTemplate(r.Context(), key, tenantFrom(r.Context())) - // a redirected retry carries no_redirect, so the owner answers for itself and never bounces if errors.Is(err, pool.ErrUnknownTemplate) && s.placer != nil && q.Get("no_redirect") == "" && - writeRedirect(w, s.templateOwners(s.placer.TemplateOwners, key.Hash(), tenantFrom(r.Context()))) { + s.writeRedirect(w, s.templateOwners(s.placer.TemplateOwners, key.Hash(), tenantFrom(r.Context()))) { return } writeResult(w, r, "delete template", req.Template, "delete template failed", err, func() { @@ -537,7 +511,6 @@ func (s *Server) handleDeleteTemplate(w http.ResponseWriter, r *http.Request) { }) } -// handleOwner answers whether this node owns the sandbox; the token both authorizes and proves it. func (s *Server) handleOwner(w http.ResponseWriter, r *http.Request) { token, ok := sandboxToken(w, r) if !ok { @@ -576,13 +549,12 @@ func (s *Server) handleUncordon(w http.ResponseWriter, r *http.Request) { s.handleInfo(w, r) } -// handlePeers lists the cluster's node addresses; topology is readable by any valid token. func (s *Server) handlePeers(w http.ResponseWriter, _ *http.Request) { var peers []string if s.placer != nil { peers = s.placer.PeerAddrs() } - writeJSON(w, http.StatusOK, map[string][]string{"peers": peers}) + writeJSON(w, http.StatusOK, map[string][]string{"peers": s.clientAddrs(peers)}) } func (s *Server) handleInfo(w http.ResponseWriter, _ *http.Request) { @@ -592,7 +564,7 @@ func (s *Server) handleInfo(w http.ResponseWriter, _ *http.Request) { Draining: g.Draining, AtCapacity: g.AtCapacity, AtCapacityReason: g.AtCapacityReason, } if s.placer != nil { - resp.Peers = s.placer.PeerAddrs() + resp.Peers = s.clientAddrs(s.placer.PeerAddrs()) } writeJSON(w, http.StatusOK, resp) } @@ -601,7 +573,6 @@ func (s *Server) handleHealthz(w http.ResponseWriter, _ *http.Request) { _, _ = io.WriteString(w, "ok") } -// requireToken resolves the caller's scope — root or a tenant name — onto the request context. func (s *Server) requireToken(next http.HandlerFunc) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { tenant, ok := s.resolveScope(r) @@ -613,7 +584,6 @@ func (s *Server) requireToken(next http.HandlerFunc) http.HandlerFunc { } } -// requireRoot guards operator-only surfaces: an authenticated tenant token gets 403, not 401. func (s *Server) requireRoot(next http.HandlerFunc) http.HandlerFunc { return s.requireToken(func(w http.ResponseWriter, r *http.Request) { if tenantFrom(r.Context()) != "" { @@ -629,12 +599,10 @@ func (s *Server) rootRequest(r *http.Request) bool { return ok && s.isRootToken(token) } -// isRootToken reports whether token is the root api_token; an unset api token matches nothing. func (s *Server) isRootToken(token string) bool { return s.apiToken != "" && subtle.ConstantTimeCompare([]byte(token), []byte(s.apiToken)) == 1 } -// sandboxCred resolves a header bearer token to a Cred; the root api_token elevates to Operator. func (s *Server) sandboxCred(token string) pool.Cred { if s.isRootToken(token) { return pool.Cred{Operator: true} @@ -642,12 +610,10 @@ func (s *Server) sandboxCred(token string) pool.Cred { return pool.Cred{Token: token} } -// bodyCred resolves a body-carried sandbox token; absent on a root request means Operator. func (s *Server) bodyCred(r *http.Request, bodyToken string) pool.Cred { return pool.Cred{Token: bodyToken, Operator: bodyToken == "" && s.rootRequest(r)} } -// resolveScope matches the bearer token to root ("") or a tenant name. func (s *Server) resolveScope(r *http.Request) (string, bool) { if s.apiToken == "" && len(s.tenants) == 0 { return "", true diff --git a/sandboxd/server/server_http.go b/sandboxd/server/server_http.go index 3ace7f56..66a99d2c 100644 --- a/sandboxd/server/server_http.go +++ b/sandboxd/server/server_http.go @@ -15,12 +15,29 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/utils" ) -// retryAfterSeconds bounds a waiting client to a few probes a minute. const retryAfterSeconds = "10" -// tenantKey carries the resolved tenant scope ("" = root) on the request context. type tenantKey struct{} +func (s *Server) writeRedirect(w http.ResponseWriter, addrs []string) bool { + if len(addrs) == 0 { + return false + } + writeJSON(w, http.StatusOK, types.ClaimResponse{Redirect: s.clientAddrs(addrs)}) + return true +} + +func (s *Server) clientAddrs(addrs []string) []string { + if s.placer == nil { + return addrs + } + clients := make([]string, len(addrs)) + for i, addr := range addrs { + clients[i] = s.placer.ClientAddr(addr) + } + return clients +} + func withTenant(ctx context.Context, tenant string) context.Context { return context.WithValue(ctx, tenantKey{}, tenant) } @@ -35,7 +52,6 @@ func bearerToken(r *http.Request) (string, bool) { return token, ok && token != "" } -// sandboxToken extracts the per-sandbox bearer token, answering 401 itself. func sandboxToken(w http.ResponseWriter, r *http.Request) (string, bool) { token, ok := bearerToken(r) if !ok { @@ -44,7 +60,6 @@ func sandboxToken(w http.ResponseWriter, r *http.Request) (string, bool) { return token, ok } -// decodeBody parses a JSON request body, answering 400 itself on failure. func decodeBody[T any](w http.ResponseWriter, r *http.Request) (T, bool) { var v T if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxBodyBytes)).Decode(&v); err != nil { @@ -54,7 +69,6 @@ func decodeBody[T any](w http.ResponseWriter, r *http.Request) (T, bool) { return v, true } -// decodeBodyStrict is decodeBody for operator bodies, where a mistyped key must fail. func decodeBodyStrict[T any](w http.ResponseWriter, r *http.Request) (T, bool) { var v T raw, err := io.ReadAll(http.MaxBytesReader(w, r.Body, maxBodyBytes)) @@ -69,11 +83,9 @@ func decodeBodyStrict[T any](w http.ResponseWriter, r *http.Request) (T, bool) { return v, true } -// writePoolErr answers a pool-sentinel error, reporting whether it handled err. func writePoolErr(w http.ResponseWriter, err error) bool { for _, m := range poolErrHTTP { if errors.Is(err, m.err) { - // a 503 is node exhaustion, not a caller error, so the client should retry if m.code == http.StatusServiceUnavailable { w.Header().Set("Retry-After", retryAfterSeconds) } @@ -95,15 +107,6 @@ func writeResult(w http.ResponseWriter, r *http.Request, op, id, failMsg string, } } -// writeRedirect answers with the claim protocol's redirect shape, reporting whether it did. -func writeRedirect(w http.ResponseWriter, addrs []string) bool { - if len(addrs) == 0 { - return false - } - writeJSON(w, http.StatusOK, types.ClaimResponse{Redirect: addrs}) - return true -} - func writeJSON(w http.ResponseWriter, code int, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(code) diff --git a/sandboxd/server/server_test.go b/sandboxd/server/server_test.go index 627274e6..b3d2c60a 100644 --- a/sandboxd/server/server_test.go +++ b/sandboxd/server/server_test.go @@ -1880,16 +1880,24 @@ func credToken(cred pool.Cred) string { return cred.Token } +type claimFunc func(context.Context, types.PoolKey, time.Duration) (*types.Sandbox, error) + +type sandboxVerbFunc func(string, string) error + +type checkpointClaimFunc func(string) (*types.Sandbox, error) + +type idActionFunc func(string) error + type fakeManager struct { ckptDir string hasCheckpoint map[string]bool - claim func(ctx context.Context, key types.PoolKey, ttl time.Duration) (*types.Sandbox, error) - warmClaim func(ctx context.Context, key types.PoolKey, ttl time.Duration) (*types.Sandbox, error) - release func(id, token string) error - releaseOp func(id string) error + claim claimFunc + warmClaim claimFunc + release sandboxVerbFunc + releaseOp idActionFunc socket func(id, token string) (string, error) - hibernate func(id, token string) error - wake func(id, token string) error + hibernate sandboxVerbFunc + wake sandboxVerbFunc fork func(id, token string, count int, ttl time.Duration) ([]*types.Sandbox, error) promote func(id, token, template string) error promoteContentDigest string @@ -1901,11 +1909,11 @@ type fakeManager struct { audited func(id string, line []byte) checkpoint func(id, token, name string) (types.Checkpoint, error) - claimCheckpoint func(ckptID string) (*types.Sandbox, error) - healCheckpoint func(ckptID string) (*types.Sandbox, error) + claimCheckpoint checkpointClaimFunc + healCheckpoint checkpointClaimFunc healCalls int checkpoints []types.Checkpoint - deleteCheckpoint func(ckptID string) error + deleteCheckpoint idActionFunc setPools func(pools []config.PoolSpec) error infoPools []pool.PoolInfo claimDeadline func(id, token string) (time.Time, error) @@ -2208,6 +2216,7 @@ func (f *fakePlacer) TemplateVolumeOwners(probe string, _ []string) []string { } func (f *fakePlacer) VolumeHolders() map[string]int { return f.volumeHolders } func (f *fakePlacer) PeerAddrs() []string { return f.addrs } +func (f *fakePlacer) ClientAddr(addr string) string { return addr } func (f *fakePlacer) ConfigMismatches() int { return 0 } type fakeProber struct { diff --git a/sdk/go/client.go b/sdk/go/client.go index 29569058..1be27fb5 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -1,12 +1,11 @@ -// Package sandbox is the Go SDK for the cocoon sandbox control plane: claim -// a microVM from a sandboxd node, run commands in it over the relayed silkd -// protocol, release it. +// Package sandbox controls sandboxes and relays guest operations through sandboxd. package sandbox import ( "bytes" "cmp" "context" + "crypto/tls" "encoding/json" "errors" "fmt" @@ -19,6 +18,8 @@ import ( ) const ( + idleConnsPerHost = 100 + templateQueryParam = "template" netQueryParam = "net" sizeQueryParam = "size" @@ -28,48 +29,130 @@ const ( // ClientOption configures Connect. type ClientOption func(*Client) -// WithAPIToken sets the operator bearer for every node-scoped call — claim and -// info, plus drain, pools, templates, checkpoints, and fork/promote/preview. -func WithAPIToken(token string) ClientOption { - return func(c *Client) { c.apiToken = token } +type claimEncoder func(noRedirect, requirePromoted bool) ([]byte, error) + +type claimPoster func(addr string, body []byte) (claimResponse, error) + +type claimResponse struct { + ID string `json:"id"` + Token string `json:"token"` + Deadline time.Time `json:"deadline"` + OwnerAddr string `json:"owner_addr,omitempty"` + FromCheckpoint string `json:"from_checkpoint,omitempty"` + TemplateDigest string `json:"template_digest,omitempty"` + Volumes []Volume `json:"volumes,omitempty"` + Redirect []string `json:"redirect,omitempty"` + RequirePromoted bool `json:"require_promoted,omitzero"` } -// WithHTTPClient replaces the control-plane HTTP client, for callers that need -// their own transport, proxy, or timeout. -func WithHTTPClient(hc *http.Client) ClientOption { - return func(c *Client) { c.hc = hc } +type volumeListResponse struct { + Volumes []VolumeInfo `json:"volumes"` +} + +type forkRequest struct { + Token string `json:"token"` + Count int `json:"count"` + TTLSeconds int `json:"ttl_seconds,omitzero"` +} + +type forkResponse struct { + Children []claimResponse `json:"children"` +} + +type promoteRequest struct { + Token string `json:"token"` + Template string `json:"template"` +} + +type promoteResponse struct { + Key struct { + Template string `json:"template"` + Net string `json:"net"` + Size string `json:"size"` + } `json:"key"` + ContentDigest string `json:"content_digest"` +} + +type errorResponse struct { + Error string `json:"error"` +} + +// APIError is a non-2xx control-plane reply. +type APIError struct { + Verb string + Status int + Message string +} + +func (e *APIError) Error() string { + if e.Message != "" { + return fmt.Sprintf("%s: %s (http %d)", e.Verb, e.Message, e.Status) + } + return fmt.Sprintf("%s: http %d", e.Verb, e.Status) +} + +type claimRequest struct { + Template string `json:"template"` + Net string `json:"net,omitempty"` + Size string `json:"size,omitempty"` + Volumes []Volume `json:"volumes,omitempty"` + VolumesAttachOnly bool `json:"volumes_attach_only,omitzero"` + TTLSeconds int `json:"ttl_seconds,omitzero"` + NoRedirect bool `json:"no_redirect,omitzero"` + RequirePromoted bool `json:"require_promoted,omitzero"` + ClaimRef string `json:"claim_ref,omitempty"` +} + +func (r claimRequest) rejectPinnedAxes() error { + if r.Net != "" || r.Size != "" { + return fmt.Errorf("network and size are pinned by the snapshot; WithNetwork/WithSize are not accepted here") + } + return nil +} + +func (r claimRequest) validateVolumes() error { + for _, v := range r.Volumes { + if v.Mode != "" && v.Mode != volumeModeRW { + return fmt.Errorf("volume %q: mode must be \"\", %q, or %q, got %q", v.Name, volumeModeRO, volumeModeRW, v.Mode) + } + if r.VolumesAttachOnly && v.Mount != "" { + return fmt.Errorf("volume %q: mount %q is meaningless with WithVolumesAttachOnly, which leaves mounting to the caller", v.Name, v.Mount) + } + } + return nil } // Client talks to one sandboxd node. type Client struct { - addr string - apiToken string - hc *http.Client + addr string + scheme string + apiToken string + hc *http.Client + tlsConfig *tls.Config } -// Connect returns a client for a sandboxd node. addr accepts a -// comma-separated seed list for forward compatibility; v0 uses the first -// entry. Calls are bounded by their ctx — checkpoint and promote run as long -// as the snapshot takes, so the client sets no blanket deadline. +// Connect returns a client for a sandboxd node. func Connect(addr string, opts ...ClientOption) (*Client, error) { first, _, _ := strings.Cut(addr, ",") first = strings.TrimSpace(first) if first == "" { return nil, fmt.Errorf("empty sandboxd address") } - c := &Client{addr: first, hc: &http.Client{}} + u, err := endpointURL(first, "http") + if err != nil { + return nil, err + } + c := &Client{addr: strings.TrimPrefix(u.String(), "http://"), scheme: u.Scheme, hc: &http.Client{}} for _, opt := range opts { opt(c) } + if err := c.configureTLS(); err != nil { + return nil, err + } return c, nil } -// New claims a sandbox for template. Without options the node serves its -// defaults: the no-network lane and the smallest size tier. New returns when -// the sandbox's silkd is reachable. Against a cluster, a warm miss redirects -// to a peer that holds one, which New follows transparently; if every -// candidate fails transiently (full, mid-heal, unreachable), New falls back -// to the origin once so it provisions or heals locally. +// New claims a sandbox for template. func (c *Client) New(ctx context.Context, template string, opts ...Option) (*Sandbox, error) { claim := claimRequest{Template: template} for _, opt := range opts { @@ -99,10 +182,7 @@ func (c *Client) Volumes(ctx context.Context) ([]VolumeInfo, error) { return resp.Volumes, nil } -// Lookup relocates a sandbox handle whose owner address was lost, given its -// id and token: it asks the entry node, then scatters across the cluster's -// peers concurrently, and returns a handle bound to whichever node confirms -// ownership first — one hung peer must not stall the whole lookup. +// Lookup finds the sandbox owner using its id and token. func (c *Client) Lookup(ctx context.Context, id, token string) (*Sandbox, error) { if owner, err := c.ownerAt(ctx, c.addr, id, token); err == nil { return &Sandbox{ID: id, token: token, c: c, owner: owner}, nil @@ -117,18 +197,12 @@ func (c *Client) Lookup(ctx context.Context, id, token string) (*Sandbox, error) return &Sandbox{ID: id, token: token, c: c, owner: owner}, nil } -// Attach binds a handle to an already-claimed sandbox whose owner data-plane -// address is already known (e.g. delivered by the L3 apiserver as annotations), -// with no lookup round-trip. ownerAddr is the node's sandboxd data-plane -// address; token is the per-sandbox credential. +// Attach binds a handle to a known owner without a lookup. func (c *Client) Attach(ownerAddr, id, token string) *Sandbox { return &Sandbox{ID: id, token: token, c: c, owner: ownerAddr} } -// DeleteTemplate removes a promoted template by name. When the entry node -// does not hold it but the mesh's gossip names an owner, the delete follows -// the redirect there (one hop); gossip lags a fresh promote by about a tick, -// so right after promoting prefer Template.Delete on the returned handle. +// DeleteTemplate removes a promoted template by name. func (c *Client) DeleteTemplate(ctx context.Context, template string, opts ...Option) error { claim := claimRequest{Template: template} for _, opt := range opts { @@ -139,7 +213,6 @@ func (c *Client) DeleteTemplate(ctx context.Context, template string, opts ...Op if err != nil || len(redirect) == 0 { return err } - // gossip named the template's owners; no_redirect makes the owner answer for itself, never a second hop u.Set(noRedirectQueryParam, "1") if tryErr := tryEach(redirect, func(addr string) error { _, retryErr := c.deleteTemplates(ctx, addr, u) @@ -150,8 +223,6 @@ func (c *Client) DeleteTemplate(ctx context.Context, template string, opts ...Op return nil } -// ownerAt asks one node whether it owns the sandbox, returning its data-plane -// address on success. func (c *Client) ownerAt(ctx context.Context, addr, id, token string) (string, error) { body, err := doJSON[struct { OwnerAddr string `json:"owner_addr"` @@ -162,8 +233,6 @@ func (c *Client) ownerAt(ctx context.Context, addr, id, token string) (string, e return cmp.Or(body.OwnerAddr, addr), nil } -// handleFrom builds a sandbox handle, defaulting the data-plane owner to the -// node that answered when a single-node deployment omits owner_addr. func (c *Client) handleFrom(dialed string, cr claimResponse) *Sandbox { return &Sandbox{ ID: cr.ID, Deadline: cr.Deadline, Volumes: cr.Volumes, @@ -176,8 +245,6 @@ func (c *Client) claimAt(ctx context.Context, addr string, body []byte) (claimRe return doJSON[claimResponse](ctx, c, http.MethodPost, addr, "/v1/claim", bytes.NewReader(body), c.apiToken, "claim") } -// deleteTemplates issues the template delete against one node; a 200 with a -// redirect list names the owners to retry at. func (c *Client) deleteTemplates(ctx context.Context, addr string, u url.Values) ([]string, error) { resp, err := c.roundTrip(ctx, http.MethodDelete, addr, "/v1/templates?"+u.Encode(), nil, c.apiToken) if err != nil { @@ -198,10 +265,15 @@ func (c *Client) deleteTemplates(ctx context.Context, addr string, u url.Values) } } -// roundTrip issues one control-plane request against addr, attaching bearer -// when non-empty; a non-nil body is JSON. +func (c *Client) requestURL(addr, path string) string { + if strings.Contains(addr, "://") { + return addr + path + } + return c.scheme + "://" + addr + path +} + func (c *Client) roundTrip(ctx context.Context, method, addr, path string, body io.Reader, bearer string) (*http.Response, error) { - req, err := http.NewRequestWithContext(ctx, method, "http://"+addr+path, body) + req, err := http.NewRequestWithContext(ctx, method, c.requestURL(addr, path), body) if err != nil { return nil, err } @@ -214,7 +286,16 @@ func (c *Client) roundTrip(ctx context.Context, method, addr, path string, body return c.hc.Do(req) //nolint:gosec // dialing the caller-configured node is the SDK's purpose } -// doJSON issues one control-plane request and decodes a 200 reply into T; any other status maps through apiError under verb. +// WithAPIToken sets the bearer for node-scoped calls. +func WithAPIToken(token string) ClientOption { + return func(c *Client) { c.apiToken = token } +} + +// WithHTTPClient replaces the control-plane HTTP client, for callers that need their own transport, proxy, or timeout. +func WithHTTPClient(hc *http.Client) ClientOption { + return func(c *Client) { c.hc = hc } +} + func doJSON[T any](ctx context.Context, c *Client, method, addr, path string, body io.Reader, bearer, verb string) (T, error) { var out T resp, err := c.roundTrip(ctx, method, addr, path, body, bearer) @@ -239,7 +320,6 @@ func doJSONPtr[T any](ctx context.Context, c *Client, method, addr, path string, return &out, nil } -// doNoContent is doJSON's reply-less twin: 204 or an apiError. func doNoContent(ctx context.Context, c *Client, method, addr, path string, body io.Reader, bearer, verb string) error { resp, err := c.roundTrip(ctx, method, addr, path, body, bearer) if err != nil { @@ -252,7 +332,6 @@ func doNoContent(ctx context.Context, c *Client, method, addr, path string, body return nil } -// tryEach walks candidates until call succeeds: an error retry accepts moves on (the last one propagates), any other returns at once. func tryEach(candidates []string, call func(addr string) error, retry func(error) bool) error { var lastErr error for _, addr := range candidates { @@ -267,14 +346,11 @@ func tryEach(candidates []string, call func(addr string) error, retry func(error return lastErr } -// retryMiss retries a miss (the next candidate may own the record) or a -// transport failure (dead peer); a served error is real and stops the walk. func retryMiss(err error) bool { he, ok := errors.AsType[*APIError](err) return !ok || he.Status == http.StatusNotFound } -// retryTransient reports whether origin can still answer differently; a served 4xx outside the list would fail there the same way. func retryTransient(err error) bool { he, ok := errors.AsType[*APIError](err) if !ok { @@ -290,11 +366,6 @@ func retryTransient(err error) bool { } } -type claimEncoder func(noRedirect, requirePromoted bool) ([]byte, error) - -type claimPoster func(addr string, body []byte) (claimResponse, error) - -// claimFollow claims at origin and follows a redirect through redirectFallback; only the fallback error carries the verb. func claimFollow(origin, verb string, encode claimEncoder, claimAt claimPoster) (string, claimResponse, error) { body, err := encode(false, false) if err != nil { @@ -319,9 +390,6 @@ func claimFollow(origin, verb string, encode claimEncoder, claimAt claimPoster) return addr, target, nil } -// redirectFallback claims at every candidate with no_redirect; when the last failure was transient -// (retryTransient) origin gets one more attempt, since it provisions or heals locally instead of -// leaving the claim on stale gossip. func redirectFallback(origin string, candidates []string, claimAt func(addr string) (claimResponse, error)) (string, claimResponse, error) { claimNoRedirect := func(target string) (claimResponse, error) { cr, err := claimAt(target) @@ -346,15 +414,11 @@ func redirectFallback(origin string, candidates []string, claimAt func(addr stri } cr, err := claimNoRedirect(origin) if err != nil { - // the peers' failures say why the claim left origin, origin's says why coming back did not help return "", claimResponse{}, fmt.Errorf("all redirect targets failed, origin fallback failed: %w", errors.Join(lastErr, err)) } return origin, cr, nil } -// scatter probes addrs concurrently, returning the first success and -// canceling the losers — one hung peer must not stall the caller. When -// every probe fails (or addrs is empty) ok is false. func scatter[T any](ctx context.Context, addrs []string, probe func(ctx context.Context, addr string) (T, error)) (result T, ok bool) { scatterCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -372,8 +436,6 @@ func scatter[T any](ctx context.Context, addrs []string, probe func(ctx context. return result, ok } -// encodeBody marshals a wire request body, wrapping failures under verb — -// the shared prelude of every POST-a-JSON-body call in this package. func encodeBody(verb string, v any) ([]byte, error) { body, err := json.Marshal(v) if err != nil { @@ -382,105 +444,9 @@ func encodeBody(verb string, v any) ([]byte, error) { return body, nil } -// APIError is a non-2xx control-plane reply. -type APIError struct { - Verb string - Status int - Message string -} - -func (e *APIError) Error() string { - if e.Message != "" { - return fmt.Sprintf("%s: %s (http %d)", e.Verb, e.Message, e.Status) - } - return fmt.Sprintf("%s: http %d", e.Verb, e.Status) -} - -// apiError surfaces the server's {"error": ...} body when present. func apiError(verb string, resp *http.Response) error { var er errorResponse _ = json.NewDecoder(io.LimitReader(resp.Body, 4096)).Decode(&er) _, _ = io.Copy(io.Discard, resp.Body) return &APIError{Verb: verb, Status: resp.StatusCode, Message: er.Error} } - -// claimRequest mirrors sandboxd's wire type; duplicated so the SDK stays -// dependency-free — the e2e module guards against drift. -type claimRequest struct { - Template string `json:"template"` - Net string `json:"net,omitempty"` - Size string `json:"size,omitempty"` - Volumes []Volume `json:"volumes,omitempty"` - VolumesAttachOnly bool `json:"volumes_attach_only,omitzero"` - TTLSeconds int `json:"ttl_seconds,omitzero"` - NoRedirect bool `json:"no_redirect,omitzero"` - RequirePromoted bool `json:"require_promoted,omitzero"` - ClaimRef string `json:"claim_ref,omitempty"` -} - -// rejectPinnedAxes fails a snapshot claim (checkpoint, template) that passed -// WithNetwork/WithSize — those axes are pinned and would silently no-op. -func (r claimRequest) rejectPinnedAxes() error { - if r.Net != "" || r.Size != "" { - return fmt.Errorf("network and size are pinned by the snapshot; WithNetwork/WithSize are not accepted here") - } - return nil -} - -// validateVolumes rejects a mode outside the wire's vocabulary and a mount -// attach-only makes meaningless; WithVolumes already normalizes "ro" to "". -func (r claimRequest) validateVolumes() error { - for _, v := range r.Volumes { - if v.Mode != "" && v.Mode != volumeModeRW { - return fmt.Errorf("volume %q: mode must be \"\", %q, or %q, got %q", v.Name, volumeModeRO, volumeModeRW, v.Mode) - } - if r.VolumesAttachOnly && v.Mount != "" { - return fmt.Errorf("volume %q: mount %q is meaningless with WithVolumesAttachOnly, which leaves mounting to the caller", v.Name, v.Mount) - } - } - return nil -} - -type claimResponse struct { - ID string `json:"id"` - Token string `json:"token"` - Deadline time.Time `json:"deadline"` - OwnerAddr string `json:"owner_addr,omitempty"` - FromCheckpoint string `json:"from_checkpoint,omitempty"` - TemplateDigest string `json:"template_digest,omitempty"` - Volumes []Volume `json:"volumes,omitempty"` - Redirect []string `json:"redirect,omitempty"` - RequirePromoted bool `json:"require_promoted,omitzero"` -} - -type volumeListResponse struct { - Volumes []VolumeInfo `json:"volumes"` -} - -type forkRequest struct { - Token string `json:"token"` - Count int `json:"count"` - TTLSeconds int `json:"ttl_seconds,omitzero"` -} - -type forkResponse struct { - Children []claimResponse `json:"children"` -} - -type promoteRequest struct { - Token string `json:"token"` - Template string `json:"template"` -} - -type promoteResponse struct { - Key struct { - Template string `json:"template"` - Net string `json:"net"` - Size string `json:"size"` - } `json:"key"` - ContentDigest string `json:"content_digest"` -} - -type errorResponse struct { - Error string `json:"error"` -} diff --git a/sdk/go/endpoint.go b/sdk/go/endpoint.go new file mode 100644 index 00000000..d40bf36f --- /dev/null +++ b/sdk/go/endpoint.go @@ -0,0 +1,66 @@ +package sandbox + +import ( + "cmp" + "crypto/tls" + "fmt" + "net/http" + "net/url" + "strconv" + "strings" +) + +const httpsScheme = "https" + +func (c *Client) configureTLS() error { + tr, ok := cmp.Or(c.hc.Transport, http.DefaultTransport).(*http.Transport) + switch { + case !ok && c.tlsConfig != nil: + return fmt.Errorf("tls configuration requires an http.Transport") + case !ok: + c.tlsConfig = &tls.Config{MinVersion: tls.VersionTLS12} + case c.tlsConfig == nil && c.hc.Transport != nil: + c.tlsConfig = cmp.Or(tr.TLSClientConfig.Clone(), &tls.Config{MinVersion: tls.VersionTLS12}) + default: + c.tlsConfig = cmp.Or(c.tlsConfig, &tls.Config{MinVersion: tls.VersionTLS12}) + hc := *c.hc + tr = tr.Clone() + tr.TLSClientConfig = c.tlsConfig + if c.hc.Transport == nil { + tr.MaxIdleConnsPerHost = idleConnsPerHost + } + hc.Transport = tr + c.hc = &hc + } + if c.tlsConfig.ClientSessionCache == nil { + c.tlsConfig.ClientSessionCache = tls.NewLRUClientSessionCache(0) + } + return nil +} + +// WithTLSConfig sets certificate verification for HTTPS requests and agent relays. +func WithTLSConfig(cfg *tls.Config) ClientOption { + return func(c *Client) { c.tlsConfig = cfg.Clone() } +} + +func endpointURL(addr, scheme string) (*url.URL, error) { + if !strings.Contains(addr, "://") { + addr = scheme + "://" + addr + } + u, err := url.Parse(addr) + if err != nil { + return nil, fmt.Errorf("parse sandboxd endpoint: %w", err) + } + if (u.Scheme != "http" && u.Scheme != httpsScheme) || u.Hostname() == "" || u.User != nil || + (u.Path != "" && u.Path != "/") || strings.ContainsAny(addr, "?#") { + return nil, fmt.Errorf("sandboxd endpoint must be an http or https origin") + } + if port := u.Port(); port != "" { + n, err := strconv.Atoi(port) + if err != nil || n < 1 || n > 65535 { + return nil, fmt.Errorf("sandboxd endpoint port must be between 1 and 65535") + } + } + u.Path = "" + return u, nil +} diff --git a/sdk/go/endpoint_test.go b/sdk/go/endpoint_test.go new file mode 100644 index 00000000..4771629d --- /dev/null +++ b/sdk/go/endpoint_test.go @@ -0,0 +1,167 @@ +package sandbox + +import ( + "context" + "crypto/tls" + "crypto/x509" + "errors" + "io" + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/cocoonstack/sandbox/sdk/go/silkd/silkdtest" +) + +func TestEndpointURL(t *testing.T) { + for _, tt := range []struct{ addr, scheme, want string }{ + {"localhost:7777", "http", "http://localhost:7777"}, + {"localhost:443", "https", "https://localhost:443"}, + {"https://example.com/", "http", "https://example.com"}, + {"http://example.com", "https", "http://example.com"}, + {"[::1]:7777", "https", "https://[::1]:7777"}, + {"https://[::1]", "http", "https://[::1]"}, + } { + t.Run(tt.addr, func(t *testing.T) { + u, err := endpointURL(tt.addr, tt.scheme) + if err != nil || u.String() != tt.want { + t.Fatalf("endpoint = %v, %v; want %s", u, err, tt.want) + } + }) + } + for _, addr := range []string{"", "https://", "ftp://node", "https://user:pass@node", "https://node/path", "https://node?", "https://node#", "https://node#frag", "https://node:0", "https://node:65536", "https://node:bad", "https://node\r\nX: bad"} { + t.Run(addr, func(t *testing.T) { + if _, err := endpointURL(addr, "http"); err == nil { + t.Fatalf("accepted %q", addr) + } + }) + } +} + +func TestAgentEndpoint(t *testing.T) { + for _, tt := range []struct { + addr, scheme, wantScheme, wantAuthority, wantTarget string + }{ + {"node:7777", "http", "http", "node:7777", "node:7777"}, + {"HTTPS://node", "http", "https", "node", "node:443"}, + {"https://[::1]", "http", "https", "[::1]", "[::1]:443"}, + {"[::1]:7777", "https", "https", "[::1]:7777", "[::1]:7777"}, + } { + t.Run(tt.addr, func(t *testing.T) { + scheme, authority, target := agentEndpoint(tt.addr, tt.scheme) + if scheme != tt.wantScheme || authority != tt.wantAuthority || target != tt.wantTarget { + t.Fatalf("agent endpoint = %q, %q, %q; want %q, %q, %q", scheme, authority, target, tt.wantScheme, tt.wantAuthority, tt.wantTarget) + } + }) + } +} + +func TestTLSControlAndRelay(t *testing.T) { + agent := newAgentServer(t, silkdtest.ServeConn) + secure := httptest.NewTLSServer(agent.Config.Handler) + t.Cleanup(secure.Close) + roots := x509.NewCertPool() + roots.AddCert(secure.Certificate()) + for _, tt := range []struct { + name string + opts []ClientOption + }{ + {"tls option", []ClientOption{WithTLSConfig(&tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12})}}, + {"http transport", []ClientOption{WithHTTPClient(secure.Client())}}, + } { + t.Run(tt.name, func(t *testing.T) { + c, err := Connect(secure.URL, tt.opts...) + if err != nil { + t.Fatal(err) + } + resp, err := c.roundTrip(t.Context(), http.MethodGet, secure.URL, "/healthz", nil, "") + if err != nil { + t.Fatal(err) + } + _ = resp.Body.Close() + for _, owner := range []string{secure.URL, strings.TrimPrefix(secure.URL, "https://")} { + sb := c.Attach(owner, "sb_test", "tok") + out, err := sb.Exec(t.Context(), "echo", "tls") + if err != nil || out != "tls\n" { + t.Fatalf("exec = %q, %v", out, err) + } + pc, err := sb.DialPort(t.Context(), 8080) + if err != nil { + t.Fatal(err) + } + defer pc.Close() + if _, err = pc.Write([]byte("tail")); err != nil { + t.Fatal(err) + } + if err = pc.CloseWrite(); err != nil { + t.Fatal(err) + } + body, err := io.ReadAll(pc) + if err != nil || string(body) != "tail" { + t.Fatalf("tail = %q, %v", body, err) + } + } + }) + } +} + +func TestTLSVerification(t *testing.T) { + agent := newAgentServer(t, silkdtest.ServeConn) + secure := httptest.NewTLSServer(agent.Config.Handler) + t.Cleanup(secure.Close) + roots := x509.NewCertPool() + roots.AddCert(secure.Certificate()) + for _, tt := range []struct { + name string + cfg *tls.Config + }{ + {"untrusted CA", &tls.Config{MinVersion: tls.VersionTLS12}}, + {"wrong name", &tls.Config{RootCAs: roots, ServerName: "wrong.example", MinVersion: tls.VersionTLS12}}, + } { + t.Run(tt.name, func(t *testing.T) { + c, err := Connect(secure.URL, WithTLSConfig(tt.cfg)) + if err != nil { + t.Fatal(err) + } + if resp, requestErr := c.roundTrip(t.Context(), http.MethodGet, secure.URL, "/healthz", nil, ""); requestErr == nil { + _ = resp.Body.Close() + t.Fatal("control request accepted invalid certificate") + } + if _, err := c.dialAgent(t.Context(), secure.URL, "sb_test", "tok"); err == nil { + t.Fatal("relay accepted invalid certificate") + } + }) + } +} + +func TestTLSHandshakeCancellation(t *testing.T) { + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer l.Close() + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + done := make(chan struct{}) + go func() { + defer close(done) + conn, acceptErr := l.Accept() + if acceptErr != nil { + return + } + defer conn.Close() + cancel() + _, _ = io.Copy(io.Discard, conn) + }() + c, err := Connect("https://" + l.Addr().String()) + if err != nil { + t.Fatal(err) + } + if _, err := c.dialAgent(ctx, c.addr, "sb_test", "tok"); !errors.Is(err, context.Canceled) { + t.Fatalf("got %v, want cancellation", err) + } + <-done +} diff --git a/sdk/go/silkd/silkdtest/fake.go b/sdk/go/silkd/silkdtest/fake.go index 19b789c0..b7c194b1 100644 --- a/sdk/go/silkd/silkdtest/fake.go +++ b/sdk/go/silkd/silkdtest/fake.go @@ -83,8 +83,6 @@ func (f *Fake) ServeConn(conn net.Conn) { ptyEcho(conn, r) case *wire.PtyResize: send(conn, wire.Done{}) - case *wire.PortForward: - portEcho(conn, r, req.Port) default: if !serveCommon(conn, r, req) { errFrame(conn, wire.KindUnimplemented, "silkdtest: "+req.Op()) diff --git a/sdk/go/silkd/silkdtest/silkdtest.go b/sdk/go/silkd/silkdtest/silkdtest.go index 7132ceb2..df29cf61 100644 --- a/sdk/go/silkd/silkdtest/silkdtest.go +++ b/sdk/go/silkd/silkdtest/silkdtest.go @@ -1,6 +1,4 @@ -// Package silkdtest fakes a silkd daemon for host-side tests: deterministic -// frame semantics over any listener, plus the hybrid-vsock muxer -// handshake for tests that dial a UDS the way sandboxd does. +// Package silkdtest fakes a silkd daemon for host-side tests: deterministic frame semantics over any listener, plus the hybrid-vsock muxer handshake for tests that dial a UDS the way sandboxd does. package silkdtest import ( @@ -13,24 +11,17 @@ import ( "github.com/cocoonstack/sandbox/protocol/wire" ) -// Serve accepts connections until l closes, speaking one RPC per connection -// and closing it after the terminal frame, like silkd. Semantics: info → -// InfoResp; exec echo → stdout of the args; exec cat → echoes stdin frames -// until stdin_close; exec false → exit 1; exec sleep → started, then blocks -// until the client disconnects; anything else → an error frame. +// Serve accepts connections until l closes, speaking one RPC per connection and closing it after the terminal frame, like silkd. func Serve(l net.Listener) { acceptLoop(l, ServeConn) } -// ServeConn speaks one RPC on an already-open connection and closes it — -// for tests that produce the conn themselves (e.g. after an HTTP hijack). +// ServeConn handles one RPC on an open connection, then closes it. func ServeConn(conn net.Conn) { handle(conn, bufio.NewReader(conn)) } -// ListenHybrid serves the muxer handshake on a UDS: each connection must -// open with "CONNECT ", is answered "OK ", then speaks the -// Serve protocol. The returned closer stops the listener. +// ListenHybrid serves the muxer handshake on a UDS: each connection must open with "CONNECT ", is answered "OK ", then speaks the Serve protocol. func ListenHybrid(sockPath string, port int) (io.Closer, error) { l, err := net.Listen("unix", sockPath) if err != nil { @@ -52,8 +43,6 @@ func ListenHybrid(sockPath string, port int) (io.Closer, error) { return l, nil } -// handle speaks one RPC; r must be the connection's only reader — a second -// buffered layer would swallow frames the first one read ahead. func handle(conn net.Conn, r *bufio.Reader) { defer func() { _ = conn.Close() }() req, err := recvRequest(r) @@ -65,14 +54,14 @@ func handle(conn net.Conn, r *bufio.Reader) { } } -// serveCommon handles the verbs shared by the stateless server and the Fake, -// reporting whether it recognized the request. func serveCommon(conn net.Conn, r *bufio.Reader, req wire.Request) bool { switch req := req.(type) { case *wire.Info: send(conn, &wire.InfoResp{Version: "silkdtest", Proto: wire.ProtoVersion}) case *wire.Exec: serveExec(conn, r, req) + case *wire.PortForward: + portEcho(conn, r, req.Port) default: return false } diff --git a/sdk/go/upgrade.go b/sdk/go/upgrade.go index c78d7f3a..c1f2241d 100644 --- a/sdk/go/upgrade.go +++ b/sdk/go/upgrade.go @@ -3,37 +3,48 @@ package sandbox import ( "bufio" "context" + "crypto/tls" "fmt" "net" "net/http" "strings" ) -// dialAgent opens the data-plane connection to the owner node: a raw TCP dial -// plus a hand-rolled HTTP Upgrade, so the result is a real net.Conn with no -// Transport pooling or proxying underneath a long-lived byte stream. func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (net.Conn, error) { + if strings.ContainsAny(id, "\r\n\x00") || strings.ContainsAny(token, "\r\n\x00") { + return nil, fmt.Errorf("agent upgrade: id or token contains a control character") + } + scheme, authority, target := agentEndpoint(addr, c.scheme) var d net.Dialer - raw, err := d.DialContext(ctx, "tcp", addr) + raw, err := d.DialContext(ctx, "tcp", target) if err != nil { return nil, err } stop := context.AfterFunc(ctx, func() { _ = raw.Close() }) defer stop() - - // id/token interpolate into the raw request; CR/LF would inject headers. - if strings.ContainsAny(id, "\r\n\x00") || strings.ContainsAny(token, "\r\n\x00") { - _ = raw.Close() - return nil, fmt.Errorf("agent upgrade: id or token contains a control character") + conn := raw + if scheme == httpsScheme { + cfg := c.tlsConfig.Clone() + if cfg.ServerName == "" { + cfg.ServerName, _, _ = net.SplitHostPort(target) + } + cfg.NextProtos = []string{"http/1.1"} + secured := tls.Client(raw, cfg) + if err = secured.HandshakeContext(ctx); err != nil { + _ = raw.Close() + return nil, fmt.Errorf("agent TLS handshake: %w", err) + } + conn = secured } - req := "GET /v1/sandboxes/" + id + "/agent HTTP/1.1\r\nHost: " + addr + + + req := "GET /v1/sandboxes/" + id + "/agent HTTP/1.1\r\nHost: " + authority + "\r\nConnection: Upgrade\r\nUpgrade: silkd\r\nAuthorization: Bearer " + token + "\r\n\r\n" - if _, err = raw.Write([]byte(req)); err != nil { + if _, err = conn.Write([]byte(req)); err != nil { _ = raw.Close() return nil, fmt.Errorf("write upgrade request: %w", err) } - br := bufio.NewReader(raw) + br := bufio.NewReader(conn) resp, err := http.ReadResponse(br, nil) if err != nil { _ = raw.Close() @@ -48,14 +59,29 @@ func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (net.Con _ = raw.Close() return nil, err } - return &upgradedConn{Conn: raw, r: br}, nil + return &upgradedConn{Conn: conn, r: br}, nil } -// upgradedConn keeps reading through the handshake reader forever, so frame -// bytes the server coalesced behind the 101 are never lost. type upgradedConn struct { net.Conn r *bufio.Reader } func (u *upgradedConn) Read(p []byte) (int, error) { return u.r.Read(p) } + +func agentEndpoint(addr, scheme string) (string, string, string) { + authority := addr + if s, a, ok := strings.Cut(addr, "://"); ok { + scheme, authority = strings.ToLower(s), a + } + target := authority + if _, _, err := net.SplitHostPort(authority); err != nil { + port := "80" + if scheme == httpsScheme { + port = "443" + } + host := strings.TrimSuffix(strings.TrimPrefix(authority, "["), "]") + target = net.JoinHostPort(host, port) + } + return scheme, authority, target +} diff --git a/sdk/python/cocoonsandbox/checkpoint.py b/sdk/python/cocoonsandbox/checkpoint.py index a8550f0e..414d51e1 100644 --- a/sdk/python/cocoonsandbox/checkpoint.py +++ b/sdk/python/cocoonsandbox/checkpoint.py @@ -1,10 +1,8 @@ -"""Checkpoint: a captured sandbox state bound to the node that holds it. -Branch any number of fresh sandboxes from the captured moment; the source -keeps running and can be checkpointed again, so captures form a tree.""" +"""Checkpoint handles bound to the node that holds their captured state.""" from __future__ import annotations -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from .client import Client @@ -14,7 +12,7 @@ class Checkpoint: """A captured sandbox state on its owner node.""" - def __init__(self, client: Client, addr: str, rec: dict): + def __init__(self, client: Client, addr: str, rec: dict[str, Any]) -> None: self._client = client self._addr = addr self.id = rec["id"] @@ -23,15 +21,10 @@ def __init__(self, client: Client, addr: str, rec: dict): self.created_at = rec.get("created_at", "") def new(self, ttl_seconds: int = 0) -> Sandbox: - """Claims a fresh sandbox branched from the checkpoint, following a - redirect to the node that actually holds it; if every candidate - fails transiently, the claim falls back to the origin once so it - heals (pulls the checkpoint) locally.""" + """Claims from the checkpoint, following redirects with one origin fallback.""" claim = {"ttl_seconds": ttl_seconds} if ttl_seconds else {} return self._client._claim_from(self._addr, claim, f"/v1/checkpoints/{self.id}/claim", "claim checkpoint") def delete(self) -> None: - """Removes the checkpoint and broadcasts the drop; cleanup is - best-effort eventual, bounded by checkpoint_ttl_hours (see - docs/sdk-python.md).""" + """Deletes the checkpoint with eventual peer cleanup bounded by checkpoint_ttl_hours.""" self._client._request(self._addr, "DELETE", f"/v1/checkpoints/{self.id}", None, "delete checkpoint") diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index bedc4f23..0febb4d1 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -1,20 +1,21 @@ -"""Client: the control-plane entry point. Dial any node of a cluster; a -claim either lands locally or follows one MOVED-style redirect to the node -that has capacity, and the returned Sandbox is bound to its owner.""" +"""Client: the control-plane entry point.""" from __future__ import annotations import http.client import json import queue +import ssl import threading import urllib.error import urllib.parse import urllib.request from collections.abc import Callable, Iterable, Mapping, Sequence -from typing import TypeVar +from typing import Any, TypeVar, cast from .checkpoint import Checkpoint +from .conn import Conn, dial_agent +from .endpoint import _endpoint_url from .errors import APIError from .sandbox import Sandbox @@ -26,8 +27,16 @@ class Client: """Talks to one sandboxd node (and, transparently, its cluster).""" - def __init__(self, addr: str, api_token: str = "", timeout: float = 120.0) -> None: - self.addr = addr.split(",")[0].strip() + def __init__( + self, addr: str, api_token: str = "", timeout: float = 120.0, *, ssl_context: ssl.SSLContext | None = None + ) -> None: + endpoint = _endpoint_url(addr.split(",")[0].strip()) + self.addr = endpoint.geturl().removeprefix("http://") + self._scheme = endpoint.scheme + self._ssl_context = ssl_context + if ssl_context is not None: + ssl_context.set_alpn_protocols(["http/1.1"]) + self._opener: urllib.request.OpenerDirector | None = None self.api_token = api_token self.timeout = timeout @@ -41,37 +50,26 @@ def new( volumes: list[str | Mapping[str, str]] | None = None, mount: bool = True, ) -> Sandbox: - """Claims a sandbox; a warm hit is milliseconds. On a cluster a warm - miss may redirect to a peer, followed transparently; if every - candidate fails transiently, the claim falls back to the origin - once so it provisions or heals locally. mount=False attaches the - volumes without mounting them, leaving that — and the flush — to - the workload: releasing without a clean self-umount discards - unsynced guest pages, since the sandbox performs no sync.""" + """Claims a sandbox; a warm hit is milliseconds.""" claim = _claim_body(template, net, size, ttl_seconds, volumes, mount, claim_ref) return self._claim_from(self.addr, claim) def delete_template(self, template: str, net: str = "", size: str = "") -> None: - """Removes a promoted template by name; on a cluster the delete - follows gossip to the owner node (one hop).""" + """Removes a promoted template by name; on a cluster the delete follows gossip to the owner node (one hop).""" query = _template_query(template, net, size) path = "/v1/templates?" + urllib.parse.urlencode(query) reply = self._request(self.addr, "DELETE", path, None, "delete template") candidates = (reply or {}).get("redirect") or [] if not candidates: return - # the owner answers for itself under no_redirect, never a second hop. query["no_redirect"] = "1" path = "/v1/templates?" + urllib.parse.urlencode(query) _try_each(candidates, lambda peer: self._request(peer, "DELETE", path, None, "delete template")) def lookup(self, id: str, token: str) -> Sandbox: - """Relocates a handle from id + token: asks the entry node and every - mesh peer concurrently, binding to whichever confirms ownership - first — one dead peer must not cost its full timeout.""" + """Finds the owner by probing the entry and peers concurrently.""" def probe(addr: str) -> Sandbox: - # bounded like _peers: a scatter loser must not hold a socket for the full timeout. reply = self._request( addr, "GET", @@ -90,28 +88,24 @@ def probe(addr: str) -> Sandbox: raise APIError("lookup", 404, f"no owner found for {id}") from None def attach(self, owner_addr: str, id: str, token: str) -> Sandbox: - """Binds a handle to an already-claimed sandbox whose owner address is - known (an apiserver annotation, say), with no lookup round-trip.""" + """Binds a handle to a known owner without a lookup.""" return Sandbox(client=self, id=id, token=token, owner=owner_addr) - def sandboxes(self) -> list[dict]: - """Lists the claims this token may see: id, key, deadline, claim_ref — - never tokens or host paths.""" + def sandboxes(self) -> list[dict[str, Any]]: + """Lists the claims this token may see: id, key, deadline, claim_ref — never tokens or host paths.""" reply = self._request(self.addr, "GET", "/v1/sandboxes", None, "list sandboxes") return [dict(sb) for sb in reply.get("sandboxes") or []] - def drain(self) -> dict: - """Cordons the node (root token): new claims are refused, live ones run - to their leases.""" + def drain(self) -> dict[str, Any]: + """Cordons the node (root token): new claims are refused, live ones run to their leases.""" return self._request(self.addr, "POST", "/v1/drain", None, "drain") - def uncordon(self) -> dict: + def uncordon(self) -> dict[str, Any]: """Lifts a drain on the node (root token).""" return self._request(self.addr, "DELETE", "/v1/drain", None, "uncordon") def checkpoint(self, id: str) -> Checkpoint: - """A handle for a known checkpoint id, bound to the entry node — no - listing round-trip; an unknown id surfaces as 404 at claim time.""" + """Binds a checkpoint handle to the entry node without a lookup.""" return Checkpoint(self, self.addr, {"id": id}) def checkpoints(self) -> list[Checkpoint]: @@ -119,16 +113,16 @@ def checkpoints(self) -> list[Checkpoint]: reply = self._request(self.addr, "GET", "/v1/checkpoints", None, "list checkpoints") return [Checkpoint(self, self.addr, rec) for rec in reply.get("checkpoints") or []] - def volumes(self) -> list[dict]: + def volumes(self) -> list[dict[str, Any]]: """Lists the caller-visible fleet catalog; availability is local.""" reply = self._request(self.addr, "GET", "/v1/volumes", None, "list volumes") return [dict(volume) for volume in reply.get("volumes") or []] - def info(self) -> dict: + def info(self) -> dict[str, Any]: """The node's pool/claim counters, as served by GET /v1/info.""" return self._request(self.addr, "GET", "/v1/info", None, "info") - def _claim_from(self, addr: str, claim: dict, path: str = "/v1/claim", verb: str = "claim") -> Sandbox: + def _claim_from(self, addr: str, claim: dict[str, Any], path: str = "/v1/claim", verb: str = "claim") -> Sandbox: reply = self._post_json(addr, path, claim, verb) redirect = reply.get("redirect") or [] if not redirect: @@ -137,14 +131,13 @@ def _claim_from(self, addr: str, claim: dict, path: str = "/v1/claim", verb: str if reply.get("require_promoted"): claim["require_promoted"] = True - def post(peer): + def post(peer: str) -> dict[str, Any]: return self._post_json(peer, path, claim, verb) owner, reply = _redirect_fallback(addr, redirect, post, verb) return self._handle_from(owner, reply) def _peers(self) -> list[str]: - # /v1/peers is tenant-accessible; /v1/info is operator-only, so a tenant cannot read peers from it. try: return ( self._request( @@ -155,7 +148,7 @@ def _peers(self) -> list[str]: except APIError: return [] - def _handle_from(self, dialed: str, reply: dict) -> Sandbox: + def _handle_from(self, dialed: str, reply: dict[str, Any]) -> Sandbox: return Sandbox( client=self, id=reply["id"], @@ -167,24 +160,29 @@ def _handle_from(self, dialed: str, reply: dict) -> Sandbox: volumes=reply.get("volumes") or [], ) - def _post_json(self, addr: str, path: str, body: dict, verb: str) -> dict: + def _post_json(self, addr: str, path: str, body: dict[str, Any], verb: str) -> dict[str, Any]: return self._request(addr, "POST", path, body, verb) def _request( - self, addr: str, method: str, path: str, body: dict | None, verb: str, bearer: str = "", timeout: float = 0.0 - ) -> dict: - """Issues one control-plane request. bearer overrides the api token — - sandbox-scoped verbs (release, hibernate) authenticate with the - per-sandbox token instead; timeout overrides the client default (0 keeps it).""" + self, + addr: str, + method: str, + path: str, + body: dict[str, Any] | None, + verb: str, + bearer: str = "", + timeout: float = 0.0, + ) -> dict[str, Any]: data = json.dumps(body).encode() if body is not None else None - req = urllib.request.Request(f"http://{addr}{path}", data=data, method=method) + url = addr + path if "://" in addr else f"{self._scheme}://{addr}{path}" + req = urllib.request.Request(url, data=data, method=method) if data is not None: req.add_header("Content-Type", "application/json") token = bearer or self.api_token if token: req.add_header("Authorization", f"Bearer {token}") try: - with urllib.request.urlopen(req, timeout=timeout or self.timeout) as resp: + with self._open(req, timeout or self.timeout) as resp: raw = resp.read() except urllib.error.HTTPError as exc: try: @@ -195,15 +193,32 @@ def _request( except urllib.error.URLError as exc: raise APIError(verb, 0, str(exc.reason)) from None except (OSError, http.client.HTTPException) as exc: - # a reset or truncated body read is neither HTTPError nor URLError raise APIError(verb, 0, str(exc)) from None if not raw: return {} try: - return json.loads(raw) + return cast(dict[str, Any], json.loads(raw)) except json.JSONDecodeError as exc: raise APIError(verb, 0, "malformed JSON in response") from exc + def _dial(self, addr: str, sandbox_id: str, token: str, deadline: float | None = None) -> Conn: + origin = addr if "://" in addr else f"{self._scheme}://{addr}" + context = self._tls() if origin.startswith("https://") else None + return dial_agent(origin, sandbox_id, token, self.timeout, deadline, ssl_context=context) + + def _open(self, req: urllib.request.Request, timeout: float) -> Any: + if req.type != "https": + return urllib.request.urlopen(req, timeout=timeout) + if self._opener is None: + self._opener = urllib.request.build_opener(urllib.request.HTTPSHandler(context=self._tls())) + return self._opener.open(req, timeout=timeout) + + def _tls(self) -> ssl.SSLContext: + if self._ssl_context is None: + self._ssl_context = ssl.create_default_context() + self._ssl_context.set_alpn_protocols(["http/1.1"]) + return self._ssl_context + def _claim_body( template: str, @@ -213,8 +228,8 @@ def _claim_body( volumes: list[str | Mapping[str, str]] | None = None, mount: bool = True, claim_ref: str = "", -) -> dict: - claim = {"template": template} +) -> dict[str, Any]: + claim: dict[str, Any] = {"template": template} if net: claim["net"] = net if size: @@ -230,7 +245,7 @@ def _claim_body( return claim -def _volume_body(volume: str | Mapping[str, str], mount: bool) -> dict: +def _volume_body(volume: str | Mapping[str, str], mount: bool) -> dict[str, str]: if isinstance(volume, str): return {"name": volume} if not isinstance(volume, Mapping): @@ -253,7 +268,7 @@ def _volume_body(volume: str | Mapping[str, str], mount: bool) -> dict: return body -def _template_query(template: str, net: str, size: str) -> dict: +def _template_query(template: str, net: str, size: str) -> dict[str, str]: query = {"template": template} if net: query["net"] = net @@ -264,22 +279,18 @@ def _template_query(template: str, net: str, size: str) -> dict: def _error_message(raw: bytes) -> str: try: - return json.loads(raw)["error"] + return cast(str, json.loads(raw)["error"]) except (ValueError, KeyError, TypeError): return raw.decode(errors="replace").strip() def _retry_miss(exc: APIError) -> bool: - """Default _try_each policy: only a miss (404) or a dead peer (status 0) moves on.""" return exc.status in (404, 0) def _try_each( candidates: Iterable[str], call: Callable[[str], T], retry: Callable[[APIError], bool] = _retry_miss ) -> T: - """An APIError for which retry(exc) is true moves on to the next - candidate and the last such error propagates; any other raises at once. - candidates must be non-empty.""" last_error = None for addr in candidates: try: @@ -288,33 +299,17 @@ def _try_each( if not retry(exc): raise last_error = exc - raise last_error + raise cast(APIError, last_error) def _retry_transient(exc: APIError) -> bool: - """Origin-fallback policy: worth the round-trip for a transport failure - (status 0), a miss (404), full (429), mid-heal (503), an engine/proxy - failure (500/502/504), or a mid-rotation 401 (the origin proved the - token valid by issuing the redirect). A served 4xx like a bad request, - a forbidden token, or an egress conflict is definitive: the origin - would fail the same way.""" return exc.status in (0, 401, 404, 429, 503, 500, 502, 504) def _redirect_fallback( - origin: str, candidates: Sequence[str], post: Callable[[str], dict], verb: str -) -> tuple[str, dict]: - """Retries broadly across candidates (any failure moves to the next) so - one wrong candidate doesn't cost one that would still succeed. If every - candidate is exhausted and the last failure was transient - (_retry_transient), gives the origin one more no_redirect attempt -- the - node that issued the redirect provisions or heals locally instead of - leaving the claim stuck on stale gossip. A definitive last failure skips - the fallback: the origin would fail the same way. A second-level redirect - (a compliant server never sends one once no_redirect is set) fails the - candidate rather than being followed.""" - - def attempt(addr): + origin: str, candidates: Sequence[str], post: Callable[[str], dict[str, Any]], verb: str +) -> tuple[str, dict[str, Any]]: + def attempt(addr: str) -> tuple[str, dict[str, Any]]: reply = post(addr) if reply.get("redirect"): raise APIError(verb, 0, f"{addr} redirected again despite no_redirect") @@ -328,18 +323,14 @@ def attempt(addr): try: return attempt(origin) except APIError as origin_exc: - # both halves matter: why the claim left the origin, and why returning did not help. combined = f"{origin_exc.message} (after redirect targets failed: {exc.message})" raise APIError(verb, origin_exc.status, combined) from origin_exc def _scatter(addrs: Sequence[str], probe: Callable[[str], T]) -> T: - """When all probes fail the last error propagates. Loser threads are - daemons whose requests die with _request's own timeout; the queue is - bounded by len(addrs) so they never block. addrs must be non-empty.""" - results = queue.Queue(maxsize=len(addrs)) + results: queue.Queue[tuple[T | None, Exception | None]] = queue.Queue(maxsize=len(addrs)) - def run(addr): + def run(addr: str) -> None: try: results.put((probe(addr), None)) except Exception as exc: # any escape would hang the drain below @@ -351,6 +342,6 @@ def run(addr): for _ in addrs: value, error = results.get() if error is None: - return value + return cast(T, value) last_error = error - raise last_error + raise cast(Exception, last_error) diff --git a/sdk/python/cocoonsandbox/conn.py b/sdk/python/cocoonsandbox/conn.py index 39a65569..5c87ffab 100644 --- a/sdk/python/cocoonsandbox/conn.py +++ b/sdk/python/cocoonsandbox/conn.py @@ -1,13 +1,14 @@ -"""One relayed silkd connection per RPC: a raw TCP socket hand-upgraded to -the sandboxd agent relay (Upgrade: silkd), then newline-JSON frames.""" +"""Silkd frames over an HTTP(S) Upgrade relay.""" from __future__ import annotations import contextlib import socket +import ssl import time +import urllib.parse from collections.abc import Iterator -from typing import BinaryIO, TypeVar +from typing import Any, BinaryIO, Protocol, TypeVar from .errors import APIError, ProtocolError, SilkdError from .frames import MAX_FRAME, decode_response, encode_request @@ -15,20 +16,22 @@ _CloseableT = TypeVar("_CloseableT", bound="_Closeable") -class _Closeable: +class _Closeable(Protocol): """Context-manager mixin for handles whose exit is just close().""" def __enter__(self: _CloseableT) -> _CloseableT: return self - def __exit__(self, *exc) -> None: + def __exit__(self, *exc: object) -> None: self.close() + def close(self) -> None: ... + class Conn(_Closeable): """A live frame stream to one sandbox's silkd, via the owner node.""" - def __init__(self, sock: socket.socket, reader: BinaryIO): + def __init__(self, sock: socket.socket, reader: BinaryIO) -> None: self._sock = sock self._reader = reader @@ -40,10 +43,8 @@ def abort(self) -> None: with contextlib.suppress(OSError): self._sock.shutdown(socket.SHUT_RDWR) - def recv(self) -> dict: - """Returns the next frame; raises SilkdError on an error frame and - ProtocolError on EOF, an oversized line, or an undecodable one — every - failure is typed.""" + def recv(self) -> dict[str, Any]: + """Reads one frame or raises a typed protocol or guest error.""" try: line = self._reader.readline(MAX_FRAME + 1) except OSError as exc: @@ -60,19 +61,14 @@ def recv(self) -> dict: raise SilkdError(frame.get("kind", "internal"), frame.get("message", "")) return frame - def recv_until(self, *terminal: str) -> Iterator[dict]: - """Yields frames until one of the terminal types arrives; the - terminal frame is yielded last.""" + def recv_until(self, *terminal: str) -> Iterator[dict[str, Any]]: + """Yields frames until one of the terminal types arrives; the terminal frame is yielded last.""" while True: frame = self.recv() yield frame if frame["type"] in terminal: return - def close_write(self) -> None: - with contextlib.suppress(OSError): - self._sock.shutdown(socket.SHUT_WR) - def close(self) -> None: self.abort() try: @@ -81,24 +77,40 @@ def close(self) -> None: self._sock.close() -def dial_agent(addr: str, sandbox_id: str, token: str, timeout: float, deadline: float | None = None) -> Conn: +def dial_agent( + addr: str, + sandbox_id: str, + token: str, + timeout: float, + deadline: float | None = None, + *, + ssl_context: ssl.SSLContext | None = None, +) -> Conn: """Opens one TCP/HTTP Upgrade relay within timeout and the optional deadline.""" - # id/token interpolate into the raw request line; CR/LF would inject headers. for name, value in (("sandbox id", sandbox_id), ("token", token)): if any(c in value for c in "\r\n\0"): raise APIError("agent upgrade", 0, f"{name} contains a control character") - host, port = addr.rsplit(":", 1) + endpoint = urllib.parse.urlsplit(addr if "://" in addr else f"http://{addr}") + host = endpoint.hostname + port = endpoint.port or (443 if endpoint.scheme == "https" else 80) try: - sock = socket.create_connection((host, int(port)), timeout=_remaining_timeout(timeout, deadline)) + sock = socket.create_connection((host, port), timeout=_remaining_timeout(timeout, deadline)) except OSError as exc: raise ProtocolError(f"dial {addr}: {exc}") from exc - # Nagle off: exec/write send small back-to-back frames before the first read. sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) reader = None try: + if endpoint.scheme == "https": + context = ssl_context or ssl.create_default_context() + sock = context.wrap_socket(sock, server_hostname=host, do_handshake_on_connect=False) + sock.settimeout(_remaining_timeout(timeout, deadline)) + try: + sock.do_handshake() + except OSError as exc: + raise ProtocolError(f"tls handshake {addr}: {exc}") from exc request = ( f"GET /v1/sandboxes/{sandbox_id}/agent HTTP/1.1\r\n" - f"Host: {addr}\r\n" + f"Host: {endpoint.netloc}\r\n" "Connection: Upgrade\r\n" "Upgrade: silkd\r\n" f"Authorization: Bearer {token}\r\n" @@ -126,7 +138,6 @@ def dial_agent(addr: str, sandbox_id: str, token: str, timeout: float, deadline: except ValueError as exc: raise ProtocolError("invalid content-length in upgrade reply") from exc if code != 101: - # a bogus content-length must not buffer unbounded bytes. sock.settimeout(_remaining_timeout(timeout, deadline)) body = reader.read(min(body_len, MAX_FRAME)).decode(errors="replace") if body_len else "" raise APIError("agent upgrade", code, body.strip() or status.strip()) @@ -134,7 +145,6 @@ def dial_agent(addr: str, sandbox_id: str, token: str, timeout: float, deadline: sock.settimeout(None) return Conn(sock, reader) except Exception: - # makefile() holds a ref on the socket; close it too or the fd lingers. if reader is not None: reader.close() sock.close() diff --git a/sdk/python/cocoonsandbox/endpoint.py b/sdk/python/cocoonsandbox/endpoint.py new file mode 100644 index 00000000..42c53e6e --- /dev/null +++ b/sdk/python/cocoonsandbox/endpoint.py @@ -0,0 +1,29 @@ +"""HTTP origins shared by control requests and agent relays.""" + +from __future__ import annotations + +import urllib.parse + + +def _endpoint_url(addr: str, scheme: str = "http") -> urllib.parse.SplitResult: + if any(ord(c) <= 32 or ord(c) == 127 for c in addr): + raise ValueError("sandboxd endpoint contains whitespace or a control character") + if "://" not in addr: + addr = f"{scheme}://{addr}" + endpoint = urllib.parse.urlsplit(addr) + if ( + endpoint.scheme not in ("http", "https") + or not endpoint.hostname + or endpoint.username is not None + or endpoint.path not in ("", "/") + or "?" in addr + or "#" in addr + ): + raise ValueError("sandboxd endpoint must be an http or https origin") + try: + port = endpoint.port + except ValueError: + port = 0 + if port == 0: + raise ValueError("sandboxd endpoint port must be between 1 and 65535") + return endpoint._replace(path="") diff --git a/sdk/python/cocoonsandbox/frames.py b/sdk/python/cocoonsandbox/frames.py index 36bd39a1..a1cea6bf 100644 --- a/sdk/python/cocoonsandbox/frames.py +++ b/sdk/python/cocoonsandbox/frames.py @@ -1,12 +1,11 @@ -"""silkd wire frames: newline-delimited JSON, requests tagged by "op", -responses by "type", binary payloads base64 in "data" fields. Mirrors the Go -and Rust implementations; all three round-trip protocol/wire/fixtures/v1.""" +"""Silkd newline-delimited JSON frames with base64 binary payloads.""" from __future__ import annotations import base64 import binascii import json +from typing import Any PROTO_VERSION = 1 MAX_FRAME = 8 * 1024 * 1024 @@ -16,8 +15,7 @@ def encode_request(op: str, **fields: object) -> bytes: - """Renders {"v":1,"op":...,fields} with a trailing newline; None fields - are omitted and bytes values ride base64 under their field name.""" + """Encodes one request, omitting None fields and base64-encoding byte values.""" frame = {"v": PROTO_VERSION, "op": op} for key, value in fields.items(): if value is None: @@ -28,9 +26,8 @@ def encode_request(op: str, **fields: object) -> bytes: return json.dumps(frame, separators=(",", ":")).encode() + b"\n" -def decode_response(line: bytes) -> dict: - """Parses one response frame; the returned dict carries its tag under - "type" and any binary payload decoded under "data".""" +def decode_response(line: bytes) -> dict[str, Any]: + """Decodes one response with binary payloads under data.""" # base64 is JSON-escape-free, so an exactly-shaped data frame slices without json.loads. if line.startswith(b'{"type":"'): te = line.find(b'"', 9) diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index a6668c3a..e8ab84cb 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -1,7 +1,4 @@ -"""Sandbox: the data-plane handle. Every RPC opens one relayed silkd -connection via the owner node; sessions and processes are server-side state, -so nothing is lost between calls — including across a transparent hibernate -wake.""" +"""Sandbox: the data-plane handle.""" from __future__ import annotations @@ -10,10 +7,10 @@ import threading import time from collections.abc import Callable, Iterator -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any, cast from .checkpoint import Checkpoint -from .conn import Conn, _Closeable, dial_agent +from .conn import Conn, _Closeable from .errors import APIError, ExitError, ProtocolError, SandboxError from .frames import BULK_CHUNK, FS_CHUNK from .template import Template @@ -34,7 +31,7 @@ def __init__( deadline: str = "", from_checkpoint: str = "", template_digest: str = "", - volumes: list[dict] | None = None, + volumes: list[dict[str, str]] | None = None, ) -> None: self._client = client self.id = id @@ -48,7 +45,7 @@ def __init__( def __enter__(self) -> Sandbox: return self - def __exit__(self, *exc) -> None: + def __exit__(self, *exc: object) -> None: try: self.close() except APIError: @@ -59,14 +56,13 @@ def exec( self, *argv: str, cwd: str = "", - env: dict | None = None, + env: dict[str, str] | None = None, user: str = "", session: str = "", stdin: bytes = b"", timeout: float | None = None, ) -> str: - """Runs argv to completion and returns stdout; a non-zero exit raises - ExitError carrying stderr. timeout is run()'s wall clock.""" + """Runs argv to completion and returns stdout; a non-zero exit raises ExitError carrying stderr.""" out, err = bytearray(), bytearray() code = self.run( list(argv), @@ -87,7 +83,7 @@ def run( self, argv: list[str], cwd: str = "", - env: dict | None = None, + env: dict[str, str] | None = None, user: str = "", session: str = "", stdin: bytes = b"", @@ -95,10 +91,7 @@ def run( on_stderr: Callable[[bytes], object] | None = None, timeout: float | None = None, ) -> int: - """Runs argv streaming stdio through the callbacks (raw bytes — chunk - boundaries may split multi-byte sequences); returns the exit code. - timeout is a wall clock over the dial and the run: at its end the - connection is cut, which kills the command, and TimeoutError is raised.""" + """Streams raw output bytes and returns the exit code; timeout bounds the entire call.""" if timeout is not None and timeout <= 0: raise ValueError("timeout must be positive") deadline = None if timeout is None else time.monotonic() + timeout @@ -121,7 +114,6 @@ def run( detach=False, session=session or None, ) - # the guest stops draining stdin while blocked on stdout, so feeding it fully first deadlocks. pump = threading.Thread(target=_feed_stdin, args=(conn, stdin), daemon=True) pump.start() code = _pump_stdio(conn, on_stdout, on_stderr) @@ -137,22 +129,19 @@ def run( raise ProtocolError("exec stream ended without an exit frame") return code - def spawn(self, *argv: str, cwd: str = "", env: dict | None = None, user: str = "") -> int: - """Starts argv detached, returning its pid immediately; the process - keeps a bounded output ring readable later via logs()/attach().""" + def spawn(self, *argv: str, cwd: str = "", env: dict[str, str] | None = None, user: str = "") -> int: + """Starts a detached process with a bounded output ring and returns its pid.""" started = self._call( "exec", "started", argv=list(argv), cwd=cwd or None, env=env, user=user or None, detach=True ) - return started["pid"] + return cast(int, started["pid"]) - def ps(self) -> list[dict]: - """Lists tracked processes: {pid, argv, detached, state, - exit_code?, started_at_epoch_secs}.""" - return self._call("ps", "procs")["procs"] + def ps(self) -> list[dict[str, Any]]: + """Lists tracked processes: {pid, argv, detached, state, exit_code?, started_at_epoch_secs}.""" + return cast(list[dict[str, Any]], self._call("ps", "procs")["procs"]) def kill(self, pid: int, signal: int | None = None) -> None: - """Signals a tracked process (default SIGKILL); killing one that - already exited is a no-op success.""" + """Signals a tracked process (default SIGKILL); killing one that already exited is a no-op success.""" self._done_rpc("kill", pid=pid, signal=signal or None) def logs( @@ -161,8 +150,7 @@ def logs( on_stdout: Callable[[bytes], object] | None = None, on_stderr: Callable[[bytes], object] | None = None, ) -> int | None: - """Replays a process's ring-buffered output through the callbacks; - returns its exit code if it already exited, else None.""" + """Replays buffered output and returns the exit code, or None if the process still runs.""" return self._drain_proc("logs", pid, on_stdout, on_stderr) def attach( @@ -171,9 +159,7 @@ def attach( on_stdout: Callable[[bytes], object] | None = None, on_stderr: Callable[[bytes], object] | None = None, ) -> int | None: - """Replays buffered output then follows live output until the - process exits, returning its exit code (None only if the proc table - dropped it mid-attach).""" + """Replays then follows output; returns None if the process record disappears.""" return self._drain_proc("attach", pid, on_stdout, on_stderr) def write_file(self, path: str, data: bytes, mode: int | None = None) -> None: @@ -189,17 +175,17 @@ def read_file(self, path: str) -> bytes: conn.send("fs_read", path=path) return _drain_data(conn) - def list_dir(self, path: str) -> list[dict]: + def list_dir(self, path: str) -> list[dict[str, Any]]: with self._dial() as conn: conn.send("fs_list", path=path) - entries: list[dict] = [] + entries: list[dict[str, Any]] = [] for frame in conn.recv_until("done"): entries.extend(frame.get("entries") or []) return entries - def stat(self, path: str) -> dict: + def stat(self, path: str) -> dict[str, Any]: """Returns {kind, size, mode, mtime_epoch_secs} for path.""" - return self._call("fs_stat", "stat", path=path)["info"] + return cast(dict[str, Any], self._call("fs_stat", "stat", path=path)["info"]) def mkdir(self, path: str, parents: bool = False) -> None: self._done_rpc("fs_mkdir", path=path, parents=parents or None) @@ -211,8 +197,7 @@ def rename(self, src: str, dst: str) -> None: self._done_rpc("fs_rename", **{"from": src, "to": dst}) def push(self, dest: str, tar_stream: bytes) -> None: - """Extracts a tar archive into dest — atomic against a truncated - stream; the only project-ingestion path on the no-network lane.""" + """Extracts a tar stream into dest; a truncated stream leaves dest untouched.""" with self._dial() as conn: conn.send("fs_push", dest=dest) _send_chunks(conn, tar_stream, chunk=BULK_CHUNK) @@ -225,30 +210,27 @@ def pull(self, path: str) -> bytes: conn.send("fs_pull", path=path) return _drain_data(conn) - def find(self, path: str, pattern: str, glob: str = "") -> list[dict]: + def find(self, path: str, pattern: str, glob: str = "") -> list[dict[str, Any]]: return list(self.find_iter(path, pattern, glob)) - def find_iter(self, path: str, pattern: str, glob: str = "") -> Iterator[dict]: - """Yields matches as they stream. Closing the generator closes the - connection and ends the walk in the guest; wrap it in contextlib.closing - for deterministic cleanup, since only CPython finalizes on refcount.""" + def find_iter(self, path: str, pattern: str, glob: str = "") -> Iterator[dict[str, Any]]: + """Yields matches as they stream.""" with self._dial() as conn: conn.send("fs_find", path=path, pattern=pattern, glob=glob or None) for f in conn.recv_until("done"): if f["type"] == "match": yield f - def replace(self, files: list[str], pattern: str, replacement: str) -> list[dict]: + def replace(self, files: list[str], pattern: str, replacement: str) -> list[dict[str, Any]]: with self._dial() as conn: conn.send("fs_replace", files=files, pattern=pattern, replacement=replacement) return [f for f in conn.recv_until("done") if f["type"] == "replaced"] def git_clone(self, url: str, path: str, branch: str = "", depth: int = 0, auth: str = "") -> None: - """Clones into path (egress lane only; the none lane answers a typed - unimplemented error pointing at push).""" + """Clones into path (egress lane only; the none lane answers a typed unimplemented error pointing at push).""" self._done_rpc("git_clone", url=url, path=path, branch=branch or None, depth=depth or None, auth=auth or None) - def git_status(self, path: str) -> dict: + def git_status(self, path: str) -> dict[str, Any]: return self._call("git_status", "git_status_result", path=path) def git_add(self, path: str, files: list[str]) -> None: @@ -256,7 +238,10 @@ def git_add(self, path: str, files: list[str]) -> None: def git_commit(self, path: str, message: str, author: str) -> str: """Commits staged changes; returns the commit hash.""" - return self._call("git_commit", "git_commit_result", path=path, message=message, author=author).get("hash", "") + return cast( + str, + self._call("git_commit", "git_commit_result", path=path, message=message, author=author).get("hash", ""), + ) def git_push(self, path: str, auth: str = "") -> None: self._done_rpc("git_push", path=path, auth=auth or None) @@ -264,7 +249,7 @@ def git_push(self, path: str, auth: str = "") -> None: def git_pull(self, path: str, auth: str = "") -> None: self._done_rpc("git_pull", path=path, auth=auth or None) - def git_branches(self, path: str) -> dict: + def git_branches(self, path: str) -> dict[str, Any]: return self._call("git_branch", "git_branches", path=path, action="list") def git_checkout(self, path: str, name: str) -> None: @@ -277,14 +262,12 @@ def git_delete_branch(self, path: str, name: str) -> None: self._done_rpc("git_branch", path=path, action="delete", name=name) def watch(self, path: str, recursive: bool = False) -> Watcher: - """Streams filesystem events under path; events after the returned - Watcher exists are guaranteed captured. Close it to stop.""" + """Streams filesystem events under path; events after the returned Watcher exists are guaranteed captured.""" conn, _ = self._open_stream("fs_watch", path=path, recursive=recursive or None) return Watcher(conn) - def session(self, cwd: str = "", env: dict | None = None) -> Session: - """Creates a persistent shell: cd/export/aliases survive across exec - calls routed into it.""" + def session(self, cwd: str = "", env: dict[str, str] | None = None) -> Session: + """Creates a persistent shell: cd/export/aliases survive across exec calls routed into it.""" created = self._call("session_create", "session_created", cwd=cwd or None, env=env) return Session(self, created["id"]) @@ -292,8 +275,7 @@ def sessions(self) -> list[str]: return self._call("session_list", "sessions").get("sessions") or [] def fork(self, count: int, ttl_seconds: int = 0) -> list[Sandbox]: - """Clones this sandbox into count independent children carrying its - exact memory and disk state; all-or-nothing.""" + """Clones memory and disk into independent children, all-or-nothing.""" body = {"token": self.token, "count": count} if ttl_seconds: body["ttl_seconds"] = ttl_seconds @@ -301,15 +283,13 @@ def fork(self, count: int, ttl_seconds: int = 0) -> list[Sandbox]: return [self._client._handle_from(self.owner, child) for child in reply.get("children") or []] def hibernate(self) -> None: - """Snapshots and stops the VM, freeing its memory; the next call that - reaches the guest wakes it transparently, state intact.""" + """Snapshots and stops the VM; the next guest call restores its state.""" self._client._request( self.owner, "POST", f"/v1/sandboxes/{self.id}/hibernate", None, "hibernate", bearer=self.token ) def checkpoint(self, name: str = "") -> Checkpoint: - """Captures full state without stopping the sandbox; the returned - Checkpoint branches fresh sandboxes from that exact moment.""" + """Captures a branchable snapshot without stopping the sandbox.""" body = {"token": self.token} if name: body["name"] = name @@ -317,8 +297,7 @@ def checkpoint(self, name: str = "") -> Checkpoint: return Checkpoint(self._client, self.owner, reply["checkpoint"]) def promote(self, template: str) -> Template: - """Publishes this sandbox's state as a claimable template on its - node; the returned handle is bound to that node.""" + """Publishes a claimable template and returns an owner-bound handle.""" reply = self._client._post_json( self.owner, f"/v1/sandboxes/{self.id}/promote", {"token": self.token, "template": template}, "promote" ) @@ -333,38 +312,33 @@ def promote(self, template: str) -> Template: ) def start_lsp(self, language: str, root: str = "") -> Lsp: - """Spawns the language server the flavor image provides for language; - the base image ships none, so it raises SilkdError(kind="not_found"). - The returned handle streams JSON-RPC over the relay.""" + """Starts the registered language server; missing servers raise not_found.""" started = self._call("lsp_start", "lsp_started", language=language, root=root or None) return Lsp(self, started["server_id"]) - def open_pty(self, cols: int = 80, rows: int = 24, cwd: str = "", env: dict | None = None, user: str = "") -> Pty: - """Runs the guest shell under a pty; returns a byte-stream handle. - A pty is a process guest-side: resize goes through its pid.""" + def open_pty( + self, cols: int = 80, rows: int = 24, cwd: str = "", env: dict[str, str] | None = None, user: str = "" + ) -> Pty: + """Runs the guest shell under a pty; returns a byte-stream handle.""" conn, started = self._open_stream( "pty_open", expect="started", cols=cols, rows=rows, cwd=cwd or None, env=env, user=user or None ) return Pty(self, conn, started["pid"]) def proxy_port(self, local_addr: str, port: int) -> socket.socket: - """Serves a guest port on a local listener for unmodified local - tools; returns the listening socket (close it to stop). local_addr - is "host:port"; port 0 picks a free one.""" + """Serves a guest port on a local socket; closing the listener stops new connections.""" host, _, lport = local_addr.rpartition(":") listener = socket.create_server((host or "127.0.0.1", int(lport))) threading.Thread(target=self._proxy_accept_loop, args=(listener, port), daemon=True).start() return listener def preview_url(self, port: int, ttl_seconds: int = 0) -> str: - """Mints a shareable URL serving the guest HTTP port from a browser, - valid for ttl_seconds (the node clamps it to the claim's lease). - Requires the node to have preview configured.""" + """Mints a guest HTTP URL whose lifetime is clamped to the claim's lease.""" body = {"token": self.token, "port": port} if ttl_seconds: body["ttl_seconds"] = ttl_seconds reply = self._client._post_json(self.owner, f"/v1/sandboxes/{self.id}/preview", body, "preview") - return reply["url"] + return cast(str, reply["url"]) def dial_port(self, port: int) -> PortConn: """Opens a byte stream to 127.0.0.1:port inside the guest.""" @@ -372,9 +346,7 @@ def dial_port(self, port: int) -> PortConn: return PortConn(conn) def close(self) -> None: - """Releases the sandbox; its VM is destroyed. Releasing one already - gone is not an error — double-release and reap races stay silent, - matching the Go SDK.""" + """Releases the sandbox; its VM is destroyed.""" try: self._client._request( self.owner, "POST", f"/v1/sandboxes/{self.id}/release", None, "release", bearer=self.token @@ -384,11 +356,9 @@ def close(self) -> None: raise def _dial(self, deadline: float | None = None) -> Conn: - return dial_agent(self.owner, self.id, self.token, self._client.timeout, deadline) + return self._client._dial(self.owner, self.id, self.token, deadline) - def _open_stream(self, op: str, expect: str = "ready", **fields) -> tuple[Conn, dict]: - """Dials, sends op, and waits for the handshake frame, closing the - conn on any failure so no socket leaks on the error path.""" + def _open_stream(self, op: str, expect: str = "ready", **fields: object) -> tuple[Conn, dict[str, Any]]: conn = self._dial() try: conn.send(op, **fields) @@ -411,14 +381,13 @@ def _proxy_conn(self, local: socket.socket, port: int) -> None: local.close() return - def pump_out(): + def pump_out() -> None: with contextlib.suppress(SandboxError, OSError): while True: chunk = guest.recv() if not chunk: break local.sendall(chunk) - # a close cannot wake the read below, and the socket outlives it with contextlib.suppress(OSError): local.shutdown(socket.SHUT_WR) @@ -431,7 +400,6 @@ def pump_out(): if not chunk: break guest.send(chunk) - # half-close, not close: the guest's reply is still in flight. with contextlib.suppress(SandboxError, OSError): guest.close_write() pump.join() @@ -440,12 +408,12 @@ def pump_out(): with contextlib.suppress(OSError): local.close() - def _call(self, op: str, expect: str, **fields) -> dict: + def _call(self, op: str, expect: str, **fields: object) -> dict[str, Any]: with self._dial() as conn: conn.send(op, **fields) return _expect(conn, expect) - def _done_rpc(self, op: str, **fields) -> None: + def _done_rpc(self, op: str, **fields: object) -> None: self._call(op, "done", **fields) def _drain_proc( @@ -482,8 +450,7 @@ def __init__(self, conn: Conn) -> None: self._conn = conn self.error: Exception | None = None - def __iter__(self) -> Iterator[dict]: - # a transport failure ends iteration and sets error; a SilkdError frame propagates. + def __iter__(self) -> Iterator[dict[str, Any]]: while True: try: frame = self._conn.recv() @@ -507,8 +474,7 @@ def __init__(self, sandbox: Sandbox, conn: Conn, pid: int) -> None: self.exit_code: int | None = None def read(self) -> bytes: - """The next output chunk; b'' once the shell exits, after which - exit_code holds the shell's status.""" + """The next output chunk; b'' once the shell exits, after which exit_code holds the shell's status.""" frame = self._conn.recv() if frame["type"] == "exit": self.exit_code = frame.get("code") @@ -526,17 +492,14 @@ def close(self) -> None: class Lsp(_Closeable): - """A language server in the sandbox, spoken to over the relay. silkd is a - broker: it pipes JSON-RPC bytes; the caller frames and correlates.""" + """A language server in the sandbox, spoken to over the relay.""" def __init__(self, sandbox: Sandbox, server_id: str) -> None: self._sandbox = sandbox self.server_id = server_id def request(self) -> PortConn: - """Opens the JSON-RPC byte stream: writes go to the server's stdin, - recv returns its stdout. A server serves one request for its - lifetime; closing the stream ends the session and reaps it.""" + """Opens the JSON-RPC byte stream: writes go to the server's stdin, recv returns its stdout.""" conn, _ = self._sandbox._open_stream("lsp_request", server_id=self.server_id) return PortConn(conn) @@ -570,7 +533,6 @@ def recv(self) -> bytes: def close_write(self) -> None: """Half-close: signals EOF to the guest side; reads keep working.""" self._conn.send("data_end") - self._conn.close_write() def close(self) -> None: self._conn.close() @@ -606,8 +568,6 @@ def _feed_stdin(conn: Conn, stdin: bytes) -> None: def _pump_stdio( conn: Conn, on_stdout: Callable[[bytes], object] | None, on_stderr: Callable[[bytes], object] | None ) -> int | None: - """Streams stdout/stderr frames into the callbacks until the terminal - frame: the exit code, or None when the stream ends with done.""" for frame in conn.recv_until("exit", "done"): t = frame["type"] if t == "stdout" and on_stdout: @@ -615,11 +575,11 @@ def _pump_stdio( elif t == "stderr" and on_stderr: on_stderr(frame["data"]) elif t == "exit": - return frame["code"] + return cast(int, frame["code"]) return None -def _expect(conn: Conn, frame_type: str) -> dict: +def _expect(conn: Conn, frame_type: str) -> dict[str, Any]: frame = conn.recv() if frame["type"] != frame_type: raise ProtocolError(f"expected {frame_type}, got {frame['type']}") diff --git a/sdk/python/pyproject.toml b/sdk/python/pyproject.toml index ff3d09bd..9ff9e834 100644 --- a/sdk/python/pyproject.toml +++ b/sdk/python/pyproject.toml @@ -13,6 +13,10 @@ readme = "README.md" [tool.setuptools.packages.find] include = ["cocoonsandbox*"] +[tool.mypy] +strict = true +files = ["cocoonsandbox"] + [tool.ruff] line-length = 120 target-version = "py39" diff --git a/sdk/python/tests/test_endpoint.py b/sdk/python/tests/test_endpoint.py new file mode 100644 index 00000000..1a20ffad --- /dev/null +++ b/sdk/python/tests/test_endpoint.py @@ -0,0 +1,64 @@ +import socket +import threading + +import pytest + +from cocoonsandbox import Client, ProtocolError +from cocoonsandbox.conn import dial_agent +from cocoonsandbox.endpoint import _endpoint_url + + +@pytest.mark.parametrize( + ("addr", "scheme", "want"), + [ + ("localhost:7777", "http", "http://localhost:7777"), + ("localhost:443", "https", "https://localhost:443"), + ("https://example.com/", "http", "https://example.com"), + ("http://example.com", "https", "http://example.com"), + ("[::1]:7777", "https", "https://[::1]:7777"), + ("https://[::1]", "http", "https://[::1]"), + ], +) +def test_endpoint_url(addr: str, scheme: str, want: str) -> None: + assert _endpoint_url(addr, scheme).geturl() == want + + +@pytest.mark.parametrize( + "addr", + [ + "", + "https://", + "ftp://node", + "https://u:p@node", + "https://node/path", + "https://node?", + "https://node#", + "https://node#x", + "https://node:0", + "https://node:65536", + "https://node:bad", + "https://node\r\nX: bad", + ], +) +def test_endpoint_rejects_invalid_origins(addr: str) -> None: + with pytest.raises(ValueError): + Client(addr) + + +def test_tls_handshake_timeout_closes_socket() -> None: + with socket.create_server(("127.0.0.1", 0)) as server: + done = threading.Event() + + def stall() -> None: + with server.accept()[0] as conn: + conn.settimeout(5) + while conn.recv(4096): + pass + done.set() + + thread = threading.Thread(target=stall, daemon=True) + thread.start() + with pytest.raises(ProtocolError): + dial_agent(f"https://127.0.0.1:{server.getsockname()[1]}", "sb_1", "token", 0.05) + assert done.wait(5), "timed-out handshake kept its socket open" + thread.join()