diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml index 7e85518e4..572872d1c 100644 --- a/.github/workflows/python.yml +++ b/.github/workflows/python.yml @@ -24,6 +24,9 @@ on: - ".github/workflows/python.yml" workflow_dispatch: +permissions: + contents: read + jobs: test: runs-on: ubuntu-latest diff --git a/Makefile b/Makefile index fe6ba27af..ff2ed85ed 100644 --- a/Makefile +++ b/Makefile @@ -43,12 +43,12 @@ help: ## show this list @grep -E '^[a-z-]+:.*##' $(MAKEFILE_LIST) | awk -F ':.*## ' '{printf "%-14s %s\n", $$1, $$2}' test: ## Rust tests: boot/init + silkd - cd boot/init && cargo test - cd silkd && cargo test + cd boot/init && cargo test --locked + cd silkd && cargo test --locked lint: ## Rust fmt --check + clippy -D warnings: boot/init + silkd - cd boot/init && cargo fmt --check && cargo clippy --all-targets -- -D warnings - cd silkd && cargo fmt --check && cargo clippy --all-targets -- -D warnings + cd boot/init && cargo fmt --check && cargo clippy --locked --all-targets -- -D warnings + cd silkd && cargo fmt --check && cargo clippy --locked --all-targets -- -D warnings sh-lint: ## shellcheck every tracked shell script git ls-files '*.sh' | xargs shellcheck @@ -78,7 +78,7 @@ boot-debug: ## boot image with busybox + /bin/sh on fatal errors docker build -t $(BOOT_IMAGE)-debug --build-arg KERNEL_VERSION=$(KERNEL_VERSION) --build-arg KERNEL_MIRROR=$(KERNEL_MIRROR) --build-arg INITRD_DEBUG=1 boot extract: ## dump /boot artifacts into dist/ for boot-bench.sh - rm -rf dist && mkdir -p dist + rm -rf dist/boot && mkdir -p dist cid=$$(docker create $(EXTRACT_IMAGE)) && docker cp $$cid:/boot dist/boot && docker rm $$cid extract-debug: ## extract from the boot-debug image diff --git a/docs/cluster.md b/docs/cluster.md index b288f83c0..00e9305d9 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -341,11 +341,12 @@ checkpoint store and keeps each claim's registry row on the latest snapshot: the next round, a release, an archive, a wake, the hourly sweep and a restart settle it by reading the row: a row on that checkpoint promotes it, another owner's row drops the claim, and an absent row means the adopter finished - when the checkpoint is gone and a write that never landed when it is still - there. A refused write reads the row: one this node still owns only had a + when the checkpoint is gone or this node confirmed an earlier row, and a + write that never landed otherwise. A refused write reads the row: one this node still owns only had a stale epoch and is refreshed, and an absent or foreign row drops the claim and - its VM. A claim gets its first replica one interval after it is claimed (or - after a restart). A hibernated, archived or failed claim, + its VM. A claim gets its first replica one interval after it is claimed, after a + restart, or after it wakes from an archive or an adoption; until a claim woken + that way commits its first replica it has no row, and it is lost with its node. A hibernated, archived or failed claim, one whose tenant was removed, and one with volumes (capture refuses them) is skipped; a replicated claim that gains volumes drops its row and replica. The claim's policy pool, not its own template key, diff --git a/docs/deploy.md b/docs/deploy.md index 80e26a274..2a250501d 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -147,7 +147,7 @@ the intersection of the two. | `preview_secret` | — | cluster-shared HMAC secret signing preview tokens (all nodes share one) | | `preview_advertise` | = `preview_listen` | the browser-facing preview base URL, minted into every preview URL, so it must name a routable host (a wildcard `preview_listen` needs it set explicitly); nodes behind one TLS proxy may share it, while signed tokens route internally through each owner's `advertise_addr` | | `checkpoint_dir` | `/checkpoints` | where checkpoints and promoted templates live. Point it at a shared FUSE mount (JuiceFS over object storage, NFS) and every node sharing the mount can branch every checkpoint — records are generation-addressed with `meta.json` as the atomic commit pointer, so no cross-node locking is required of the filesystem. A re-publish retains its superseded export generation for at least ~1h and until a following hourly sweep so an in-flight clone that resolved the old metadata can finish; budget the current generation plus every generation retained across that grace-and-sweep window. An explicit delete can make a concurrent clone fail visibly. One contract on any shared root (mount or bucket): a template key has a single writer — promotes go to the sandbox's owner node, and operators must not race promotes of one name from different nodes (checkpoint ids are node-generated and never collide). Abandoned staging (`*.tmp`) is removed only once it is older than an hour, at startup and on the hourly generation sweep, so a publish in flight on another node of a shared root survives; the `meta.json` commit pointer is fsynced with its directory | -| `checkpoint_store` | dir | checkpoint AND promoted-template backend (both live in one store root, id-namespaced ck_/tp_): `{"kind": "s3", "s3": {"bucket": "…", "prefix": "ck/", "endpoint": "…", "region": "…", "force_path_style": true}}` stores checkpoints in object storage (any node claims any checkpoint, no shared mount needed). Credentials come from the standard AWS chain (env/IAM role), never this file. Holes and all-zero chunks are never uploaded, and a fetch recreates the file at its full size and writes only the present chunks, so a mostly-empty disk costs its data, not its size, on the wire and on the fetching node. Consecutive captures of one claim upload only the chunks that changed (the first capture after a wake uploads everything again when no other live record of that claim remains, because the wake deletes the record it restored from). sandboxd never deletes a chunk or a manifest from the bucket; [Reclaiming S3 storage](#reclaiming-s3-storage) has the key layout and the operator's job. A publish or fetch is aborted after 30 minutes. Do not put an expiry lifecycle rule on the prefix: it would delete chunks a live record still names. Budget storage for every generation and chunk until your reclaim runs. When caching a new generation, each node prunes other cached generations of that record unused for at least an hour. The periodic generation sweep drops a node's cache of a record once the record is gone from the bucket and the cache is unused for at least an hour. Absent = the dir backend at `checkpoint_dir` | +| `checkpoint_store` | dir | checkpoint AND promoted-template backend (both live in one store root, id-namespaced ck_/tp_): `{"kind": "s3", "s3": {"bucket": "…", "prefix": "ck/", "endpoint": "…", "region": "…", "force_path_style": true}}` stores checkpoints in object storage (any node claims any checkpoint, no shared mount needed). Credentials come from the standard AWS chain (env/IAM role), never this file. Holes and all-zero chunks are never uploaded, and a fetch recreates the file at its full size and writes only the present chunks, so a mostly-empty disk costs its data, not its size, on the wire and on the fetching node. Consecutive captures of one claim upload only the chunks that changed (the first capture after a wake uploads everything again when no other live record of that claim remains, because the wake deletes the record it restored from). sandboxd never deletes a chunk or a manifest from the bucket; [Reclaiming S3 storage](#reclaiming-s3-storage) has the key layout and the operator's job. A publish or fetch is aborted after 30 minutes. Each publish in flight holds up to 192 MiB of chunk buffers and sandboxd does not cap concurrent checkpoints or promotes, so the layer above it limits how many a tenant starts. Do not put an expiry lifecycle rule on the prefix: it would delete chunks a live record still names. Budget storage for every generation and chunk until your reclaim runs. When caching a new generation, each node prunes other cached generations of that record unused for at least an hour. The periodic generation sweep drops a node's cache of a record once the record is gone from the bucket and the cache is unused for at least an hour. Absent = the dir backend at `checkpoint_dir` | | `checkpoint_ttl_hours` | 0 (keep forever) | ages out checkpoints older than this; the sweep runs hourly and at startup. Explicit deletes never wait for it. Must be nonzero and match fleet-wide when `checkpoint_peer_heal` is on ([why](cluster.md#delete-is-eventual-not-a-fleet-wide-revocation)) | | `checkpoint_peer_heal` | false | on a cluster, lets a node pull a checkpoint it lacks from a peer — found via a live probe, not gossip — rather than failing the branch; see [placement lifecycle](cluster.md#checkpoints-on-a-cluster). Three requirements, all enforced at config load: a nonempty `api_token` (the blob transfer between peers authenticates with it; without one the raw record stream would be open), `mesh.cluster_key` set (the pull presents the fleet `api_token` to an address learned from the peer probe, so the gossip layer carrying that address must itself be authenticated), and `checkpoint_ttl_hours` nonzero. A shared checkpoint store (`checkpoint_store` kind `s3`) ignores this setting — every node already resolves every checkpoint directly, so there is nothing to heal | | `warm_max` (pool entry) | 0 (static) | turns on the demand-adaptive watermark for that pool: the warm target rises from `warm` toward `warm_max` while claims arrive faster than the measured provision lead covers, and decays back over ~a minute of silence. The pool follows the target down: warm VMs above it are destroyed once the count has stayed above it for a full decay period, so a burst does not leave the pool parked at its high-water mark | @@ -655,6 +655,7 @@ ExecStart=/usr/local/bin/sandboxd -config /etc/sandboxd/config.json Restart=on-failure RestartSec=5s KillMode=process +TimeoutStopSec=27min Environment=SANDBOXD_LOG_LEVEL=info [Install] @@ -673,8 +674,12 @@ test the output for emptiness. Stopping sandboxd leaves VMs alive; the next start reconciles them. The unit's `KillMode=process` is what guarantees it: systemd signals sandboxd alone, so a -VMM that cocoon could not move into its own scope, and any `cocoon` call still -in flight, outlive the stop instead of dying with the service's cgroup. Claimed +VMM that cocoon could not move into its own scope outlives the stop instead of +dying with the service's cgroup. sandboxd starts no new `cocoon` command once +stopped and waits for the running ones, since cocoon dies on its next log line +once sandboxd's end of its stderr closes, and a hibernate killed that way leaves +the guest paused. A save or hibernate can take 26 minutes, hence +`TimeoutStopSec=27min`; stops normally take seconds. Claimed sandboxes are reaped when their TTL expires (default 5m, capped at 24h). A snapshot save, hibernate, export, restore or clone has 15 minutes. Once canceled, a diff --git a/docs/egress.md b/docs/egress.md index 5717bdd82..2ef9eae75 100644 --- a/docs/egress.md +++ b/docs/egress.md @@ -159,7 +159,8 @@ domain policy first; the allow-list widens the IP gate only. A [config reload](deploy.md#reloading-the-config) applies policy, internal allow, secret and upstream changes without a restart: a live claim's next -request or connection is evaluated against the new policy. +request or connection is evaluated against the new policy. Whether a claim has +the SOCKS5 door is settled when it is claimed. Policy is per pool and per egress class; a tenant claim's effective policy is the intersection of its pool's and its tenant's class (a request must pass diff --git a/docs/live-volumes.md b/docs/live-volumes.md index 3d959a48b..6dac28bed 100644 --- a/docs/live-volumes.md +++ b/docs/live-volumes.md @@ -62,7 +62,7 @@ with client.new("rt:24.04", ttl_seconds=900) as sb: ``` A successful run reads `v1`, then `v2`, and preserves every identity assertion. -The repeated full target verifies that a settled mount can be retained. +The repeated full target is a no-op on a settled mount. This example starts without other volumes. Retain any volumes you intend to keep in a full PUT, or use PATCH to replace only the dataset names. All mounted and attach-only entries support live removal in either mode. @@ -72,16 +72,14 @@ mean an empty target. Go uses the same owning handle and client API token. ```go -view, err := sb.PatchVolumes(ctx, sandbox.VolumePatch{ +v2 := sandbox.Volume{Name: "dataset-v2", Mount: "/datasets/current"} +_, err := sb.PatchVolumes(ctx, sandbox.VolumePatch{ Remove: []string{"dataset-v1"}, - Add: []sandbox.Volume{{Name: "dataset-v2", Mount: "/datasets/current"}}, + Add: []sandbox.Volume{v2}, }) if err != nil { return err } -if view.Converging { - return errors.New("volume request still running") -} ``` For a 409 mount conflict, use the guest path in `states[].error` to close diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index 286b58252..5af584792 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -477,7 +477,8 @@ for a blocked device call. Capture operations (hibernate, fork, checkpoint, and promote) are refused while either set is nonempty or a convergence request still owns admitted volume holds. -Device calls honor cancellation and have a 30-second ceiling per volume; a +Guest device calls honor cancellation and have a 30-second ceiling per volume, and a +`cocoon` attach or detach can take 30 more to exit; a caller that disconnects cancels the run, leaving interrupted states to retry. Journal writes are atomic. Host-crash durability follows `sync_claims`. An older daemon ignores `VolumesWant`; finish pending volume changes before downgrading. @@ -651,8 +652,8 @@ the export they were set on, which the new one replaces. 400 invalid name, 401 bad api token, 409 when the name collides with a configured pool, the template is owned by another tenant, or the sandbox is -on the egress lane, has volumes attached (see [egress](egress.md)), or is -archived (`POST /wake` or an exec wakes it first), 404 unknown id or wrong +on the egress lane, has volumes attached (see [egress](egress.md)), has failed, +or is archived (`POST /wake` or an exec wakes it first), 404 unknown id or wrong sandbox token, 429 while the disk is under [`disk_reserve`](deploy.md), 503 `lease lost; retry` for a replicated sandbox while its node's lease is overdue. @@ -665,7 +666,8 @@ success, 404 unknown template, 409 when the key belongs to a configured pool (those goldens are owned by the node config). `digest=` deletes only while the record still has that digest and answers 412 otherwise, so a caller that observed an older build never removes the newer one that replaced -it. On a cluster, a node that does not +it on this node; on a shared store a promote committing on another node at the +same moment is not covered. On a cluster, a node that does not hold the template but sees an owner in gossip answers `200 {"redirect": [addrs]}` — the claim redirect shape — and the SDK retries the delete at the owner. The retry carries `no_redirect=1`, mirroring the claim @@ -684,7 +686,7 @@ Labels follow the [claim `metadata`](#post-v1claim) rules — at most 16 pairs, the same key and value grammar, at most 4 KiB as JSON — and replace the whole map; `{}` clears it. `digest=` writes only while the record still has that digest and answers 412 otherwise, so a writer that observed an -older build never relabels the newer one. They are stored beside the template's record, so they +older build never relabels the newer one on this node. They are stored beside the template's record, so they survive a restart, are listed by the node that promoted the template or loaded it at startup, and go with the template when it is deleted; a re-promote starts with none. A tenant may label only templates it promoted — anything else is 404, root labels anything. 204 on @@ -872,8 +874,8 @@ answers `200 {"checkpoint": {id, name?, sandbox_id, key, tenant?, created_at, policy_source?, no_egress?, guest_env?}}` — `tenant` records the calling tenant, absent for root. 400 bad body or name, 401 bad api token, 404 unknown id or wrong sandbox token, 409 egress-lane sandbox, one with volumes attached (see -[egress](egress.md)), or an archived one (`POST /wake` or an exec wakes it -first), 429 while the disk is under [`disk_reserve`](deploy.md), 503 `lease lost; retry` +[egress](egress.md)), a failed one, or an archived one (`POST /wake` or an exec +wakes it first), 429 while the disk is under [`disk_reserve`](deploy.md), 503 `lease lost; retry` for a replicated sandbox while its node's lease is overdue. ## POST /v1/checkpoints/{id}/claim @@ -1062,7 +1064,7 @@ template clone starts with the document its source held. The node writes it to is a no-op. silkd serves the address only in images that alias it on `lo`. 204 on success; 400 when the body is not one JSON object within the bound; 404 -unknown id; 409 when the sandbox is hibernated, mid-transition or archived (this +unknown id; 409 when the sandbox is failed, hibernated, mid-transition or archived (this verb never wakes one) or its image serves no instance metadata. ## GET /metrics diff --git a/e2e/e2e_test.go b/e2e/e2e_test.go index 07413a1f6..cf17c4045 100644 --- a/e2e/e2e_test.go +++ b/e2e/e2e_test.go @@ -37,11 +37,11 @@ const nodeToken = "node-token" var testKey = types.PoolKey{Template: "rt:24.04", Net: types.NetNone, Size: types.SizeSmall} -func TestClaimColdThenWarmThenCloseThroughTheWholeStack(t *testing.T) { +func TestClaimThenWarmClaimThenCloseThroughTheWholeStack(t *testing.T) { st := startStack(t, config.PoolSpec{PoolKey: testKey, Warm: 1}) var sb *sandbox.Sandbox - t.Run("cold claim before any refill", func(t *testing.T) { + t.Run("first claim", func(t *testing.T) { var err error sb, err = st.client.New(t.Context(), "rt:24.04") if err != nil { @@ -58,11 +58,15 @@ func TestClaimColdThenWarmThenCloseThroughTheWholeStack(t *testing.T) { t.Run("refill then warm claim", func(t *testing.T) { waitFor(t, st.oneWarm) + warmBefore := st.mgr.Counters().ClaimsWarm warm, err := st.client.New(t.Context(), "rt:24.04") if err != nil { t.Fatalf("New: %v", err) } defer warm.Close() + if got := st.mgr.Counters().ClaimsWarm; got != warmBefore+1 { + t.Errorf("warm claims %d, want %d", got, warmBefore+1) + } if out, err := warm.Exec(t.Context(), "echo", "warm"); err != nil || out != "warm\n" { t.Errorf("exec on warm claim: %q, %v", out, err) } @@ -70,7 +74,7 @@ func TestClaimColdThenWarmThenCloseThroughTheWholeStack(t *testing.T) { t.Run("close releases and revokes access", func(t *testing.T) { if sb == nil { - t.Skip("cold claim subtest failed") + t.Skip("first claim subtest failed") } if err := sb.Close(); err != nil { t.Fatalf("Close: %v", err) diff --git a/mcp/server_test.go b/mcp/server_test.go index 07f7cb568..428938ac5 100644 --- a/mcp/server_test.go +++ b/mcp/server_test.go @@ -197,12 +197,14 @@ func TestCloseBoxesWaitsForEveryReleaseOnce(t *testing.T) { } var wg sync.WaitGroup for range 2 { - wg.Go(srv.closeBoxes) + wg.Go(func() { + srv.closeBoxes() + if got := releases.Load(); got != 3 { + t.Errorf("releases = %d after closeBoxes returned, want 3", got) + } + }) } wg.Wait() - if got := releases.Load(); got != 3 { - t.Errorf("releases = %d after closeBoxes returned, want 3", got) - } } func TestExecRejectsAnEmptyCommand(t *testing.T) { diff --git a/packaging/sandboxd.service b/packaging/sandboxd.service index 8318e67b7..8b6d3077c 100644 --- a/packaging/sandboxd.service +++ b/packaging/sandboxd.service @@ -8,6 +8,7 @@ ExecStart=/usr/local/bin/sandboxd -config /etc/sandboxd/config.json Restart=on-failure RestartSec=5s KillMode=process +TimeoutStopSec=27min Environment=SANDBOXD_LOG_LEVEL=info [Install] diff --git a/sandboxd/egress/proxy.go b/sandboxd/egress/proxy.go index e9d1e57e7..b89183c5c 100644 --- a/sandboxd/egress/proxy.go +++ b/sandboxd/egress/proxy.go @@ -44,7 +44,8 @@ const ( var ( // ErrRateLimited refuses a new upstream connection over the claim's connection rate. - ErrRateLimited = errors.New("egress: connection rate limit") + ErrRateLimited = errors.New("egress: connection rate limit") + ErrInternalAddr = errors.New("egress: blocked internal address") // hopHeaders are hop-by-hop and proxy-scoped headers this hop owns, never passed on. hopHeaders = []string{ @@ -96,8 +97,11 @@ type Event struct { } func (ev *Event) attach(conn net.Conn, err error) { - if errors.Is(err, ErrRateLimited) { + switch { + case errors.Is(err, ErrRateLimited): ev.Decision = DecisionRateLimited + case errors.Is(err, ErrInternalAddr): + ev.Decision = DecisionDeny } if ue, ok := errors.AsType[*upstreamError](err); ok { ev.Upstream = ue.host @@ -387,12 +391,15 @@ func (p *Proxy) relay(w http.ResponseWriter, r *http.Request, ev Event, rule Rul out.Body = body } var received int64 + var failed bool defer func() { var sent int64 if body != nil { sent = body.n.Load() } - p.reportTransfer(ev, sent, received) + if !failed || sent > 0 { + p.reportTransfer(ev, sent, received) + } }() resp, err := p.transport(route, mitm).RoundTrip(out) @@ -403,6 +410,7 @@ func (p *Proxy) relay(w http.ResponseWriter, r *http.Request, ev Event, rule Rul ev.attach(conn, err) p.record(ev) if err != nil { + failed = true upstreamFailed(w, err) return } diff --git a/sandboxd/egress/proxy_test.go b/sandboxd/egress/proxy_test.go index 574098441..b0e539fd0 100644 --- a/sandboxd/egress/proxy_test.go +++ b/sandboxd/egress/proxy_test.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "errors" + "fmt" "io" "net" "net/http" @@ -554,28 +555,46 @@ func TestAuditNamesTheConnectionEnds(t *testing.T) { } } -func TestARateLimitedConnectionAnswers429AndIsAudited(t *testing.T) { - events := make(chan Event, 4) - dial := DialFunc(func(context.Context, string, string) (net.Conn, error) { return nil, ErrRateLimited }) - p := New(Policy{Allow: []Rule{{Host: "*"}}}, nil, nil, dial, func(ev Event) { events <- ev }, nil) - front := serveFront(t, p) - - resp := proxiedGet(t, front.URL, "http://api.internal/x") - _ = resp.Body.Close() - if resp.StatusCode != http.StatusTooManyRequests { - t.Errorf("forward status %d, want 429", resp.StatusCode) - } - if ev := recvEvent(t, events); ev.Decision != DecisionRateLimited { - t.Errorf("forward audit %+v, want rate limited", ev) - } - - conn := dialConnect(t, front.Listener.Addr().String(), "echo.internal:443") - defer func() { _ = conn.Close() }() - if status := readStatus(t, bufio.NewReader(conn)); status != "HTTP/1.1 429 Too Many Requests" { - t.Errorf("CONNECT status %q, want 429", status) - } - if ev := recvEvent(t, events); ev.Decision != DecisionRateLimited { - t.Errorf("CONNECT audit %+v, want rate limited", ev) +func TestARefusedDialAnswersItsStatusAndIsAuditedWithNoTransfer(t *testing.T) { + for _, tt := range []struct { + name string + err error + status int + connect string + decision Decision + }{ + {"rate limited", ErrRateLimited, http.StatusTooManyRequests, "HTTP/1.1 429 Too Many Requests", DecisionRateLimited}, + {"internal address", fmt.Errorf("%w 127.0.0.1", ErrInternalAddr), http.StatusBadGateway, "HTTP/1.1 502 Bad Gateway", DecisionDeny}, + } { + t.Run(tt.name, func(t *testing.T) { + events, transfers := make(chan Event, 4), make(chan Event, 4) + dial := DialFunc(func(context.Context, string, string) (net.Conn, error) { return nil, tt.err }) + p := New(Policy{Allow: []Rule{{Host: "*"}}}, nil, nil, dial, func(ev Event) { events <- ev }, nil) + p.OnTransfer(func(ev Event, _, _ int64) { transfers <- ev }) + front := serveFront(t, p) + + resp := proxiedGet(t, front.URL, "http://api.internal/x") + _ = resp.Body.Close() + if resp.StatusCode != tt.status { + t.Errorf("forward status %d, want %d", resp.StatusCode, tt.status) + } + if ev := recvEvent(t, events); ev.Decision != tt.decision { + t.Errorf("forward audit %+v, want decision %v", ev, tt.decision) + } + conn := dialConnect(t, front.Listener.Addr().String(), "echo.internal:443") + defer func() { _ = conn.Close() }() + if status := readStatus(t, bufio.NewReader(conn)); status != tt.connect { + t.Errorf("CONNECT status %q, want %q", status, tt.connect) + } + if ev := recvEvent(t, events); ev.Decision != tt.decision { + t.Errorf("CONNECT audit %+v, want decision %v", ev, tt.decision) + } + select { + case ev := <-transfers: + t.Errorf("a refused dial reported a transfer for %s", ev.Host) + case <-time.After(200 * time.Millisecond): + } + }) } } diff --git a/sandboxd/egress/socks_test.go b/sandboxd/egress/socks_test.go index 3a32bb801..c980cc384 100644 --- a/sandboxd/egress/socks_test.go +++ b/sandboxd/egress/socks_test.go @@ -3,8 +3,10 @@ package egress import ( "bufio" "encoding/binary" + "errors" "io" "net" + "os" "testing" "time" ) @@ -136,7 +138,7 @@ func TestCloseEndsSocksTunnelAndHandshake(t *testing.T) { t.Error("established SOCKS tunnel survived proxy Close") } _ = pending.SetReadDeadline(time.Now().Add(2 * time.Second)) - if _, err := pending.Read(make([]byte, 1)); err == nil { + if _, err := pending.Read(make([]byte, 1)); err == nil || errors.Is(err, os.ErrDeadlineExceeded) { t.Error("mid-handshake SOCKS connection survived proxy Close") } } diff --git a/sandboxd/engine/engine.go b/sandboxd/engine/engine.go index a8d437675..ad5432fe1 100644 --- a/sandboxd/engine/engine.go +++ b/sandboxd/engine/engine.go @@ -16,6 +16,7 @@ import ( "runtime" "strconv" "strings" + "sync" "sync/atomic" "syscall" "time" @@ -66,11 +67,15 @@ const ( cocoonMaxProcs = 4 ) -// capacitySignatures mean the node cannot attach another VM, not that this VM failed. -var capacitySignatures = []string{ - "exchange full", - "no space left on device", -} +var ( + errStopping = errors.New("sandboxd is stopping") + + // capacitySignatures mean the node cannot attach another VM, not that this VM failed. + capacitySignatures = []string{ + "exchange full", + "no space left on device", + } +) // StaleCreateOutcome reports what reconcile-stale-create did with a record. type StaleCreateOutcome string @@ -84,6 +89,9 @@ type Engine struct { noBalloon bool restoreMode types.RestoreMode dumps chan struct{} + mu sync.Mutex + closed bool + running sync.WaitGroup // version caches the first successful query; replacing cocoon takes a restart. version atomic.Pointer[string] } @@ -93,6 +101,13 @@ func New(bin string, bridges, networks []string, noDirectIO, noBalloon bool, res return &Engine{bin: bin, bridges: bridges, networks: networks, noDirectIO: noDirectIO, noBalloon: noBalloon, restoreMode: restoreMode, dumps: make(chan struct{}, memoryDumps)} } +func (e *Engine) Close() { + e.mu.Lock() + e.closed = true + e.mu.Unlock() + e.running.Wait() +} + // Version reports cocoon's version string: a "vX.Y.Z" release or a "master-" dev build. func (e *Engine) Version(ctx context.Context) (string, error) { if v := e.version.Load(); v != nil { @@ -422,6 +437,10 @@ func (e *Engine) run(ctx context.Context, args ...string) ([]byte, error) { } func (e *Engine) runFor(ctx context.Context, timeout, termGrace time.Duration, args ...string) ([]byte, error) { + if !e.begin() { + return nil, errStopping + } + defer e.running.Done() ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() log.WithFunc("engine.runFor").Debugf(ctx, "cocoon %s", strings.Join(args, " ")) @@ -437,6 +456,15 @@ func (e *Engine) runFor(ctx context.Context, timeout, termGrace time.Duration, a return stdout.Bytes(), nil } +func (e *Engine) begin() bool { + e.mu.Lock() + defer e.mu.Unlock() + if !e.closed { + e.running.Add(1) + } + return !e.closed +} + func (e *Engine) runDump(ctx context.Context, args ...string) error { select { case e.dumps <- struct{}{}: diff --git a/sandboxd/engine/engine_test.go b/sandboxd/engine/engine_test.go index 5912de033..d70d60269 100644 --- a/sandboxd/engine/engine_test.go +++ b/sandboxd/engine/engine_test.go @@ -277,11 +277,7 @@ func TestAMemoryCommandGetsItsLongerTermGraceAndAColdBootTheCommandOne(t *testin defer cancel() done := make(chan error, 1) go func() { done <- tt.run(ctx, New(bin, nil, nil, false, false, "")) }() - for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) { - if _, err := os.Stat(ready); err == nil { - break - } - } + waitFile(ready) cancel() if err := <-done; err == nil || !strings.Contains(err.Error(), tt.want) { t.Errorf("got %v, want %q", err, tt.want) @@ -290,6 +286,40 @@ func TestAMemoryCommandGetsItsLongerTermGraceAndAColdBootTheCommandOne(t *testin } } +func TestCloseWaitsForARunningCommandAndRefusesNewOnes(t *testing.T) { + t.Parallel() + dir := t.TempDir() + bin, ready, release := filepath.Join(dir, "cocoon"), filepath.Join(dir, "ready"), filepath.Join(dir, "release") + script := "#!/bin/sh\n: > " + ready + "\nwhile [ ! -e " + release + " ]; do sleep 0.05; done\n" + if err := os.WriteFile(bin, []byte(script), 0o700); err != nil { + t.Fatal(err) + } + e := New(bin, nil, nil, false, false, "") + done := make(chan error, 1) + go func() { done <- e.Hibernate(t.Context(), "vm", "snap") }() + waitFile(ready) + closed := make(chan struct{}) + go func() { + e.Close() + close(closed) + }() + select { + case <-closed: + t.Fatal("Close returned while a cocoon command was running") + case <-time.After(200 * time.Millisecond): + } + if err := os.WriteFile(release, nil, 0o600); err != nil { + t.Fatal(err) + } + if err := <-done; err != nil { + t.Fatalf("the running hibernate: %v", err) + } + <-closed + if _, err := e.List(t.Context()); !errors.Is(err, errStopping) { + t.Errorf("List after Close = %v, want %v", err, errStopping) + } +} + func TestADialOfAGuestThatNeverAnswersFails(t *testing.T) { t.Parallel() path := sockPath(t) @@ -401,6 +431,14 @@ func sockPath(t *testing.T) string { return filepath.Join(dir, "mux.sock") } +func waitFile(path string) { + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) { + if _, err := os.Stat(path); err == nil { + return + } + } +} + func fakeCocoon(t *testing.T, script string) { t.Helper() dir := t.TempDir() diff --git a/sandboxd/engine/volume_live_test.go b/sandboxd/engine/volume_live_test.go index 3ad5635d7..4c0ea1726 100644 --- a/sandboxd/engine/volume_live_test.go +++ b/sandboxd/engine/volume_live_test.go @@ -186,7 +186,7 @@ func TestLiveDetachKeepsBusyAndUnobservableDevices(t *testing.T) { } func TestLiveDetachMissingGuestStillChecksHost(t *testing.T) { - for _, hostErr := range []string{"", `disk "imagenet" not attached`, "vm not running", "backend unavailable"} { + for _, hostErr := range []string{"", "vm not running", "backend unavailable"} { t.Run(hostErr, func(t *testing.T) { sock := sockPath(t) serveFakeSilkd(t, sock) diff --git a/sandboxd/engine/volume_test.go b/sandboxd/engine/volume_test.go index d6230cae8..e3783a5b6 100644 --- a/sandboxd/engine/volume_test.go +++ b/sandboxd/engine/volume_test.go @@ -356,10 +356,7 @@ func TestMountVolumeStopsAtFailedStage(t *testing.T) { } } -func TestMountVolumeDeviceProbeIsBoundedAndCancelable(t *testing.T) { - if volumeProbeTimeout != 2*time.Second { - t.Fatalf("volume probe timeout = %s, want 2s", volumeProbeTimeout) - } +func TestMountVolumeDeviceProbeIsCancelable(t *testing.T) { path := sockPath(t) fake := serveFakeSilkd(t, path) configureVolumeDevices(fake) @@ -376,10 +373,14 @@ func TestMountVolumeDevNodeNeverAppearsTimesOut(t *testing.T) { fake := serveFakeSilkd(t, path) configureVolumeDevices(fake) fake.set(func(f *fakeSilkd) { delete(f.stat, "/dev/vdc") }) + start := time.Now() err := newTestEngine().MountVolume(t.Context(), path, "imagenet", "/datasets/training", false) if !errors.Is(err, context.DeadlineExceeded) || !strings.Contains(err.Error(), "wait for volume device imagenet") { t.Errorf("got %v, want a bounded wait-for-device timeout", err) } + if elapsed := time.Since(start); elapsed >= volumeSetupTimeout { + t.Errorf("the probe took %v, want it to end before the %v setup budget", elapsed, volumeSetupTimeout) + } } func TestSilkdListAndReadAreBounded(t *testing.T) { diff --git a/sandboxd/main.go b/sandboxd/main.go index bb34632bb..3c74db4d6 100644 --- a/sandboxd/main.go +++ b/sandboxd/main.go @@ -198,6 +198,8 @@ func main() { logger.Fatalf(ctx, err, "serve") } <-drained + logger.Info(ctx, "waiting for running cocoon commands") + eng.Close() if err := mgr.FlushClaims(); err != nil { logger.Error(ctx, err, "flush claims") } diff --git a/sandboxd/outbound/dial.go b/sandboxd/outbound/dial.go index 79b2a6de3..00ad507f5 100644 --- a/sandboxd/outbound/dial.go +++ b/sandboxd/outbound/dial.go @@ -101,7 +101,7 @@ func destVerdict(allow []egress.InternalAllow, self []netip.Addr, ip netip.Addr, return true, nil } if internalAddr(self, ip) { - return false, fmt.Errorf("egress: blocked internal address %s", ip) + return false, fmt.Errorf("%w %s", egress.ErrInternalAddr, ip) } return false, nil } diff --git a/sandboxd/outbound/outbound_test.go b/sandboxd/outbound/outbound_test.go index f51160870..a31930899 100644 --- a/sandboxd/outbound/outbound_test.go +++ b/sandboxd/outbound/outbound_test.go @@ -2,6 +2,7 @@ package outbound import ( "context" + "errors" "io" "net" "net/http" @@ -109,6 +110,19 @@ func TestArmEgressFailsClosedWhenNICUnlockable(t *testing.T) { } } +func TestArmEgressFailsClosedWhenTheLockFails(t *testing.T) { + lockErr := errors.New("nft down") + h := newHost(t, &fakeEngine{}, poolConfig(egKey, egPolicy), func(o *Options) { + o.LockNIC = true + o.Lock = func(string) error { return lockErr } + }) + + sb := &types.Sandbox{ID: "sb_eg_refused", Key: egKey, VMName: "sbx-refused-1", TAP: "tap-x"} + if armErr := h.Arm(t.Context(), sb); !errors.Is(armErr, lockErr) { + t.Fatalf("Arm = %v, want the lock failure", armErr) + } +} + func TestArmEgressLocksEgressLaneWithoutPolicy(t *testing.T) { cfg := &config.Config{Pools: []config.PoolSpec{ {PoolKey: testKey, Egress: &egress.Policy{Allow: []egress.Rule{{Host: "a.test"}}}}, diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index 3491f0e7b..a89cb1e55 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -5,6 +5,7 @@ import ( "encoding/json/v2" "errors" "fmt" + "io/fs" "maps" "os" "path/filepath" @@ -276,6 +277,10 @@ func (m *Manager) runHeal(ctx context.Context, ckptID string) (types.Checkpoint, } return types.Checkpoint{}, fmt.Errorf("heal checkpoint: %w", err) } + if err := syncFiles(staging); err != nil { + m.heals.clear(ckptID) + return types.Checkpoint{}, fmt.Errorf("sync healed checkpoint: %w", err) + } defer m.recs.lockRec(ckptID)() if aborted := m.heals.clear(ckptID); aborted { @@ -423,3 +428,16 @@ func parseCheckpoint(raw []byte) (types.Checkpoint, error) { } return ckpt, nil } + +func syncFiles(dir string) error { + return filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error { + if err != nil || !d.Type().IsRegular() { + return err + } + f, err := os.Open(path) //nolint:gosec // walked under our staging dir + if err != nil { + return err + } + return errors.Join(f.Sync(), f.Close()) + }) +} diff --git a/sandboxd/pool/egress_test.go b/sandboxd/pool/egress_test.go index cebd64000..aa31c0ba8 100644 --- a/sandboxd/pool/egress_test.go +++ b/sandboxd/pool/egress_test.go @@ -99,8 +99,8 @@ func TestEgressLaneWakeFailsClosed(t *testing.T) { } for _, tt := range cases { t.Run(tt.name, func(t *testing.T) { - if _, err := m.wakeResolved(t.Context(), tt.sb); err == nil { - t.Fatal("egress-lane wake must fail closed, not resume unguarded") + if _, err := m.wakeResolved(t.Context(), tt.sb); err == nil || !strings.Contains(err.Error(), "egress lane cannot resume") { + t.Fatalf("wake = %v, want the egress lane to fail closed before it resumes", err) } }) } diff --git a/sandboxd/pool/failover_test.go b/sandboxd/pool/failover_test.go index 84e286b1d..98e66b430 100644 --- a/sandboxd/pool/failover_test.go +++ b/sandboxd/pool/failover_test.go @@ -1822,6 +1822,9 @@ func TestFailoverAWakeWithoutARegistryKeepsItsPendingIntent(t *testing.T) { sb := mustClaim(t, a, testKey) hibernateClaim(t, a, sb) c.archiveLosingPutReply(t, a, sb) + a.mu.Lock() + pending := sb.ArchivePending + a.mu.Unlock() a.archives = nil if _, _, _, err := a.WakeAgentSocket(t.Context(), sb.ID, sb.Token); err != nil { t.Fatalf("wake with failover off: %v", err) @@ -1829,6 +1832,12 @@ func TestFailoverAWakeWithoutARegistryKeepsItsPendingIntent(t *testing.T) { if hibernated(a) != 0 { t.Error("the claim stayed hibernated after the wake") } + a.mu.Lock() + kept := sb.ArchivePending + a.mu.Unlock() + if pending == "" || kept != pending || !ckExists(t, a, pending) { + t.Errorf("intent %q then %q, want it and its export kept", pending, kept) + } } func TestFailoverAReleaseFindsItsUnsettledArchiveTakenElsewhere(t *testing.T) { diff --git a/sandboxd/pool/liveness.go b/sandboxd/pool/liveness.go index 083a81d36..ffd347273 100644 --- a/sandboxd/pool/liveness.go +++ b/sandboxd/pool/liveness.go @@ -264,10 +264,11 @@ func (m *Manager) restartLocked(ctx context.Context, sb *types.Sandbox) error { } m.mu.Lock() sb.VsockSocket = sock + env := sb.Env m.mu.Unlock() err = m.out.ArmProxy(ctx, sb) - if err == nil { - err = m.deliverEnv(ctx, []*types.Sandbox{sb}, false) + if err == nil && env.HasGuest() { + err = m.eng.WriteGuestEnv(ctx, sock, env.GuestFile()) } if err != nil { m.out.Disarm(sb.ID, false) diff --git a/sandboxd/pool/pool_test.go b/sandboxd/pool/pool_test.go index fd4311282..99a72b3b5 100644 --- a/sandboxd/pool/pool_test.go +++ b/sandboxd/pool/pool_test.go @@ -228,7 +228,7 @@ func TestReleaseAfterResolveTearsDownOnce(t *testing.T) { if err := m.releaseResolved(t.Context(), sb.ID, second); !errors.Is(err, ErrUnknownSandbox) { t.Errorf("second release: %v, want ErrUnknownSandbox", err) } - if got := slices.Compact(slices.Clone(eng.removes)); len(got) != 1 || got[0] != sb.VMName { + if !slices.Equal(eng.removes, []string{sb.VMName}) { t.Errorf("removes=%v, want %s torn down exactly once", eng.removes, sb.VMName) } } @@ -965,6 +965,9 @@ func TestRenewClampsAndDefaults(t *testing.T) { m := newTestManager(t, newFakeEngine()) sb := mustClaim(t, m, testKey) + if _, err := m.Renew(t.Context(), sb.ID, Cred{Token: sb.Token}, time.Hour, ""); err != nil { + t.Fatalf("renew: %v", err) + } if _, err := m.Renew(t.Context(), sb.ID, Cred{Token: sb.Token}, 0, ""); err != nil { t.Fatalf("renew: %v", err) } diff --git a/sandboxd/pool/poolstore_test.go b/sandboxd/pool/poolstore_test.go index 9a56b7683..3df1f5861 100644 --- a/sandboxd/pool/poolstore_test.go +++ b/sandboxd/pool/poolstore_test.go @@ -152,9 +152,6 @@ func TestNodesOfACellShareThePoolSet(t *testing.T) { if !tenantstest.Eventually(t, func() bool { return hasAPISet(b) }) { t.Fatal("b never applied the pool set put on a") } - if hasAPISet(other) { - t.Error("a node of another cell applied the pool set") - } cached := func(m *Manager) bool { set, err := m.poolStore.load() return err == nil && set != nil && len(set.Pools) == 1 && set.Pools[0].PoolKey == apiKey @@ -162,6 +159,9 @@ func TestNodesOfACellShareThePoolSet(t *testing.T) { if !tenantstest.Eventually(t, func() bool { return cached(b) }) { t.Error("b never cached the cell set in pools.json") } + if hasAPISet(other) { + t.Error("a node of another cell applied the pool set") + } if joiner := metaStoreNode(t, "c1"); !hasAPISet(joiner) || !cached(joiner) { t.Error("a node joining the cell did not boot with its pool set and cache it") } diff --git a/sandboxd/pool/replica.go b/sandboxd/pool/replica.go index 20427d9a8..5e8099bad 100644 --- a/sandboxd/pool/replica.go +++ b/sandboxd/pool/replica.go @@ -96,6 +96,12 @@ func (m *Manager) replicate(ctx context.Context, sb *types.Sandbox) error { if _, _, err := m.publishCheckpoint(ctx, sb, ckID, "replica", sb.Tenant, true); err != nil { return err } + var prev string + defer func() { + if prev != "" { + m.deleteOrphanArchiveCk(ctx, prev) + } + }() sb.Transition.Lock() defer sb.Transition.Unlock() m.mu.Lock() @@ -130,7 +136,6 @@ func (m *Manager) replicate(ctx context.Context, sb *types.Sandbox) error { case err != nil: return fmt.Errorf("replicate %s: registry: %w", sb.ID, err) } - var prev string live, _ := m.commitIfLive(ctx, sb, func() { prev, sb.ReplicaCk, sb.ReplicaPending, sb.ArchiveEpoch = sb.ReplicaCk, ckID, "", epoch }) @@ -138,9 +143,6 @@ func (m *Manager) replicate(ctx context.Context, sb *types.Sandbox) error { m.abandonArchive(ctx, sb, epoch, ckID) return errWokeMeanwhile } - if prev != "" { - m.deleteOrphanArchiveCk(ctx, prev) - } return nil } diff --git a/sandboxd/server/reader.go b/sandboxd/server/reader.go index 64ba9e581..8269b56dd 100644 --- a/sandboxd/server/reader.go +++ b/sandboxd/server/reader.go @@ -200,6 +200,7 @@ func (r *router) startPump() { _, _ = io.Copy(r.client, reader) if !r.superseded.Load() { _ = r.client.Close() + _ = reader.Close() } }() } diff --git a/sandboxd/server/reader_test.go b/sandboxd/server/reader_test.go index 84e5bbd3e..fac1be27a 100644 --- a/sandboxd/server/reader_test.go +++ b/sandboxd/server/reader_test.go @@ -331,6 +331,29 @@ func TestSleepingAwakeClaimRelaysToItsGuest(t *testing.T) { } } +func TestSleepingReaderClosesWhenItsClientLeavesMidReply(t *testing.T) { + env := newSleepingEnv(t) + env.reader.serve = func(c net.Conn) { + defer func() { + _ = c.Close() + env.readerClosed <- struct{}{} + }() + if _, err := bufio.NewReader(c).ReadString('\n'); err != nil { + return + } + _, _ = c.Write(make([]byte, 64<<20)) + } + conn, _ := env.upgrade(t) + send(t, conn, readFrame) + send(t, conn, statFrame) + _ = conn.Close() + select { + case <-env.readerClosed: + case <-time.After(5 * time.Second): + t.Fatal("the reader connection outlived its client") + } +} + func TestSleepingCloseRelaysEndsReaderConnections(t *testing.T) { env := newSleepingEnv(t) conn, r := env.upgrade(t) diff --git a/sandboxd/store/dir/dir_test.go b/sandboxd/store/dir/dir_test.go index 1f87e289d..80200f02e 100644 --- a/sandboxd/store/dir/dir_test.go +++ b/sandboxd/store/dir/dir_test.go @@ -80,6 +80,10 @@ func TestPlainPublishReplacesDigestWithEmpty(t *testing.T) { if _, statErr := os.Stat(firstDigestPath); statErr != nil { t.Fatalf("superseded digest removed inside the grace: %v", statErr) } + mustPublish(t, st, testID, "digested") + if _, _, digest = mustFetch(t, st, testID); digest != "" { + t.Errorf("a plain publish of the digested meta kept digest %q, want empty", digest) + } } func TestPublishDigestedChunkBoundaries(t *testing.T) { diff --git a/sandboxd/store/peer/peer.go b/sandboxd/store/peer/peer.go index 5f0613096..759482a84 100644 --- a/sandboxd/store/peer/peer.go +++ b/sandboxd/store/peer/peer.go @@ -42,14 +42,14 @@ func (h *Healer) Pull(ctx context.Context, id, staging string, validate Validate } func (h *Healer) pullFrom(ctx context.Context, id, staging string, addrs []string, validate Validate) error { - perOwner := healBudget / time.Duration(len(addrs)) + deadline, _ := ctx.Deadline() var errs []error - for _, addr := range addrs { + for i, addr := range addrs { // One peer's rejected or partial transfer must not linger into the next's. if err := utils.RemoveDirEntries(staging, nil); err != nil { return fmt.Errorf("reset staging: %w", err) } - attemptCtx, cancel := context.WithTimeout(ctx, perOwner) + attemptCtx, cancel := context.WithTimeout(ctx, time.Until(deadline)/time.Duration(len(addrs)-i)) err := h.puller.Pull(attemptCtx, addr, id, staging) cancel() if err != nil { diff --git a/sandboxd/store/peer/peer_test.go b/sandboxd/store/peer/peer_test.go index 41486c06b..6e2b15b8e 100644 --- a/sandboxd/store/peer/peer_test.go +++ b/sandboxd/store/peer/peer_test.go @@ -153,6 +153,15 @@ func TestPullBudgetBoundsSlowOwners(t *testing.T) { }) } +func TestPullGivesTheNextOwnerTheTimeAFailedOneLeft(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + h := healerOf(&sleepyPuller{sleep: healBudget * 3 / 4, failAddr: "peer-a:7777"}, "peer-a:7777", "peer-b:7777") + if err := h.Pull(t.Context(), testID, t.TempDir(), nil); err != nil { + t.Errorf("Pull = %v, want peer-b to finish inside the budget peer-a left", err) + } + }) +} + type fakePuller struct { mu sync.Mutex records map[string]map[string]string @@ -184,10 +193,14 @@ func (p *fakePuller) Pull(_ context.Context, addr, _, dst string) error { } type sleepyPuller struct { - sleep time.Duration + sleep time.Duration + failAddr string } -func (p *sleepyPuller) Pull(ctx context.Context, _, _, _ string) error { +func (p *sleepyPuller) Pull(ctx context.Context, addr, _, _ string) error { + if addr == p.failAddr { + return errors.New("refused") + } select { case <-time.After(p.sleep): return nil diff --git a/sandboxd/store/peer/transport_test.go b/sandboxd/store/peer/transport_test.go index 56a6e19ea..45f1ffbc2 100644 --- a/sandboxd/store/peer/transport_test.go +++ b/sandboxd/store/peer/transport_test.go @@ -10,6 +10,7 @@ import ( "net/http/httptest" "os" "path/filepath" + "slices" "strings" "syscall" "testing" @@ -78,6 +79,9 @@ func TestUntarRejectsPathTraversal(t *testing.T) { if _, err := tw.Write(body); err != nil { t.Fatalf("Write: %v", err) } + if err := tw.WriteHeader(&tar.Header{Name: recordTrailer, Mode: 0o600, Typeflag: tar.TypeReg}); err != nil { + t.Fatalf("WriteHeader: %v", err) + } if err := tw.Close(); err != nil { t.Fatalf("Close: %v", err) } @@ -87,8 +91,8 @@ func TestUntarRejectsPathTraversal(t *testing.T) { if err == nil { t.Fatalf("Untar accepted %q; it must refuse to write outside the destination", name) } - if !strings.Contains(err.Error(), "untar") { - t.Errorf("error = %v, want an untar rejection", err) + if !strings.Contains(err.Error(), "refusing") { + t.Errorf("error = %v, want the path refused", err) } }) } @@ -144,14 +148,16 @@ func TestTarSkipsSymlinksAtSource(t *testing.T) { t.Fatalf("tarDir: %v", err) } tr := tar.NewReader(&buf) + var names []string for { hdr, err := tr.Next() if err != nil { break } - if hdr.Typeflag == tar.TypeSymlink { - t.Errorf("Tar emitted a symlink entry %q", hdr.Name) - } + names = append(names, hdr.Name) + } + if !slices.Contains(names, "real") || slices.Contains(names, "link") { + t.Errorf("Tar emitted %q, want real and no link", names) } } diff --git a/sandboxd/tenants/pg/pg.go b/sandboxd/tenants/pg/pg.go index 69243b1a7..ba21160fe 100644 --- a/sandboxd/tenants/pg/pg.go +++ b/sandboxd/tenants/pg/pg.go @@ -56,8 +56,10 @@ type Source struct { byName map[string]entry misses map[[sha256.Size]byte]time.Time inflight map[string]struct{} - // seq counts evictions; a read caches only if none landed after it started. - seq uint64 + // seq counts evictions; a read caches only if neither its name nor the whole cache was evicted after it started. + seq uint64 + evicted map[string]uint64 + floor uint64 stop context.CancelFunc done chan struct{} @@ -77,6 +79,7 @@ func Open(ctx context.Context, dsn string) (*Source, error) { byName: map[string]entry{}, misses: map[[sha256.Size]byte]time.Time{}, inflight: map[string]struct{}{}, + evicted: map[string]uint64{}, stop: stop, done: make(chan struct{}), } @@ -118,7 +121,7 @@ func (s *Source) Resolve(ctx context.Context, sum [sha256.Size]byte) (*tenants.T case err != nil: return nil, unavailable(err) } - if s.seq == start { + if s.fresh(r.Name, start) { s.cache(&r, sum) } return &r, nil @@ -335,10 +338,14 @@ func (s *Source) write(ctx context.Context, key string, f func(tx pgx.Tx) error) return nil } +func (s *Source) fresh(name string, start uint64) bool { + return start >= s.floor && s.evicted[name] <= start +} + // settle caches a by-name read that no eviction overtook; callers hold s.mu. func (s *Source) settle(name string, start uint64, r tenants.TenantRecord, ok bool, err error) { switch { - case err != nil || s.seq != start: + case err != nil || !s.fresh(name, start): case ok: sum, _ := hex.DecodeString(r.TokenSHA256) s.cache(&r, [sha256.Size]byte(sum)) @@ -360,14 +367,17 @@ func (s *Source) evict(key string) { s.mu.Lock() defer s.mu.Unlock() s.seq++ - if key == flushAll { + if key == flushAll || len(s.evicted) >= maxCached { s.clearCache() + clear(s.evicted) + s.floor = s.seq return } if e := s.byName[key]; e.rec != nil { delete(s.byToken, e.sum) } delete(s.byName, key) + s.evicted[key] = s.seq clear(s.misses) } diff --git a/sandboxd/tenants/pg/pg_test.go b/sandboxd/tenants/pg/pg_test.go index 817e67b60..4c65c1b76 100644 --- a/sandboxd/tenants/pg/pg_test.go +++ b/sandboxd/tenants/pg/pg_test.go @@ -152,6 +152,26 @@ func TestARoleWithoutCreateServesAPreCreatedTable(t *testing.T) { } } +func TestAnUnrelatedEvictionKeepsALookupCacheable(t *testing.T) { + s := &Source{evicted: map[string]uint64{}} + start := s.seq + s.seq++ + s.evicted["other"] = s.seq + if !s.fresh("u-1", start) { + t.Error("another tenant's eviction discarded a read of u-1") + } + s.seq++ + s.evicted["u-1"] = s.seq + if s.fresh("u-1", start) { + t.Error("a read that raced u-1's own eviction stays cacheable") + } + s.seq++ + s.floor = s.seq + if s.fresh("u-2", start) { + t.Error("a read that raced a full flush stays cacheable") + } +} + func openIn(t *testing.T, dsn string) *Source { t.Helper() s, err := Open(t.Context(), dsn) diff --git a/sandboxd/utils/clone_test.go b/sandboxd/utils/clone_test.go index 8c46c3336..98c2e1e3d 100644 --- a/sandboxd/utils/clone_test.go +++ b/sandboxd/utils/clone_test.go @@ -4,6 +4,8 @@ import ( "bytes" "os" "path/filepath" + "runtime" + "syscall" "testing" ) @@ -54,6 +56,13 @@ func TestCopyDataCopiesDataRangesOnly(t *testing.T) { t.Fatalf("copyData: %v", err) } assertSameFile(t, srcPath, dstPath) + info, err := dst.Stat() + if err != nil { + t.Fatalf("stat dst: %v", err) + } + if allocated := info.Sys().(*syscall.Stat_t).Blocks * 512; runtime.GOOS == "linux" && allocated >= info.Size()/2 { + t.Errorf("copy allocates %d of %d bytes, want the holes kept", allocated, info.Size()) + } } func writeSparse(t *testing.T) (src, dst string) { diff --git a/scripts/credentials-e2e.sh b/scripts/credentials-e2e.sh index d615401bc..c24465a58 100755 --- a/scripts/credentials-e2e.sh +++ b/scripts/credentials-e2e.sh @@ -125,7 +125,7 @@ echo "== inject pool: conditional interception, query and body placeholders" -inject-size medium | tee "$DATA/inject.log" grep -q '"ev":"egress".*"secret":"claim:SERP"' "$DATA/state/usage.jsonl" || { echo "usage names no claim:SERP"; exit 1; } while IFS= read -r v; do - if grep '"op":"egress"' "$DATA/state/audit.jsonl" | grep -qF "$v" || grep -qF "$v" "$DATA/state/usage.jsonl"; then + if grep '"op":"egress"' "$DATA/state/audit.jsonl" | grep -F "$v" >/dev/null || grep -qF "$v" "$DATA/state/usage.jsonl"; then echo "an inject value reached an egress audit record or the usage journal" exit 1 fi diff --git a/sdk/go/silkd/conn_test.go b/sdk/go/silkd/conn_test.go index 9e7034450..6dbffdbc9 100644 --- a/sdk/go/silkd/conn_test.go +++ b/sdk/go/silkd/conn_test.go @@ -1,7 +1,9 @@ package silkd_test import ( + "bufio" "bytes" + "errors" "io" "net" "strings" @@ -123,8 +125,8 @@ func TestConnRejectsOversizedFrame(t *testing.T) { _, _ = io.WriteString(server, huge+"\n") }() - if _, err := conn.Recv(); err == nil { - t.Error("oversized frame accepted") + if _, err := conn.Recv(); !errors.Is(err, bufio.ErrTooLong) { + t.Errorf("Recv = %v, want the frame cap", err) } } diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index ec6c928dd..3e059b95a 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -184,6 +184,8 @@ def run( if watchdog is not None: watchdog.cancel() if pump is not None: + if pump.is_alive(): + conn.abort() pump.join() # the pump's frames must not land on the next RPC if code is None: raise ProtocolError("exec stream ended without an exit frame") diff --git a/sdk/python/tests/conftest.py b/sdk/python/tests/conftest.py index 65fdaf984..fbb169ed6 100644 --- a/sdk/python/tests/conftest.py +++ b/sdk/python/tests/conftest.py @@ -82,6 +82,9 @@ def recv_until(self, *terminal): if frame["type"] in terminal: return + def abort(self): + pass + def close(self): pass diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index dffbf9cd8..6b4d5a4e2 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -149,6 +149,40 @@ def serve(): server.close() +def test_run_stops_feeding_stdin_once_the_command_exits(): + server = socket.create_server(("127.0.0.1", 0)) + release = threading.Event() + + def serve(): + conn, _ = server.accept() + with conn, contextlib.suppress(OSError): + reader = accept_upgrade(conn) + for line in iter(reader.readline, b""): + op = json.loads(line)["op"] + if op == "info": + info = {"type": "info", "version": "fake", "proto": KEEP_ALIVE_PROTO, "uptime_secs": 0} + conn.sendall(json.dumps(info).encode() + b"\n") + elif op == "exec": + conn.sendall(b'{"type":"started","pid":1}\n{"type":"exit","code":0}\n') + release.wait(10) + return + + threading.Thread(target=serve, daemon=True).start() + sb = sandbox_at(f"127.0.0.1:{server.getsockname()[1]}") + codes = [] + call = threading.Thread( + target=lambda: codes.append(sb.run(["head", "-c", "10"], stdin=bytes(64 << 20), timeout=0.5)), daemon=True + ) + try: + call.start() + call.join(2) + assert codes == [0], "run outlived its timeout behind a stdin the guest stopped reading" + assert sb._pool.take() is None + finally: + release.set() + server.close() + + def test_close_drains_the_parked_connection(monkeypatch, relay): agent, sb = relay() monkeypatch.setattr(sb._client, "_request", lambda *args, **kwargs: {}) diff --git a/silkd/src/server.rs b/silkd/src/server.rs index 4db98512d..7d3460e05 100644 --- a/silkd/src/server.rs +++ b/silkd/src/server.rs @@ -3,7 +3,7 @@ use std::borrow::Cow; use std::time::{Instant, SystemTime, UNIX_EPOCH}; -use tokio::io::{AsyncBufRead, AsyncWrite}; +use tokio::io::{AsyncBufRead, AsyncBufReadExt, AsyncWrite}; use tokio::sync::mpsc; use tokio::task::JoinHandle; @@ -35,26 +35,27 @@ impl State { W: AsyncWrite + Unpin, { let mut line = Vec::new(); - let mut next: Option = None; + let mut next: Option = None; let mut served = false; loop { - let req = match next.take() { - Some(req) => req, + let parsed = match next.take() { + Some(parsed) => parsed, None => { if !proto::read_frame_into(&mut reader, &mut line).await? { return Ok(()); } - match serde_json::from_slice(&line) { - Ok(req) => req, - Err(e) => { - return proto::error_frame( - &mut writer, - ErrorKind::BadRequest, - format!("parse: {e}"), - ) - .await; - } - } + serde_json::from_slice(&line) + } + }; + let req = match parsed { + Ok(req) => req, + Err(e) => { + return proto::error_frame( + &mut writer, + ErrorKind::BadRequest, + format!("parse: {e}"), + ) + .await; } }; // a finished RPC's input frames can still be in flight @@ -72,7 +73,7 @@ impl State { req: Request, mut reader: R, writer: &mut W, - ) -> std::io::Result<(R, Option)> + ) -> std::io::Result<(R, Option)> where R: AsyncBufRead + Unpin + Send + 'static, W: AsyncWrite + Unpin, @@ -82,7 +83,7 @@ impl State { Request::Ps => self.ps(writer).await, Request::Kill { pid, signal } => self.kill(writer, pid, signal).await, Request::Logs { pid } => self.logs(writer, pid).await, - Request::Attach { pid } => self.attach(writer, pid).await, + Request::Attach { pid } => self.attach(&mut reader, writer, pid).await, Request::Exec(e) => return self.exec(e, reader, writer).await, Request::SessionCreate { id, cwd, env } => { match self.sessions.create(id, cwd, &env).await { @@ -191,7 +192,7 @@ impl State { e: ExecReq, reader: R, writer: &mut W, - ) -> std::io::Result<(R, Option)> + ) -> std::io::Result<(R, Option)> where R: AsyncBufRead + Unpin + Send + 'static, W: AsyncWrite + Unpin, @@ -277,7 +278,11 @@ impl State { proto::write_frame(w, &Response::Done).await } - async fn attach(&self, w: &mut W, pid: u32) -> std::io::Result<()> { + async fn attach(&self, reader: &mut R, w: &mut W, pid: u32) -> std::io::Result<()> + where + R: AsyncBufRead + Unpin, + W: AsyncWrite + Unpin, + { let Some(proc) = self.table.get_or_not_found(w, pid).await? else { return Ok(()); }; @@ -290,15 +295,25 @@ impl State { if let Some(code) = exit { return proto::write_frame(w, &Response::Exit { code }).await; } + let mut pipelined = false; loop { - match rx.recv().await { - Ok(Chunk::Exit(code)) => { - return proto::write_frame(w, &Response::Exit { code }).await; - } - Ok(chunk) => write_chunk(w, &mut frame, chunk).await?, - Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, - Err(tokio::sync::broadcast::error::RecvError::Closed) => { - return proto::write_frame(w, &Response::Done).await; + tokio::select! { + biased; + recv = rx.recv() => match recv { + Ok(Chunk::Exit(code)) => { + return proto::write_frame(w, &Response::Exit { code }).await; + } + Ok(chunk) => write_chunk(w, &mut frame, chunk).await?, + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + return proto::write_frame(w, &Response::Done).await; + } + }, + r = reader.fill_buf(), if !pipelined => { + if !matches!(r, Ok(buf) if !buf.is_empty()) { + return Ok(()); + } + pipelined = true; } } } @@ -311,7 +326,8 @@ impl Default for State { } } -type Feeder = JoinHandle<(R, Option)>; +type Parsed = serde_json::Result; +type Feeder = JoinHandle<(R, Option)>; fn spawn_feeder(reader: R) -> (mpsc::Receiver, Feeder) where @@ -325,7 +341,7 @@ where async fn finish_feeder( feeder: Feeder, res: std::io::Result<()>, -) -> std::io::Result<(R, Option)> { +) -> std::io::Result<(R, Option)> { if let Err(e) = res { feeder.abort(); return Err(e); @@ -334,19 +350,17 @@ async fn finish_feeder( } /// Reads the client's frames during a streaming RPC: input frames go to the handler, the next request ends the feed. -async fn feed_client(mut reader: R, tx: mpsc::Sender) -> (R, Option) +async fn feed_client(mut reader: R, tx: mpsc::Sender) -> (R, Option) where R: AsyncBufRead + Unpin, { let mut line = Vec::new(); let mut data = Vec::new(); while let Ok(true) = proto::read_frame_into(&mut reader, &mut line).await { - let Ok(req) = proto::parse_request(&line, &mut data) else { - break; + let req = match proto::parse_request(&line, &mut data) { + Ok(req) if req.is_continuation() => req, + parsed => return (reader, Some(parsed)), }; - if !req.is_continuation() { - return (reader, Some(req)); - } // a closed channel is a finished handler, whose late input is dropped let _ = tx.send(req).await; } diff --git a/silkd/tests/exec_e2e.rs b/silkd/tests/exec_e2e.rs index b9d560d95..7ff6ac8dd 100644 --- a/silkd/tests/exec_e2e.rs +++ b/silkd/tests/exec_e2e.rs @@ -26,6 +26,15 @@ async fn detached_pid(state: &Arc, script: &str) -> u64 { started[0]["pid"].as_u64().unwrap() } +async fn attach_frames(state: &Arc, pid: u64) -> Vec { + let (mut cw, mut lines, _) = connect(state); + send(&mut cw, json!({"op":"attach","pid":pid})).await; + frames_until(&mut lines, |f| { + matches!(type_of(f), "exit" | "done" | "error") + }) + .await +} + async fn wait_for_exit(state: &Arc, pid: u64) -> Vec { for _ in 0..250 { let logs = one(state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; @@ -143,12 +152,9 @@ async fn daemonizer_exit_is_the_last_frame() { async fn attach_streams_live_output_then_exit() { let state = Arc::new(State::new()); let pid = detached_pid(&state, "sleep 0.3; echo live-chunk").await; - let frames = tokio::time::timeout( - Duration::from_secs(8), - one(&state, &format!(r#"{{"op":"attach","pid":{pid}}}"#)), - ) - .await - .expect("attach hung"); + let frames = tokio::time::timeout(Duration::from_secs(8), attach_frames(&state, pid)) + .await + .expect("attach hung"); assert!( stdout_body(&frames).contains("live-chunk"), "attach missed live output: {frames:?}" @@ -208,6 +214,23 @@ async fn attach_keeps_a_tail_emitted_during_the_replay() { assert_eq!(type_of(frames.last().unwrap()), "exit"); } +#[tokio::test] +async fn attach_ends_when_its_client_hangs_up() { + let state = Arc::new(State::new()); + let pid = detached_pid(&state, "sleep 30").await; + let (mut cw, lines, handle) = connect(&state); + send(&mut cw, json!({"op":"attach","pid":pid})).await; + + drop(lines); + drop(cw); + tokio::time::timeout(Duration::from_secs(5), handle) + .await + .expect("attach did not end on hang-up") + .unwrap() + .unwrap(); + one(&state, &json!({"op":"kill","pid":pid}).to_string()).await; +} + #[tokio::test] async fn attach_unknown_pid_is_not_found() { let frames = roundtrip(r#"{"op":"attach","pid":999999}"#).await; @@ -310,8 +333,7 @@ async fn disconnect_during_drain_publishes_real_exit_code() { let pid = started["pid"].as_u64().unwrap(); let st = Arc::clone(&state); - let attach = - tokio::spawn(async move { one(&st, &json!({"op":"attach","pid":pid}).to_string()).await }); + let attach = tokio::spawn(async move { attach_frames(&st, pid).await }); tokio::time::sleep(Duration::from_millis(200)).await; drop(lines); drop(cw); diff --git a/silkd/tests/keepalive_e2e.rs b/silkd/tests/keepalive_e2e.rs index 209bea571..345548a33 100644 --- a/silkd/tests/keepalive_e2e.rs +++ b/silkd/tests/keepalive_e2e.rs @@ -3,6 +3,7 @@ mod common; use std::sync::Arc; +use std::time::Duration; use serde_json::json; use silkd::server::State; @@ -83,6 +84,32 @@ async fn late_stdin_after_exit_is_dropped() { assert_eq!(type_of(&next_frame(&mut lines).await), "info"); } +#[tokio::test] +async fn a_malformed_request_after_exec_is_bad_request() { + let state = Arc::new(State::new()); + let (mut cw, mut lines, _) = connect(&state); + + send( + &mut cw, + json!({"v":1,"op":"exec","argv":["true"],"detach":false}), + ) + .await; + assert_eq!( + types_until(&mut lines, "exit").await, + vec!["started", "exit"] + ); + + send(&mut cw, json!({"op":"exec","argv":"ls -la"})).await; + let err = timeout(Duration::from_secs(3), next_frame(&mut lines)) + .await + .expect("a malformed request after exec went unanswered"); + assert_error(&err, "bad_request"); + assert!( + err["message"].as_str().unwrap().starts_with("parse: "), + "{err}" + ); +} + #[tokio::test] async fn upload_hands_the_connection_back_after_done() { let dir = tempfile::tempdir().unwrap();