diff --git a/docs/deploy.md b/docs/deploy.md index 8fd11307..91c3da9b 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -125,7 +125,7 @@ gets 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) | -| `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. Re-publish retains prior export generations until Delete so an in-flight fetch that selected old metadata can finish; budget storage for those generations. An explicit S3 Delete can still make a concurrent fetch that has not finished materializing fail visibly. A crash between upload and the meta.json commit marker leaves orphan objects invisible to listings — add an S3 lifecycle rule to reclaim them. 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. Re-publish retains prior export generations until Delete so an in-flight fetch that selected old metadata can finish; budget storage for those generations. When caching a new generation, each node prunes other cached generations of that record unused for at least an hour. An explicit S3 Delete can still make a concurrent fetch that has not finished materializing fail visibly. A crash between upload and the meta.json commit marker leaves orphan objects invisible to listings — add an S3 lifecycle rule to reclaim them. 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 — it is the expiry eligibility point for a healed replica a delete broadcast missed, after which its next successful hourly sweep removes it; persistent sweep failure extends retention until one succeeds, so it is not a hard ceiling | | `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 replica a delete broadcast missed becomes eligible for expiry after it, and its next successful hourly sweep removes it — so it is the finite eligibility point, not an exact ceiling). 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 | diff --git a/docs/egress.md b/docs/egress.md index 22872f17..ca140ff3 100644 --- a/docs/egress.md +++ b/docs/egress.md @@ -167,13 +167,17 @@ as if it were that pool's claim: the template records its source pool, a re-promote records the new one, and a fork, checkpoint or branch of such a clone keeps it. So a clone of a template promoted from a pool with no policy has no egress, as that pool's claims have none, even for a tenant with its own -policy; a template promoted from a sandbox of no pool records no source. Any other key with no configured pool — an image cold-booted on -demand — has no pool layer, so a tenant's claim of one takes the tenant's -policy alone (a root claim of one stays denied). A claim with `"egress": false` +policy; a template promoted from a sandbox of no pool records no source. A pool's +`egress` block belongs to the config file and stays bound to its key: when a +`PUT /v1/pools` drops a config pool that has one, the claims made after it, +now cold-booted on demand, still take that policy. Any other key that is in +no pool at claim time — an image cold-booted on demand — has no pool layer, so +a tenant's claim of one takes the tenant's policy alone (a root claim of one +stays denied). A claim with `"egress": false` takes no policy at all, whatever its pool's: no door is bound, so the guest's dial is refused. Which layers a claim has is settled when it is made and kept for its life: a `PUT /v1/pools` that adds or drops a pool -changes the claims made after it, never a live one (a record from before this +without a config-file policy changes the claims made after it, never a live one (a record from before this rule resolves against the live pool set). Secrets are registered separately and referenced by name — the value comes from the environment, never the config file: a claim's own `guest: false` env first, the node's second diff --git a/docs/mcp.md b/docs/mcp.md index 22f1c4a7..cd3cd589 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -42,12 +42,14 @@ in the guest, which can hold its reply up to 5 s past the cap. | `hibernate` | snapshot + stop, freeing memory while keeping id, files and processes; the next call that reaches the guest wakes it | | `promote` | publish the sandbox as a named template on its node; re-promoting replaces it | | `release` | destroy the sandbox and its files; the session forgets the id, so a second release is rejected as unknown | -| `node_info` | warm pools, promoted templates, live claims, drain state, capacity and mesh peers | +| `node_info` | warm pools, promoted templates, live claims, drain state, capacity and mesh peers; needs the operator token | Sandbox handles (and their tokens) are held by the server process for the session, and released with it: when the MCP client disconnects or the server is signalled, it destroys every sandbox that session claimed, whatever -`ttl_seconds` asked for. `checkpoint` first if the state must survive. +`ttl_seconds` asked for; a claim still in flight at the signal is left to its +lease (one hour unless `ttl_seconds` says otherwise). `checkpoint` first if the +state must survive. Checkpoints outlive sessions: `branch_checkpoint` accepts any known id without a listing round-trip. If the connected node does not hold it, the claim follows a live owner probe and redirect, or heals the checkpoint locally diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index 05d22a3c..0fade356 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -466,7 +466,7 @@ The URL's life is clamped to the claim's remaining lease; an archived claim kept forever (`archive_delete_after_seconds: 0`) has no lease, so there the requested `ttl_seconds` stands unclamped and `ttl_seconds: 0` mints a one-hour URL; releasing the sandbox ends it early either way. -501 when the node +400 a port of 0, 501 when the node has no `preview_listen`. The signed token embeds the sandbox id, port, and owner `advertise_addr`, so any node's preview listener can serve it (forwarding to the owner's main listener) and a released sandbox's URL simply stops @@ -642,7 +642,7 @@ The env is persisted with the claim. A `guest: false` change takes effect on the egress proxy's next request at any time, a hibernated or archived claim included, and never touches the guest. A change to the guest entries is written to the guest at once, so it needs a running guest: on a hibernated, -archived or transitioning sandbox it answers 409 (this verb never wakes one). +archived, failed or transitioning sandbox it answers 409 (this verb never wakes one). A `PUT` identical to the stored env is written to a running guest again, so resending it repairs a guest file that was lost or not updated. A `PATCH` writes the guest file only when it changes a guest entry or sets one, so @@ -653,7 +653,7 @@ are already live, but the guest write failed; resend the same request to deliver it. The guest side is described in [silkd](silkd.md#claim-env). `PUT`/`PATCH`/`DELETE` answer 204, `GET` 200; 400 a bad entry or an unknown body field; 404 unknown id or another tenant's claim; 409 a guest change on a -paused sandbox. +paused or failed sandbox. ## PUT /v1/sandboxes/{id}/instance-metadata diff --git a/e2e/cmd/androidsmoke/main.go b/e2e/cmd/androidsmoke/main.go index 2aaafc3c..6a5456c9 100644 --- a/e2e/cmd/androidsmoke/main.go +++ b/e2e/cmd/androidsmoke/main.go @@ -152,8 +152,5 @@ func dialAdb(ctx context.Context, sb *sandbox.Sandbox) (string, error) { func head(s string) string { lines := strings.Split(strings.TrimSpace(s), "\n") - if len(lines) > 12 { - lines = lines[:12] - } - return strings.Join(lines, "\n") + return strings.Join(lines[:min(len(lines), 12)], "\n") } diff --git a/e2e/cmd/interceptsmoke/main.go b/e2e/cmd/interceptsmoke/main.go index 61ca61e0..dac09fa1 100644 --- a/e2e/cmd/interceptsmoke/main.go +++ b/e2e/cmd/interceptsmoke/main.go @@ -93,8 +93,5 @@ func grepLine(out, needle string) string { func tail(s string) string { lines := strings.Split(strings.TrimRight(s, "\n"), "\n") - if len(lines) > 20 { - lines = lines[len(lines)-20:] - } - return strings.Join(lines, "\n") + return strings.Join(lines[max(0, len(lines)-20):], "\n") } diff --git a/e2e/cmd/lifecycle/main.go b/e2e/cmd/lifecycle/main.go index d3bec0bc..a629da75 100644 --- a/e2e/cmd/lifecycle/main.go +++ b/e2e/cmd/lifecycle/main.go @@ -53,8 +53,7 @@ func run(addr, token, template, netShape string, wait time.Duration) error { } fmt.Printf("idle-hibernated then archived in %.1fs (local VM dropped)\n", elapsed.Seconds()) - // the first access on an archived id is the cold wake: fetch the - // checkpoint, provision a fresh local VM, keep the id/token. + // The first access on an archived id is the cold wake: fetch the checkpoint into a fresh VM under the same id/token. wakeStart := time.Now() got, err := sb.ReadFile(ctx, markerPath) if err != nil { @@ -82,8 +81,7 @@ func run(addr, token, template, netShape string, wait time.Duration) error { return nil } -// waitArchived polls the node until archived reaches n, returning how long the -// reaper's idle→hibernate→archive ladder took. +// waitArchived returns how long the reaper's idle→hibernate→archive ladder took to archive n claims. func waitArchived(ctx context.Context, client *sandbox.Client, n int, timeout time.Duration) (time.Duration, error) { start := time.Now() deadline := start.Add(timeout) diff --git a/mcp/e2e.py b/mcp/e2e.py index 5bebb731..b7408e04 100644 --- a/mcp/e2e.py +++ b/mcp/e2e.py @@ -94,7 +94,7 @@ def main() -> int: mcp.tool("hibernate", sandbox_id=sandbox_id) out = mcp.tool("exec", sandbox_id=sandbox_id, command="cat /root/m.txt") - assert out["stdout"] == "v2", out # transparent wake + assert out["stdout"] == "v2", out print(" hibernate + transparent wake ok") forks = mcp.tool("fork", sandbox_id=sandbox_id, count=2)["sandbox_ids"] diff --git a/mcp/server.go b/mcp/server.go index 50115d8a..7702be9a 100644 --- a/mcp/server.go +++ b/mcp/server.go @@ -47,7 +47,6 @@ type server struct { template string mu sync.Mutex - closed bool boxes map[string]*sandbox.Sandbox ckpts map[string]*sandbox.Checkpoint release sync.Once @@ -72,9 +71,7 @@ func newServer(addr, token, template string) (*server, error) { }, nil } -// serve answers newline-delimited JSON-RPC 2.0 over r/w until EOF — the MCP -// stdio transport. Requests are handled strictly in order: sandbox tools are -// stateful, and an agent's tool calls arrive sequentially anyway. +// serve answers the MCP stdio transport's JSON-RPC 2.0 lines until EOF, in order, since sandbox tools are stateful. func (s *server) serve(ctx context.Context, r *bufio.Reader, w io.Writer) error { defer s.closeBoxes() for { @@ -165,7 +162,6 @@ func (s *server) box(id string) (*sandbox.Sandbox, error) { func (s *server) closeBoxes() { s.release.Do(func() { s.mu.Lock() - s.closed = true boxes := slices.Collect(maps.Values(s.boxes)) clear(s.boxes) s.mu.Unlock() @@ -181,14 +177,8 @@ func (s *server) closeBoxes() { func (s *server) trackBox(sb *sandbox.Sandbox) { s.mu.Lock() - closed := s.closed - if !closed { - s.boxes[sb.ID] = sb - } - s.mu.Unlock() - if closed { - _ = sb.Close() // claimed after the session's release ran; nothing else would ever release it - } + defer s.mu.Unlock() + s.boxes[sb.ID] = sb } func (s *server) dropBox(id string) { diff --git a/mcp/server_test.go b/mcp/server_test.go index b391a7ff..c86f3fa8 100644 --- a/mcp/server_test.go +++ b/mcp/server_test.go @@ -174,35 +174,6 @@ func TestToolRepliesReplaceInvalidUTF8(t *testing.T) { } } -func TestTrackBoxAfterCloseReleasesTheClaim(t *testing.T) { - released := make(chan string, 1) - node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if strings.HasSuffix(r.URL.Path, "/release") { - released <- r.URL.Path - } - w.WriteHeader(http.StatusNoContent) - })) - t.Cleanup(node.Close) - addr := strings.TrimPrefix(node.URL, "http://") - srv, err := newServer(addr, "", "rt:24.04") - if err != nil { - t.Fatalf("server: %v", err) - } - srv.closeBoxes() - srv.trackBox(srv.client.Attach(addr, "sb_late", "tok")) - select { - case path := <-released: - if !strings.Contains(path, "sb_late") { - t.Errorf("released %s, want sb_late", path) - } - case <-time.After(5 * time.Second): - t.Fatal("a claim tracked after closeBoxes was never released") - } - if len(srv.boxes) != 0 { - t.Errorf("boxes = %d after a late track, want 0", len(srv.boxes)) - } -} - func TestCloseBoxesWaitsForEveryReleaseOnce(t *testing.T) { var releases atomic.Int32 node := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/mcp/tools.go b/mcp/tools.go index d2763b84..cc8e7359 100644 --- a/mcp/tools.go +++ b/mcp/tools.go @@ -10,7 +10,6 @@ import ( "strings" "time" - "github.com/cocoonstack/sandbox/protocol/wire" sandbox "github.com/cocoonstack/sandbox/sdk/go" ) @@ -77,7 +76,7 @@ var ( schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint"), "template_name": str("template name to publish as")}, "sandbox_id", "template_name"), toolPromote, }, {"release", "Destroy a sandbox and free its resources; files and processes inside it are lost. This session forgets the id, so a second release of it is rejected as unknown.", schema(props{"sandbox_id": str("id returned by create_sandbox, fork, or branch_checkpoint")}, "sandbox_id"), toolRelease}, - {"node_info", "Report the connected node's warm pools, promoted templates, live claims, drain state, capacity, and mesh peers as JSON.", schema(props{}), toolNodeInfo}, + {"node_info", "Report the connected node's warm pools, promoted templates, live claims, drain state, capacity, and mesh peers as JSON. Needs the operator token; a tenant token is refused.", schema(props{}), toolNodeInfo}, } ) @@ -290,14 +289,6 @@ func toolReadFile(ctx context.Context, s *server, raw jsontext.Value) (string, e if err != nil { return "", err } - // a device or a FIFO is not a file: /dev/null reads as empty and a writerless FIFO blocks in open - info, err := sb.Stat(ctx, args.Path) - if err != nil { - return "", err - } - if info.Kind != wire.FileKindFile { - return "", fmt.Errorf("%s is a %s; read_file reads a regular file", args.Path, info.Kind) - } out := cappedOutput{stopAtCap: true} if err := sb.ReadFileTo(ctx, args.Path, &out); errors.Is(err, errOutputCap) { return "", fmt.Errorf("%s exceeds the %d-byte read_file cap; use exec with head, tail, or a checksum", args.Path, execOutputCap) diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 783338dd..2779dc10 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -30,9 +30,6 @@ const ( // PortWriteChunk keeps a port data frame (payload x4/3 base64 plus envelope) well under MaxFrame. PortWriteChunk = 1 << 20 - // maxSkipDepth bounds a tag scan past nested siblings; wire values nest two deep. - maxSkipDepth = 16 - // GitBranch.Action values (silkd's GitBranchOp). BranchList = "list" BranchCreate = "create" @@ -682,16 +679,13 @@ func DecodeResponse(line []byte) (Response, error) { } // NewFrameScanner wraps r for newline-delimited frames capped at MaxFrame. -// No pre-sized buffer: bulk frames outgrow any fixed start anyway. func NewFrameScanner(r io.Reader) *bufio.Scanner { sc := bufio.NewScanner(r) sc.Buffer(nil, MaxFrame) return sc } -// AppendBulkRequest renders a data-carrying request frame — -// {"v":1,"op":,"data":""} plus newline — into buf, reused across -// calls on the bulk send paths (base64's alphabet needs no JSON escaping). +// AppendBulkRequest renders {"v":1,"op":,"data":""} and a newline into buf's storage; base64 needs no JSON escaping. func AppendBulkRequest(buf []byte, op string, data []byte) []byte { buf = append(buf[:0], requestHead...) buf = append(buf, op...) @@ -713,8 +707,7 @@ func IsContinuation(line []byte) bool { return false } -// fastBulk slices the base64 data out of a canonical bulk frame, skipping the -// json.Unmarshal that dominates downloads; any other shape falls back to slow. +// fastBulk decodes a canonical bulk frame without json.Unmarshal; any other shape falls back to slow. func fastBulk(tag string, slow respDecoder, mk func([]byte) Response) respDecoder { head := []byte(`{"type":"` + tag + `","data":"`) return func(line []byte) (Response, error) { @@ -785,35 +778,13 @@ func scanTag(line []byte, key string) (string, error) { } return tok.String(), nil } - if err = skipValue(dec); err != nil { + if err = dec.SkipValue(); err != nil { return "", err } } return "", nil } -// skipValue consumes one JSON value; a guest cannot grow the host stack with nesting. -func skipValue(dec *jsontext.Decoder) error { - depth := 0 - for { - tok, err := dec.ReadToken() - if err != nil { - return err - } - switch tok.Kind() { - case '{', '[': - if depth++; depth > maxSkipDepth { - return fmt.Errorf("value nested deeper than %d", maxSkipDepth) - } - case '}', ']': - depth-- - } - if depth == 0 { - return nil - } - } -} - func decodeAs[T any](line []byte) (*T, error) { v := new(T) if err := json.Unmarshal(line, v); err != nil { diff --git a/protocol/wire/frame_test.go b/protocol/wire/frame_test.go index d45c1141..77c69262 100644 --- a/protocol/wire/frame_test.go +++ b/protocol/wire/frame_test.go @@ -265,12 +265,12 @@ func TestTagAfterOtherKeys(t *testing.T) { } func TestTagScanBoundsNesting(t *testing.T) { - shallow := `{"x":` + strings.Repeat("[", maxSkipDepth) + strings.Repeat("]", maxSkipDepth) + `,"type":"done"}` + shallow := `{"x":` + strings.Repeat("[", 16) + strings.Repeat("]", 16) + `,"type":"done"}` if _, err := DecodeResponse([]byte(shallow)); err != nil { t.Fatalf("nesting at the cap rejected: %v", err) } deep := `{"x":` + strings.Repeat("[", 1<<20) + strings.Repeat("]", 1<<20) + `,"type":"done"}` - if _, err := DecodeResponse([]byte(deep)); err == nil || !strings.Contains(err.Error(), "nested deeper") { + if _, err := DecodeResponse([]byte(deep)); err == nil || !strings.Contains(err.Error(), "max depth") { t.Fatalf("deep nesting: err %v, want the depth cap", err) } } diff --git a/sandboxd/egress/ca.go b/sandboxd/egress/ca.go index a2d1c6ea..bced95d2 100644 --- a/sandboxd/egress/ca.go +++ b/sandboxd/egress/ca.go @@ -36,7 +36,7 @@ type CA struct { // LoadCA builds the node CA; the intermediate must be signed by the bundle's first cert. func LoadCA(rootCertPEM, interCertPEM, interKeyPEM []byte) (*CA, error) { - root, err := parseRootBundle(rootCertPEM) + root, err := parseCert(rootCertPEM, "root") if err != nil { return nil, err } @@ -123,23 +123,8 @@ func (c *CA) SignLeaf(host string) (*tls.Certificate, error) { return &tls.Certificate{Certificate: [][]byte{der, c.interDER}, PrivateKey: c.leafKey, Leaf: leaf}, nil } +// parseCert returns the file's first certificate and rejects any block that is not a CA certificate: guests trust the whole file. func parseCert(pemBytes []byte, what string) (*x509.Certificate, error) { - block, _ := pem.Decode(pemBytes) - if block == nil { - return nil, fmt.Errorf("%s cert: no pem block", what) - } - cert, err := x509.ParseCertificate(block.Bytes) - if err != nil { - return nil, fmt.Errorf("parse %s cert: %w", what, err) - } - if !cert.IsCA { - return nil, fmt.Errorf("%s cert: not a certificate authority", what) - } - return cert, nil -} - -// parseRootBundle returns the anchor and rejects a non-CA block: guests trust the whole file. -func parseRootBundle(pemBytes []byte) (*x509.Certificate, error) { var anchor *x509.Certificate for rest := pemBytes; ; { var block *pem.Block @@ -147,19 +132,19 @@ func parseRootBundle(pemBytes []byte) (*x509.Certificate, error) { break } if block.Type != pemTypeCertificate { - return nil, fmt.Errorf("root bundle: unexpected %s pem block", block.Type) + return nil, fmt.Errorf("%s cert: unexpected %s pem block", what, block.Type) } cert, err := x509.ParseCertificate(block.Bytes) if err != nil { - return nil, fmt.Errorf("parse root bundle cert: %w", err) + return nil, fmt.Errorf("parse %s cert: %w", what, err) } if !cert.IsCA { - return nil, fmt.Errorf("root bundle: %q is not a certificate authority", cert.Subject.CommonName) + return nil, fmt.Errorf("%s cert: %q is not a certificate authority", what, cert.Subject.CommonName) } anchor = cmp.Or(anchor, cert) } if anchor == nil { - return nil, fmt.Errorf("root cert: no pem block") + return nil, fmt.Errorf("%s cert: no pem block", what) } return anchor, nil } diff --git a/sandboxd/egress/internal_allow.go b/sandboxd/egress/internal_allow.go index a228037d..14c44cca 100644 --- a/sandboxd/egress/internal_allow.go +++ b/sandboxd/egress/internal_allow.go @@ -17,10 +17,10 @@ type InternalAllow struct { // ParseInternalAllow reads "prefix" or "prefix:port,port"; the colon after the prefix length is unambiguous in both families. func ParseInternalAllow(s string) (InternalAllow, error) { cidr, list, scoped := s, "", false - if slash := strings.LastIndexByte(s, '/'); slash >= 0 { + if addr, rest, ok := strings.CutLast(s, "/"); ok { var bits string - bits, list, scoped = strings.Cut(s[slash:], ":") - cidr = s[:slash] + bits + bits, list, scoped = strings.Cut(rest, ":") + cidr = addr + "/" + bits } p, err := netip.ParsePrefix(cidr) if err != nil { diff --git a/sandboxd/egress/policy.go b/sandboxd/egress/policy.go index ab4da0fc..ff7dd257 100644 --- a/sandboxd/egress/policy.go +++ b/sandboxd/egress/policy.go @@ -1,7 +1,4 @@ -// Package egress is the host-side guarded-egress data plane: a forward proxy -// a sandbox reaches as its only route out. Every request is evaluated against -// a per-sandbox policy (domain allow-list, methods) before it leaves the node, -// and a matched rule may inject a node-side credential the guest never holds. +// Package egress is a sandbox's only route out: a forward proxy that enforces its policy and injects node-side credentials. package egress import ( diff --git a/sandboxd/egress/proxy.go b/sandboxd/egress/proxy.go index 98bac4d6..37d14c67 100644 --- a/sandboxd/egress/proxy.go +++ b/sandboxd/egress/proxy.go @@ -223,7 +223,6 @@ func (p *Proxy) serveForward(w http.ResponseWriter, r *http.Request) { p.relay(w, r, Event{Method: r.Method, Host: host, Port: port, Decision: decision}, rule, p.tr, nil) } -// relay forwards one policy-evaluated request upstream and mirrors the response. func (p *Proxy) relay(w http.ResponseWriter, r *http.Request, ev Event, rule Rule, tr *http.Transport, prepare func(*http.Request)) { if ev.Decision == DecisionDeny { p.record(ev) @@ -299,7 +298,6 @@ func splice(a, b net.Conn) { <-done } -// denied answers a policy rejection with a 403 rather than a hang. func denied(w http.ResponseWriter, host string) { http.Error(w, fmt.Sprintf("egress denied: %s", host), http.StatusForbidden) } diff --git a/sandboxd/engine/portconn.go b/sandboxd/engine/portconn.go index 59122937..a2d3b84b 100644 --- a/sandboxd/engine/portconn.go +++ b/sandboxd/engine/portconn.go @@ -5,6 +5,7 @@ import ( "bytes" "encoding/base64" "encoding/json/v2" + "errors" "fmt" "io" "net" @@ -32,7 +33,7 @@ func newGuestPortConn(conn net.Conn, r *bufio.Reader) *guestPortConn { func (g *guestPortConn) Read(p []byte) (int, error) { for len(g.pending) == 0 { - line, err := g.r.ReadBytes('\n') + line, err := g.readFrame() if err != nil { return 0, err } @@ -86,6 +87,22 @@ func (g *guestPortConn) Write(p []byte) (int, error) { return written, nil } +func (g *guestPortConn) readFrame() ([]byte, error) { + line, err := g.r.ReadSlice('\n') + if !errors.Is(err, bufio.ErrBufferFull) { + return line, err + } + long := bytes.Clone(line) + for errors.Is(err, bufio.ErrBufferFull) { + line, err = g.r.ReadSlice('\n') + long = append(long, line...) + if len(long) > wire.MaxFrame { + return nil, fmt.Errorf("port frame exceeds %d bytes", wire.MaxFrame) + } + } + return long, err +} + // fastPortData slices the canonical data frame's base64 out without a JSON parse. func (g *guestPortConn) fastPortData(line []byte) ([]byte, bool) { after, ok := bytes.CutPrefix(line, portDataHead) diff --git a/sandboxd/engine/portconn_test.go b/sandboxd/engine/portconn_test.go index 2a8f8822..6c73b8bf 100644 --- a/sandboxd/engine/portconn_test.go +++ b/sandboxd/engine/portconn_test.go @@ -9,6 +9,7 @@ import ( "net" "os" "path/filepath" + "strings" "testing" "github.com/cocoonstack/sandbox/protocol/wire" @@ -125,6 +126,22 @@ func TestPortReadBufHoldsOneBulkFrame(t *testing.T) { } } +func TestGuestPortConnRefusesAFrameOverMaxFrame(t *testing.T) { + for name, tail := range map[string]string{"unterminated": "", "terminated": "\n"} { + t.Run(name, func(t *testing.T) { + silk, gc := newTestGuestPortConn(t) + + go func() { + _, _ = silk.Write(append(bytes.Repeat([]byte("a"), wire.MaxFrame+1), tail...)) + _ = silk.Close() + }() + if _, err := gc.Read(make([]byte, 64)); err == nil || !strings.Contains(err.Error(), "exceeds") { + t.Fatalf("oversized frame: %v, want a frame size error", err) + } + }) + } +} + func newTestGuestPortConn(t *testing.T) (net.Conn, *guestPortConn) { t.Helper() client, silk := net.Pipe() diff --git a/sandboxd/engine/silkd.go b/sandboxd/engine/silkd.go index e32bac80..16fa2134 100644 --- a/sandboxd/engine/silkd.go +++ b/sandboxd/engine/silkd.go @@ -10,6 +10,11 @@ import ( "github.com/cocoonstack/sandbox/protocol/wire" ) +const ( + silkdListMax = 4096 + silkdReadMax = 64 << 10 +) + // silkdSession is a dialed silkd conn bound to a ctx, with wire-typed request/reply helpers. type silkdSession struct { conn net.Conn @@ -91,6 +96,9 @@ func (e *Engine) silkdList(ctx context.Context, vsockSocket, path string) ([]wir return fmt.Errorf("unexpected silkd frame %q", frame.RespType()) } entries = append(entries, resp.Entries...) + if len(entries) > silkdListMax { + return fmt.Errorf("silkd list of %s exceeds %d entries", path, silkdListMax) + } return nil }) if err != nil { @@ -107,6 +115,9 @@ func (e *Engine) silkdReadFile(ctx context.Context, vsockSocket, path string) ([ return fmt.Errorf("unexpected silkd frame %q", frame.RespType()) } data = append(data, resp.Data...) + if len(data) > silkdReadMax { + return fmt.Errorf("silkd read of %s exceeds %d bytes", path, silkdReadMax) + } return nil }) if err != nil { diff --git a/sandboxd/engine/volume_test.go b/sandboxd/engine/volume_test.go index 4aee874a..f9825acc 100644 --- a/sandboxd/engine/volume_test.go +++ b/sandboxd/engine/volume_test.go @@ -254,6 +254,22 @@ func TestMountVolumeDevNodeNeverAppearsTimesOut(t *testing.T) { } } +func TestSilkdListAndReadAreBounded(t *testing.T) { + path := sockPath(t) + fake := serveFakeSilkd(t, path) + fake.mu.Lock() + fake.block = make([]wire.DirEntry, silkdListMax+1) + fake.serial["/sys/block/vda/serial"] = make([]byte, silkdReadMax+1) + fake.mu.Unlock() + e := New("cocoon", nil, nil, false, false, "") + if _, err := e.silkdList(t.Context(), path, "/sys/block"); err == nil || !strings.Contains(err.Error(), "exceeds") { + t.Errorf("list of %d entries: %v, want a bound error", silkdListMax+1, err) + } + if _, err := e.silkdReadFile(t.Context(), path, "/sys/block/vda/serial"); err == nil || !strings.Contains(err.Error(), "exceeds") { + t.Errorf("read of %d bytes: %v, want a bound error", silkdReadMax+1, err) + } +} + func configureVolumeDevices(f *fakeSilkd) { f.mu.Lock() defer f.mu.Unlock() diff --git a/sandboxd/log.go b/sandboxd/log.go index 5c3c47c4..5b3969ea 100644 --- a/sandboxd/log.go +++ b/sandboxd/log.go @@ -29,6 +29,10 @@ type journalWriter struct { json bool } +func newJournalWriter(out io.Writer, json bool) journalWriter { + return journalWriter{console: zerolog.ConsoleWriter{TimeFormat: time.RFC822, NoColor: true}, out: out, json: json} +} + func (j journalWriter) Write(p []byte) (int, error) { return j.WriteLevel(zerolog.NoLevel, p) } func (j journalWriter) WriteLevel(level zerolog.Level, p []byte) (int, error) { @@ -57,10 +61,6 @@ func setupLog(ctx context.Context, level string) error { return nil } -func newJournalWriter(out io.Writer, json bool) journalWriter { - return journalWriter{console: zerolog.ConsoleWriter{TimeFormat: time.RFC822, NoColor: true}, out: out, json: json} -} - func syslogPrefix(level zerolog.Level) string { switch level { case zerolog.PanicLevel, zerolog.FatalLevel: diff --git a/sandboxd/netfilter/nft.go b/sandboxd/netfilter/nft.go index 484abd9b..c0a8a4a9 100644 --- a/sandboxd/netfilter/nft.go +++ b/sandboxd/netfilter/nft.go @@ -1,6 +1,2 @@ -// Package netfilter locks an egress-lane VM's NIC via nftables so a NIC-backed -// guest cannot bypass the host egress proxy. cocoon's bridge backend keeps the -// tap in the root netns, so the drop rule is applied there in a per-tap table -// and removed again (Unlock) when the sandbox is released — the kernel does not -// reap a netdev table when its device is deleted. +// Package netfilter locks an egress-lane tap to the proxy with a per-tap nftables table that Unlock removes, since the kernel keeps it after the device is gone. package netfilter diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index c8e4d151..4c4a779a 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -149,18 +149,9 @@ func (m *Manager) wakeArchived(ctx context.Context, sb *types.Sandbox) (string, } ctx = context.WithoutCancel(ctx) ck := sb.ArchiveCk - // evicting while the lock is held would split a concurrent recLock onto a different mutex l := m.recLock(ck) l.Lock() - consumed := false - defer func() { - l.Unlock() - if consumed { - m.recDoneEvict(ck) - } else { - m.recDone(ck) - } - }() + defer func() { l.Unlock(); m.recDone(ck) }() dir, _, _, err := m.ckpts.Fetch(ctx, ck) if errors.Is(err, store.ErrNotFound) { return "", ErrUnknownSandbox // record disagrees with the store @@ -187,11 +178,8 @@ func (m *Manager) wakeArchived(ctx context.Context, sb *types.Sandbox) (string, // the pre-commit marker keeps a failed delete retryable after the journal update. if delErr := m.ckpts.Delete(ctx, ck); delErr != nil { log.WithFunc("pool.wakeArchived").Warnf(ctx, "delete consumed archive ck %s: %v", ck, delErr) - } else { - consumed = true - if clearErr := m.clearArchiveCk(ck); clearErr != nil { - log.WithFunc("pool.wakeArchived").Warnf(ctx, "clear archive ck %s: %v", ck, clearErr) - } + } else if clearErr := m.clearArchiveCk(ck); clearErr != nil { + log.WithFunc("pool.wakeArchived").Warnf(ctx, "clear archive ck %s: %v", ck, clearErr) } // only the none lane reaches here, so there is no NIC to re-lock if proxyErr := m.armEgressProxy(ctx, sb); proxyErr != nil { @@ -328,7 +316,7 @@ func (m *Manager) retryArchiveDelete(ctx context.Context, ckID string) { log.WithFunc("pool.retryArchiveDelete").Warnf(ctx, "delete %s: %v", ckID, err) return } - m.recDoneEvict(ckID) + m.recDone(ckID) if err := m.clearArchiveCk(ckID); err != nil { log.WithFunc("pool.retryArchiveDelete").Warnf(ctx, "clear %s: %v", ckID, err) } diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index dad81cd4..78832f69 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -115,7 +115,7 @@ func (m *Manager) DeleteCheckpoint(ctx context.Context, ckptID, tenant string, s } l := m.recLock(ckptID) l.Lock() - defer func() { l.Unlock(); m.recDoneEvict(ckptID) }() + defer func() { l.Unlock(); m.recDone(ckptID) }() ckpt, err := m.loadCheckpoint(ctx, ckptID) if err != nil { m.vetoIfHealPending(ckptID) @@ -181,7 +181,7 @@ func (m *Manager) publishCheckpoint(ctx context.Context, sb *types.Sandbox, ckID return types.Checkpoint{}, "", fmt.Errorf("stage checkpoint: %w", err) } defer func() { _ = os.RemoveAll(staging) }() - srcSnap, guestEnv, err := m.exportSource(ctx, sb, filepath.Join(staging, store.ExportDir)) + srcSnap, _, guestEnv, err := m.exportSource(ctx, sb, filepath.Join(staging, store.ExportDir)) if err != nil { return types.Checkpoint{}, "", fmt.Errorf("checkpoint %s: %w", sb.ID, err) } @@ -362,7 +362,7 @@ func (m *Manager) deleteCkLocked(ctx context.Context, ckID string) error { m.recDone(ckID) return err } - m.recDoneEvict(ckID) + m.recDone(ckID) return nil } diff --git a/sandboxd/pool/checkpoint_heal_test.go b/sandboxd/pool/checkpoint_heal_test.go index 32c7672a..589b8260 100644 --- a/sandboxd/pool/checkpoint_heal_test.go +++ b/sandboxd/pool/checkpoint_heal_test.go @@ -168,7 +168,7 @@ func TestRecLockEvictionWaitsForAllHolders(t *testing.T) { t.Fatalf("recLock returned a different mutex while the first holder is still outstanding") } l1.Unlock() - m.recDoneEvict(id) + m.recDone(id) if !hasRecLock(m, id) { t.Fatal("entry evicted while a second holder still references it") } @@ -179,7 +179,7 @@ func TestRecLockEvictionWaitsForAllHolders(t *testing.T) { } m.recDone(id) - m.recDoneEvict(id) + m.recDone(id) if hasRecLock(m, id) { t.Error("entry not evicted once every holder released") } @@ -194,7 +194,7 @@ func TestRecLockEvictDeferredToLastHolder(t *testing.T) { if l2 != l1 { t.Fatal("recLock diverged for the same id") } - m.recDoneEvict(id) + m.recDone(id) if !hasRecLock(m, id) { t.Fatal("entry evicted while an ordinary holder still references it") } diff --git a/sandboxd/pool/checkpoint_test.go b/sandboxd/pool/checkpoint_test.go index 677e3e56..a6cb210b 100644 --- a/sandboxd/pool/checkpoint_test.go +++ b/sandboxd/pool/checkpoint_test.go @@ -56,6 +56,23 @@ func TestCheckpointThenBranch(t *testing.T) { } } +func TestCheckpointClaimsLeaveNoRecordLock(t *testing.T) { + m := newTestManager(t, newFakeEngine()) + src := mustClaim(t, m, testKey) + ckpt, err := m.Checkpoint(t.Context(), src.ID, Cred{Token: src.Token}, "step-1", "") + if err != nil { + t.Fatalf("Checkpoint: %v", err) + } + for range 3 { + if _, err := m.ClaimCheckpoint(t.Context(), ckpt.ID, ClaimOptions{TTL: time.Hour}); err != nil { + t.Fatalf("ClaimCheckpoint: %v", err) + } + } + if hasRecLock(m, ckpt.ID) { + t.Errorf("recLocks keeps %s after every claim released it (%d entries)", ckpt.ID, lockCount(m)) + } +} + func TestCheckpointBranchTakesItsOwnMetadata(t *testing.T) { m := newTestManager(t, newFakeEngine()) src, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Metadata: types.Metadata{"role": "source"}}) diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index e0cbba85..49fdefc1 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -660,6 +660,25 @@ func (m *Manager) recommit(ctx context.Context, snap claimSnapshot) { }() } +func (m *Manager) commitIfLive(ctx context.Context, sb *types.Sandbox, mutate func()) (bool, error) { + m.mu.Lock() + live := m.claimed[sb.ID] == sb + var js claimSnapshot + if live { + mutate() + js = m.store.set(sb) + } + m.mu.Unlock() + if !live { + return false, nil + } + if err := m.store.commit(js); err != nil { + m.recommit(ctx, js) + return true, err + } + return true, nil +} + func (m *Manager) claim(id, token string) (*types.Sandbox, bool) { m.mu.Lock() defer m.mu.Unlock() @@ -691,22 +710,6 @@ func (m *Manager) admitOptions(ctx context.Context, key types.PoolKey, o ClaimOp return volumeSpecs, applied, err } -func stampIdentity(sb *types.Sandbox, ttl time.Duration) { - now := time.Now() - sb.ID = "sb_" + randHex(8) - sb.Token = randHex(16) - sb.ClaimedAt = now - sb.Deadline = now.Add(ttl) - sb.LeaseSeconds = int(ttl / time.Second) -} - -func clampTTL(ttl time.Duration) time.Duration { - if ttl <= 0 { - return defaultTTL - } - return min(ttl, maxTTL) -} - // heldConn keeps its sandbox held until the guest connection closes. type heldConn struct { net.Conn @@ -727,6 +730,22 @@ func (c *heldConn) CloseWrite() error { return cw.CloseWrite() } +func stampIdentity(sb *types.Sandbox, ttl time.Duration) { + now := time.Now() + sb.ID = "sb_" + randHex(8) + sb.Token = randHex(16) + sb.ClaimedAt = now + sb.Deadline = now.Add(ttl) + sb.LeaseSeconds = int(ttl / time.Second) +} + +func clampTTL(ttl time.Duration) time.Duration { + if ttl <= 0 { + return defaultTTL + } + return min(ttl, maxTTL) +} + func archivable(key types.PoolKey, volumes bool, onExpire types.ExpireAction) error { switch { case onExpire != types.ExpireArchive: diff --git a/sandboxd/pool/env.go b/sandboxd/pool/env.go index d2d1d43c..e2c6bf1d 100644 --- a/sandboxd/pool/env.go +++ b/sandboxd/pool/env.go @@ -50,7 +50,7 @@ func (m *Manager) writeEnv(ctx context.Context, id, tenant, verb string, next fu m.mu.Unlock() return ErrUnknownSandbox } - prev, sock := sb.Env, sb.VsockSocket + prev, sock, failed := sb.Env, sb.VsockSocket, sb.Failed != "" env, resend := next(prev) if len(env) == 0 { env = nil @@ -60,9 +60,12 @@ func (m *Manager) writeEnv(ctx context.Context, id, tenant, verb string, next fu return fmt.Errorf("%w: %w", ErrBadEnv, err) } changed := !env.SameGuest(prev) - paused := !locked || sb.HibernateSnap != "" || sb.PendingSnap != "" || sb.ArchiveCk != "" + paused := !locked || sb.HibernateSnap != "" || sb.PendingSnap != "" || sb.ArchiveCk != "" || failed if changed && paused { m.mu.Unlock() + if failed { + return ErrFailed + } return ErrPaused } deliver := !paused && (changed || resend && env.Equal(prev)) diff --git a/sandboxd/pool/env_test.go b/sandboxd/pool/env_test.go index 79aab07f..8ccc46d0 100644 --- a/sandboxd/pool/env_test.go +++ b/sandboxd/pool/env_test.go @@ -301,6 +301,24 @@ func TestAGuestEnvChangeNeedsARunningGuest(t *testing.T) { } } +func TestAGuestEnvChangeOnAFailedClaimIsRefused(t *testing.T) { + eng := newFakeEngine() + m := envManager(t, eng, t.TempDir()) + sb, err := m.ClaimProvision(t.Context(), testKey, ClaimOptions{Env: types.Env{"G": {Value: "1"}}}) + if err != nil { + t.Fatalf("claim: %v", err) + } + sb.Transition.Lock() + m.failLocked(t.Context(), sb, "vmm exited") + sb.Transition.Unlock() + if err := m.SetEnv(t.Context(), sb.ID, types.Env{"G": {Value: "2"}}, ""); !errors.Is(err, ErrFailed) { + t.Errorf("guest change on a failed claim: %v, want ErrFailed", err) + } + if got := sb.Env["G"].Value; got != "1" { + t.Errorf("refused change stored G=%q, want 1", got) + } +} + func TestAFailedEnvDeliveryDestroysTheClaim(t *testing.T) { eng := newFakeEngine() eng.guestEnvErr = errors.New("guest write failed") diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index 5a803fa0..7d00a6d3 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -235,27 +235,14 @@ func (m *Manager) idleHibernate(ctx context.Context, id, token string, sweepStar // commitTransition publishes a hibernate/wake result only if the claim is still live. func (m *Manager) commitTransition(ctx context.Context, sb *types.Sandbox, snap, sock string) (live bool, err error) { - m.mu.Lock() - live = m.claimed[sb.ID] == sb - var js claimSnapshot - if live { + return m.commitIfLive(ctx, sb, func() { sb.HibernateSnap = snap sb.VsockSocket = sock sb.PendingSnap = "" if now := time.Now(); snap == "" && sb.OnExpire == types.ExpireArchive && now.After(sb.Deadline) { sb.Deadline = now.Add(clampTTL(types.Seconds(sb.LeaseSeconds))) } - js = m.store.set(sb) - } - m.mu.Unlock() - if !live { - return false, nil - } - if err := m.store.commit(js); err != nil { - m.recommit(ctx, js) - return true, err - } - return true, nil + }) } // syncClaims flushes a lagging journal so a hibernate retry cannot report a false success. diff --git a/sandboxd/pool/liveness.go b/sandboxd/pool/liveness.go index ee3a784c..842dbb09 100644 --- a/sandboxd/pool/liveness.go +++ b/sandboxd/pool/liveness.go @@ -226,24 +226,15 @@ func (m *Manager) restartLocked(ctx context.Context, sb *types.Sandbox) error { m.stopVM(ctx, sb.VMName) return fmt.Errorf("restore egress and env after cold boot: %w", err) } - m.mu.Lock() - live := m.claimed[sb.ID] == sb - var js claimSnapshot - if live { + if live, _ := m.commitIfLive(ctx, sb, func() { sb.Failed = "" sb.Restarts++ sb.RestartedAt = time.Now() - js = m.store.set(sb) - } - m.mu.Unlock() - if !live { + }); !live { // the release that dropped the claim owns the VM teardown m.disarmEgress(sb.ID, true) return ErrUnknownSandbox } - if err := m.store.commit(js); err != nil { - m.recommit(ctx, js) - } if m.disarmIfReleased(sb) { return ErrUnknownSandbox } @@ -256,20 +247,9 @@ func (m *Manager) restartLocked(ctx context.Context, sb *types.Sandbox) error { // failLocked parks sb as failed with reason; the caller holds sb.Transition. func (m *Manager) failLocked(ctx context.Context, sb *types.Sandbox, reason string) { - m.mu.Lock() - live := m.claimed[sb.ID] == sb - var js claimSnapshot - if live { - sb.Failed = reason - js = m.store.set(sb) - } - m.mu.Unlock() - if !live { + if live, _ := m.commitIfLive(ctx, sb, func() { sb.Failed = reason }); !live { return } - if err := m.store.commit(js); err != nil { - m.recommit(ctx, js) - } m.recordUsage(ctx, usageEvent{Event: "vmm_failed", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant, Reference: reason}) log.WithFunc("pool.failLocked").Warnf(ctx, "sandbox %s failed: %s", sb.ID, reason) } diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 7b58cde0..9b67171c 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -1,6 +1,4 @@ -// Package pool owns a node's warm pools and claimed sandboxes: refill keeps -// every configured pool topped up with claim-ready VMs (cloned, reseeded, -// probed), so a claim is ownership transfer only. +// Package pool owns a node's warm pools and claimed sandboxes; refill keeps pools claim-ready, so a claim is ownership transfer only. package pool import ( @@ -341,12 +339,10 @@ type Manager struct { tplMu sync.Mutex tplSet map[string]TemplateInfo - // recLocks serializes same-id record mutations; an entry evicts only when no holder remains, never on a bare delete, since a peer heal can republish the id. + // recLocks serializes same-id record mutations; an entry lives exactly as long as a recLock reference to it. recLocks map[string]*sync.RWMutex recLocksMu sync.Mutex recRefs map[string]int - // recEvict defers a delete's eviction to the last holder, else the recLocks entry leaks. - recEvict map[string]struct{} // notifyTemplates is set before serving starts and fires after a promote or template delete. notifyTemplates func() @@ -408,7 +404,6 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg egressTaps: map[string]string{}, recLocks: map[string]*sync.RWMutex{}, recRefs: map[string]int{}, - recEvict: map[string]struct{}{}, healPending: map[string]struct{}{}, healAbort: map[string]struct{}{}, egressSecrets: secrets, diff --git a/sandboxd/pool/promote_test.go b/sandboxd/pool/promote_test.go index b7c72d6b..c5c9cb26 100644 --- a/sandboxd/pool/promote_test.go +++ b/sandboxd/pool/promote_test.go @@ -189,9 +189,6 @@ func TestTemplateRecordLockEvictsWithTheRecord(t *testing.T) { if _, _, err := m.Promote(t.Context(), parent.ID, Cred{Token: parent.Token}, key.Template, ""); err != nil { t.Fatalf("Promote: %v", err) } - if !hasRecLock(m, id) { - t.Error("recLocks dropped the entry of a live template") - } if err := m.DeleteTemplate(t.Context(), key, "", ""); err != nil { t.Fatalf("DeleteTemplate: %v", err) } diff --git a/sandboxd/pool/reclock.go b/sandboxd/pool/reclock.go index e86d680c..5f4c83a4 100644 --- a/sandboxd/pool/reclock.go +++ b/sandboxd/pool/reclock.go @@ -22,29 +22,6 @@ func (m *Manager) recDone(id string) { m.recRefs[id]-- if m.recRefs[id] <= 0 { delete(m.recRefs, id) - m.evictIfPending(id) - } -} - -// recDoneEvict is recDone for a just-deleted record: the lock slot goes at zero references. -func (m *Manager) recDoneEvict(id string) { - m.recLocksMu.Lock() - defer m.recLocksMu.Unlock() - m.recRefs[id]-- - if m.recRefs[id] <= 0 { - delete(m.recRefs, id) - delete(m.recEvict, id) - delete(m.recLocks, id) - return - } - // a holder or waiter remains; the call that drops the last reference evicts instead - m.recEvict[id] = struct{}{} -} - -// evictIfPending drops id's lock entry if a delete asked for eviction; callers hold recLocksMu. -func (m *Manager) evictIfPending(id string) { - if _, ok := m.recEvict[id]; ok { - delete(m.recEvict, id) delete(m.recLocks, id) } } diff --git a/sandboxd/pool/refill.go b/sandboxd/pool/refill.go index f69f7029..6e71cac1 100644 --- a/sandboxd/pool/refill.go +++ b/sandboxd/pool/refill.go @@ -354,17 +354,17 @@ func (m *Manager) sourceSnap(ctx context.Context, sb *types.Sandbox) (string, fu return snap, func() { m.dropSnap(ctx, snap) }, nil } -// exportSource captures a claimed sandbox's state into exportDir, returning the snapshot name and whether it holds a guest env file. -func (m *Manager) exportSource(ctx context.Context, sb *types.Sandbox, exportDir string) (snap string, guestEnv bool, err error) { +// exportSource captures a claimed sandbox into exportDir and returns its snapshot, VM name and guest-env presence. +func (m *Manager) exportSource(ctx context.Context, sb *types.Sandbox, exportDir string) (snap, vmName string, guestEnv bool, err error) { sb.Transition.Lock() defer sb.Transition.Unlock() m.trimForCapture(ctx, sb) snap, cleanup, err := m.sourceSnap(ctx, sb) if err != nil { - return "", false, err + return "", "", false, err } defer cleanup() - return snap, m.heldGuestEnv(sb), m.eng.SnapshotExport(ctx, snap, exportDir) + return snap, sb.VMName, m.heldGuestEnv(sb), m.eng.SnapshotExport(ctx, snap, exportDir) } // trimForCapture trims a live sandbox's copy-on-write disk when its pool asks for it; a failed trim only leaves the capture larger. diff --git a/sandboxd/pool/template.go b/sandboxd/pool/template.go index 7b0ae05c..2d9a9df9 100644 --- a/sandboxd/pool/template.go +++ b/sandboxd/pool/template.go @@ -62,22 +62,22 @@ func (m *Manager) Promote(ctx context.Context, id string, cred Cred, template, t if err := m.checkTemplateOwner(ctx, store.TemplateID(key.Hash()), tenant); err != nil { return types.PoolKey{}, "", err } - // the transition lock pins the source snapshot; a started promote must finish uncanceled - sb.Transition.Lock() - defer sb.Transition.Unlock() + // a started promote must finish uncanceled ctx = context.WithoutCancel(ctx) - m.trimForCapture(ctx, sb) - - snap, cleanup, err := m.sourceSnap(ctx, sb) + staging, err := m.tpls.Stage(store.TemplateID(key.Hash())) + if err != nil { + return types.PoolKey{}, "", fmt.Errorf("stage template: %w", err) + } + defer func() { _ = os.RemoveAll(staging) }() + _, vmName, guestEnv, err := m.exportSource(ctx, sb, filepath.Join(staging, store.ExportDir)) if err != nil { return types.PoolKey{}, "", fmt.Errorf("promote %s: %w", sb.ID, err) } - defer cleanup() - rec := templateRecord{Key: key, Tenant: tenant, GuestEnv: m.heldGuestEnv(sb)} + rec := templateRecord{Key: key, Tenant: tenant, GuestEnv: guestEnv} if sb.Layer == types.LayerPooled { rec.PolicySource = sb.PolicyKey() } - digest, err := m.publishTemplate(ctx, snap, rec) + digest, err := m.commitTemplate(ctx, staging, rec) if err != nil { return types.PoolKey{}, "", fmt.Errorf("promote %s: %w", sb.ID, err) } @@ -85,7 +85,7 @@ func (m *Manager) Promote(ctx context.Context, id string, cred Cred, template, t m.notifyTemplates() } m.counters.promotes.Add(1) - m.recordUsage(ctx, usageEvent{Event: "promote", ID: sb.ID, VMName: sb.VMName, Reference: key.Template}) + m.recordUsage(ctx, usageEvent{Event: "promote", ID: sb.ID, VMName: vmName, Reference: key.Template}) return key, digest, nil } @@ -100,7 +100,7 @@ func (m *Manager) DeleteTemplate(ctx context.Context, key types.PoolKey, tenant, id := store.TemplateID(key.Hash()) l := m.recLock(id) l.Lock() - defer func() { l.Unlock(); m.recDoneEvict(id) }() + defer func() { l.Unlock(); m.recDone(id) }() if err := m.ownedTemplate(ctx, id, tenant); err != nil { return err } @@ -147,7 +147,7 @@ func (m *Manager) SetTemplateLabels(ctx context.Context, key types.PoolKey, labe id := store.TemplateID(key.Hash()) l := m.recLock(id) l.Lock() - defer func() { l.Unlock(); m.recDoneEvict(id) }() + defer func() { l.Unlock(); m.recDone(id) }() err := m.ownedTemplate(ctx, id, tenant) if err != nil { return err @@ -292,19 +292,6 @@ func (m *Manager) checkTemplateOwner(ctx context.Context, id, tenant string) err return nil } -func (m *Manager) publishTemplate(ctx context.Context, snap string, rec templateRecord) (string, error) { - id := store.TemplateID(rec.Key.Hash()) - staging, err := m.tpls.Stage(id) - if err != nil { - return "", fmt.Errorf("stage template: %w", err) - } - defer func() { _ = os.RemoveAll(staging) }() - if err = m.eng.SnapshotExport(ctx, snap, filepath.Join(staging, store.ExportDir)); err != nil { - return "", fmt.Errorf("export template: %w", err) - } - return m.commitTemplate(ctx, staging, rec) -} - func (m *Manager) commitTemplate(ctx context.Context, staging string, rec templateRecord) (string, error) { id := store.TemplateID(rec.Key.Hash()) l := m.recLock(id) @@ -369,7 +356,7 @@ func (m *Manager) resolveGolden(ctx context.Context, key types.PoolKey, tenant s dir, meta, digest, err := m.tpls.Fetch(ctx, id) if errors.Is(err, store.ErrNotFound) { l.RUnlock() - m.recDoneEvict(id) + m.recDone(id) return goldenResolution{}, nil } if err != nil { diff --git a/sandboxd/server/preview.go b/sandboxd/server/preview.go index 11f7e45d..f38bbe99 100644 --- a/sandboxd/server/preview.go +++ b/sandboxd/server/preview.go @@ -185,6 +185,10 @@ func (s *Server) handlePreview(w http.ResponseWriter, r *http.Request) { if !ok { return } + if req.Port == 0 { + writeErr(w, http.StatusBadRequest, "port must be 1-65535") + return + } id := r.PathValue("id") deadline, err := s.mgr.ClaimDeadline(id, req.Token) if err != nil { @@ -195,7 +199,7 @@ func (s *Server) handlePreview(w http.ResponseWriter, r *http.Request) { switch { case !deadline.IsZero(): if lease := time.Until(deadline); ttl <= 0 || ttl > lease { - ttl = lease // never outlive the claim + ttl = lease } case ttl <= 0: ttl = previewTTL diff --git a/sandboxd/server/relay.go b/sandboxd/server/relay.go index bccf98d1..dae6c883 100644 --- a/sandboxd/server/relay.go +++ b/sandboxd/server/relay.go @@ -213,10 +213,7 @@ func (t *auditTee) take(b []byte) { if t.skip { return } - if room := pool.AuditLineCap + 1 - len(t.buf); len(b) > room { - b = b[:room] - } - t.buf = append(t.buf, b...) + t.buf = append(t.buf, b[:min(len(b), pool.AuditLineCap+1-len(t.buf))]...) if len(t.buf) <= pool.AuditLineCap { return } diff --git a/sandboxd/server/server.go b/sandboxd/server/server.go index 18f6624d..49e94561 100644 --- a/sandboxd/server/server.go +++ b/sandboxd/server/server.go @@ -460,9 +460,7 @@ func (s *Server) handleInstanceMetadata(w http.ResponseWriter, r *http.Request) } id := r.PathValue("id") err = s.mgr.SetInstanceMetadata(r.Context(), id, doc) - writeResult(w, r, "instance metadata", id, "instance metadata failed", err, func() { - w.WriteHeader(http.StatusNoContent) - }) + writeResult(w, r, "instance metadata", id, "instance metadata failed", err, noContent(w)) } func (s *Server) handleGetEnv(w http.ResponseWriter, r *http.Request) { @@ -488,9 +486,7 @@ func (s *Server) handlePatchEnv(w http.ResponseWriter, r *http.Request) { } id := r.PathValue("id") err := s.mgr.PatchEnv(r.Context(), id, req.Env, tenantFrom(r.Context())) - writeResult(w, r, "env", id, "patch env failed", err, func() { - w.WriteHeader(http.StatusNoContent) - }) + writeResult(w, r, "env", id, "patch env failed", err, noContent(w)) } func (s *Server) handleClearEnv(w http.ResponseWriter, r *http.Request) { @@ -500,9 +496,7 @@ func (s *Server) handleClearEnv(w http.ResponseWriter, r *http.Request) { func (s *Server) setEnv(w http.ResponseWriter, r *http.Request, env types.Env) { id := r.PathValue("id") err := s.mgr.SetEnv(r.Context(), id, env, tenantFrom(r.Context())) - writeResult(w, r, "env", id, "set env failed", err, func() { - w.WriteHeader(http.StatusNoContent) - }) + writeResult(w, r, "env", id, "set env failed", err, noContent(w)) } func (s *Server) handleSandboxVerb(verb string, do func(ctx context.Context, id string, cred pool.Cred) error) http.HandlerFunc { @@ -513,9 +507,7 @@ func (s *Server) handleSandboxVerb(verb string, do func(ctx context.Context, id } id := r.PathValue("id") err := do(r.Context(), id, s.sandboxCred(token)) - writeResult(w, r, verb, id, verb+" failed", err, func() { - w.WriteHeader(http.StatusNoContent) - }) + writeResult(w, r, verb, id, verb+" failed", err, noContent(w)) } } @@ -640,9 +632,7 @@ func (s *Server) handleDeleteCheckpoint(w http.ResponseWriter, r *http.Request) if s.prober != nil { s.prober.Forget(id) } - writeResult(w, r, "delete checkpoint", id, "delete checkpoint failed", err, func() { - w.WriteHeader(http.StatusNoContent) - }) + writeResult(w, r, "delete checkpoint", id, "delete checkpoint failed", err, noContent(w)) } func (s *Server) handleDeleteTemplate(w http.ResponseWriter, r *http.Request) { @@ -653,9 +643,7 @@ func (s *Server) handleDeleteTemplate(w http.ResponseWriter, r *http.Request) { s.writeRedirect(w, s.templateOwners(s.placer.TemplateOwners, key.Hash(), tenantFrom(r.Context()))) { return } - writeResult(w, r, "delete template", key.Template, "delete template failed", err, func() { - w.WriteHeader(http.StatusNoContent) - }) + writeResult(w, r, "delete template", key.Template, "delete template failed", err, noContent(w)) } func (s *Server) handleSetTemplateLabels(w http.ResponseWriter, r *http.Request) { @@ -666,9 +654,7 @@ func (s *Server) handleSetTemplateLabels(w http.ResponseWriter, r *http.Request) q := r.URL.Query() key := templateKey(q) err := s.mgr.SetTemplateLabels(r.Context(), key, req.Labels, tenantFrom(r.Context()), q.Get("digest")) - writeResult(w, r, "template labels", key.Template, "set template labels failed", err, func() { - w.WriteHeader(http.StatusNoContent) - }) + writeResult(w, r, "template labels", key.Template, "set template labels failed", err, noContent(w)) } func (s *Server) handleOwner(w http.ResponseWriter, r *http.Request) { diff --git a/sandboxd/server/server_http.go b/sandboxd/server/server_http.go index dbf5ef2f..17209a25 100644 --- a/sandboxd/server/server_http.go +++ b/sandboxd/server/server_http.go @@ -177,6 +177,10 @@ func writeResult(w http.ResponseWriter, r *http.Request, op, id, failMsg string, } } +func noContent(w http.ResponseWriter) func() { + return func() { w.WriteHeader(http.StatusNoContent) } +} + 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 edeb1313..440f6439 100644 --- a/sandboxd/server/server_test.go +++ b/sandboxd/server/server_test.go @@ -2068,6 +2068,26 @@ func TestPreviewHandlerZeroDeadlineMintsLiveToken(t *testing.T) { } } +func TestPreviewHandlerRejectsPortZero(t *testing.T) { + mgr := &fakeManager{claimDeadline: func(string, string) (time.Time, error) { + return time.Now().Add(time.Hour), nil + }} + ps := NewPreviewServer("secret", "node:7777", "node:7777", &fakePreviewMgr{}) + srv := New("", nil, "node:7777", mgr, &fakeDialer{}, nil, nil, nil, ps) + ts := httptest.NewServer(srv.Handler()) + t.Cleanup(func() { ts.Close(); srv.CloseRelays() }) + + resp, err := http.Post(ts.URL+"/v1/sandboxes/sb_1/preview", "application/json", + strings.NewReader(`{"token":"tok","port":0}`)) + if err != nil { + t.Fatalf("preview: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusBadRequest { + t.Errorf("status %d, want 400 for port 0", resp.StatusCode) + } +} + func TestCheckpointClaimRedirectsToOwner(t *testing.T) { mgr := &fakeManager{} prober := &fakeProber{owners: []string{"owner-a:7777"}} diff --git a/sandboxd/store/dir/dir.go b/sandboxd/store/dir/dir.go index ebbf2df1..82a7b4ef 100644 --- a/sandboxd/store/dir/dir.go +++ b/sandboxd/store/dir/dir.go @@ -28,9 +28,6 @@ const ( digestPrefix = "digest-" digestWorkerLimit = 8 digestReadBufferSize = 128 << 10 - - // generationGrace bounds the race where another node still clones a generation being re-published. - generationGrace = time.Hour ) type digestSource struct { @@ -219,7 +216,7 @@ func (d *Store) publish(_ context.Context, staging, id, digest string) error { return readErr } // a sweep may already have selected an expired path for removal. - if time.Since(genInfo.ModTime()) >= generationGrace { + if time.Since(genInfo.ModTime()) >= store.GenerationGrace { return fmt.Errorf("generation %s expired before commit; retry after sweep", filepath.Base(genDir)) } } @@ -333,7 +330,7 @@ func removeAgedEntry(final string, entry fs.DirEntry) error { return err } // Publish refreshes a generation before it becomes visible or superseded. - if time.Since(info.ModTime()) < generationGrace { + if time.Since(info.ModTime()) < store.GenerationGrace { return nil } return os.RemoveAll(filepath.Join(final, entry.Name())) diff --git a/sandboxd/store/dir/dir_test.go b/sandboxd/store/dir/dir_test.go index d2c82127..ba608c08 100644 --- a/sandboxd/store/dir/dir_test.go +++ b/sandboxd/store/dir/dir_test.go @@ -353,8 +353,8 @@ func TestPublishSweepsGenerationsBySupersessionAge(t *testing.T) { t.Fatalf("superseded generation reclaimed inside the grace: %v", statErr) } - setAge(t, gen1, generationGrace+time.Minute) - setAge(t, gen2, generationGrace/2) + setAge(t, gen1, store.GenerationGrace+time.Minute) + setAge(t, gen2, store.GenerationGrace/2) mustPublish(t, st, id, "fourth") if _, statErr := os.Stat(gen1); !os.IsNotExist(statErr) { t.Fatalf("generation past its supersession grace survived publish: %v", statErr) @@ -623,7 +623,7 @@ func seedRecord(t *testing.T, dir, metaID string) { func backdate(t *testing.T, path string) { t.Helper() - setAge(t, path, 2*generationGrace) + setAge(t, path, 2*store.GenerationGrace) } func setAge(t *testing.T, path string, age time.Duration) { diff --git a/sandboxd/store/peer/transport.go b/sandboxd/store/peer/transport.go index 22704007..46052632 100644 --- a/sandboxd/store/peer/transport.go +++ b/sandboxd/store/peer/transport.go @@ -64,10 +64,7 @@ func (p *HTTPPuller) Pull(ctx context.Context, addr, id, dst string) error { if err != nil { return fmt.Errorf("pull %s from %s: %w", id, addr, err) } - defer func() { - _, _ = io.Copy(io.Discard, resp.Body) - _ = resp.Body.Close() - }() + defer func() { _ = resp.Body.Close() }() switch resp.StatusCode { case http.StatusOK: diff --git a/sandboxd/store/peer/transport_test.go b/sandboxd/store/peer/transport_test.go index 15638394..2fcd6a22 100644 --- a/sandboxd/store/peer/transport_test.go +++ b/sandboxd/store/peer/transport_test.go @@ -11,6 +11,7 @@ import ( "path/filepath" "strings" "testing" + "time" ) func TestTarUntarRoundTrip(t *testing.T) { @@ -162,6 +163,25 @@ func TestHTTPPullerNotFound(t *testing.T) { } } +func TestHTTPPullerStopsReadingAfterAFailedUntar(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write(bytes.Repeat([]byte("x"), 4096)) + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) && r.Context().Err() == nil { + if _, err := w.Write(make([]byte, 64<<10)); err != nil { + return + } + w.(http.Flusher).Flush() + } + })) + t.Cleanup(srv.Close) + start := time.Now() + err := (&HTTPPuller{Client: srv.Client()}).Pull(t.Context(), srv.URL, "ck_0000000000000001", t.TempDir()) + if err == nil || time.Since(start) > 5*time.Second { + t.Fatalf("pull of a non-tar body: %v after %s, want a prompt untar error", err, time.Since(start)) + } +} + func TestHTTPPullerPresentsToken(t *testing.T) { var gotAuth, gotPath string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/sandboxd/store/s3/s3.go b/sandboxd/store/s3/s3.go index e8cf2147..cbfd0ad7 100644 --- a/sandboxd/store/s3/s3.go +++ b/sandboxd/store/s3/s3.go @@ -92,7 +92,7 @@ func New(ctx context.Context, cfg Config, stagingRoot string, idRe *regexp.Regex } client := awss3.NewFromConfig(awsCfg, func(o *awss3.Options) { if cfg.Endpoint != "" { - o.BaseEndpoint = aws.String(cfg.Endpoint) + o.BaseEndpoint = new(cfg.Endpoint) } o.UsePathStyle = cfg.ForcePathStyle }) @@ -125,6 +125,8 @@ func (s *Store) Fetch(ctx context.Context, id string) (string, []byte, string, e gen := filepath.Join(s.staging, "cache", id, store.ExportGenHash(meta)) export := filepath.Join(gen, store.ExportDir) if _, statErr := os.Stat(export); statErr == nil { + now := time.Now() + _ = os.Chtimes(gen, now, now) return export, meta, digest, nil } flight := s.fetches.DoChan(gen, func() (any, error) { @@ -152,7 +154,7 @@ func (s *Store) Metas(ctx context.Context) ([]store.Record, error) { // delimiter listing yields one CommonPrefix per record instead of every export object. var ids []string p := awss3.NewListObjectsV2Paginator(s.client, &awss3.ListObjectsV2Input{ - Bucket: &s.bucket, Prefix: &s.prefix, Delimiter: aws.String("/"), + Bucket: &s.bucket, Prefix: &s.prefix, Delimiter: new("/"), }) for p.HasMorePages() { page, err := p.NextPage(ctx) @@ -252,7 +254,7 @@ func (s *Store) publish(ctx context.Context, staging, id string, digested bool) if err != nil { return "", fmt.Errorf("staging has no %s: %w", store.MetaFile, err) } - files, err := s.publishFiles(staging, id, store.ExportGen(metaRaw), digested) + files, err := s.publishFiles(staging, id, store.ExportGen(metaRaw)) if err != nil { return "", err } @@ -293,11 +295,8 @@ func (s *Store) publish(ctx context.Context, staging, id string, digested bool) return digest, nil } -func (s *Store) publishFiles(staging, id, gen string, digested bool) ([]publishFile, error) { - root := staging - if digested { - root = filepath.Join(staging, store.ExportDir) - } +func (s *Store) publishFiles(staging, id, gen string) ([]publishFile, error) { + root := filepath.Join(staging, store.ExportDir) var files []publishFile err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error { if err != nil || entry.IsDir() { @@ -307,16 +306,6 @@ func (s *Store) publishFiles(staging, id, gen string, digested bool) ([]publishF if err != nil { return err } - if !digested { - if rel == store.MetaFile { - return nil - } - files = append(files, publishFile{ - path: path, - key: s.key(id, gen+"/"+strings.TrimPrefix(rel, store.ExportDir+"/")), - }) - return nil - } info, err := entry.Info() if err != nil { return err @@ -369,7 +358,11 @@ func (s *Store) populate(ctx context.Context, id string, meta []byte, gen string if err := os.MkdirAll(filepath.Dir(gen), 0o750); err != nil { return err } - return os.Rename(local, gen) + if err := os.Rename(local, gen); err != nil { + return err + } + pruneIdleGenerations(filepath.Dir(gen), gen) + return nil } func (s *Store) key(id, rest string) string { @@ -417,7 +410,7 @@ func (s *Store) uploadReader(ctx context.Context, key string, body io.Reader, si Bucket: &s.bucket, Key: &key, Body: body, - ContentLength: aws.Int64(size), + ContentLength: new(size), Metadata: metadata, } if _, err := s.tm.UploadObject(ctx, input); err != nil { @@ -450,7 +443,7 @@ func (s *Store) deleteKeys(ctx context.Context, keys []string) error { } out, err := s.client.DeleteObjects(ctx, &awss3.DeleteObjectsInput{ Bucket: &s.bucket, - Delete: &s3types.Delete{Objects: objs, Quiet: aws.Bool(true)}, + Delete: &s3types.Delete{Objects: objs, Quiet: new(true)}, }) if err != nil { return fmt.Errorf("delete %d objects: %w", len(chunk), err) @@ -480,3 +473,17 @@ func (s *Store) list(ctx context.Context, prefix string) ([]string, error) { } return keys, nil } + +// pruneIdleGenerations drops cached generations of one record that no Fetch used within the grace; a hit refreshes the mtime. +func pruneIdleGenerations(dir, keep string) { + entries, err := os.ReadDir(dir) + if err != nil { + return + } + for _, entry := range entries { + path := filepath.Join(dir, entry.Name()) + if info, err := entry.Info(); err == nil && path != keep && time.Since(info.ModTime()) >= store.GenerationGrace { + _ = os.RemoveAll(path) + } + } +} diff --git a/sandboxd/store/s3/s3_test.go b/sandboxd/store/s3/s3_test.go index 58202742..63bf7d73 100644 --- a/sandboxd/store/s3/s3_test.go +++ b/sandboxd/store/s3/s3_test.go @@ -383,6 +383,57 @@ func TestRepublishRetainsGenerationSelectedByAnotherStore(t *testing.T) { } } +func TestFetchPrunesSupersededGenerationsIdlePastTheGrace(t *testing.T) { + const id = "ck_00000000000000aa" + st := newTestStore(t, &fakeS3{objects: map[string][]byte{}}) + fetchGen := func(gen int) string { + t.Helper() + publishRecord(t, st, id, []byte(fmt.Sprintf(`{"id":"%s","gen":%d}`, id, gen)), fmt.Sprint(gen)) + export, _, _, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("fetch generation %d: %v", gen, err) + } + return filepath.Dir(export) + } + first := fetchGen(1) + second := fetchGen(2) + if _, err := os.Stat(first); err != nil { + t.Fatalf("a generation used within the grace was pruned: %v", err) + } + old := time.Now().Add(-store.GenerationGrace - time.Minute) + if err := os.Chtimes(first, old, old); err != nil { + t.Fatal(err) + } + fetchGen(3) + if _, err := os.Stat(first); !errors.Is(err, os.ErrNotExist) { + t.Errorf("an idle superseded generation survived the next fetch: %v", err) + } + if _, err := os.Stat(second); err != nil { + t.Errorf("a generation used within the grace was pruned: %v", err) + } +} + +func TestFetchHitRefreshesTheGenerationsUse(t *testing.T) { + const id = "ck_00000000000000aa" + st := newTestStore(t, &fakeS3{objects: map[string][]byte{}}) + publishRecord(t, st, id, []byte(`{"id":"`+id+`","gen":1}`), "first") + export, _, _, err := st.Fetch(t.Context(), id) + if err != nil { + t.Fatalf("fetch: %v", err) + } + old := time.Now().Add(-2 * store.GenerationGrace) + if err = os.Chtimes(filepath.Dir(export), old, old); err != nil { + t.Fatal(err) + } + if _, _, _, err = st.Fetch(t.Context(), id); err != nil { + t.Fatalf("fetch hit: %v", err) + } + info, err := os.Stat(filepath.Dir(export)) + if err != nil || time.Since(info.ModTime()) > time.Minute { + t.Errorf("a cache hit left the generation looking idle: %v, %v", info, err) + } +} + func TestS3BackendContractRealEndpoint(t *testing.T) { endpoint := os.Getenv("SANDBOX_S3_E2E") if endpoint == "" { diff --git a/sandboxd/store/store.go b/sandboxd/store/store.go index 94a3b194..97791640 100644 --- a/sandboxd/store/store.go +++ b/sandboxd/store/store.go @@ -14,6 +14,7 @@ import ( "io/fs" "regexp" "strings" + "time" ) const ( @@ -23,6 +24,9 @@ const ( MetaFile = "meta.json" // LabelsFile holds a record's labels beside its meta, which names the export generation and so cannot change. LabelsFile = "labels.json" + + // GenerationGrace bounds the race where a node still clones a generation a re-publish superseded. + GenerationGrace = time.Hour ) var ( diff --git a/sandboxd/types/types.go b/sandboxd/types/types.go index c656236f..17a96b0d 100644 --- a/sandboxd/types/types.go +++ b/sandboxd/types/types.go @@ -497,7 +497,7 @@ type Sandbox struct { // Touch stamps last data-plane activity; lock-free, called on the relay hot path. func (s *Sandbox) Touch() { s.lastActivity.Store(time.Now().UnixNano()) } -// TouchAt stamps last-activity at a caller-supplied instant (batch claim/adoption, tests). +// TouchAt stamps the last data-plane activity at t. func (s *Sandbox) TouchAt(t time.Time) { s.lastActivity.Store(t.UnixNano()) } // LastSeen returns the last data-plane activity time. diff --git a/sdk/go/files_test.go b/sdk/go/files_test.go index 6f0dc24e..667b3451 100644 --- a/sdk/go/files_test.go +++ b/sdk/go/files_test.go @@ -1,8 +1,10 @@ package sandbox import ( + "bufio" "bytes" "errors" + "net" "testing" "github.com/cocoonstack/sandbox/protocol/wire" @@ -130,6 +132,30 @@ func TestReadFileMultiChunk(t *testing.T) { } } +func TestUploadToAnOldDaemonReturnsItsRejection(t *testing.T) { + sb := testSandbox(t, newAgentServer(t, func(conn net.Conn) { + defer func() { _ = conn.Close() }() + line, err := bufio.NewReader(conn).ReadBytes('\n') + if err != nil { + return + } + req, err := wire.DecodeRequest(line) + if err != nil { + return + } + var resp wire.Response = &wire.ErrorResp{Kind: wire.KindNotFound, Message: "create /missing/f: no such file or directory"} + if _, ok := req.(*wire.Info); ok { + resp = &wire.InfoResp{Proto: 1} + } + frame, _ := wire.EncodeResponse(resp) + _, _ = conn.Write(append(frame, '\n')) + })) + err := sb.WriteFile(t.Context(), "/missing/f", make([]byte, 64<<20), nil) + if e, ok := errors.AsType[*wire.ErrorResp](err); !ok || e.Kind != wire.KindNotFound { + t.Fatalf("upload rejected by a proto 1 daemon: %v, want its not_found frame", err) + } +} + func fakeSandbox(t *testing.T) *Sandbox { t.Helper() fake := silkdtest.NewFake(t.TempDir()) diff --git a/sdk/go/info.go b/sdk/go/info.go index 1bfdff56..eb659b37 100644 --- a/sdk/go/info.go +++ b/sdk/go/info.go @@ -78,7 +78,7 @@ type sandboxListResponse struct { Sandboxes []SandboxSummary `json:"sandboxes"` } -// Info reports the entry node's pools, claims, capacity state, and mesh peers. +// Info reports the entry node's pools, claims, capacity state, and mesh peers. Requires the operator token. func (c *Client) Info(ctx context.Context) (*NodeInfo, error) { return doJSONPtr[NodeInfo](ctx, c, http.MethodGet, c.addr, "/v1/info", nil, c.apiToken, "info") } diff --git a/sdk/go/pools.go b/sdk/go/pools.go index 3bec953d..8542a349 100644 --- a/sdk/go/pools.go +++ b/sdk/go/pools.go @@ -9,8 +9,7 @@ import ( "sync" ) -// PoolSpec is a desired warm pool for SetPools. Egress is config-owned on the -// node and rejected by the API, so it has no field here. +// PoolSpec is a desired warm pool for SetPools; egress, warmup, capture_trim and storage are config-owned, so it has no field for them. type PoolSpec struct { Template string `json:"template"` Net NetShape `json:"net,omitempty"` diff --git a/sdk/go/port.go b/sdk/go/port.go index 00375db5..b8bc2166 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -55,8 +55,7 @@ func (p *PortConn) CloseWrite() error { func (p *PortConn) Close() error { p.closeOnce.Do(func() { p.stop() - // drain can be parked in a pipe write on an unread tail; only the - // reader side unblocks it. + // A drain parked in a pipe write on an unread tail unblocks only from the reader side. _ = p.out.CloseWithError(net.ErrClosed) }) return nil @@ -67,8 +66,7 @@ func (p *PortConn) LocalAddr() net.Addr { return portAddr("sandbox-relay") } func (p *PortConn) RemoteAddr() net.Addr { return portAddr("sandbox-guest") } -// SetDeadline implements net.Conn; deadlines are unsupported — bound the -// DialPort ctx instead. +// SetDeadline implements net.Conn; deadlines are unsupported, so bound the DialPort ctx instead. func (p *PortConn) SetDeadline(time.Time) error { return errors.ErrUnsupported } // SetReadDeadline implements net.Conn; unsupported, see SetDeadline. @@ -77,9 +75,7 @@ func (p *PortConn) SetReadDeadline(time.Time) error { return errors.ErrUnsupport // SetWriteDeadline implements net.Conn; unsupported, see SetDeadline. func (p *PortConn) SetWriteDeadline(time.Time) error { return errors.ErrUnsupported } -// drain relays guest bytes into the pipe until Done (clean server close → -// EOF), an error frame, or teardown; a relay dropped before its terminal -// frame surfaces as drainData's truncation error, not a clean EOF. +// drain relays guest bytes into the pipe; a relay dropped before its terminal frame ends in drainData's truncation error, not EOF. func (p *PortConn) drain(ctx context.Context, pw *io.PipeWriter) { _ = pw.CloseWithError(drainData(ctx, p.conn, func(b []byte) error { _, err := pw.Write(b) diff --git a/sdk/go/upgrade.go b/sdk/go/upgrade.go index 976f77c8..ab80e506 100644 --- a/sdk/go/upgrade.go +++ b/sdk/go/upgrade.go @@ -10,6 +10,14 @@ import ( "strings" ) +type upgradedConn struct { + net.Conn + tcp net.Conn // under any TLS, for the parked-connection probe + r *bufio.Reader +} + +func (u *upgradedConn) Read(p []byte) (int, error) { return u.r.Read(p) } + func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (*upgradedConn, 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") @@ -62,14 +70,6 @@ func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (*upgrad return &upgradedConn{Conn: conn, tcp: raw, r: br}, nil } -type upgradedConn struct { - net.Conn - tcp net.Conn // under any TLS, for the parked-connection probe - 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 { diff --git a/sdk/go/utils.go b/sdk/go/utils.go index 8c11cb33..99a824e9 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -46,7 +46,7 @@ func (s *Sandbox) uploadRPC(ctx context.Context, req wire.Request, r io.Reader) n, readErr := r.Read(buf) if n > 0 { if err := conn.Send(&wire.Data{Data: buf[:n]}); err != nil { - return err + return cmp.Or(<-terminal, err) } } if errors.Is(readErr, io.EOF) { @@ -57,7 +57,7 @@ func (s *Sandbox) uploadRPC(ctx context.Context, req wire.Request, r io.Reader) } } if err := conn.Send(wire.DataEnd{}); err != nil { - return err + return cmp.Or(<-terminal, err) } return l.done(<-terminal) } @@ -160,8 +160,7 @@ func streamRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.R } } -// drainData consumes Data frames into sink until Done; an error frame or an -// unexpected frame is a Go error. +// drainData consumes Data frames into sink until Done; any other frame is an error. func drainData(ctx context.Context, conn *silkd.Conn, sink func([]byte) error) error { for { resp, err := recv(ctx, conn) @@ -188,8 +187,7 @@ func terminalErr(ctx context.Context, conn *silkd.Conn) error { return err } -// expect reads one frame and requires it to be a *T, mapping an error frame -// to a Go error. +// expect reads one frame as a *T; an error frame becomes a Go error. func expect[T any, PT respPtr[T]](ctx context.Context, conn *silkd.Conn) (*T, error) { resp, err := recv(ctx, conn) if err != nil { diff --git a/sdk/langchain/cocoonsandbox_langchain/toolkit.py b/sdk/langchain/cocoonsandbox_langchain/toolkit.py index e3ff3282..f164fe13 100644 --- a/sdk/langchain/cocoonsandbox_langchain/toolkit.py +++ b/sdk/langchain/cocoonsandbox_langchain/toolkit.py @@ -1,7 +1,4 @@ -"""LangChain toolkit over cocoon microVM sandboxes: one claimed sandbox -behind a set of StructuredTools (exec, file read/write, list). The claim is -lazy — building an agent costs nothing until a tool actually runs — and the -toolkit is a context manager whose exit releases the sandbox.""" +"""LangChain toolkit: one lazily claimed cocoon sandbox behind StructuredTools, released when the context exits.""" from __future__ import annotations @@ -11,7 +8,7 @@ import time from collections.abc import Callable -from cocoonsandbox import Client, Sandbox, SandboxError +from cocoonsandbox import APIError, Client, Sandbox, SandboxError from langchain_core.tools import StructuredTool, ToolException from pydantic import BaseModel, Field @@ -58,7 +55,11 @@ def __enter__(self) -> CocoonToolkit: return self def __exit__(self, *exc: object) -> None: - self.close() + try: + self.close() + except APIError: + if exc[0] is None: + raise def get_tools(self) -> list[StructuredTool]: """The sandbox tool set; sync-native (_run), async via to_thread.""" diff --git a/sdk/langchain/tests/test_toolkit.py b/sdk/langchain/tests/test_toolkit.py index 761137d0..04480b18 100644 --- a/sdk/langchain/tests/test_toolkit.py +++ b/sdk/langchain/tests/test_toolkit.py @@ -7,7 +7,7 @@ import time import pytest -from cocoonsandbox import Sandbox, SilkdError +from cocoonsandbox import APIError, Sandbox, SilkdError from cocoonsandbox_langchain import CocoonToolkit from cocoonsandbox_langchain.toolkit import CALL_TIMEOUT @@ -116,6 +116,18 @@ def test_use_after_close_raises(monkeypatch): kit.sandbox() +def test_a_failed_release_never_hides_the_block_error(monkeypatch): + kit, fake = hooked(monkeypatch) + fake.close_error = APIError("release", 500, "boom") + with pytest.raises(ValueError), kit: + kit.get_tools()[0].invoke({"command": "x"}) + raise ValueError("agent failed") + kit, fake = hooked(monkeypatch) + fake.close_error = APIError("release", 500, "boom") + with pytest.raises(APIError), kit: + kit.get_tools()[0].invoke({"command": "x"}) + + def test_read_file_reports_a_missing_path_as_a_tool_error(monkeypatch): kit, fake = hooked(monkeypatch) @@ -144,6 +156,7 @@ def test_a_stalled_agent_upgrade_comes_back_as_a_tool_error(): class FakeSandbox: def __init__(self): self.closed = 0 + self.close_error = None self.files = {} def run(self, argv, cwd="", on_stdout=None, on_stderr=None, timeout=None, **_): @@ -169,6 +182,8 @@ def list_dir(self, path): def close(self): self.closed += 1 + if self.close_error is not None: + raise self.close_error def hooked(monkeypatch): diff --git a/sdk/openai/cocoonsandbox_openai/adapter.py b/sdk/openai/cocoonsandbox_openai/adapter.py index b5efff73..7f91a641 100644 --- a/sdk/openai/cocoonsandbox_openai/adapter.py +++ b/sdk/openai/cocoonsandbox_openai/adapter.py @@ -20,6 +20,7 @@ from agents.sandbox.session.sandbox_session_state import SandboxSessionState from agents.sandbox.snapshot import SnapshotBase, SnapshotSpec, resolve_snapshot from agents.sandbox.types import ExecResult, ExposedPortEndpoint, User +from agents.sandbox.util.tar_utils import strip_tar_member_prefix from cocoonsandbox import Client, Sandbox, SandboxError, SilkdError @@ -85,8 +86,9 @@ async def running(self) -> bool: async def persist_workspace(self) -> io.IOBase: sb = self._sandbox() - tar = await asyncio.to_thread(sb.pull, str(self.state.manifest.root)) - return io.BytesIO(tar) + root = Path(self.state.manifest.root) + tar = await asyncio.to_thread(sb.pull, str(root)) + return await asyncio.to_thread(strip_tar_member_prefix, io.BytesIO(tar), prefix=root.name) async def hydrate_workspace(self, data: io.IOBase) -> None: sb = self._sandbox() diff --git a/sdk/openai/tests/test_adapter.py b/sdk/openai/tests/test_adapter.py index 87ff58a4..6c4cb9c8 100644 --- a/sdk/openai/tests/test_adapter.py +++ b/sdk/openai/tests/test_adapter.py @@ -6,6 +6,7 @@ import asyncio import io import json +import tarfile import threading from http.server import BaseHTTPRequestHandler, HTTPServer from pathlib import Path @@ -178,7 +179,7 @@ def write_file(self, path, data): def pull(self, path): calls["pull"] = path - return b"tar-bytes" + return _silkd_pull_tar("workspace", {"a.txt": b"body"}) def push(self, dest, data): calls["push"] = (dest, data) @@ -191,7 +192,9 @@ async def go(): await inner.write(Path("/workspace/a.txt"), io.BytesIO(b"body")) assert calls["write"] == ("/workspace/a.txt", b"body") tar = await inner.persist_workspace() - assert tar.read() == b"tar-bytes" and calls["pull"] == "/workspace" + with tarfile.open(fileobj=tar) as archive: + names = sorted(archive.getnames()) + assert calls["pull"] == "/workspace" and names == [".", "a.txt"], names await inner.hydrate_workspace(io.BytesIO(b"tar-in")) assert calls["push"] == ("/workspace", b"tar-in") @@ -232,3 +235,16 @@ async def go(): asyncio.run(go()) assert released == ["sb_1"] + + +def _silkd_pull_tar(root: str, files: dict[str, bytes]) -> bytes: + out = io.BytesIO() + with tarfile.open(fileobj=out, mode="w") as archive: + top = tarfile.TarInfo(root) + top.type = tarfile.DIRTYPE + archive.addfile(top) + for name, body in files.items(): + info = tarfile.TarInfo(f"{root}/{name}") + info.size = len(body) + archive.addfile(info, io.BytesIO(body)) + return out.getvalue() diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index e50bb6e8..23de7017 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -5,12 +5,11 @@ import http.client import json import queue +import select import ssl import threading import time -import urllib.error import urllib.parse -import urllib.request from collections.abc import Callable, Iterable, Mapping, Sequence from typing import Any, TypeVar, cast @@ -23,6 +22,7 @@ T = TypeVar("T") _PEERS_TIMEOUT = 5.0 +_IDLE_CONNS = 8 class Client: @@ -47,7 +47,8 @@ def __init__( 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._idle: dict[tuple[str, str], list[tuple[http.client.HTTPConnection, float]]] = {} + self._idle_lock = threading.Lock() self.api_token = api_token self.timeout = timeout self.keep_alive = keep_alive @@ -200,27 +201,19 @@ def _request( ) -> dict[str, Any]: timeout = remaining_timeout(timeout or self.timeout, deadline, verb) data = json.dumps(body).encode() if body is not None else None - 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") + url = urllib.parse.urlsplit(addr + path if "://" in addr else f"{self._scheme}://{addr}{path}") + headers = {"Content-Type": "application/json"} if data is not None else {} token = bearer or self.api_token if token: - req.add_header("Authorization", f"Bearer {token}") + headers["Authorization"] = f"Bearer {token}" try: - with self._open(req, timeout) as resp: - raw = resp.read() - except urllib.error.HTTPError as exc: - try: - detail = _error_message(exc.read()) - except (OSError, http.client.HTTPException) as read_exc: - detail = str(read_exc) - raise APIError(verb, exc.code, detail) from None - except (urllib.error.URLError, OSError, http.client.HTTPException) as exc: + status, raw = self._roundtrip(url, method, data, headers, timeout) + except (OSError, http.client.HTTPException) as exc: if deadline is not None and time.monotonic() >= deadline: raise SandboxTimeout(f"{verb} timed out") from None - detail = str(exc.reason) if isinstance(exc, urllib.error.URLError) else str(exc) - raise APIError(verb, 0, detail) from None + raise APIError(verb, 0, str(exc)) from None + if status >= 400: + raise APIError(verb, status, _error_message(raw)) if not raw: return {} try: @@ -233,12 +226,47 @@ def _dial(self, addr: str, sandbox_id: str, token: str, deadline: float | None = 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 _roundtrip( + self, url: urllib.parse.SplitResult, method: str, data: bytes | None, headers: dict[str, str], timeout: float + ) -> tuple[int, bytes]: + key = (url.scheme, url.netloc) + conn = self._take(key, timeout) + status = 0 + try: + conn.request(method, url.path + (f"?{url.query}" if url.query else ""), body=data, headers=headers) + resp = conn.getresponse() + status = resp.status + raw = resp.read() + except (OSError, http.client.HTTPException) as exc: + conn.close() + if status < 400: + raise + return status, str(exc).encode() + except BaseException: + conn.close() + raise + with self._idle_lock: + idle = self._idle.setdefault(key, []) + if resp.will_close or not self.keep_alive or len(idle) >= _IDLE_CONNS: + conn.close() + else: + idle.append((conn, time.monotonic())) + return resp.status, raw + + def _take(self, key: tuple[str, str], timeout: float) -> http.client.HTTPConnection: + """Reuses a parked connection the node has not closed within keep_alive, else dials.""" + with self._idle_lock: + idle = self._idle.get(key, []) + while idle: + conn, parked = idle.pop() + if time.monotonic() - parked < self.keep_alive and _quiet(conn.sock): + conn.sock.settimeout(timeout) + return conn + conn.close() + scheme, netloc = key + if scheme == "https": + return http.client.HTTPSConnection(netloc, timeout=timeout, context=self._tls()) + return http.client.HTTPConnection(netloc, timeout=timeout) def _tls(self) -> ssl.SSLContext: if self._ssl_context is None: @@ -247,6 +275,16 @@ def _tls(self) -> ssl.SSLContext: return self._ssl_context +def _quiet(sock: Any) -> bool: + if sock is None: + return False + try: + readable, _, _ = select.select([sock], [], [], 0) + except (OSError, ValueError): + return False + return not readable + + def _claim_body( template: str, net: str, diff --git a/sdk/python/cocoonsandbox/errors.py b/sdk/python/cocoonsandbox/errors.py index 41f58f4a..e81f1473 100644 --- a/sdk/python/cocoonsandbox/errors.py +++ b/sdk/python/cocoonsandbox/errors.py @@ -14,7 +14,7 @@ class SandboxTimeout(SandboxError, TimeoutError): class APIError(SandboxError): """A sandboxd control-plane call failed.""" - def __init__(self, verb: str, status: int, message: str): + def __init__(self, verb: str, status: int, message: str) -> None: super().__init__(f"{verb}: {message} (HTTP {status})") self.verb = verb self.status = status @@ -22,13 +22,9 @@ def __init__(self, verb: str, status: int, message: str): class SilkdError(SandboxError): - """The in-guest daemon answered an error frame. + """The in-guest daemon answered an error frame; kind is its typed wire kind.""" - kind is the typed wire kind: bad_request, not_found, unimplemented, - internal. - """ - - def __init__(self, kind: str, message: str): + def __init__(self, kind: str, message: str) -> None: super().__init__(f"{kind}: {message}") self.kind = kind self.message = message @@ -38,7 +34,7 @@ class ExitError(SandboxError): """A command exited non-zero; carries the exit code, stderr, and whatever stdout it had produced — a failing build's log is on stdout.""" - def __init__(self, code: int, stderr: str, stdout: str = ""): + def __init__(self, code: int, stderr: str, stdout: str = "") -> None: super().__init__(f"exit status {code}: {stderr.strip()}") self.code = code self.stderr = stderr diff --git a/sdk/python/cocoonsandbox/frames.py b/sdk/python/cocoonsandbox/frames.py index 137bec13..f9e15af9 100644 --- a/sdk/python/cocoonsandbox/frames.py +++ b/sdk/python/cocoonsandbox/frames.py @@ -13,7 +13,7 @@ KEEP_ALIVE_PROTO = 2 MAX_FRAME = 8 * 1024 * 1024 FS_CHUNK = 256 * 1024 -# tar and port streams chunk at 1 MiB: fewer frames per byte, still under MAX_FRAME after base64 +# tar and port streams chunk at 1 MiB, still under MAX_FRAME after base64 BULK_CHUNK = 1 << 20 _FAST_DATA = sys.version_info >= (3, 11) diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 6f1f1716..d25107e7 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -184,9 +184,7 @@ def write_file(self, path: str, data: bytes, mode: int | None = None) -> None: """Writes data to path atomically (temp + rename on the guest).""" with self._lease() as conn: conn.send("fs_write", path=path, mode=mode) - _send_chunks(conn, data) - conn.send("data_end") - _expect(conn, "done") + _upload(conn, data, FS_CHUNK) def read_file(self, path: str) -> bytes: with self._lease() as conn: @@ -218,9 +216,7 @@ def push(self, dest: str, tar_stream: bytes) -> None: """Extracts a tar stream into dest; a truncated stream leaves dest untouched.""" with self._lease() as conn: conn.send("fs_push", dest=dest) - _send_chunks(conn, tar_stream, chunk=BULK_CHUNK) - conn.send("data_end") - _expect(conn, "done") + _upload(conn, tar_stream, BULK_CHUNK) def pull(self, path: str) -> bytes: """Returns path (file or tree) as a tar archive.""" @@ -356,7 +352,9 @@ def open_pty( def proxy_port(self, local_addr: str, port: int) -> socket.socket: """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))) + host = host.strip("[]") or "127.0.0.1" + family = socket.AF_INET6 if ":" in host else socket.AF_INET + listener = socket.create_server((host, int(lport)), family=family) threading.Thread(target=self._proxy_accept_loop, args=(listener, port), daemon=True).start() return listener @@ -646,6 +644,17 @@ def _send_chunks(conn: Conn, data: bytes, op: str = "data", chunk: int = FS_CHUN conn.send(op, data=view[off : off + chunk]) +def _upload(conn: Conn, data: bytes, chunk: int) -> None: + try: + _send_chunks(conn, data, chunk=chunk) + conn.send("data_end") + except ProtocolError: + with contextlib.suppress(ProtocolError): + _expect(conn, "done") + raise + _expect(conn, "done") + + def _arm_watchdog(conn: Conn, deadline: float, expired: threading.Event) -> threading.Timer: def cut() -> None: expired.set() diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 8dbfc492..920c302b 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -2,7 +2,7 @@ import threading import time -from http.server import HTTPServer +from http.server import HTTPServer, ThreadingHTTPServer import pytest from conftest import FakeNode, sandbox_at @@ -340,3 +340,45 @@ def claim(body, path): FakeNode.routes[("POST", "/v1/claim")] = claim return seen + + +@pytest.mark.parametrize(("keep_alive", "conns"), [(30.0, 1), (0, 2)]) +def test_control_calls_reuse_one_connection(kept_node, keep_alive, conns): + KeptNode.routes[("POST", "/v1/claim")] = lambda body, path: (200, {"id": "sb_1", "token": "tok"}) + client = Client(kept_node, keep_alive=keep_alive) + client.new("rt:24.04") + client.new("rt:24.04") + assert KeptNode.conns == conns + + +def test_a_parked_connection_the_node_closed_is_not_reused(kept_node): + KeptNode.routes[("POST", "/v1/claim")] = lambda body, path: (200, {"id": "sb_1", "token": "tok"}) + KeptNode.drop_after_reply = True + client = Client(kept_node) + client.new("rt:24.04") + time.sleep(0.1) + assert client.new("rt:24.04").id == "sb_1" + assert KeptNode.conns == 2 + + +@pytest.fixture +def kept_node(): + server = ThreadingHTTPServer(("127.0.0.1", 0), KeptNode) + KeptNode.routes, KeptNode.conns, KeptNode.drop_after_reply = {}, 0, False + threading.Thread(target=server.serve_forever, daemon=True).start() + yield f"127.0.0.1:{server.server_port}" + server.shutdown() + + +class KeptNode(FakeNode): + protocol_version = "HTTP/1.1" + conns = 0 + drop_after_reply = False + + def setup(self): + super().setup() + KeptNode.conns += 1 + + def _reply(self, code, payload): + super()._reply(code, payload) + self.close_connection = self.drop_after_reply diff --git a/sdk/python/tests/test_keepalive.py b/sdk/python/tests/test_keepalive.py index 37e53e6f..22b5ce90 100644 --- a/sdk/python/tests/test_keepalive.py +++ b/sdk/python/tests/test_keepalive.py @@ -120,12 +120,23 @@ def test_close_drains_the_parked_connection(monkeypatch): agent.stop() +def test_an_upload_an_old_daemon_rejects_raises_its_error_frame(): + agent = FakeAgent(proto=1, slam=True) + sb = sandbox_at(agent.addr) + try: + with pytest.raises(SilkdError): + sb.write_file("/missing/f", bytes(64 << 20)) + finally: + agent.stop() + + class FakeAgent: - """A relay-side silkd stand-in; proto 1 half-closes after one RPC and drains like the relay does.""" + """A relay-side silkd stand-in; proto 1 half-closes after one RPC and drains; slam closes as a failed relay does.""" - def __init__(self, proto: int = KEEP_ALIVE_PROTO, hang_up_after: int = 0) -> None: + def __init__(self, proto: int = KEEP_ALIVE_PROTO, hang_up_after: int = 0, slam: bool = False) -> None: self.proto = proto self.hang_up_after = hang_up_after + self.slam = slam self.upgrades = 0 self.hangups = 0 self.closed = 0 @@ -157,6 +168,8 @@ def _serve(self, conn: socket.socket) -> None: for frame in self._answer(op): conn.sendall(json.dumps(frame).encode() + b"\n") replies += 1 + if self.slam: + break if self.proto < KEEP_ALIVE_PROTO or replies == self.hang_up_after: conn.shutdown(socket.SHUT_WR) self.hangups += 1 diff --git a/sdk/python/tests/test_proxy.py b/sdk/python/tests/test_proxy.py index 03ee24cf..6d52a211 100644 --- a/sdk/python/tests/test_proxy.py +++ b/sdk/python/tests/test_proxy.py @@ -6,6 +6,7 @@ import threading import time +import pytest from conftest import sandbox_at @@ -54,13 +55,14 @@ def close(self) -> None: pass -def test_proxy_port_accepts_and_pipes(monkeypatch): +@pytest.mark.parametrize("local", ["127.0.0.1:0", "[::1]:0"]) +def test_proxy_port_accepts_and_pipes(monkeypatch, local): sb = sandbox_at("127.0.0.1:1") monkeypatch.setattr(sb, "dial_port", lambda port: FakePortConn()) - listener = sb.proxy_port("127.0.0.1:0", 8080) + listener = sb.proxy_port(local, 8080) try: - c = socket.create_connection(listener.getsockname(), timeout=5) + c = socket.create_connection(listener.getsockname()[:2], timeout=5) c.sendall(b"hello") c.settimeout(5) got, deadline = b"", time.monotonic() + 5 diff --git a/silkd/Cargo.toml b/silkd/Cargo.toml index c73355bf..63714e1f 100644 --- a/silkd/Cargo.toml +++ b/silkd/Cargo.toml @@ -10,7 +10,7 @@ publish = false base64 = "0.23" libc = "0.2" memchr = "2" -nix = { version = "0.31", features = ["term", "user"] } +nix = { version = "0.31", features = ["dir", "fs", "term", "user"] } notify = "8" regex = "1" serde = { version = "1", features = ["derive", "rc"] } diff --git a/silkd/src/fs.rs b/silkd/src/fs.rs index 90455ea0..fe993a74 100644 --- a/silkd/src/fs.rs +++ b/silkd/src/fs.rs @@ -26,9 +26,8 @@ where W: AsyncWrite + Unpin, { let path = Path::new(&path); - let tmp = tmp_name(path); - let mut file = match fs::File::create(&tmp).await { - Ok(f) => f, + let (tmp, mut file) = match create_tmp(path, mode).await { + Ok(created) => created, Err(e) => return err_frame(w, &e, "create").await, }; @@ -42,7 +41,7 @@ where let _ = fs::remove_file(&tmp).await; return proto::write_feed_error(w, fail).await; } - if let Err(e) = commit_tmp(&tmp, path, mode).await { + if let Err(e) = commit_tmp(&tmp, path).await { return err_frame(w, &e, "commit").await; } proto::write_frame(w, &Response::Done).await @@ -100,12 +99,17 @@ pub async fn list(w: &mut W, path: String) -> io::Result< /// Writes `bytes` to `path` via a sibling temp file, so a crash never leaves a truncated file. pub async fn write_atomic(path: &Path, bytes: &[u8]) -> io::Result<()> { - let tmp = tmp_name(path); - if let Err(e) = fs::write(&tmp, bytes).await { + let (tmp, mut file) = create_tmp(path, None).await?; + let written = match file.write_all(bytes).await { + Ok(()) => file.flush().await, + Err(e) => Err(e), + }; + drop(file); + if let Err(e) = written { let _ = fs::remove_file(&tmp).await; return Err(e); } - commit_tmp(&tmp, path, None).await + commit_tmp(&tmp, path).await } /// Reports metadata for `path` (following symlinks). @@ -208,22 +212,35 @@ fn tmp_name(path: &Path) -> PathBuf { PathBuf::from(tmp) } -/// Commits a temp file over `path`; without an explicit mode an overwrite inherits the destination's bits. -async fn commit_tmp(tmp: &Path, path: &Path, mode: Option) -> io::Result<()> { - let outcome = async { - let effective = match mode { - Some(m) => Some(m), - None => fs::metadata(path) - .await - .ok() - .map(|meta| meta.permissions().mode() & 0o7777), - }; - if let Some(m) = effective { - fs::set_permissions(tmp, std::fs::Permissions::from_mode(m)).await?; - } - fs::rename(tmp, path).await +/// Creates the temp file exclusively under its final mode; without an explicit mode an overwrite inherits the destination's bits. +async fn create_tmp(path: &Path, mode: Option) -> io::Result<(PathBuf, fs::File)> { + let effective = match mode { + Some(m) => Some(m), + None => fs::metadata(path) + .await + .ok() + .map(|meta| meta.permissions().mode() & 0o7777), + }; + let tmp = tmp_name(path); + let file = fs::OpenOptions::new() + .write(true) + .create_new(true) + .mode(effective.unwrap_or(0o666)) + .open(&tmp) + .await?; + if let Some(m) = effective + && let Err(e) = file + .set_permissions(std::fs::Permissions::from_mode(m)) + .await + { + let _ = fs::remove_file(&tmp).await; + return Err(e); } - .await; + Ok((tmp, file)) +} + +async fn commit_tmp(tmp: &Path, path: &Path) -> io::Result<()> { + let outcome = fs::rename(tmp, path).await; if outcome.is_err() { let _ = fs::remove_file(tmp).await; } @@ -242,3 +259,39 @@ fn file_kind(meta: &std::fs::Metadata) -> FileKind { FileKind::Other } } + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn write_atomic_refuses_a_planted_temp_name() { + let dir = tempfile::tempdir().unwrap(); + let victim = dir.path().join("victim"); + std::fs::write(&victim, "original").unwrap(); + let target = dir.path().join("x"); + let next: u64 = crate::sysutil::tmp_suffix() + .rsplit('-') + .next() + .unwrap() + .parse() + .unwrap(); + for seq in next + 1..next + 1000 { + let name = format!("x.silkd-{}-{seq}.tmp", std::process::id()); + std::os::unix::fs::symlink(&victim, dir.path().join(name)).unwrap(); + } + let err = write_atomic(&target, b"planted").await.unwrap_err(); + assert_eq!(err.kind(), io::ErrorKind::AlreadyExists); + assert_eq!(std::fs::read_to_string(&victim).unwrap(), "original"); + } + + #[tokio::test] + async fn a_temp_file_is_born_with_its_mode() { + let dir = tempfile::tempdir().unwrap(); + let (tmp, _file) = create_tmp(&dir.path().join("secret"), Some(0o600)) + .await + .unwrap(); + let mode = std::fs::metadata(&tmp).unwrap().permissions().mode() & 0o7777; + assert_eq!(mode, 0o600); + } +} diff --git a/silkd/src/git.rs b/silkd/src/git.rs index 1a3dada6..cc4c43ea 100644 --- a/silkd/src/git.rs +++ b/silkd/src/git.rs @@ -146,6 +146,7 @@ async fn list_branches(w: &mut W, path: &str) -> std::io: fn git_cmd(dir: &str, auth: Option<&str>) -> Command { let mut cmd = Command::new("git"); sysutil::align_proxy_env(&mut cmd); + sysutil::apply_owner(&mut cmd, dir); cmd.arg("-C").arg(dir); apply_config(&mut cmd, auth); cmd.stdin(Stdio::null()) @@ -155,7 +156,6 @@ fn git_cmd(dir: &str, auth: Option<&str>) -> Command { cmd } -/// Runs `git`, capturing output. async fn git( dir: &str, auth: Option<&str>, diff --git a/silkd/src/proc.rs b/silkd/src/proc.rs index 2616c9f3..30be0708 100644 --- a/silkd/src/proc.rs +++ b/silkd/src/proc.rs @@ -55,10 +55,6 @@ impl Table { proc } - pub(crate) fn get(&self, pid: u32) -> Option> { - sysutil::lock(&self.inner).get(&pid).cloned() - } - /// Looks up a pid, writing a NotFound frame and returning None on a miss. pub async fn get_or_not_found( &self, @@ -89,6 +85,10 @@ impl Table { } } + pub(crate) fn get(&self, pid: u32) -> Option> { + sysutil::lock(&self.inner).get(&pid).cloned() + } + pub(crate) fn len(&self) -> usize { sysutil::lock(&self.inner).len() } @@ -249,7 +249,7 @@ impl Ring { } } -/// Stdout/Stderr ride the reused-buffer bulk path; serde's per-chunk allocations dominate replay otherwise. +/// Writes a Stdout/Stderr chunk on the reused-buffer bulk path. pub async fn write_chunk( w: &mut W, buf: &mut Vec, diff --git a/silkd/src/proto.rs b/silkd/src/proto.rs index 4b180b47..6cab7d6f 100644 --- a/silkd/src/proto.rs +++ b/silkd/src/proto.rs @@ -23,7 +23,7 @@ pub const PROTO_VERSION: u32 = 2; /// Chunk size for streaming a file back over `fs.read`. pub const READ_CHUNK: usize = 32 * 1024; -/// Bulk streams chunk larger: fewer frames and flushes per byte, still under MAX_FRAME after base64. +/// Chunk size for bulk streams; stays under MAX_FRAME after base64. pub const BULK_CHUNK: usize = 256 * 1024; const DATA_FRAME_HEAD: &[u8] = b"{\"v\":1,\"op\":\"data\",\"data\":\""; @@ -567,7 +567,7 @@ pub async fn write_feed_error(w: &mut W, err: FeedError) } } -fn decode_data_frame(frame: &[u8], out: &mut Vec) -> Option { +pub(crate) fn decode_data_frame(frame: &[u8], out: &mut Vec) -> Option { let b64 = frame.strip_prefix(DATA_FRAME_HEAD)?.strip_suffix(b"\"}")?; out.resize(base64::decoded_len_estimate(b64.len()), 0); STANDARD.decode_slice(b64, out).ok() diff --git a/silkd/src/server.rs b/silkd/src/server.rs index eb6787b1..c8b55f5d 100644 --- a/silkd/src/server.rs +++ b/silkd/src/server.rs @@ -340,7 +340,13 @@ where { let mut line = Vec::new(); while let Ok(true) = proto::read_frame_into(&mut reader, &mut line).await { - let Ok(req) = serde_json::from_slice::(&line) else { + let mut data = Vec::new(); + let req = if let Some(n) = proto::decode_data_frame(&line, &mut data) { + data.truncate(n); + Request::Data { data } + } else if let Ok(req) = serde_json::from_slice::(&line) { + req + } else { break; }; if !req.is_continuation() { diff --git a/silkd/src/session.rs b/silkd/src/session.rs index 301c2039..ccf2c447 100644 --- a/silkd/src/session.rs +++ b/silkd/src/session.rs @@ -127,10 +127,6 @@ impl Table { removed.is_some() } - pub(crate) fn len(&self) -> usize { - sysutil::lock(&self.inner).len() - } - /// Removes and kills sessions idle longer than `ttl`; a session holding the io lock is never idle. pub fn reap_idle(&self, ttl: Duration) -> usize { let mut map = sysutil::lock(&self.inner); @@ -144,6 +140,10 @@ impl Table { }); before - map.len() } + + pub(crate) fn len(&self) -> usize { + sysutil::lock(&self.inner).len() + } } /// One persistent shell; `io` serializes commands so a session runs one at a time. diff --git a/silkd/src/sysutil.rs b/silkd/src/sysutil.rs index 5fc3ca17..ffd902a1 100644 --- a/silkd/src/sysutil.rs +++ b/silkd/src/sysutil.rs @@ -2,6 +2,7 @@ use std::io::Read; use std::os::fd::{AsRawFd, OwnedFd, RawFd}; +use std::os::unix::fs::MetadataExt; use std::os::unix::process::ExitStatusExt; use std::path::PathBuf; use std::process::ExitStatus; @@ -98,6 +99,30 @@ pub fn apply_user(cmd: &mut Command, user: &str) -> Result<(), String> { Ok(()) } +pub fn apply_owner(cmd: &mut Command, path: &str) { + let Ok(meta) = std::fs::metadata(path) else { + return; + }; + if meta.uid() == 0 || !nix::unistd::geteuid().is_root() { + return; + } + cmd.uid(meta.uid()).gid(meta.gid()); + if let Ok(Some(pw)) = nix::unistd::User::from_uid(nix::unistd::Uid::from_raw(meta.uid())) { + cmd.env("HOME", pw.dir).env("USER", pw.name); + } +} + +pub fn start_in(cmd: &mut Command, dir: &OwnedFd) { + let fd = dir.as_raw_fd(); + // SAFETY: fchdir is async-signal-safe and `dir` outlives the spawn that runs this hook. + unsafe { + cmd.pre_exec(move || { + nix::unistd::fchdir(std::os::fd::BorrowedFd::borrow_raw(fd)) + .map_err(std::io::Error::from) + }) + }; +} + /// Opens a pty, returning the (master, slave) fds; the master is non-blocking for async I/O. pub fn openpty(cols: u16, rows: u16) -> std::io::Result<(OwnedFd, OwnedFd)> { let ws = nix::pty::Winsize { diff --git a/silkd/src/tree.rs b/silkd/src/tree.rs index cde49f2b..4b89cec4 100644 --- a/silkd/src/tree.rs +++ b/silkd/src/tree.rs @@ -3,9 +3,14 @@ //! network-failure atomic: it stages inside `dest`, then merges by rename. use std::io; +use std::os::fd::OwnedFd; use std::path::Path; use std::process::Stdio; +use nix::dir::Dir; +use nix::fcntl::{self, AtFlags, OFlag}; +use nix::sys::stat::{self, Mode, SFlag}; +use nix::unistd::{self, UnlinkatFlags}; use tokio::io::{AsyncBufRead, AsyncReadExt, AsyncWrite}; use tokio::process::Command; @@ -13,6 +18,9 @@ use crate::proto::{self, ErrorKind, Response, err_frame}; use crate::sysutil; const STDERR_CAP: usize = 16 * 1024; +const DIR_FLAGS: OFlag = OFlag::O_DIRECTORY + .union(OFlag::O_NOFOLLOW) + .union(OFlag::O_CLOEXEC); /// Extracts a client tar stream into `dest`, creating it; a stream failure leaves `dest` unchanged. pub async fn push(mut reader: R, w: &mut W, dest: String) -> io::Result<()> @@ -23,13 +31,14 @@ where if let Err(e) = tokio::fs::create_dir_all(&dest).await { return err_frame(w, &e, "mkdir dest").await; } - let staging = match stage_dir(&dest).await { - Ok(s) => s, + let name = format!(".silkd-push-{}", sysutil::tmp_suffix()); + let (dest_fd, staging) = match stage_dir(&dest, &name) { + Ok(fds) => fds, Err(e) => return err_frame(w, &e, "stage push").await, }; - let res = push_staged(&mut reader, w, &staging, dest).await; + let res = push_staged(&mut reader, w, staging, dest_fd).await; // a failed push can leave a large partial tree, so cleanup runs on every exit path. - let _ = tokio::fs::remove_dir_all(&staging).await; + let _ = tokio::fs::remove_dir_all(Path::new(&dest).join(&name)).await; res } @@ -75,17 +84,19 @@ pub async fn pull(w: &mut W, path: String) -> io::Result< } /// The caller owns removing `staging` on every path. -async fn push_staged(reader: &mut R, w: &mut W, staging: &str, dest: String) -> io::Result<()> +async fn push_staged( + reader: &mut R, + w: &mut W, + staging: OwnedFd, + dest: OwnedFd, +) -> io::Result<()> where R: AsyncBufRead + Unpin, W: AsyncWrite + Unpin, { let mut cmd = Command::new("tar"); - cmd.arg("-x") - .arg("-C") - .arg(staging) - .stdin(Stdio::piped()) - .stdout(Stdio::null()); + cmd.arg("-x").stdin(Stdio::piped()).stdout(Stdio::null()); + sysutil::start_in(&mut cmd, &staging); let (mut child, err_task) = match spawn_tar(cmd) { Ok(pair) => pair, Err(e) => return err_frame(w, &e, "spawn tar").await, @@ -99,15 +110,16 @@ where let status = child.wait().await?; let msg = err_task.await.unwrap_or_default(); - if let Err(fail) = feed { + if let Err(fail) = feed + && (status.success() || !matches!(fail, proto::FeedError::Io(_, "write"))) + { return proto::write_feed_error(w, fail).await; } if !status.success() { return proto::subprocess_result(w, status.success(), "tar extract", msg.as_bytes()).await; } - let src = staging.to_owned(); // spawn_blocking: a big tree is thousands of local fs syscalls. - let merged = tokio::task::spawn_blocking(move || merge_tree(Path::new(&src), Path::new(&dest))) + let merged = tokio::task::spawn_blocking(move || merge_tree(&staging, &dest)) .await .map_err(io::Error::other) .and_then(|r| r); @@ -129,33 +141,51 @@ fn spawn_tar( Ok((child, tokio::spawn(drain(stderr)))) } -/// Creates the staging dir inside `dest` so the merge is same-filesystem renames. -async fn stage_dir(dest: &str) -> io::Result { - let staging = format!("{dest}/.silkd-push-{}", sysutil::tmp_suffix()); - tokio::fs::create_dir(&staging).await?; - Ok(staging) +/// Creates the staging dir inside `dest` (so the merge is same-filesystem renames) and holds both by fd, so renaming either path cannot redirect the push. +fn stage_dir(dest: &str, name: &str) -> io::Result<(OwnedFd, OwnedFd)> { + let dest = fcntl::open(dest, OFlag::O_DIRECTORY | OFlag::O_CLOEXEC, Mode::empty())?; + stat::mkdirat(&dest, name, Mode::S_IRWXU)?; + let staging = fcntl::openat(&dest, name, DIR_FLAGS, Mode::empty())?; + if stat::fstat(&staging)?.st_uid != unistd::geteuid().as_raw() { + return Err(io::Error::other("push staging dir was replaced")); + } + Ok((dest, staging)) } -/// Overlays `src` into `dst` with tar's semantics: dirs merge, a file over a dir fails. -fn merge_tree(src: &Path, dst: &Path) -> io::Result<()> { - for entry in std::fs::read_dir(src)? { - let entry = entry?; - let from = entry.path(); - let to = dst.join(entry.file_name()); - let to_meta = std::fs::symlink_metadata(&to).ok(); - match (entry.file_type()?.is_dir(), to_meta) { - (true, Some(m)) if m.is_dir() => merge_tree(&from, &to)?, +/// Overlays `src` into `dst` with tar's semantics through fds that never follow a link; `src` is listed through a fresh open, as overlayfs lists stale through a fd opened before the extract. +fn merge_tree(src: &OwnedFd, dst: &OwnedFd) -> io::Result<()> { + let mut dir = Dir::openat(src, ".", DIR_FLAGS, Mode::empty())?; + let mut names = Vec::new(); + for entry in dir.iter() { + let name = entry?.file_name().to_owned(); + if name.as_bytes() != b"." && name.as_bytes() != b".." { + names.push(name); + } + } + for name in &names { + let from_dir = + is_dir(stat::fstatat(&dir, name.as_c_str(), AtFlags::AT_SYMLINK_NOFOLLOW)?.st_mode); + match stat::fstatat(dst, name.as_c_str(), AtFlags::AT_SYMLINK_NOFOLLOW) { + Ok(to) if from_dir && is_dir(to.st_mode) => { + let from = fcntl::openat(&dir, name.as_c_str(), DIR_FLAGS, Mode::empty())?; + let to = fcntl::openat(dst, name.as_c_str(), DIR_FLAGS, Mode::empty())?; + merge_tree(&from, &to)?; + } // mirrors tar's unlink-before-extract. - (true, Some(_)) => { - std::fs::remove_file(&to)?; - std::fs::rename(&from, &to)?; + Ok(_) if from_dir => { + unistd::unlinkat(dst, name.as_c_str(), UnlinkatFlags::NoRemoveDir)?; + fcntl::renameat(&dir, name.as_c_str(), dst, name.as_c_str())?; } - _ => std::fs::rename(&from, &to)?, + _ => fcntl::renameat(&dir, name.as_c_str(), dst, name.as_c_str())?, } } Ok(()) } +fn is_dir(mode: stat::mode_t) -> bool { + SFlag::from_bits_truncate(mode) & SFlag::S_IFMT == SFlag::S_IFDIR +} + /// Reads a child's stderr up to STDERR_CAP; tar's first error is the useful one. async fn drain(mut stderr: tokio::process::ChildStderr) -> String { let mut out = Vec::new(); @@ -170,3 +200,60 @@ async fn drain(mut stderr: tokio::process::ChildStderr) -> String { } String::from_utf8(out).unwrap_or_else(|e| String::from_utf8_lossy(e.as_bytes()).into_owned()) } + +#[cfg(test)] +mod tests { + use base64::Engine; + use serde_json::json; + + use super::*; + + #[tokio::test] + async fn push_holds_its_dirs_by_fd_when_the_staging_path_is_swapped() { + let root = tempfile::tempdir().unwrap(); + let (dest, victim, src) = ( + root.path().join("dest"), + root.path().join("victim"), + root.path().join("src"), + ); + for dir in [&dest, &victim, &src] { + std::fs::create_dir_all(dir).unwrap(); + } + std::fs::write(victim.join("keep"), "victim").unwrap(); + std::fs::write(src.join("pushed"), "client").unwrap(); + let archive = std::process::Command::new("tar") + .arg("-c") + .arg("-C") + .arg(&src) + .arg(".") + .output() + .unwrap() + .stdout; + + let name = ".silkd-push-swap"; + let (dest_fd, staging) = stage_dir(dest.to_str().unwrap(), name).unwrap(); + std::fs::rename(dest.join(name), root.path().join("moved")).unwrap(); + std::os::unix::fs::symlink(&victim, dest.join(name)).unwrap(); + + let data = base64::engine::general_purpose::STANDARD.encode(&archive); + let frames = format!( + "{}\n{}\n", + json!({"op":"data","data":data}), + json!({"op":"data_end"}) + ); + let mut reader = tokio::io::BufReader::new(frames.as_bytes()); + let mut out = Vec::new(); + push_staged(&mut reader, &mut out, staging, dest_fd) + .await + .unwrap(); + + let reply = String::from_utf8_lossy(&out); + assert_eq!( + std::fs::read_to_string(dest.join("pushed")).ok().as_deref(), + Some("client"), + "{reply}" + ); + assert!(!victim.join("pushed").exists()); + assert!(victim.join("keep").exists() && !dest.join("keep").exists()); + } +} diff --git a/silkd/tests/git_e2e.rs b/silkd/tests/git_e2e.rs index 9c82c5e5..b65f7382 100644 --- a/silkd/tests/git_e2e.rs +++ b/silkd/tests/git_e2e.rs @@ -31,6 +31,18 @@ fn init_repo() -> tempfile::TempDir { dir } +fn chown_tree(dir: &std::path::Path, uid: u32) { + std::os::unix::fs::lchown(dir, Some(uid), Some(uid)).unwrap(); + for entry in std::fs::read_dir(dir).unwrap() { + let path = entry.unwrap().path(); + if path.is_dir() && !path.is_symlink() { + chown_tree(&path, uid); + } else { + std::os::unix::fs::lchown(&path, Some(uid), Some(uid)).unwrap(); + } + } +} + #[tokio::test] async fn add_commit_status_reports_structured_state() { let repo = init_repo(); @@ -194,3 +206,21 @@ async fn commit_works_without_preconfigured_identity() { "commit failed: {commit:?}" ); } + +#[tokio::test] +async fn git_runs_as_the_repo_owner_under_root() { + if !nix::unistd::geteuid().is_root() { + return; + } + let repo = init_repo(); + let p = repo.path().to_str().unwrap(); + std::fs::write(repo.path().join("a.txt"), "hello\n").unwrap(); + chown_tree(repo.path(), 65534); + + let st = exchange(&[json!({"op":"git_status","path":p}).to_string()]).await; + assert_eq!(type_of(&st[0]), "git_status_result", "{st:?}"); + let add = exchange(&[json!({"op":"git_add","path":p,"files":["a.txt"]}).to_string()]).await; + assert_eq!(type_of(add.last().unwrap()), "done", "{add:?}"); + let index = std::fs::metadata(repo.path().join(".git/index")).unwrap(); + assert_eq!(std::os::unix::fs::MetadataExt::uid(&index), 65534); +} diff --git a/silkd/tests/tree_e2e.rs b/silkd/tests/tree_e2e.rs index 9aa42eb4..100778b4 100644 --- a/silkd/tests/tree_e2e.rs +++ b/silkd/tests/tree_e2e.rs @@ -134,6 +134,27 @@ async fn push_bad_archive_reports_tar_failure() { assert_eq!(frames.last().unwrap()["kind"], "internal"); } +#[tokio::test] +async fn push_large_bad_archive_reports_tar_not_the_pipe() { + let dest = tempfile::tempdir().unwrap(); + let mut gzip = vec![0u8; 256 * 1024]; + gzip[..3].copy_from_slice(&[0x1f, 0x8b, 0x08]); + let chunk = json!({"op":"data","data":b64(&vec![0u8; 256 * 1024])}).to_string(); + let mut lines = vec![ + json!({"op":"fs_push","dest":dest.path().join("p").to_str().unwrap()}).to_string(), + json!({"op":"data","data":b64(&gzip)}).to_string(), + ]; + lines.extend(std::iter::repeat_n(chunk, 32)); + lines.push(json!({"op":"data_end"}).to_string()); + let frames = exchange(&lines).await; + let last = frames.last().unwrap(); + assert_eq!(type_of(last), "error"); + assert!( + last["message"].as_str().unwrap().starts_with("tar extract"), + "{last:?}" + ); +} + #[tokio::test] async fn pull_missing_path_errors() { let dir = tempfile::tempdir().unwrap();