diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml index 4b82aaa7..484f416c 100644 --- a/.github/workflows/python.yml +++ b/.github/workflows/python.yml @@ -6,6 +6,8 @@ on: - "sdk/python/**" - "sdk/openai/**" - "sdk/langchain/**" + - "mcp/**/*.py" + - "ruff.toml" - "protocol/**" - ".github/workflows/python.yml" pull_request: @@ -13,6 +15,8 @@ on: - "sdk/python/**" - "sdk/openai/**" - "sdk/langchain/**" + - "mcp/**/*.py" + - "ruff.toml" - "protocol/**" jobs: @@ -24,9 +28,9 @@ jobs: with: python-version: "3.12" - name: Install tooling - run: pip install ruff pytest && pip install -e sdk/python -e sdk/openai -e sdk/langchain + run: pip install ruff==0.15.20 pytest==8.4.2 && pip install -e sdk/python -e sdk/openai -e sdk/langchain - name: Lint - run: ruff check sdk/python && ruff check sdk/openai && ruff check sdk/langchain + run: ruff format --check . && ruff check . - name: Test cocoonsandbox run: cd sdk/python && pytest -q - name: Test openai adapter diff --git a/boot/init/src/boot.rs b/boot/init/src/boot.rs index 622be227..db2df6c2 100644 --- a/boot/init/src/boot.rs +++ b/boot/init/src/boot.rs @@ -122,7 +122,7 @@ fn assemble(cfg: &BootCfg, marks: &mut Marks) -> Result<(), String> { if let Some(hostname) = &cfg.hostname { sys::sethostname(hostname)?; } - // Empty machine-id => systemd generates a fresh one per VM (clone identity). + // empty machine-id => systemd generates a fresh one per VM (clone identity). let _ = fs::write(format!("{NEWROOT}/etc/machine-id"), ""); persist_network(cfg); @@ -139,7 +139,7 @@ fn assemble(cfg: &BootCfg, marks: &mut Marks) -> Result<(), String> { Ok(()) } -/// Persists kernel ip= params as MAC-matched networkd units in the new root; a missing NIC degrades to the DHCP fallback. +/// Persists kernel ip= params as MAC-matched networkd units in the new root. fn persist_network(cfg: &BootCfg) { if cfg.ips.is_empty() { return; @@ -228,7 +228,7 @@ fn scan_serials(ids: &[&str], found: &mut [Option]) { if !name.starts_with("vd") { continue; } - // The serial attribute location varies by kernel version. + // the serial attribute location varies by kernel version. let paths = [ format!("/sys/block/{name}/serial"), format!("/sys/block/{name}/device/serial"), diff --git a/boot/init/src/cfg.rs b/boot/init/src/cfg.rs index a5e58fee..ad466c9f 100644 --- a/boot/init/src/cfg.rs +++ b/boot/init/src/cfg.rs @@ -59,7 +59,7 @@ pub fn parse(cmdline: &str) -> Result { .collect(); } "cocoon.cow" => cfg.cow = val.to_string(), - // A junk value keeps the default, matching the old initramfs hook. + // a junk value keeps the default, matching the old initramfs hook. "cocoon.timeout" => { if let Ok(secs) = val.parse::() { cfg.timeout = Duration::from_secs(secs.min(MAX_TIMEOUT_SECS)); @@ -155,7 +155,7 @@ fn parse_ip_param(val: &str) -> Option { fn mask_to_prefix(mask: &str) -> Option { let bits = u32::from(mask.parse::().ok()?); let prefix = bits.leading_ones(); - // Reject non-contiguous masks. + // reject non-contiguous masks. (bits == u32::MAX.checked_shl(32 - prefix).unwrap_or(0)).then_some(prefix as u8) } diff --git a/docs/langchain.md b/docs/langchain.md index aafdb362..112a1cbd 100644 --- a/docs/langchain.md +++ b/docs/langchain.md @@ -19,7 +19,7 @@ schemas, sync-native with `asyncio.to_thread` async bridges): | tool | what it does | |---|---| -| `sandbox_exec` | run a shell command, cut off after 5 minutes; stdout/stderr/exit code; disk state persists across calls | +| `sandbox_exec` | run a shell command, cut off after 5 minutes without output; stdout/stderr/exit code; disk state persists across calls | | `sandbox_write_file` | write a text file (atomic on the guest) | | `sandbox_read_file` | read a text file | | `sandbox_list_dir` | list a directory as JSON | diff --git a/e2e/cmd/crossnode/main.go b/e2e/cmd/crossnode/main.go index 93131cfe..cfe4680b 100644 --- a/e2e/cmd/crossnode/main.go +++ b/e2e/cmd/crossnode/main.go @@ -55,6 +55,7 @@ func run(addrA, addrB, token, template string) error { if err != nil { return fmt.Errorf("checkpoint on A: %w", err) } + defer func() { _ = ck.Delete(context.WithoutCancel(ctx)) }() fmt.Printf(" A: checkpoint %s published to the store in %.1fs\n", ck.ID, time.Since(t0).Seconds()) cb, err := sandbox.Connect(addrB, sandbox.WithAPIToken(token)) diff --git a/e2e/cmd/egresssmoke/main.go b/e2e/cmd/egresssmoke/main.go index 00046888..c59f41e7 100644 --- a/e2e/cmd/egresssmoke/main.go +++ b/e2e/cmd/egresssmoke/main.go @@ -14,6 +14,7 @@ package main import ( "context" + "errors" "flag" "fmt" "io" @@ -49,6 +50,9 @@ func main() { } func run(addr, token, template, wantToken, netShape, reach, nicAddr, echo string, guarded bool) error { + if guarded && wantToken == "" { + return errors.New("-secret is required for the injection check") + } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) defer cancel() diff --git a/e2e/cmd/interceptsmoke/main.go b/e2e/cmd/interceptsmoke/main.go index 5f59f934..46f4a405 100644 --- a/e2e/cmd/interceptsmoke/main.go +++ b/e2e/cmd/interceptsmoke/main.go @@ -7,6 +7,7 @@ package main import ( "context" + "errors" "flag" "fmt" "os" @@ -34,6 +35,9 @@ func main() { } func run(addr, token, template, echo, secret, issuer string) error { + if secret == "" { + return errors.New("-secret is required for the injection check") + } ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) defer cancel() diff --git a/e2e/cmd/rpcbench/main.go b/e2e/cmd/rpcbench/main.go index a5c1d18d..ae313cc0 100644 --- a/e2e/cmd/rpcbench/main.go +++ b/e2e/cmd/rpcbench/main.go @@ -141,6 +141,8 @@ func dialAgent(ctx context.Context, addr, id, token string) (net.Conn, error) { if err != nil { return nil, err } + stop := context.AfterFunc(ctx, func() { _ = raw.Close() }) + defer stop() req, err := http.NewRequest(http.MethodGet, "http://"+addr+"/v1/sandboxes/"+id+"/agent", nil) if err != nil { _ = raw.Close() diff --git a/e2e/cmd/smoke/main.go b/e2e/cmd/smoke/main.go index 4ff229b6..9e9a309a 100644 --- a/e2e/cmd/smoke/main.go +++ b/e2e/cmd/smoke/main.go @@ -14,6 +14,7 @@ import ( "flag" "fmt" "io" + "net/textproto" "os" "slices" "strconv" @@ -274,8 +275,11 @@ func smokeGit(ctx context.Context, sb *sandbox.Sandbox) error { if err = sb.GitCheckout(ctx, "/work", "feature"); err != nil { return err } - if br, err = sb.GitBranches(ctx, "/work"); err != nil || br.Current != "feature" { - return fmt.Errorf("after checkout: current=%q err=%v", br.Current, err) + if br, err = sb.GitBranches(ctx, "/work"); err != nil { + return fmt.Errorf("branches after checkout: %w", err) + } + if br.Current != "feature" { + return fmt.Errorf("after checkout: current=%q, want feature", br.Current) } // This sandbox is on the no-network lane: push must fail with the typed @@ -706,25 +710,15 @@ func lspWrite(w io.Writer, body string) error { // id arrives (server-initiated requests also carry ids but have a method; // notifications have no id — both are skipped), returning its result. func lspReadResponse(r *bufio.Reader, id int) (json.RawMessage, error) { + tp := textproto.NewReader(r) for { - length := 0 - for { - line, err := r.ReadString('\n') - if err != nil { - return nil, err - } - line = strings.TrimRight(line, "\r\n") - if line == "" { - break - } - if v, ok := strings.CutPrefix(line, "Content-Length: "); ok { - if length, err = strconv.Atoi(v); err != nil { - return nil, fmt.Errorf("content-length %q: %w", v, err) - } - } + hdr, err := tp.ReadMIMEHeader() + if err != nil { + return nil, err } - if length <= 0 { - return nil, fmt.Errorf("lsp frame without content-length") + length, err := strconv.Atoi(hdr.Get("Content-Length")) + if err != nil || length <= 0 { + return nil, fmt.Errorf("lsp frame content-length %q", hdr.Get("Content-Length")) } body := make([]byte, length) if _, err := io.ReadFull(r, body); err != nil { diff --git a/mcp/e2e.py b/mcp/e2e.py index 8f576374..5bebb731 100644 --- a/mcp/e2e.py +++ b/mcp/e2e.py @@ -5,6 +5,8 @@ python3 e2e.py --bin ./sandbox-mcp --addr 127.0.0.1:7777 --token e2e --template rt2:24.04 """ +from __future__ import annotations + import argparse import json import subprocess @@ -12,13 +14,17 @@ class McpClient: - def __init__(self, argv): - self.proc = subprocess.Popen( - argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE - ) + def __init__(self, argv: list[str]) -> None: + self.proc = subprocess.Popen(argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE) self.seq = 0 - def call(self, method, params=None): + def __enter__(self) -> McpClient: + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + def call(self, method: str, params: dict | None = None) -> dict: self.seq += 1 req = {"jsonrpc": "2.0", "id": self.seq, "method": method} if params is not None: @@ -31,7 +37,7 @@ def call(self, method, params=None): assert "error" not in resp, resp return resp["result"] - def tool(self, tool_name, **arguments): + def tool(self, tool_name: str, **arguments: object) -> object: result = self.call("tools/call", {"name": tool_name, "arguments": arguments}) text = result["content"][0]["text"] if result.get("isError"): @@ -41,9 +47,14 @@ def tool(self, tool_name, **arguments): except ValueError: return text - def close(self): + def close(self) -> None: self.proc.stdin.close() - self.proc.wait(timeout=10) + self.proc.stdout.close() + try: + self.proc.wait(timeout=10) + except subprocess.TimeoutExpired: + self.proc.kill() + self.proc.wait() def main() -> int: @@ -54,18 +65,11 @@ def main() -> int: parser.add_argument("--template", default="rt:24.04") args = parser.parse_args() - mcp = McpClient( - [args.bin, "-addr", args.addr, "-token", args.token, "-template", args.template] - ) - try: - init = mcp.call( - "initialize", {"protocolVersion": "2024-11-05", "capabilities": {}} - ) + with McpClient([args.bin, "-addr", args.addr, "-token", args.token, "-template", args.template]) as mcp: + init = mcp.call("initialize", {"protocolVersion": "2024-11-05", "capabilities": {}}) assert init["serverInfo"]["name"] == "sandbox-mcp", init tools = {t["name"] for t in mcp.call("tools/list")["tools"]} - assert {"create_sandbox", "exec", "checkpoint", "branch_checkpoint"} <= tools, ( - tools - ) + assert {"create_sandbox", "exec", "checkpoint", "branch_checkpoint"} <= tools, tools print(f" initialize + tools/list ok ({len(tools)} tools)") sandbox_id = mcp.tool("create_sandbox")["sandbox_id"] @@ -75,15 +79,11 @@ def main() -> int: mcp.tool("write_file", sandbox_id=sandbox_id, path="/root/m.txt", content="v1") assert mcp.tool("read_file", sandbox_id=sandbox_id, path="/root/m.txt") == "v1" - names = { - e["name"] for e in mcp.tool("list_dir", sandbox_id=sandbox_id, path="/root") - } + names = {e["name"] for e in mcp.tool("list_dir", sandbox_id=sandbox_id, path="/root")} assert "m.txt" in names, names print(" files ok") - ckpt = mcp.tool("checkpoint", sandbox_id=sandbox_id, name="mcp-step")[ - "checkpoint_id" - ] + ckpt = mcp.tool("checkpoint", sandbox_id=sandbox_id, name="mcp-step")["checkpoint_id"] mcp.tool("write_file", sandbox_id=sandbox_id, path="/root/m.txt", content="v2") branch = mcp.tool("branch_checkpoint", checkpoint_id=ckpt)["sandbox_id"] assert mcp.tool("read_file", sandbox_id=branch, path="/root/m.txt") == "v1" @@ -109,8 +109,6 @@ def main() -> int: info = mcp.tool("node_info") assert "pools" in info, info print(" cleanup + node_info ok") - finally: - mcp.close() print("MCP-E2E PASS") return 0 diff --git a/mcp/tools.go b/mcp/tools.go index 9bc168fd..8666d272 100644 --- a/mcp/tools.go +++ b/mcp/tools.go @@ -438,11 +438,7 @@ func jsonText(v any) string { type props map[string]map[string]any func str(description string) map[string]any { - p := map[string]any{"type": "string"} - if description != "" { - p["description"] = description - } - return p + return map[string]any{"type": "string", "description": description} } func integer(description string) map[string]any { diff --git a/protocol/wire/frame.go b/protocol/wire/frame.go index 74ba6233..474235bf 100644 --- a/protocol/wire/frame.go +++ b/protocol/wire/frame.go @@ -22,6 +22,10 @@ const ( ProtoVersion = 1 // MaxFrame mirrors silkd's frame cap. MaxFrame = 8 << 20 + // BulkChunk mirrors silkd's BULK_CHUNK, the payload of one bulk data frame. + BulkChunk = 256 * 1024 + // PortWriteChunk keeps a port data frame (payload x4/3 base64 plus envelope) well under MaxFrame. + PortWriteChunk = 1 << 20 // GitBranch.Action values (silkd's GitBranchOp). BranchList = "list" diff --git a/ruff.toml b/ruff.toml new file mode 100644 index 00000000..66f33c4f --- /dev/null +++ b/ruff.toml @@ -0,0 +1,5 @@ +line-length = 120 +target-version = "py310" + +[lint] +select = ["E", "F", "W", "I", "UP", "B", "SIM"] diff --git a/sandboxd/engine/installca.go b/sandboxd/engine/installca.go index e0d8876a..9749319a 100644 --- a/sandboxd/engine/installca.go +++ b/sandboxd/engine/installca.go @@ -41,7 +41,7 @@ func (e *Engine) silkdWriteFile(ctx context.Context, vsockSocket, path string, m if serr := s.send(wire.FsWrite{Path: path, Mode: &mode}); serr != nil { return serr } - for chunk := range slices.Chunk(data, silkdChunk) { + for chunk := range slices.Chunk(data, wire.BulkChunk) { if serr := s.send(wire.Data{Data: chunk}); serr != nil { return serr } diff --git a/sandboxd/engine/portconn.go b/sandboxd/engine/portconn.go index 16176a95..5458c356 100644 --- a/sandboxd/engine/portconn.go +++ b/sandboxd/engine/portconn.go @@ -13,9 +13,6 @@ import ( ) const ( - // portWriteChunk keeps each data frame well under silkd's 8MiB frame cap. - portWriteChunk = 1 << 20 - // portReadBuf fits silkd's data frames in one buffered read. portReadBuf = 64 << 10 ) @@ -69,7 +66,7 @@ func (g *guestPortConn) Read(p []byte) (int, error) { func (g *guestPortConn) Write(p []byte) (int, error) { written := 0 for len(p) > 0 { - n := min(len(p), portWriteChunk) + n := min(len(p), wire.PortWriteChunk) // the hot relay path reuses one buffer instead of allocating per chunk. g.wbuf = wire.AppendBulkRequest(g.wbuf, "data", p[:n]) if _, err := g.Conn.Write(g.wbuf); err != nil { diff --git a/sandboxd/engine/silkd.go b/sandboxd/engine/silkd.go index c9afd760..cbd06f77 100644 --- a/sandboxd/engine/silkd.go +++ b/sandboxd/engine/silkd.go @@ -10,8 +10,6 @@ import ( "github.com/cocoonstack/sandbox/protocol/wire" ) -const silkdChunk = 256 * 1024 - // silkdSession is a dialed silkd conn bound to a ctx, with wire-typed request/reply helpers. type silkdSession struct { conn net.Conn diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index cae92656..1a256fb6 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -247,7 +247,7 @@ func (m *Manager) commitTransition(ctx context.Context, sb *types.Sandbox, snap, // syncClaims flushes a lagging journal so a hibernate retry cannot report a false success. func (m *Manager) syncClaims(ctx context.Context, sb *types.Sandbox) error { if !m.store.synced() { - if err := m.store.commit(m.claimsSnapshot()); err != nil { + if err := m.store.commit(m.store.mark()); err != nil { return err } } diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 5d45793b..1f1b654a 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -11,6 +11,7 @@ import ( "encoding/json" "errors" "fmt" + "maps" "net" "os" "path/filepath" @@ -428,20 +429,12 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg } } m.idleDefault = time.Duration(cfg.IdleHibernateSeconds) * time.Second - m.idleEnabled = m.idleDefault > 0 m.archiveAfterDefault = time.Duration(cfg.ArchiveAfterSeconds) * time.Second m.archiveDeleteDefault = time.Duration(cfg.ArchiveDeleteAfterSeconds) * time.Second - m.archiveEnabled = m.archiveAfterDefault > 0 for _, spec := range cfg.Pools { p := newPool(spec.PoolKey) p.applySpec(spec) m.pools[spec.PoolKey] = p - if spec.IdleHibernateSeconds > 0 { - m.idleEnabled = true - } - if spec.ArchiveAfterSeconds > 0 { - m.archiveEnabled = true - } if spec.Egress != nil { m.poolEgress[spec.PoolKey] = spec.Egress m.guardedEgress = true @@ -461,6 +454,7 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg if err := m.adoptPersistedPools(ctx); err != nil { return nil, err } + m.recomputeSweepFlags() return m, nil } @@ -510,7 +504,7 @@ func (m *Manager) Run(ctx context.Context) { // FlushClaims synchronously persists the current claim set at shutdown. func (m *Manager) FlushClaims() error { - return m.store.commit(m.claimsSnapshot()) + return m.store.commit(m.store.mark()) } // SetTemplateNotifier wires the immediate-republish hook; call it before serving starts. @@ -523,15 +517,11 @@ func (m *Manager) Info() ([]PoolInfo, Gauges) { m.mu.Lock() defer m.mu.Unlock() now := time.Now() - live := make([]*pool, 0, len(m.pools)) - for _, p := range m.pools { - if !p.removed { - live = append(live, p) + pools := make([]PoolInfo, 0, len(m.pools)) + for _, p := range slices.SortedFunc(maps.Values(m.pools), func(a, b *pool) int { return strings.Compare(a.hash, b.hash) }) { + if p.removed { + continue } - } - slices.SortFunc(live, func(a, b *pool) int { return strings.Compare(a.hash, b.hash) }) - pools := make([]PoolInfo, 0, len(live)) - for _, p := range live { pools = append(pools, PoolInfo{ Key: p.key, Warm: len(p.warm), @@ -606,8 +596,14 @@ func (m *Manager) sweepStoreGenerations(ctx context.Context) { } } -func (m *Manager) claimsSnapshot() claimSnapshot { - return m.store.mark() +// recomputeSweepFlags derives the sweep switches from the live pool set rather than latching them: removing every idle pool turns the sweep off again. +func (m *Manager) recomputeSweepFlags() { + m.idleEnabled = m.idleDefault > 0 + m.archiveEnabled = m.archiveAfterDefault > 0 + for _, p := range m.pools { + m.idleEnabled = m.idleEnabled || p.idle > 0 + m.archiveEnabled = m.archiveEnabled || p.archiveAfter > 0 + } } func (m *Manager) untrack(set map[string]struct{}, key string) { diff --git a/sandboxd/pool/poolstore.go b/sandboxd/pool/poolstore.go index c8e642dc..40c86546 100644 --- a/sandboxd/pool/poolstore.go +++ b/sandboxd/pool/poolstore.go @@ -87,8 +87,6 @@ func (m *Manager) adoptPersistedPools(ctx context.Context) error { logger.Warnf(ctx, "config.json pools differ from the API-applied set and are overridden; delete %s to return to config-owned pools", m.poolStore.path) } clear(m.pools) - m.idleEnabled = m.idleDefault > 0 - m.archiveEnabled = m.archiveAfterDefault > 0 for _, spec := range pf.Pools { spec = normalizePoolSpec(spec) if err := m.validate(spec.PoolKey); err != nil { @@ -101,12 +99,6 @@ func (m *Manager) adoptPersistedPools(ctx context.Context) error { p.applySpec(spec) m.adoptGolden(p) m.pools[spec.PoolKey] = p - if spec.IdleHibernateSeconds > 0 { - m.idleEnabled = true - } - if spec.ArchiveAfterSeconds > 0 { - m.archiveEnabled = true - } } logger.Infof(ctx, "restored %d API-applied pools from pools.json", len(pf.Pools)) return nil diff --git a/sandboxd/pool/refill.go b/sandboxd/pool/refill.go index e84b0a03..c794e614 100644 --- a/sandboxd/pool/refill.go +++ b/sandboxd/pool/refill.go @@ -213,14 +213,10 @@ func (m *Manager) buildGoldenSteps(ctx context.Context, key types.PoolKey, name, return writeGoldenSidecar(final+warmupSidecarSuffix, warmupStamp(warmup)) } -func (m *Manager) poolWarmup(key types.PoolKey) []string { - m.mu.Lock() - defer m.mu.Unlock() - return m.poolWarmups[key] -} - func (m *Manager) runWarmup(ctx context.Context, key types.PoolKey, sock string) ([]string, error) { - warmup := m.poolWarmup(key) + m.mu.Lock() + warmup := m.poolWarmups[key] + m.mu.Unlock() if len(warmup) == 0 { return nil, nil } diff --git a/sandboxd/pool/remove.go b/sandboxd/pool/remove.go index 8404ea56..c3887656 100644 --- a/sandboxd/pool/remove.go +++ b/sandboxd/pool/remove.go @@ -77,9 +77,6 @@ func (m *Manager) retryRemovals(ctx context.Context) *sync.WaitGroup { } } m.mu.Unlock() - if len(batch) == 0 { - return new(sync.WaitGroup) - } names := slices.Collect(maps.Keys(batch)) return m.runBounded(ctx, len(names), func(ctx context.Context, i int) { m.retryRemoval(ctx, names[i], batch[names[i]]) diff --git a/sandboxd/pool/setpools.go b/sandboxd/pool/setpools.go index e327bca7..a45437be 100644 --- a/sandboxd/pool/setpools.go +++ b/sandboxd/pool/setpools.go @@ -59,17 +59,7 @@ func (m *Manager) SetPools(ctx context.Context, specs []config.PoolSpec) error { m.adoptGolden(p) m.pools[key] = p } - // recompute rather than latch: removing every idle pool turns the sweep off again - m.idleEnabled = m.idleDefault > 0 - m.archiveEnabled = m.archiveAfterDefault > 0 - for _, p := range m.pools { - if p.idle > 0 { - m.idleEnabled = true - } - if p.archiveAfter > 0 { - m.archiveEnabled = true - } - } + m.recomputeSweepFlags() m.mu.Unlock() runCtx := context.WithoutCancel(ctx) diff --git a/sdk/go/checkpoint.go b/sdk/go/checkpoint.go index cf452213..7cd534ae 100644 --- a/sdk/go/checkpoint.go +++ b/sdk/go/checkpoint.go @@ -32,8 +32,8 @@ func (ck *Checkpoint) New(ctx context.Context, opts ...Option) (*Sandbox, error) for _, opt := range opts { opt(&claim) } - if len(claim.Volumes) > 0 { - return nil, errors.New("checkpoint claims do not accept WithVolumes") + if len(claim.Volumes) > 0 || claim.ClaimRef != "" { + return nil, errors.New("checkpoint claims do not accept WithVolumes or WithClaimRef") } if err := claim.rejectPinnedAxes(); err != nil { return nil, err diff --git a/sdk/go/checkpoint_test.go b/sdk/go/checkpoint_test.go index 705ef5cf..493c24a3 100644 --- a/sdk/go/checkpoint_test.go +++ b/sdk/go/checkpoint_test.go @@ -45,6 +45,17 @@ func TestCheckpointNewFollowsRedirect(t *testing.T) { } } +func TestCheckpointNewRejectsClaimRefLocally(t *testing.T) { + ck := &Checkpoint{} + sb, err := ck.New(t.Context(), WithClaimRef("ns/claim")) + if err == nil || !strings.Contains(err.Error(), "WithClaimRef") { + t.Errorf("err %v, want local WithClaimRef rejection", err) + } + if sb != nil { + t.Errorf("sandbox %+v, want nil", sb) + } +} + func TestCheckpointNewRedirectAllCandidatesFail(t *testing.T) { broken := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusInternalServerError) diff --git a/sdk/go/client.go b/sdk/go/client.go index 89823837..8bbecd6e 100644 --- a/sdk/go/client.go +++ b/sdk/go/client.go @@ -107,7 +107,8 @@ func (c *Client) Lookup(ctx context.Context, id, token string) (*Sandbox, error) if owner, err := c.ownerAt(ctx, c.addr, id, token); err == nil { return &Sandbox{ID: id, token: token, c: c, owner: owner}, nil } - owner, ok := scatter(ctx, c.peers(ctx), func(ctx context.Context, addr string) (string, error) { + addrs, _ := c.peersOrErr(ctx) + owner, ok := scatter(ctx, addrs, func(ctx context.Context, addr string) (string, error) { return c.ownerAt(ctx, addr, id, token) }) if !ok { diff --git a/sdk/go/info.go b/sdk/go/info.go index c0f44ff1..ba23c3c2 100644 --- a/sdk/go/info.go +++ b/sdk/go/info.go @@ -6,8 +6,7 @@ import ( "time" ) -// peersTimeout bounds the best-effort peer discovery inside Lookup so one -// slow entry node cannot stall the scatter. +// peersTimeout bounds peer discovery so one slow entry node cannot stall the caller. const peersTimeout = 5 * time.Second // NodeInfo is one node's operational state from GET /v1/info. @@ -69,12 +68,6 @@ func (c *Client) Sandboxes(ctx context.Context) ([]SandboxSummary, error) { return reply.Sandboxes, nil } -// peers fetches the cluster's node addresses, best-effort (nil on failure). -func (c *Client) peers(ctx context.Context) []string { - addrs, _ := c.peersOrErr(ctx) - return addrs -} - // peersOrErr fetches the node addresses, surfacing a discovery failure. It reads // /v1/peers (tenant-accessible), so it works under a tenant token. func (c *Client) peersOrErr(ctx context.Context) ([]string, error) { diff --git a/sdk/go/lsp.go b/sdk/go/lsp.go index aa9ccf1a..46d748e4 100644 --- a/sdk/go/lsp.go +++ b/sdk/go/lsp.go @@ -33,12 +33,7 @@ func (l *Lsp) Stop(ctx context.Context) error { // (rooted at root), returning a handle. On the base image, which ships no // language servers, this fails with silkd's typed not_found. func (s *Sandbox) StartLsp(ctx context.Context, language, root string) (*Lsp, error) { - conn, done, err := s.call(ctx, &wire.LspStart{Language: language, Root: root}) - if err != nil { - return nil, err - } - defer done() - started, err := expect[wire.LspStarted](ctx, conn) + started, err := oneShotRPC[wire.LspStarted](ctx, s, &wire.LspStart{Language: language, Root: root}) if err != nil { return nil, err } diff --git a/sdk/go/pools.go b/sdk/go/pools.go index ad6c9ad6..3bec953d 100644 --- a/sdk/go/pools.go +++ b/sdk/go/pools.go @@ -5,6 +5,7 @@ import ( "context" "fmt" "net/http" + "slices" "sync" ) @@ -44,15 +45,7 @@ func (c *Client) SetPools(ctx context.Context, pools []PoolSpec) (*NodeInfo, err // was reached — an incomplete apply to retry, not a single-node cluster (nil). func (c *Client) SetPoolsCluster(ctx context.Context, pools []PoolSpec) ([]PoolResult, error) { peers, peersErr := c.peersOrErr(ctx) - seen := map[string]struct{}{} - var addrs []string - for _, a := range append([]string{c.addr}, peers...) { - if _, dup := seen[a]; dup { - continue - } - seen[a] = struct{}{} - addrs = append(addrs, a) - } + addrs := slices.Compact(slices.Sorted(slices.Values(append([]string{c.addr}, peers...)))) results := make([]PoolResult, len(addrs)) var wg sync.WaitGroup for i, addr := range addrs { diff --git a/sdk/go/port.go b/sdk/go/port.go index b0ffd909..81f233c5 100644 --- a/sdk/go/port.go +++ b/sdk/go/port.go @@ -14,10 +14,6 @@ import ( "github.com/cocoonstack/sandbox/sdk/go/silkd" ) -// portWriteChunk keeps a data frame (payload ×4/3 base64 + envelope) well -// under wire.MaxFrame. -const portWriteChunk = 1 << 20 - var _ net.Conn = (*PortConn)(nil) // PortConn is a net.Conn to a TCP port inside the sandbox, relayed over the @@ -39,7 +35,7 @@ func (p *PortConn) Read(b []byte) (int, error) { return p.out.Read(b) } func (p *PortConn) Write(b []byte) (int, error) { sent := 0 for len(b) > 0 { - chunk := b[:min(len(b), portWriteChunk)] + chunk := b[:min(len(b), wire.PortWriteChunk)] if err := p.conn.Send(&wire.Data{Data: chunk}); err != nil { return sent, err } diff --git a/sdk/go/proc.go b/sdk/go/proc.go index 28610d51..433ac01f 100644 --- a/sdk/go/proc.go +++ b/sdk/go/proc.go @@ -1,7 +1,6 @@ package sandbox import ( - "cmp" "context" "fmt" "io" @@ -63,38 +62,11 @@ func (s *Sandbox) Attach(ctx context.Context, pid uint32, stdout, stderr io.Writ return s.drainProc(ctx, &wire.Attach{PID: pid}, stdout, stderr) } -// drainProc pumps stdout/stderr frames until the terminal frame: exit -// carries the code, done means the stream ended without one. func (s *Sandbox) drainProc(ctx context.Context, req wire.Request, stdout, stderr io.Writer) (int32, bool, error) { conn, done, err := s.call(ctx, req) if err != nil { return 0, false, err } defer done() - stdout = cmp.Or(stdout, io.Discard) - stderr = cmp.Or(stderr, io.Discard) - for { - resp, err := recv(ctx, conn) - if err != nil { - return 0, false, err - } - switch resp := resp.(type) { - case *wire.Stdout: - if _, err := stdout.Write(resp.Data); err != nil { - return 0, false, err - } - case *wire.Stderr: - if _, err := stderr.Write(resp.Data); err != nil { - return 0, false, err - } - case *wire.Exit: - return resp.Code, true, nil - case *wire.Done: - return 0, false, nil - case *wire.ErrorResp: - return 0, false, resp - default: - return 0, false, unexpected(resp) - } - } + return pumpStdio(ctx, conn, stdout, stderr) } diff --git a/sdk/go/sandbox.go b/sdk/go/sandbox.go index ddf5418d..f786b1c0 100644 --- a/sdk/go/sandbox.go +++ b/sdk/go/sandbox.go @@ -2,7 +2,6 @@ package sandbox import ( "bytes" - "cmp" "context" "fmt" "io" @@ -110,38 +109,21 @@ func (s *Sandbox) Run(ctx context.Context, cmd Cmd) (int, error) { defer done() if cmd.Stdin == nil { - if err := conn.Send(wire.StdinClose{}); err != nil { + if err = conn.Send(wire.StdinClose{}); err != nil { return 0, fmt.Errorf("close stdin: %w", err) } } else { go pumpStdin(conn, cmd.Stdin) } - stdout := cmp.Or(cmd.Stdout, io.Discard) - stderr := cmp.Or(cmd.Stderr, io.Discard) - for { - resp, err := recv(ctx, conn) - if err != nil { - return 0, err - } - switch resp := resp.(type) { - case *wire.Started: - case *wire.Stdout: - if _, err := stdout.Write(resp.Data); err != nil { - return 0, err - } - case *wire.Stderr: - if _, err := stderr.Write(resp.Data); err != nil { - return 0, err - } - case *wire.Exit: - return int(resp.Code), nil - case *wire.ErrorResp: - return 0, resp - default: - return 0, unexpected(resp) - } + code, exited, err := pumpStdio(ctx, conn, cmd.Stdout, cmd.Stderr) + if err != nil { + return 0, err + } + if !exited { + return 0, unexpected(&wire.Done{}) } + return int(code), nil } // Fork clones the sandbox into count children — memory, disk, and guest diff --git a/sdk/go/silkd/silkdtest/fake.go b/sdk/go/silkd/silkdtest/fake.go index 0560852a..9f5cf0ae 100644 --- a/sdk/go/silkd/silkdtest/fake.go +++ b/sdk/go/silkd/silkdtest/fake.go @@ -19,9 +19,6 @@ import ( "github.com/cocoonstack/sandbox/protocol/wire" ) -// readChunk mirrors silkd's BULK_CHUNK so downloads exercise real framing. -const readChunk = 256 * 1024 - // Fake is a stateful silkd fake backing the fs verbs with a real directory // and tracking sessions, so an SDK write-then-read round-trips through it. // exec/info reuse the stateless handlers. It exists for host-side unit tests; @@ -474,7 +471,7 @@ func drainUpload(r *bufio.Reader) ([]byte, error) { func sendChunked(conn net.Conn, data []byte) { for len(data) > 0 { - n := min(readChunk, len(data)) + n := min(wire.BulkChunk, len(data)) send(conn, &wire.DataResp{Data: data[:n]}) data = data[n:] } diff --git a/sdk/go/template.go b/sdk/go/template.go index 3e8a58db..2bb2b786 100644 --- a/sdk/go/template.go +++ b/sdk/go/template.go @@ -33,20 +33,9 @@ func (t *Template) New(ctx context.Context, opts ...Option) (*Sandbox, error) { return nil, err } claim.Net, claim.Size = t.net, t.size - if len(claim.Volumes) == 0 { - claim.NoRedirect = true - body, err := encodeBody("claim", claim) - if err != nil { - return nil, err - } - cr, err := t.c.claimAt(ctx, t.addr, body) - if err != nil { - return nil, err - } - return t.c.handleFrom(t.addr, cr), nil - } + noVolumes := len(claim.Volumes) == 0 addr, cr, err := claimFollow(t.addr, "claim", func(noRedirect, requirePromoted bool) ([]byte, error) { - claim.NoRedirect, claim.RequirePromoted = noRedirect, requirePromoted + claim.NoRedirect, claim.RequirePromoted = noRedirect || noVolumes, requirePromoted return encodeBody("claim", claim) }, func(addr string, body []byte) (claimResponse, error) { return t.c.claimAt(ctx, addr, body) diff --git a/sdk/go/upgrade.go b/sdk/go/upgrade.go index 05398617..c78d7f3a 100644 --- a/sdk/go/upgrade.go +++ b/sdk/go/upgrade.go @@ -7,7 +7,6 @@ import ( "net" "net/http" "strings" - "time" ) // dialAgent opens the data-plane connection to the owner node: a raw TCP dial @@ -21,9 +20,6 @@ func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (net.Con } stop := context.AfterFunc(ctx, func() { _ = raw.Close() }) defer stop() - if deadline, ok := ctx.Deadline(); ok { - _ = raw.SetDeadline(deadline) - } // id/token interpolate into the raw request; CR/LF would inject headers. if strings.ContainsAny(id, "\r\n\x00") || strings.ContainsAny(token, "\r\n\x00") { @@ -52,7 +48,6 @@ func (c *Client) dialAgent(ctx context.Context, addr, id, token string) (net.Con _ = raw.Close() return nil, err } - _ = raw.SetDeadline(time.Time{}) return &upgradedConn{Conn: raw, r: br}, nil } diff --git a/sdk/go/utils.go b/sdk/go/utils.go index 00b943d0..84de3509 100644 --- a/sdk/go/utils.go +++ b/sdk/go/utils.go @@ -1,6 +1,7 @@ package sandbox import ( + "cmp" "context" "fmt" "io" @@ -10,9 +11,6 @@ import ( "github.com/cocoonstack/sandbox/sdk/go/silkd" ) -// fsChunk matches silkd's BULK_CHUNK. -const fsChunk = 256 * 1024 - // respPtr is a frame type's pointer form, so a non-Response T fails to compile. type respPtr[T any] interface { *T @@ -52,6 +50,37 @@ func (s *Sandbox) downloadRPC(ctx context.Context, req wire.Request, sink func([ return drainData(ctx, conn, sink) } +// pumpStdio copies stdout/stderr frames to the writers until the terminal frame: exit carries the code, done means the stream ended without one. +func pumpStdio(ctx context.Context, conn *silkd.Conn, stdout, stderr io.Writer) (code int32, exited bool, err error) { + stdout = cmp.Or(stdout, io.Discard) + stderr = cmp.Or(stderr, io.Discard) + for { + resp, err := recv(ctx, conn) + if err != nil { + return 0, false, err + } + switch resp := resp.(type) { + case *wire.Started: + case *wire.Stdout: + if _, err := stdout.Write(resp.Data); err != nil { + return 0, false, err + } + case *wire.Stderr: + if _, err := stderr.Write(resp.Data); err != nil { + return 0, false, err + } + case *wire.Exit: + return resp.Code, true, nil + case *wire.Done: + return 0, false, nil + case *wire.ErrorResp: + return 0, false, resp + default: + return 0, false, unexpected(resp) + } + } +} + // oneShotRPC sends req and returns its single typed reply frame. func oneShotRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.Request) (*T, error) { conn, done, err := s.call(ctx, req) @@ -111,7 +140,7 @@ func streamRPC[T any, PT respPtr[T]](ctx context.Context, s *Sandbox, req wire.R // uploadStream chunks r into Data frames terminated by DataEnd; shared by the // FsWrite payload and the FsPush tar stream. func uploadStream(conn *silkd.Conn, r io.Reader) error { - buf := make([]byte, fsChunk) + buf := make([]byte, wire.BulkChunk) for { n, readErr := r.Read(buf) if n > 0 { diff --git a/sdk/langchain/cocoonsandbox_langchain/toolkit.py b/sdk/langchain/cocoonsandbox_langchain/toolkit.py index 5d36a113..bf9d1e81 100644 --- a/sdk/langchain/cocoonsandbox_langchain/toolkit.py +++ b/sdk/langchain/cocoonsandbox_langchain/toolkit.py @@ -8,12 +8,13 @@ import asyncio import json import threading +from collections.abc import Callable from cocoonsandbox import Client, Sandbox from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field -# Mirrors the MCP exec contract the tool description states. +# a socket inactivity bound, not a wall clock; the tool description states it CALL_TIMEOUT = 300.0 @@ -72,7 +73,7 @@ def get_tools(self) -> list[StructuredTool]: "Returns stdout; a non-empty stderr is appended as a 'stderr:' " "line and a non-zero status as an 'exit code: N' line; a " "command that prints nothing and exits 0 returns '(no output)'. " - "The call is cut off after 5 minutes. " + "The call is cut off after 5 minutes without output. " "Files and installed packages persist across calls; environment " "variables and the working directory do not.", ExecInput, @@ -126,7 +127,7 @@ def _claim(self) -> Sandbox: return self._client.checkpoint(self._from_checkpoint).new(ttl_seconds=self._ttl) return self._client.new(self._template, net=self._net, ttl_seconds=self._ttl) - def _tool(self, name: str, description: str, schema: type[BaseModel], func) -> StructuredTool: + def _tool(self, name: str, description: str, schema: type[BaseModel], func: Callable[..., str]) -> StructuredTool: async def arun(**kwargs): return await asyncio.to_thread(func, **kwargs) diff --git a/sdk/langchain/tests/test_toolkit.py b/sdk/langchain/tests/test_toolkit.py index 08e8c1f8..ab604229 100644 --- a/sdk/langchain/tests/test_toolkit.py +++ b/sdk/langchain/tests/test_toolkit.py @@ -8,39 +8,6 @@ from cocoonsandbox_langchain import CocoonToolkit -class FakeSandbox: - def __init__(self): - self.closed = 0 - self.files = {} - - def run(self, argv, cwd="", on_stdout=None, on_stderr=None, **_): - assert argv[:2] == ["sh", "-c"] - if argv[2] == "boom": - on_stderr(b"kaboom\n") - return 3 - on_stdout(f"ran: {argv[2]}\n".encode()) - return 0 - - def write_file(self, path, data): - self.files[path] = data - - def read_file(self, path): - return self.files[path] - - def list_dir(self, path): - return [{"name": "a.txt", "kind": "file", "size": 3}] - - def close(self): - self.closed += 1 - - -def hooked(monkeypatch): - kit = CocoonToolkit("127.0.0.1:1") - fake = FakeSandbox() - monkeypatch.setattr(kit, "_claim", lambda: fake) - return kit, fake - - def test_tools_shape(monkeypatch): kit, _ = hooked(monkeypatch) tools = kit.get_tools() @@ -86,3 +53,36 @@ def test_use_after_close_raises(monkeypatch): kit.close() with pytest.raises(RuntimeError): kit.sandbox() + + +class FakeSandbox: + def __init__(self): + self.closed = 0 + self.files = {} + + def run(self, argv, cwd="", on_stdout=None, on_stderr=None, **_): + assert argv[:2] == ["sh", "-c"] + if argv[2] == "boom": + on_stderr(b"kaboom\n") + return 3 + on_stdout(f"ran: {argv[2]}\n".encode()) + return 0 + + def write_file(self, path, data): + self.files[path] = data + + def read_file(self, path): + return self.files[path] + + def list_dir(self, path): + return [{"name": "a.txt", "kind": "file", "size": 3}] + + def close(self): + self.closed += 1 + + +def hooked(monkeypatch): + kit = CocoonToolkit("127.0.0.1:1") + fake = FakeSandbox() + monkeypatch.setattr(kit, "_claim", lambda: fake) + return kit, fake diff --git a/sdk/openai/cocoonsandbox_openai/adapter.py b/sdk/openai/cocoonsandbox_openai/adapter.py index 9ea611b3..989aa467 100644 --- a/sdk/openai/cocoonsandbox_openai/adapter.py +++ b/sdk/openai/cocoonsandbox_openai/adapter.py @@ -19,7 +19,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 cocoonsandbox import APIError, Client, Sandbox, SilkdError +from cocoonsandbox import Client, Sandbox, SandboxError, SilkdError class CocoonSandboxClientOptions(BaseSandboxClientOptions): @@ -59,12 +59,10 @@ def from_state(cls, state: CocoonSandboxSessionState) -> CocoonSandboxSession: async def _prepare_backend_workspace(self) -> None: sb = self._sandbox() - await asyncio.to_thread(sb.mkdir, str(self.state.manifest.root), True) + await asyncio.to_thread(sb.mkdir, str(self.state.manifest.root), parents=True) async def _exec_internal(self, *command: str | Path, timeout: float | None = None) -> ExecResult: - # Bound the blocking SDK call by the socket timeout too, so a - # wait_for cancellation is matched by the worker thread actually - # unblocking (recv wakes) instead of lingering. + # the socket timeout bounds the blocking SDK call too, so a wait_for cancellation unblocks the worker thread sb = self._sandbox(timeout=timeout) argv = [str(part) for part in command] stdout, stderr = bytearray(), bytearray() @@ -95,7 +93,7 @@ async def write(self, path: Path, data: io.IOBase, *, user: str | User | None = async def running(self) -> bool: try: await asyncio.to_thread(self._sandbox().stat, "/") - except (APIError, SilkdError, OSError): + except (SandboxError, OSError): return False return True @@ -115,8 +113,7 @@ async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint: return ExposedPortEndpoint(host="127.0.0.1", port=listener.getsockname()[1], tls=False) async def _shutdown_backend(self) -> None: - # The sandbox itself outlives shutdown (delete releases it); only the - # local port proxies belong to this process. + # the sandbox outlives shutdown (delete releases it); only the local port proxies belong to this process for listener in self._proxies: listener.close() self._proxies.clear() @@ -179,7 +176,6 @@ def deserialize_session_state(self, payload: dict[str, object]) -> SandboxSessio def _reject_user(user: str | User | None) -> None: - # fs_read/fs_write carry no per-call user on the wire; fail loud rather - # than silently running as the guest default. + # fs_read/fs_write carry no per-call user on the wire; fail loud rather than run as the guest default if user is not None: raise NotImplementedError("per-call user impersonation is not supported by the cocoon backend") diff --git a/sdk/openai/tests/test_adapter.py b/sdk/openai/tests/test_adapter.py index cd44f448..ccf3ca9c 100644 --- a/sdk/openai/tests/test_adapter.py +++ b/sdk/openai/tests/test_adapter.py @@ -11,7 +11,7 @@ from pathlib import Path import pytest -from cocoonsandbox import SilkdError +from cocoonsandbox import ProtocolError, SilkdError from cocoonsandbox_openai import CocoonSandboxClient, CocoonSandboxClientOptions, CocoonSandboxSessionState @@ -68,7 +68,6 @@ async def go(): def test_exec_maps_stdio_and_exit(node, monkeypatch): - class FakeSandbox: def __init__(self, **kw): self.id = "sb_1" @@ -106,6 +105,21 @@ async def go(): asyncio.run(go()) +def test_running_is_false_when_the_dial_fails(node, monkeypatch): + class FakeSandbox: + def stat(self, path): + raise ProtocolError("dial 127.0.0.1:1: connection refused") + + async def go(): + client = CocoonSandboxClient() + session = await client.create(options=CocoonSandboxClientOptions(addr=node)) + inner = session._inner + monkeypatch.setattr(inner, "_sandbox", lambda timeout=None: FakeSandbox()) + assert await inner.running() is False + + asyncio.run(go()) + + def test_write_and_persist_use_tree_verbs(node, monkeypatch): calls = {} diff --git a/sdk/python/cocoonsandbox/client.py b/sdk/python/cocoonsandbox/client.py index 44d8e8cc..943f5d51 100644 --- a/sdk/python/cocoonsandbox/client.py +++ b/sdk/python/cocoonsandbox/client.py @@ -11,12 +11,15 @@ import urllib.error import urllib.parse import urllib.request -from collections.abc import Mapping +from collections.abc import Callable, Iterable, Mapping, Sequence +from typing import TypeVar from .checkpoint import Checkpoint from .errors import APIError from .sandbox import Sandbox +T = TypeVar("T") + _PEERS_TIMEOUT = 5.0 @@ -192,7 +195,7 @@ def _request( except urllib.error.URLError as exc: raise APIError(verb, 0, str(exc.reason)) from None except (OSError, http.client.HTTPException) as exc: - # A reset or truncated body read is neither HTTPError nor URLError. + # a reset or truncated body read is neither HTTPError nor URLError raise APIError(verb, 0, str(exc)) from None if not raw: return {} @@ -266,11 +269,16 @@ def _error_message(raw: bytes) -> str: return raw.decode(errors="replace").strip() -def _try_each(candidates, call, retry=lambda exc: exc.status in (404, 0)): - """Calls call against each candidate in turn, returning the first - success. An APIError for which retry(exc) is true moves on to the next +def _retry_miss(exc: APIError) -> bool: + """Default _try_each policy: only a miss (404) or a dead peer (status 0) moves on.""" + return exc.status in (404, 0) + + +def _try_each( + candidates: Iterable[str], call: Callable[[str], T], retry: Callable[[APIError], bool] = _retry_miss +) -> T: + """An APIError for which retry(exc) is true moves on to the next candidate and the last such error propagates; any other raises at once. - The default retries only a miss (404 or a dead peer, status 0); candidates must be non-empty.""" last_error = None for addr in candidates: @@ -293,17 +301,18 @@ def _retry_transient(exc: APIError) -> bool: return exc.status in (0, 401, 404, 429, 503, 500, 502, 504) -def _redirect_fallback(origin: str, candidates: list, post, verb: str): - """Walks candidates via post(addr) -> raw reply dict, retrying broadly - (any candidate failure moves to the next) so one wrong candidate doesn't - cost a candidate that would still succeed. If every candidate is - exhausted and the last failure was transient (_retry_transient), gives - the origin one more no_redirect attempt -- the node that issued the - redirect provisions or heals locally instead of leaving the claim stuck - on stale gossip. A definitive last failure skips the fallback: the origin - would fail the same way. A second-level redirect (a compliant server - never sends one once no_redirect is set) fails the candidate rather than - being followed. Returns (addr, reply).""" +def _redirect_fallback( + origin: str, candidates: Sequence[str], post: Callable[[str], dict], verb: str +) -> tuple[str, dict]: + """Retries broadly across candidates (any failure moves to the next) so + one wrong candidate doesn't cost one that would still succeed. If every + candidate is exhausted and the last failure was transient + (_retry_transient), gives the origin one more no_redirect attempt -- the + node that issued the redirect provisions or heals locally instead of + leaving the claim stuck on stale gossip. A definitive last failure skips + the fallback: the origin would fail the same way. A second-level redirect + (a compliant server never sends one once no_redirect is set) fails the + candidate rather than being followed.""" def attempt(addr): reply = post(addr) @@ -324,11 +333,10 @@ def attempt(addr): raise APIError(verb, origin_exc.status, combined) from origin_exc -def _scatter(addrs, probe): - """Probes every addr concurrently and returns the first success; when - all probes fail the last error propagates. Loser threads are daemons - whose requests die with _request's own timeout; the queue is bounded by - len(addrs) so they never block. addrs must be non-empty.""" +def _scatter(addrs: Sequence[str], probe: Callable[[str], T]) -> T: + """When all probes fail the last error propagates. Loser threads are + daemons whose requests die with _request's own timeout; the queue is + bounded by len(addrs) so they never block. addrs must be non-empty.""" results = queue.Queue(maxsize=len(addrs)) def run(addr): diff --git a/sdk/python/cocoonsandbox/sandbox.py b/sdk/python/cocoonsandbox/sandbox.py index 7470c983..7d57855a 100644 --- a/sdk/python/cocoonsandbox/sandbox.py +++ b/sdk/python/cocoonsandbox/sandbox.py @@ -406,7 +406,13 @@ def _call(self, op: str, expect: str, **fields) -> dict: def _done_rpc(self, op: str, **fields) -> None: self._call(op, "done", **fields) - def _drain_proc(self, op: str, pid: int, on_stdout, on_stderr) -> int | None: + def _drain_proc( + self, + op: str, + pid: int, + on_stdout: Callable[[bytes], object] | None, + on_stderr: Callable[[bytes], object] | None, + ) -> int | None: with self._dial() as conn: conn.send(op, pid=pid) return _pump_stdio(conn, on_stdout, on_stderr) @@ -541,7 +547,9 @@ def _feed_stdin(conn: Conn, stdin: bytes) -> None: conn.send("stdin_close") -def _pump_stdio(conn: Conn, on_stdout, on_stderr) -> int | None: +def _pump_stdio( + conn: Conn, on_stdout: Callable[[bytes], object] | None, on_stderr: Callable[[bytes], object] | None +) -> int | None: """Streams stdout/stderr frames into the callbacks until the terminal frame: the exit code, or None when the stream ends with done.""" for frame in conn.recv_until("exit", "done"): diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 08130f74..44020ee9 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -64,11 +64,8 @@ def test_claim_happy_path(node): def test_claim_sends_volumes(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, { + seen = recording_claim( + { "id": "sb_1", "token": "tok", "volumes": [ @@ -76,8 +73,7 @@ def claim(body, path): {"name": "weights-llama", "mount": "/models"}, ], } - - FakeNode.routes[("POST", "/v1/claim")] = claim + ) sb = Client(node).new("rt:24.04", volumes=["imagenet", {"name": "weights-llama", "mount": "/models"}]) assert seen == [ { @@ -108,19 +104,9 @@ def test_claim_rejects_invalid_volumes(node, volumes, match): def test_claim_sends_volume_mode_rw(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, { - "id": "sb_1", - "token": "tok", - "volumes": [ - {"name": "scratch", "mount": "/data", "mode": "rw"}, - ], - } - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim( + {"id": "sb_1", "token": "tok", "volumes": [{"name": "scratch", "mount": "/data", "mode": "rw"}]} + ) sb = Client(node).new("rt:24.04", volumes=[{"name": "scratch", "mount": "/data", "mode": "rw"}]) assert seen == [ { @@ -134,33 +120,16 @@ def claim(body, path): def test_claim_omits_volume_mode_ro(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, {"id": "sb_1", "token": "tok"} - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim({"id": "sb_1", "token": "tok"}) Client(node).new("rt:24.04", volumes=[{"name": "imagenet", "mode": "ro"}]) Client(node).new("rt:24.04", volumes=[{"name": "imagenet", "mode": ""}]) assert seen[0]["volumes"] == seen[1]["volumes"] == [{"name": "imagenet"}], seen def test_claim_attaches_volumes_without_mounting(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, { - "id": "sb_1", - "token": "tok", - "volumes": [ - {"name": "imagenet"}, - {"name": "scratch", "mode": "rw"}, - ], - } - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim( + {"id": "sb_1", "token": "tok", "volumes": [{"name": "imagenet"}, {"name": "scratch", "mode": "rw"}]} + ) sb = Client(node).new("rt:24.04", volumes=["imagenet", {"name": "scratch", "mode": "rw"}], mount=False) assert seen == [ { @@ -173,13 +142,7 @@ def claim(body, path): def test_template_claim_attaches_volumes_without_mounting(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, {"id": "sb_2", "token": "tok", "volumes": [{"name": "imagenet"}]} - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim({"id": "sb_2", "token": "tok", "volumes": [{"name": "imagenet"}]}) sb = Template(Client(node), node, "task:v1", "none", "small").new(volumes=["imagenet"], mount=False) assert seen[0]["volumes_attach_only"] is True assert seen[0]["volumes"] == [{"name": "imagenet"}] @@ -196,31 +159,15 @@ def test_claim_rejects_mount_without_mounting(node): def test_claim_keeps_mounting_by_default(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, {"id": "sb_1", "token": "tok"} - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim({"id": "sb_1", "token": "tok"}) Client(node).new("rt:24.04", volumes=["imagenet"]) assert "volumes_attach_only" not in seen[0] def test_template_claim_sends_volumes(node): - seen = [] - - def claim(body, path): - seen.append(body) - return 200, { - "id": "sb_2", - "token": "tok", - "volumes": [ - {"name": "imagenet", "mount": "/datasets/imagenet"}, - ], - } - - FakeNode.routes[("POST", "/v1/claim")] = claim + seen = recording_claim( + {"id": "sb_2", "token": "tok", "volumes": [{"name": "imagenet", "mount": "/datasets/imagenet"}]} + ) sb = Template(Client(node), node, "task:v1", "none", "small").new( volumes=[{"name": "imagenet", "mount": "/datasets/imagenet"}] ) @@ -357,3 +304,14 @@ def test_checkpoint_listing_binds_handles(node): ) branch = ckpts[0].new() assert branch.id == "sb_branch" + + +def recording_claim(reply): + seen = [] + + def claim(body, path): + seen.append(body) + return 200, reply + + FakeNode.routes[("POST", "/v1/claim")] = claim + return seen diff --git a/sdk/python/tests/test_proc.py b/sdk/python/tests/test_proc.py index 2f54b452..8cb4bd64 100644 --- a/sdk/python/tests/test_proc.py +++ b/sdk/python/tests/test_proc.py @@ -7,6 +7,51 @@ from cocoonsandbox.frames import FS_CHUNK +def test_spawn_returns_pid(monkeypatch): + sb, conn = fake_sandbox(monkeypatch, [{"type": "started", "pid": 41}]) + assert sb.spawn("sleep", "30") == 41 + op, fields = conn.sent[0] + assert op == "exec" and fields["detach"] is True and fields["argv"] == ["sleep", "30"] + + +def test_ps_lists_procs(monkeypatch): + procs = [{"pid": 41, "argv": ["sleep"], "detached": True, "state": "running", "started_at_epoch_secs": 1}] + sb, _ = fake_sandbox(monkeypatch, [{"type": "procs", "procs": procs}]) + assert sb.ps() == procs + + +def test_kill_sends_signal(monkeypatch): + sb, conn = fake_sandbox(monkeypatch, [{"type": "done"}]) + sb.kill(41, signal=15) + assert conn.sent[0] == ("kill", {"pid": 41, "signal": 15}) + + +def test_logs_running_proc_has_no_code(monkeypatch): + frames = [{"type": "stdout", "data": b"hello"}, {"type": "stderr", "data": b"oops"}, {"type": "done"}] + sb, _ = fake_sandbox(monkeypatch, frames) + out, errs = [], [] + assert sb.logs(41, on_stdout=out.append, on_stderr=errs.append) is None + assert out == [b"hello"] and errs == [b"oops"] + + +def test_attach_returns_exit_code(monkeypatch): + frames = [{"type": "stdout", "data": b"late"}, {"type": "exit", "code": 7}] + sb, _ = fake_sandbox(monkeypatch, frames) + out = [] + assert sb.attach(41, on_stdout=out.append) == 7 + assert out == [b"late"] + + +def test_run_pumps_stdin_while_reading_output(monkeypatch): + sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1") + blocking = BlockingStdinConn([{"type": "exit", "code": 0}], buffer_frames=1) + monkeypatch.setattr(sb, "_dial", lambda: blocking) + + assert sb.run(["cat"], stdin=b"x" * (FS_CHUNK * 3)) == 0 + assert [op for op, _ in blocking.sent].count("stdin") == 3 + assert blocking.sent[-1][0] == "stdin_close" + + class FakeConn: """Answers one RPC with a canned frame list.""" @@ -62,48 +107,3 @@ def fake_sandbox(monkeypatch, frames): conn = FakeConn(frames) monkeypatch.setattr(sb, "_dial", lambda: conn) return sb, conn - - -def test_spawn_returns_pid(monkeypatch): - sb, conn = fake_sandbox(monkeypatch, [{"type": "started", "pid": 41}]) - assert sb.spawn("sleep", "30") == 41 - op, fields = conn.sent[0] - assert op == "exec" and fields["detach"] is True and fields["argv"] == ["sleep", "30"] - - -def test_ps_lists_procs(monkeypatch): - procs = [{"pid": 41, "argv": ["sleep"], "detached": True, "state": "running", "started_at_epoch_secs": 1}] - sb, _ = fake_sandbox(monkeypatch, [{"type": "procs", "procs": procs}]) - assert sb.ps() == procs - - -def test_kill_sends_signal(monkeypatch): - sb, conn = fake_sandbox(monkeypatch, [{"type": "done"}]) - sb.kill(41, signal=15) - assert conn.sent[0] == ("kill", {"pid": 41, "signal": 15}) - - -def test_logs_running_proc_has_no_code(monkeypatch): - frames = [{"type": "stdout", "data": b"hello"}, {"type": "stderr", "data": b"oops"}, {"type": "done"}] - sb, _ = fake_sandbox(monkeypatch, frames) - out, errs = [], [] - assert sb.logs(41, on_stdout=out.append, on_stderr=errs.append) is None - assert out == [b"hello"] and errs == [b"oops"] - - -def test_attach_returns_exit_code(monkeypatch): - frames = [{"type": "stdout", "data": b"late"}, {"type": "exit", "code": 7}] - sb, _ = fake_sandbox(monkeypatch, frames) - out = [] - assert sb.attach(41, on_stdout=out.append) == 7 - assert out == [b"late"] - - -def test_run_pumps_stdin_while_reading_output(monkeypatch): - sb, conn = fake_sandbox(monkeypatch, [{"type": "exit", "code": 0}]) - blocking = BlockingStdinConn([{"type": "exit", "code": 0}], buffer_frames=1) - monkeypatch.setattr(sb, "_dial", lambda: blocking) - - assert sb.run(["cat"], stdin=b"x" * (FS_CHUNK * 3)) == 0 - assert [op for op, _ in blocking.sent].count("stdin") == 3 - assert blocking.sent[-1][0] == "stdin_close" diff --git a/sdk/python/tests/test_wire_binding.py b/sdk/python/tests/test_wire_binding.py index f7e1de60..0622c826 100644 --- a/sdk/python/tests/test_wire_binding.py +++ b/sdk/python/tests/test_wire_binding.py @@ -60,7 +60,7 @@ ), ("req_git_push", [{"type": "done"}], lambda sb, f: sb.git_push(f["path"], auth=f["auth"])), ("req_git_pull", [{"type": "done"}], lambda sb, f: sb.git_pull(f["path"], auth=f["auth"])), - ("req_pty_resize", [{"type": "done"}], lambda sb, f: _pty_stub(sb, f["pid"]).resize(f["cols"], f["rows"])), + ("req_pty_resize", [{"type": "done"}], lambda sb, f: Pty(sb, None, f["pid"]).resize(f["cols"], f["rows"])), ("req_exec_detach", [{"type": "started", "pid": 7}], lambda sb, f: sb.spawn(*f["argv"])), ("req_ps", [{"type": "procs", "procs": []}], lambda sb, f: sb.ps()), ("req_kill", [{"type": "done"}], lambda sb, f: sb.kill(f["pid"], signal=f["signal"])), @@ -68,14 +68,14 @@ ("req_attach", [{"type": "exit", "code": 0}], lambda sb, f: sb.attach(f["pid"])), ("req_fs_watch", [{"type": "ready"}], lambda sb, f: sb.watch(f["path"], recursive=f["recursive"]).close()), ("req_git_branch", [{"type": "done"}], lambda sb, f: sb.git_create_branch(f["path"], f["name"])), - ("req_session_rm", [{"type": "done"}], lambda sb, f: _session_stub(sb, f["id"]).close()), + ("req_session_rm", [{"type": "done"}], lambda sb, f: Session(sb, f["id"]).close()), ( "req_lsp_start", [{"type": "lsp_started", "server_id": "lsp-1"}], lambda sb, f: sb.start_lsp(f["language"], root=f["root"]), ), - ("req_lsp_request", [{"type": "ready"}], lambda sb, f: _lsp_stub(sb, f["server_id"]).request().close()), - ("req_lsp_stop", [{"type": "done"}], lambda sb, f: _lsp_stub(sb, f["server_id"]).stop()), + ("req_lsp_request", [{"type": "ready"}], lambda sb, f: Lsp(sb, f["server_id"]).request().close()), + ("req_lsp_stop", [{"type": "done"}], lambda sb, f: Lsp(sb, f["server_id"]).stop()), ("req_port_forward", [{"type": "ready"}], lambda sb, f: sb.dial_port(f["port"]).close()), ( "req_pty_open", @@ -88,6 +88,48 @@ UNSENT = {"req_session_create": {"id"}} +@pytest.mark.parametrize("stem,replies,invoke", CASES, ids=lambda c: c if isinstance(c, str) else "") +def test_call_site_matches_fixture(monkeypatch, stem, replies, invoke): + fixture = json.loads((FIXTURES / f"{stem}.json").read_text()) + sb, sent, thread = fake_sandbox(monkeypatch, replies) + invoke(sb, fixture) + thread.join(timeout=5) + + assert sent, "no frame reached the guest" + frame = sent[0] + assert frame["op"] == fixture["op"] + assert frame["v"] == PROTO_VERSION + for key, value in frame.items(): + assert key in fixture, f"field {key!r} not in the golden corpus (typo?)" + if key != "data": + assert value == fixture[key], f"field {key!r}: {value!r} != {fixture[key]!r}" + for key in fixture: + if key in UNSENT.get(stem, ()): + continue + assert key in frame, f"fixture field {key!r} was never emitted (dropped field?)" + + +def test_enum_value_sets_match_corpus(): + enums = json.loads((FIXTURES / "enums.json").read_text()) + assert set(enums["error_kind"]) == {"bad_request", "not_found", "unimplemented", "internal"} + assert set(enums["event_kind"]) == {"created", "modified", "deleted", "renamed"} + assert set(enums["file_kind"]) == {"file", "dir", "symlink", "other"} + assert set(enums["git_branch_action"]) == {"list", "create", "delete", "checkout"} + + +def test_git_branch_actions_come_from_the_corpus(monkeypatch): + enums = json.loads((FIXTURES / "enums.json").read_text()) + sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1") + sent = [] + monkeypatch.setattr(sb, "_dial", lambda: BranchActionConn(sent)) + + sb.git_branches("/w") + sb.git_create_branch("/w", "b") + sb.git_delete_branch("/w", "b") + sb.git_checkout("/w", "b") + assert set(sent) == set(enums["git_branch_action"]) + + class BranchActionConn: """Records the action each git_branch verb puts on the wire.""" @@ -114,19 +156,7 @@ def recv_until(self, *terminal): yield self.recv() -def _pty_stub(sb, pid): - return Pty(sb, None, pid) - - -def _session_stub(sb, id): - return Session(sb, id) - - -def _lsp_stub(sb, server_id): - return Lsp(sb, server_id) - - -def _fake_sandbox(monkeypatch, replies): +def fake_sandbox(monkeypatch, replies): """A Sandbox whose _dial yields a real Conn over a socketpair; a guest thread records inbound frames and answers the scripted replies.""" sent = [] @@ -154,45 +184,3 @@ def guest(): sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1") monkeypatch.setattr(sb, "_dial", lambda: Conn(client_sock, client_sock.makefile("rb"))) return sb, sent, thread - - -@pytest.mark.parametrize("stem,replies,invoke", CASES, ids=lambda c: c if isinstance(c, str) else "") -def test_call_site_matches_fixture(monkeypatch, stem, replies, invoke): - fixture = json.loads((FIXTURES / f"{stem}.json").read_text()) - sb, sent, thread = _fake_sandbox(monkeypatch, replies) - invoke(sb, fixture) - thread.join(timeout=5) - - assert sent, "no frame reached the guest" - frame = sent[0] - assert frame["op"] == fixture["op"] - assert frame["v"] == PROTO_VERSION - for key, value in frame.items(): - assert key in fixture, f"field {key!r} not in the golden corpus (typo?)" - if key != "data": - assert value == fixture[key], f"field {key!r}: {value!r} != {fixture[key]!r}" - for key in fixture: - if key in UNSENT.get(stem, ()): - continue - assert key in frame, f"fixture field {key!r} was never emitted (dropped field?)" - - -def test_enum_value_sets_match_corpus(): - enums = json.loads((FIXTURES / "enums.json").read_text()) - assert set(enums["error_kind"]) == {"bad_request", "not_found", "unimplemented", "internal"} - assert set(enums["event_kind"]) == {"created", "modified", "deleted", "renamed"} - assert set(enums["file_kind"]) == {"file", "dir", "symlink", "other"} - assert set(enums["git_branch_action"]) == {"list", "create", "delete", "checkout"} - - -def test_git_branch_actions_come_from_the_corpus(monkeypatch): - enums = json.loads((FIXTURES / "enums.json").read_text()) - sb = Sandbox(client=Client("127.0.0.1:1"), id="sb_1", token="tok", owner="127.0.0.1:1") - sent = [] - monkeypatch.setattr(sb, "_dial", lambda: BranchActionConn(sent)) - - sb.git_branches("/w") - sb.git_create_branch("/w", "b") - sb.git_delete_branch("/w", "b") - sb.git_checkout("/w", "b") - assert set(sent) == set(enums["git_branch_action"]) diff --git a/silkd/src/find.rs b/silkd/src/find.rs index ca054fc1..104a2039 100644 --- a/silkd/src/find.rs +++ b/silkd/src/find.rs @@ -12,8 +12,8 @@ use tokio::sync::{Semaphore, SemaphorePermit, mpsc}; use crate::proto::{self, ErrorKind, Response, err_frame}; -/// Size above which find skips a file as binary or huge. -const FIND_MAX_FILE: u64 = 8 * 1024 * 1024; +/// Size above which find skips a file unread. +pub const FIND_MAX_FILE: u64 = 8 * 1024 * 1024; /// Match frames in flight between the walking thread and the writer. const MATCH_QUEUE: usize = 256; diff --git a/silkd/src/session.rs b/silkd/src/session.rs index 3b63db03..b0f2ecfc 100644 --- a/silkd/src/session.rs +++ b/silkd/src/session.rs @@ -79,7 +79,7 @@ impl Table { } for (k, v) in env { init.push_str("export "); - init.push_str(k); + shell_quote_into(&mut init, k); init.push('='); shell_quote_into(&mut init, v); init.push('\n'); @@ -89,7 +89,7 @@ impl Table { io.converse::(&init, None).await?; } - // reserve the id atomically: a concurrent create must not orphan the loser's shell. + // reserve the id under the table lock so a concurrent create with the same id sees AlreadyExists. let mut map = sysutil::lock(&self.inner); if map.contains_key(&id) { return Err(io::Error::new( @@ -219,7 +219,7 @@ impl Io { if let Some(pos) = memchr::memmem::find(&self.acc, mb) { emit(&mut out, &mut self.frame, &self.acc[..pos]).await; self.acc.drain(..pos + mb.len()); - while !self.acc.contains(&b'\n') && self.acc.len() < EXIT_TAIL_MAX { + while memchr::memchr(b'\n', &self.acc).is_none() && self.acc.len() < EXIT_TAIL_MAX { let m = self.stdout.read(&mut self.buf).await?; if m == 0 { break; @@ -227,11 +227,7 @@ impl Io { self.acc.extend_from_slice(&self.buf[..m]); } // parse only the exit-code line: a background writer can land bytes after the newline. - let end = self - .acc - .iter() - .position(|&b| b == b'\n') - .unwrap_or(self.acc.len()); + let end = memchr::memchr(b'\n', &self.acc).unwrap_or(self.acc.len()); return Ok(String::from_utf8_lossy(&self.acc[..end]) .trim() .parse() diff --git a/silkd/src/sysutil.rs b/silkd/src/sysutil.rs index 21f20878..ca42a065 100644 --- a/silkd/src/sysutil.rs +++ b/silkd/src/sysutil.rs @@ -1,7 +1,9 @@ //! Small OS helpers; the crate's unsafe work lives here, except pty.rs's pre_exec registration. +use std::ffi::{CStr, CString}; use std::io::Read; use std::os::fd::{AsRawFd, FromRawFd, OwnedFd, RawFd}; +use std::os::unix::process::ExitStatusExt; use std::process::ExitStatus; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{LazyLock, Mutex}; @@ -46,26 +48,18 @@ pub fn rand_token() -> String { .collect() } -/// SIGKILLs the group led by `pgid`, so a session's external command dies with its shell. +/// SIGKILLs the group led by `pgid`, so a session's external command dies with its shell; a synthetic id misses with ESRCH. pub fn kill_group(pgid: u32) { - // kill(-0) would target silkd's own group; a synthetic id passes the guard and misses with ESRCH. - if !is_valid_pid(pgid) { - return; + if is_valid_pid(pgid) { + send_signal(-(pgid as libc::pid_t), libc::SIGKILL); } - // SAFETY: kill(2) takes no pointers; the guards above keep the pid_t cast - // in range and away from silkd's own group. - unsafe { libc::kill(-(pgid as libc::pid_t), libc::SIGKILL) }; } /// Sends `sig` to `pid`, ignoring ESRCH against a just-exited pid. pub fn signal_pid(pid: u32, sig: i32) { - // pid 0 means silkd's whole process group to kill(2). - if !is_valid_pid(pid) { - return; + if is_valid_pid(pid) { + send_signal(pid as libc::pid_t, sig); } - // SAFETY: kill(2) takes no pointers; the guards above keep the pid_t cast - // in range and away from silkd's own group. - unsafe { libc::kill(pid as libc::pid_t, sig) }; } /// The environment every exec starts from; the proxy snapshot rides only on the no-network lane. @@ -97,11 +91,8 @@ pub fn openpty(cols: u16, rows: u16) -> std::io::Result<(OwnedFd, OwnedFd)> { ws_xpixel: 0, ws_ypixel: 0, }; - // SAFETY: openpty writes two valid fds into master/slave on success; ws is a - // fully-initialized winsize (openpty only reads it). We take ownership of - // both fds immediately. A raw `*mut` pointer for winp works across libc's - // platform-varying signature (*mut on macOS, *const on Linux) without the - // &mut that clippy flags as unnecessary on Linux. + // a raw `*mut` for winp fits both libc signatures (*mut on macOS, *const on Linux). + // SAFETY: openpty writes two valid fds into master/slave on success and only reads ws; both fds are owned immediately. let rc = unsafe { libc::openpty( &mut master, @@ -185,7 +176,6 @@ pub fn lock(m: &Mutex) -> std::sync::MutexGuard<'_, T> { /// Maps a wait() status to the shell convention (128 + signal when killed). pub fn exit_code(status: ExitStatus) -> i32 { - use std::os::unix::process::ExitStatusExt; if let Some(code) = status.code() { return code; } @@ -212,11 +202,16 @@ fn fill_random(b: &mut [u8]) -> bool { .is_ok() } -/// Rejects pid 0 and anything that would go negative through the pid_t cast. +/// Rejects pid 0, which kill(2) reads as silkd's own process group, and anything that would go negative through the pid_t cast. fn is_valid_pid(id: u32) -> bool { id != 0 && id <= i32::MAX as u32 } +fn send_signal(target: libc::pid_t, sig: i32) { + // SAFETY: kill(2) takes no pointers; every caller vets the target with is_valid_pid first. + unsafe { libc::kill(target, sig) }; +} + fn align_proxy_env_for(cmd: &mut Command, nic: bool) { if !nic { return; @@ -258,9 +253,7 @@ fn snapshot_proxy(get: impl Fn(&str) -> Option) -> Vec<(&'static str, St } fn lookup_user(user: &str) -> Result<(u32, u32, String), String> { - use std::ffi::CString; let cname = CString::new(user).map_err(|_| format!("invalid user {user:?}"))?; - // hold NSS_LOCK across every read of the static buffer, else a concurrent lookup clobbers it. let _guard = lock(&NSS_LOCK); // SAFETY: cname is a live NUL-terminated CString for the call. let pw = unsafe { libc::getpwnam(cname.as_ptr()) }; @@ -271,7 +264,7 @@ fn lookup_user(user: &str) -> Result<(u32, u32, String), String> { // a valid passwd whose pw_dir is a NUL-terminated string it owns. let pw = unsafe { &*pw }; // SAFETY: pw_dir is NUL-terminated and stays valid while NSS_LOCK is held. - let home = unsafe { std::ffi::CStr::from_ptr(pw.pw_dir) } + let home = unsafe { CStr::from_ptr(pw.pw_dir) } .to_string_lossy() .into_owned(); Ok((pw.pw_uid, pw.pw_gid, home)) diff --git a/silkd/tests/common/mod.rs b/silkd/tests/common/mod.rs index 28901d11..895e1c19 100644 --- a/silkd/tests/common/mod.rs +++ b/silkd/tests/common/mod.rs @@ -1,17 +1,20 @@ -//! Shared harness: drives silkd's server over an in-memory duplex as a relayed host connection would. -//! Each test binary compiles it separately, so #![allow(dead_code)] covers the helpers one binary skips. +//! Shared harness: drives silkd's server over an in-memory duplex as a relayed host connection would; each test binary compiles it separately, hence allow(dead_code). #![allow(clippy::unwrap_used, clippy::expect_used)] #![allow(dead_code)] use std::sync::Arc; +use std::time::Duration; use base64::Engine; -use serde_json::Value; +use serde_json::{Value, json}; use silkd::server::State; use tokio::io::{ AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream, Lines, ReadHalf, WriteHalf, }; use tokio::task::JoinHandle; +use tokio::time::timeout; + +pub const DEADLINE: Duration = Duration::from_secs(30); pub type FrameWriter = WriteHalf; pub type FrameLines = Lines>>; @@ -25,6 +28,20 @@ pub fn connect(state: &Arc) -> (FrameWriter, FrameLines, JoinHandle Value { + let line = timeout(DEADLINE, lines.next_line()) + .await + .expect("deadline") + .unwrap() + .expect("stream closed"); + serde_json::from_str(&line).unwrap() +} + pub async fn roundtrip(request_line: &str) -> Vec { request_on(&Arc::new(State::new()), &[request_line.to_string()]).await } @@ -82,7 +99,7 @@ pub fn stdout_body(frames: &[Value]) -> String { pub fn data_frames(bytes: &[u8]) -> Vec { let mut lines: Vec = bytes .chunks(16 * 1024) - .map(|c| serde_json::json!({"op":"data","data":b64(c)}).to_string()) + .map(|c| json!({"op":"data","data":b64(c)}).to_string()) .collect(); lines.push(r#"{"op":"data_end"}"#.to_string()); lines diff --git a/silkd/tests/exec_e2e.rs b/silkd/tests/exec_e2e.rs index 89c04f59..d902f9ad 100644 --- a/silkd/tests/exec_e2e.rs +++ b/silkd/tests/exec_e2e.rs @@ -6,21 +6,51 @@ mod common; use std::sync::Arc; use std::time::Duration; +use serde_json::{Value, json}; use silkd::server::State; -use common::{decode, exchange, one, roundtrip, stdout_body, type_of}; +use common::{ + b64, connect, decode, exchange, next_frame, one, roundtrip, send, stdout_body, type_of, +}; async fn detached_pid(state: &Arc, script: &str) -> u64 { - let argv = serde_json::json!(["/bin/sh", "-c", script]); + let argv = json!(["/bin/sh", "-c", script]); let started = one( state, - &serde_json::json!({"op":"exec","argv":argv,"detach":true}).to_string(), + &json!({"op":"exec","argv":argv,"detach":true}).to_string(), ) .await; assert_eq!(type_of(&started[0]), "started"); started[0]["pid"].as_u64().unwrap() } +async fn wait_for_exit(state: &Arc, pid: u64) -> Vec { + for _ in 0..250 { + let logs = one(state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; + if logs.iter().any(|f| type_of(f) == "exit") { + return logs; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("pid {pid} never exited"); +} + +async fn wait_listed(state: &Arc, pid: u64) -> Value { + for _ in 0..250 { + let ps = one(state, r#"{"op":"ps"}"#).await; + if let Some(p) = ps[0]["procs"] + .as_array() + .unwrap() + .iter() + .find(|p| p["pid"].as_u64() == Some(pid)) + { + return p.clone(); + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + panic!("pid {pid} never appeared in ps"); +} + #[tokio::test] async fn exec_streams_stdout_then_exit() { let frames = roundtrip(r#"{"op":"exec","argv":["/bin/echo","-n","hello"]}"#).await; @@ -133,13 +163,7 @@ async fn attach_streams_live_output_then_exit() { async fn attach_to_exited_process_returns_exit_immediately() { let state = Arc::new(State::new()); let pid = detached_pid(&state, "echo done-fast").await; - for _ in 0..100 { - let logs = one(&state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; - if logs.iter().any(|f| type_of(f) == "exit") { - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } + wait_for_exit(&state, pid).await; let frames = tokio::time::timeout( Duration::from_secs(3), one(&state, &format!(r#"{{"op":"attach","pid":{pid}}}"#)), @@ -161,30 +185,10 @@ async fn attach_unknown_pid_is_not_found() { async fn kill_actually_terminates_a_live_process() { let state = Arc::new(State::new()); let pid = detached_pid(&state, "sleep 30").await; - for _ in 0..50 { - let ps = one(&state, r#"{"op":"ps"}"#).await; - if ps[0]["procs"] - .as_array() - .unwrap() - .iter() - .any(|p| p["pid"].as_u64() == Some(pid)) - { - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } + wait_listed(&state, pid).await; let killed = one(&state, &format!(r#"{{"op":"kill","pid":{pid}}}"#)).await; assert_eq!(type_of(&killed[0]), "done"); - let mut exited = false; - for _ in 0..150 { - let logs = one(&state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; - if logs.iter().any(|f| type_of(f) == "exit") { - exited = true; - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } - assert!(exited, "killed process never reached exited state"); + wait_for_exit(&state, pid).await; } #[tokio::test] @@ -207,7 +211,7 @@ async fn exec_missing_binary_is_an_internal_error() { async fn exec_stdin_is_piped_to_the_child() { let frames = exchange(&[ r#"{"op":"exec","argv":["/bin/cat"]}"#.to_string(), - serde_json::json!({"op":"stdin","data":common::b64(b"piped-in\n")}).to_string(), + json!({"op":"stdin","data":b64(b"piped-in\n")}).to_string(), r#"{"op":"stdin_close"}"#.to_string(), ]) .await; @@ -218,19 +222,8 @@ async fn exec_stdin_is_piped_to_the_child() { #[tokio::test] async fn kill_of_an_exited_process_is_a_noop_success() { let state = Arc::new(State::new()); - let started = one( - &state, - r#"{"op":"exec","argv":["/bin/echo","x"],"detach":true}"#, - ) - .await; - let pid = started[0]["pid"].as_u64().unwrap(); - for _ in 0..100 { - let logs = one(&state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; - if logs.iter().any(|f| type_of(f) == "exit") { - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } + let pid = detached_pid(&state, "echo x").await; + wait_for_exit(&state, pid).await; let killed = one(&state, &format!(r#"{{"op":"kill","pid":{pid}}}"#)).await; assert_eq!( type_of(&killed[0]), @@ -242,68 +235,33 @@ async fn kill_of_an_exited_process_is_a_noop_success() { #[tokio::test] async fn detached_exec_is_listed_then_logs_replay_output_and_exit() { let state = Arc::new(State::new()); - let started = one( - &state, - r#"{"op":"exec","argv":["/bin/echo","detached-hi"],"detach":true}"#, - ) - .await; - assert_eq!(type_of(&started[0]), "started"); - let pid = started[0]["pid"].as_u64().unwrap(); - - let mut listed = false; - for _ in 0..250 { - let ps = one(&state, r#"{"op":"ps"}"#).await; - if ps[0]["procs"] - .as_array() - .unwrap() - .iter() - .any(|p| p["pid"].as_u64() == Some(pid) && p["detached"] == true) - { - listed = true; - break; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } - assert!(listed, "detached pid {pid} never appeared in ps"); - - for _ in 0..250 { - let logs = one(&state, &format!(r#"{{"op":"logs","pid":{pid}}}"#)).await; - if logs.iter().any(|f| type_of(f) == "exit") { - assert_eq!(stdout_body(&logs), "detached-hi\n"); - return; - } - tokio::time::sleep(Duration::from_millis(20)).await; - } - panic!("logs never reported an exit for pid {pid}"); + let pid = detached_pid(&state, "echo detached-hi").await; + let listed = wait_listed(&state, pid).await; + assert_eq!(listed["detached"], true, "{listed:?}"); + let logs = wait_for_exit(&state, pid).await; + assert_eq!(stdout_body(&logs), "detached-hi\n"); } #[tokio::test] async fn disconnect_during_drain_publishes_real_exit_code() { - use tokio::io::AsyncWriteExt; - let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); - let req = serde_json::json!({ - "op": "exec", - "argv": ["/bin/sh", "-c", "{ sleep 0.5; echo late; sleep 30; } & exit 7"] - }) - .to_string(); - cw.write_all(req.as_bytes()).await.unwrap(); - cw.write_all(b"\n").await.unwrap(); + let (mut cw, mut lines, _) = connect(&state); + send( + &mut cw, + json!({ + "op": "exec", + "argv": ["/bin/sh", "-c", "{ sleep 0.5; echo late; sleep 30; } & exit 7"] + }), + ) + .await; - let started: serde_json::Value = - serde_json::from_str(&lines.next_line().await.unwrap().unwrap()).unwrap(); + let started = next_frame(&mut lines).await; assert_eq!(type_of(&started), "started"); let pid = started["pid"].as_u64().unwrap(); let st = Arc::clone(&state); - let attach = tokio::spawn(async move { - one( - &st, - &serde_json::json!({"op":"attach","pid":pid}).to_string(), - ) - .await - }); + let attach = + tokio::spawn(async move { one(&st, &json!({"op":"attach","pid":pid}).to_string()).await }); tokio::time::sleep(Duration::from_millis(200)).await; drop(lines); drop(cw); diff --git a/silkd/tests/find_e2e.rs b/silkd/tests/find_e2e.rs index 45f1a99a..6684d253 100644 --- a/silkd/tests/find_e2e.rs +++ b/silkd/tests/find_e2e.rs @@ -3,20 +3,17 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; +use std::os::unix::fs::PermissionsExt; use std::sync::Arc; use serde_json::{Value, json}; use silkd::server::State; use tokio::io::AsyncWriteExt; -use common::{exchange, type_of}; - -async fn write_file(dir: &std::path::Path, name: &str, body: &str) { - tokio::fs::write(dir.join(name), body).await.unwrap(); -} +use common::{connect, exchange, type_of}; async fn find_frames(req: Value) -> Vec { - let (mut cw, mut out, handle) = common::connect(&Arc::new(State::new())); + let (mut cw, mut out, handle) = connect(&Arc::new(State::new())); cw.write_all(format!("{req}\n").as_bytes()).await.unwrap(); let mut frames = Vec::new(); while let Some(l) = out.next_line().await.unwrap() { @@ -35,10 +32,19 @@ async fn find_frames(req: Value) -> Vec { #[tokio::test] async fn find_streams_line_matches() { let dir = tempfile::tempdir().unwrap(); - write_file(dir.path(), "a.rs", "fn main() {\n // TODO: fix\n}\n").await; - write_file(dir.path(), "b.txt", "TODO ignore me (wrong glob)\n").await; + tokio::fs::write( + dir.path().join("a.rs"), + "fn main() {\n // TODO: fix\n}\n", + ) + .await + .unwrap(); + tokio::fs::write(dir.path().join("b.txt"), "TODO ignore me (wrong glob)\n") + .await + .unwrap(); tokio::fs::create_dir(dir.path().join("sub")).await.unwrap(); - write_file(&dir.path().join("sub"), "c.rs", "// TODO nested\n").await; + tokio::fs::write(dir.path().join("sub/c.rs"), "// TODO nested\n") + .await + .unwrap(); let frames = find_frames(json!({ "op": "fs_find", @@ -63,8 +69,12 @@ async fn find_streams_line_matches() { #[tokio::test] async fn find_glob_is_a_real_glob_not_a_substring() { let dir = tempfile::tempdir().unwrap(); - write_file(dir.path(), "a.rs", "TODO\n").await; - write_file(dir.path(), "a.rs.bak", "TODO\n").await; + tokio::fs::write(dir.path().join("a.rs"), "TODO\n") + .await + .unwrap(); + tokio::fs::write(dir.path().join("a.rs.bak"), "TODO\n") + .await + .unwrap(); let frames = find_frames(json!({ "op": "fs_find", @@ -144,7 +154,7 @@ async fn replace_rewrites_and_counts() { async fn replace_skips_a_file_over_the_size_bound() { let dir = tempfile::tempdir().unwrap(); let big = dir.path().join("big.log"); - let body = format!("foo\n{}", "x".repeat(8 * 1024 * 1024)); + let body = format!("foo\n{}", "x".repeat(silkd::find::FIND_MAX_FILE as usize)); tokio::fs::write(&big, &body).await.unwrap(); let frames = exchange(&[json!({ @@ -172,7 +182,7 @@ async fn find_skips_a_file_over_the_size_bound() { let dir = tempfile::tempdir().unwrap(); tokio::fs::write( dir.path().join("big.log"), - format!("foo\n{}", "x".repeat(8 * 1024 * 1024)), + format!("foo\n{}", "x".repeat(silkd::find::FIND_MAX_FILE as usize)), ) .await .unwrap(); @@ -190,7 +200,6 @@ async fn find_skips_a_file_over_the_size_bound() { #[tokio::test] async fn replace_preserves_exec_bit() { - use std::os::unix::fs::PermissionsExt; let dir = tempfile::tempdir().unwrap(); let script = dir.path().join("run.sh"); tokio::fs::write(&script, "#!/bin/sh\nfoo\n").await.unwrap(); diff --git a/silkd/tests/forward_e2e.rs b/silkd/tests/forward_e2e.rs index 24316d7c..9226a9b7 100644 --- a/silkd/tests/forward_e2e.rs +++ b/silkd/tests/forward_e2e.rs @@ -4,16 +4,14 @@ mod common; use std::sync::Arc; -use std::time::Duration; -use base64::Engine; -use base64::engine::general_purpose::STANDARD; +use serde_json::json; use silkd::server::State; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; use tokio::time::timeout; -const DEADLINE: Duration = Duration::from_secs(30); +use common::{DEADLINE, FrameLines, FrameWriter, b64, connect, decode, next_frame, send, type_of}; async fn echo_listener() -> u16 { let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); @@ -38,10 +36,12 @@ async fn echo_listener() -> u16 { port } -fn data_payload(line: &str) -> Option<&str> { - let start = line.find("\"data\":\"")? + "\"data\":\"".len(); - let rest = &line[start..]; - Some(&rest[..rest.find('"')?]) +async fn forwarded(state: &Arc, port: u16) -> (FrameWriter, FrameLines) { + let (mut cw, mut lines, _) = connect(state); + send(&mut cw, json!({"v":1,"op":"port_forward","port":port})).await; + let ready = next_frame(&mut lines).await; + assert_eq!(type_of(&ready), "ready", "got {ready}"); + (cw, lines) } #[tokio::test] @@ -49,28 +49,16 @@ async fn forward_round_trips_and_done_on_server_close() { timeout(DEADLINE, async { let port = echo_listener().await; let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); - - cw.write_all(format!("{{\"v\":1,\"op\":\"port_forward\",\"port\":{port}}}\n").as_bytes()) - .await - .expect("send request"); - let ready = lines.next_line().await.expect("read").expect("ready"); - assert!(ready.contains("\"ready\""), "got {ready}"); - - cw.write_all(b"{\"v\":1,\"op\":\"data\",\"data\":\"aGk=\"}\n") - .await - .expect("send data"); - let echoed = lines.next_line().await.expect("read").expect("data"); - assert!( - echoed.contains("\"data\"") && echoed.contains("aGk="), - "got {echoed}" - ); - - cw.write_all(b"{\"v\":1,\"op\":\"data_end\"}\n") - .await - .expect("send data_end"); - let done = lines.next_line().await.expect("read").expect("done"); - assert!(done.contains("\"done\""), "got {done}"); + let (mut cw, mut lines) = forwarded(&state, port).await; + + send(&mut cw, json!({"v":1,"op":"data","data":b64(b"hi")})).await; + let echoed = next_frame(&mut lines).await; + assert_eq!(type_of(&echoed), "data", "got {echoed}"); + assert_eq!(decode(&echoed), b"hi"); + + send(&mut cw, json!({"v":1,"op":"data_end"})).await; + let done = next_frame(&mut lines).await; + assert_eq!(type_of(&done), "done", "got {done}"); }) .await .expect("test deadline"); @@ -84,34 +72,21 @@ async fn forward_bidirectional_bulk_no_deadlock() { timeout(DEADLINE, async { let port = echo_listener().await; let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); - - cw.write_all(format!("{{\"v\":1,\"op\":\"port_forward\",\"port\":{port}}}\n").as_bytes()) - .await - .expect("send request"); - let ready = lines.next_line().await.expect("read").expect("ready"); - assert!(ready.contains("\"ready\""), "got {ready}"); + let (mut cw, mut lines) = forwarded(&state, port).await; - let payload = STANDARD.encode(vec![b'x'; CHUNK]); - let frame = format!("{{\"v\":1,\"op\":\"data\",\"data\":\"{payload}\"}}\n"); + let frame = json!({"v":1,"op":"data","data":b64(&[b'x'; CHUNK])}).to_string() + "\n"; let writer = tokio::spawn(async move { for _ in 0..FRAMES { cw.write_all(frame.as_bytes()).await.expect("write frame"); } - cw.write_all(b"{\"v\":1,\"op\":\"data_end\"}\n") - .await - .expect("data_end"); + send(&mut cw, json!({"v":1,"op":"data_end"})).await; }); let mut got = 0usize; while got < CHUNK * FRAMES { - let line = lines - .next_line() - .await - .expect("read") - .expect("stream ended before all bytes echoed"); - if let Some(b64) = data_payload(&line) { - got += STANDARD.decode(b64).expect("decode").len(); + let frame = next_frame(&mut lines).await; + if type_of(&frame) == "data" { + got += decode(&frame).len(); } } writer.await.expect("writer"); @@ -125,16 +100,12 @@ async fn forward_bidirectional_bulk_no_deadlock() { async fn forward_refused_port_is_not_found() { timeout(DEADLINE, async { let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); - - cw.write_all(b"{\"v\":1,\"op\":\"port_forward\",\"port\":1}\n") - .await - .expect("send request"); - let err = lines.next_line().await.expect("read").expect("error"); - assert!( - err.contains("\"error\"") && err.contains("not_found"), - "got {err}" - ); + let (mut cw, mut lines, _) = connect(&state); + + send(&mut cw, json!({"v":1,"op":"port_forward","port":1})).await; + let err = next_frame(&mut lines).await; + assert_eq!(type_of(&err), "error", "got {err}"); + assert_eq!(err["kind"], "not_found", "got {err}"); }) .await .expect("test deadline"); @@ -145,22 +116,12 @@ async fn forward_stray_frame_is_bad_request() { timeout(DEADLINE, async { let port = echo_listener().await; let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); - - cw.write_all(format!("{{\"v\":1,\"op\":\"port_forward\",\"port\":{port}}}\n").as_bytes()) - .await - .expect("send request"); - let ready = lines.next_line().await.expect("read").expect("ready"); - assert!(ready.contains("\"ready\""), "got {ready}"); - - cw.write_all(b"{\"v\":1,\"op\":\"ps\"}\n") - .await - .expect("send stray"); - let err = lines.next_line().await.expect("read").expect("error"); - assert!( - err.contains("\"error\"") && err.contains("bad_request"), - "got {err}" - ); + let (mut cw, mut lines) = forwarded(&state, port).await; + + send(&mut cw, json!({"v":1,"op":"ps"})).await; + let err = next_frame(&mut lines).await; + assert_eq!(type_of(&err), "error", "got {err}"); + assert_eq!(err["kind"], "bad_request", "got {err}"); }) .await .expect("test deadline"); diff --git a/silkd/tests/fs_e2e.rs b/silkd/tests/fs_e2e.rs index 917c5960..2f92c712 100644 --- a/silkd/tests/fs_e2e.rs +++ b/silkd/tests/fs_e2e.rs @@ -3,9 +3,11 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; +use std::os::unix::fs::PermissionsExt; + use serde_json::json; -use common::{b64, exchange, type_of}; +use common::{b64, exchange, payload, type_of}; #[tokio::test] async fn write_then_read_roundtrips_bytes() { @@ -26,7 +28,7 @@ async fn write_then_read_roundtrips_bytes() { ); let read = exchange(&[json!({"op":"fs_read","path":path}).to_string()]).await; - assert_eq!(common::payload(&read, "data"), b"silk file body"); + assert_eq!(payload(&read, "data"), b"silk file body"); assert_eq!(type_of(read.last().unwrap()), "done"); } @@ -172,7 +174,6 @@ async fn truncated_write_leaves_no_file_and_reports_error() { #[tokio::test] async fn overwrite_preserves_destination_mode() { - use std::os::unix::fs::PermissionsExt; let dir = tempfile::tempdir().unwrap(); let path = dir.path().join("script.sh"); std::fs::write(&path, b"old").unwrap(); diff --git a/silkd/tests/git_e2e.rs b/silkd/tests/git_e2e.rs index 3d63ce6d..4851573f 100644 --- a/silkd/tests/git_e2e.rs +++ b/silkd/tests/git_e2e.rs @@ -3,7 +3,7 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; -use serde_json::{Value, json}; +use serde_json::json; use common::{exchange, type_of}; @@ -29,10 +29,6 @@ fn init_repo() -> tempfile::TempDir { dir } -fn last(frames: &[Value]) -> &Value { - frames.last().unwrap() -} - #[tokio::test] async fn add_commit_status_reports_structured_state() { let repo = init_repo(); @@ -50,7 +46,7 @@ async fn add_commit_status_reports_structured_state() { ); let add = exchange(&[json!({"op":"git_add","path":p,"files":["a.txt"]}).to_string()]).await; - assert_eq!(type_of(last(&add)), "done"); + assert_eq!(type_of(add.last().unwrap()), "done"); let commit = exchange(&[json!({ "op":"git_commit","path":p,"message":"first","author":"Dev " }) @@ -78,7 +74,7 @@ async fn branch_create_list_checkout() { json!({"op":"git_branch","path":p,"action":"create","name":"feature"}).to_string(), ]) .await; - assert_eq!(type_of(last(&created)), "done"); + assert_eq!(type_of(created.last().unwrap()), "done"); let list = exchange(&[json!({"op":"git_branch","path":p,"action":"list"}).to_string()]).await; assert_eq!(type_of(&list[0]), "git_branches"); @@ -95,7 +91,7 @@ async fn branch_create_list_checkout() { json!({"op":"git_branch","path":p,"action":"checkout","name":"feature"}).to_string(), ]) .await; - assert_eq!(type_of(last(&co)), "done"); + assert_eq!(type_of(co.last().unwrap()), "done"); } #[tokio::test] @@ -135,7 +131,7 @@ async fn clone_lane_behavior() { .to_string()]) .await; silkd::net::override_egress_for_tests(None); - assert_eq!(type_of(last(&ok)), "done", "clone failed: {ok:?}"); + assert_eq!(type_of(ok.last().unwrap()), "done", "clone failed: {ok:?}"); assert!(target.join("r.txt").exists()); } @@ -183,7 +179,7 @@ async fn commit_works_without_preconfigured_identity() { std::fs::write(dir.path().join("f"), "x").unwrap(); let p = dir.path().to_str().unwrap(); let add = exchange(&[json!({"op":"git_add","path":p,"files":["f"]}).to_string()]).await; - assert_eq!(type_of(last(&add)), "done"); + assert_eq!(type_of(add.last().unwrap()), "done"); let commit = exchange(&[json!({ "op":"git_commit","path":p,"message":"m","author":"Dev " }) diff --git a/silkd/tests/lsp_e2e.rs b/silkd/tests/lsp_e2e.rs index 276a538b..e45184e4 100644 --- a/silkd/tests/lsp_e2e.rs +++ b/silkd/tests/lsp_e2e.rs @@ -4,8 +4,8 @@ mod common; use std::io::Write; +use std::os::unix::fs::PermissionsExt; use std::sync::Arc; -use std::time::Duration; use serde_json::json; use silkd::server::State; @@ -13,33 +13,56 @@ use tempfile::TempDir; use tokio::io::AsyncWriteExt; use tokio::time::timeout; -use common::{b64, connect, decode, one, type_of}; - -const DEADLINE: Duration = Duration::from_secs(30); - -static ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); +use common::{DEADLINE, b64, connect, decode, next_frame, one, send, type_of}; const FAKE_SERVER: &str = r#"#!/bin/sh IFS= read -r line printf 'reply:%s\n' "$line" "#; +static ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + fn manifest_env(server_body: &str) -> TempDir { let dir = tempfile::tempdir().unwrap(); let bin = dir.path().join("fake-lsp"); let mut f = std::fs::File::create(&bin).unwrap(); f.write_all(server_body.as_bytes()).unwrap(); - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap(); - } + std::fs::set_permissions(&bin, std::fs::Permissions::from_mode(0o755)).unwrap(); // SAFETY: ENV_LOCK, held by the caller for its whole test, serializes every // writer and reader of SILKD_LSP_DIR in this binary. unsafe { std::env::set_var("SILKD_LSP_DIR", dir.path()) }; dir } +async fn started_server(lang: &str, server_body: &str) -> (TempDir, Arc, String) { + let env = manifest_env(server_body); + std::fs::write( + env.path().join(lang), + env.path().join("fake-lsp").to_string_lossy().as_bytes(), + ) + .unwrap(); + let state = Arc::new(State::new()); + let start = one( + &state, + &json!({"op":"lsp_start","language":lang}).to_string(), + ) + .await; + let server_id = start.last().unwrap()["server_id"] + .as_str() + .unwrap() + .to_string(); + (env, state, server_id) +} + +async fn assert_stopped(state: &Arc, server_id: &str) { + let gone = one( + state, + &json!({"op":"lsp_stop","server_id":server_id}).to_string(), + ) + .await; + assert_eq!(gone.last().unwrap()["kind"], "not_found", "{gone:?}"); +} + #[tokio::test] async fn lsp_start_missing_manifest_is_not_found() { let _env_lock = ENV_LOCK.lock().await; @@ -59,6 +82,7 @@ async fn lsp_start_missing_manifest_is_not_found() { #[tokio::test] async fn lsp_start_language_name_cannot_escape() { + let _env_lock = ENV_LOCK.lock().await; for bad in ["../etc/passwd", "a/b", ".."] { let frames = one( &Arc::new(State::new()), @@ -76,118 +100,40 @@ async fn lsp_start_language_name_cannot_escape() { #[tokio::test] async fn lsp_broker_relays_to_the_server() { let _env_lock = ENV_LOCK.lock().await; - let env = manifest_env(FAKE_SERVER); - let bin = env.path().join("fake-lsp"); - std::fs::write( - env.path().join("faketest"), - bin.to_string_lossy().as_bytes(), - ) - .unwrap(); - - let state = Arc::new(State::new()); - let start = one( - &state, - &json!({"op":"lsp_start","language":"faketest"}).to_string(), - ) - .await; - let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); + let (_env, state, server_id) = started_server("faketest", FAKE_SERVER).await; let (mut cw, mut out, handle) = connect(&state); - cw.write_all( - json!({"op":"lsp_request","server_id":server_id}) - .to_string() - .as_bytes(), - ) - .await - .unwrap(); - cw.write_all(b"\n").await.unwrap(); - let ready: serde_json::Value = - serde_json::from_str(&out.next_line().await.unwrap().unwrap()).unwrap(); - assert_eq!(type_of(&ready), "ready"); - cw.write_all( - json!({"op":"data","data":b64(b"ping\n")}) - .to_string() - .as_bytes(), - ) - .await - .unwrap(); - cw.write_all(b"\n").await.unwrap(); + send(&mut cw, json!({"op":"lsp_request","server_id":server_id})).await; + assert_eq!(type_of(&next_frame(&mut out).await), "ready"); + send(&mut cw, json!({"op":"data","data":b64(b"ping\n")})).await; - let reply = timeout(DEADLINE, out.next_line()) - .await - .expect("deadline") - .unwrap() - .unwrap(); - let frame: serde_json::Value = serde_json::from_str(&reply).unwrap(); + let frame = next_frame(&mut out).await; assert_eq!(type_of(&frame), "data"); assert_eq!(decode(&frame), b"reply:ping\n"); - - let done = timeout(DEADLINE, out.next_line()) - .await - .expect("deadline") - .unwrap() - .unwrap(); - assert_eq!( - type_of(&serde_json::from_str::(&done).unwrap()), - "done" - ); + assert_eq!(type_of(&next_frame(&mut out).await), "done"); cw.shutdown().await.unwrap(); let _ = handle.await; - let gone = one( - &state, - &json!({"op":"lsp_stop","server_id":server_id}).to_string(), - ) - .await; - assert_eq!(gone.last().unwrap()["kind"], "not_found"); + assert_stopped(&state, &server_id).await; } #[tokio::test] async fn lsp_stop_kills_an_idle_server() { let _env_lock = ENV_LOCK.lock().await; - let env = manifest_env("#!/bin/sh\nsleep 60\n"); - std::fs::write( - env.path().join("idletest"), - env.path().join("fake-lsp").to_string_lossy().as_bytes(), - ) - .unwrap(); - let state = Arc::new(State::new()); - let start = one( - &state, - &json!({"op":"lsp_start","language":"idletest"}).to_string(), - ) - .await; - let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); + let (_env, state, server_id) = started_server("idletest", "#!/bin/sh\nsleep 60\n").await; let stop = one( &state, &json!({"op":"lsp_stop","server_id":server_id}).to_string(), ) .await; assert_eq!(type_of(stop.last().unwrap()), "done"); - let gone = one( - &state, - &json!({"op":"lsp_stop","server_id":server_id}).to_string(), - ) - .await; - assert_eq!(gone.last().unwrap()["kind"], "not_found"); + assert_stopped(&state, &server_id).await; } #[tokio::test] async fn lsp_request_reaps_when_the_client_vanishes_before_ready() { let _env_lock = ENV_LOCK.lock().await; - let env = manifest_env("#!/bin/sh\nsleep 60\n"); - std::fs::write( - env.path().join("gonetest"), - env.path().join("fake-lsp").to_string_lossy().as_bytes(), - ) - .unwrap(); - let state = Arc::new(State::new()); - let start = one( - &state, - &json!({"op":"lsp_start","language":"gonetest"}).to_string(), - ) - .await; - let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); + let (_env, state, server_id) = started_server("gonetest", "#!/bin/sh\nsleep 60\n").await; let (mut client, server) = tokio::io::duplex(1 << 20); let request = json!({"op":"lsp_request","server_id":server_id}).to_string(); @@ -200,79 +146,25 @@ async fn lsp_request_reaps_when_the_client_vanishes_before_ready() { .expect("deadline"); assert!(served.is_err(), "the ready write must fail: {served:?}"); - let gone = one( - &state, - &json!({"op":"lsp_stop","server_id":server_id}).to_string(), - ) - .await; - assert_eq!( - gone.last().unwrap()["kind"], - "not_found", - "a server whose client vanished before ready must be reaped: {gone:?}" - ); + assert_stopped(&state, &server_id).await; } #[tokio::test] async fn lsp_data_end_half_closes_stdin() { let _env_lock = ENV_LOCK.lock().await; - let env = manifest_env("#!/bin/sh\nprintf 'ate:%s\\n' $(cat | wc -c)\n"); - std::fs::write( - env.path().join("eoftest"), - env.path().join("fake-lsp").to_string_lossy().as_bytes(), - ) - .unwrap(); - let state = Arc::new(State::new()); - let start = one( - &state, - &json!({"op":"lsp_start","language":"eoftest"}).to_string(), - ) - .await; - let server_id = start.last().unwrap()["server_id"].as_str().unwrap(); + let (_env, state, server_id) = + started_server("eoftest", "#!/bin/sh\nprintf 'ate:%s\\n' $(cat | wc -c)\n").await; let (mut cw, mut out, handle) = connect(&state); - cw.write_all( - json!({"op":"lsp_request","server_id":server_id}) - .to_string() - .as_bytes(), - ) - .await - .unwrap(); - cw.write_all(b"\n").await.unwrap(); - let ready: serde_json::Value = - serde_json::from_str(&out.next_line().await.unwrap().unwrap()).unwrap(); - assert_eq!(type_of(&ready), "ready"); - - cw.write_all( - json!({"op":"data","data":b64(b"12345")}) - .to_string() - .as_bytes(), - ) - .await - .unwrap(); - cw.write_all(b"\n").await.unwrap(); - cw.write_all(json!({"op":"data_end"}).to_string().as_bytes()) - .await - .unwrap(); - cw.write_all(b"\n").await.unwrap(); + send(&mut cw, json!({"op":"lsp_request","server_id":server_id})).await; + assert_eq!(type_of(&next_frame(&mut out).await), "ready"); + send(&mut cw, json!({"op":"data","data":b64(b"12345")})).await; + send(&mut cw, json!({"op":"data_end"})).await; - let reply = timeout(DEADLINE, out.next_line()) - .await - .expect("data_end never closed stdin") - .unwrap() - .unwrap(); - let frame: serde_json::Value = serde_json::from_str(&reply).unwrap(); + let frame = next_frame(&mut out).await; assert_eq!(type_of(&frame), "data"); assert_eq!(decode(&frame), b"ate:5\n"); - - let done = timeout(DEADLINE, out.next_line()) - .await - .expect("deadline") - .unwrap() - .unwrap(); - assert_eq!( - type_of(&serde_json::from_str::(&done).unwrap()), - "done" - ); + assert_eq!(type_of(&next_frame(&mut out).await), "done"); cw.shutdown().await.unwrap(); let _ = handle.await; } diff --git a/silkd/tests/pty_e2e.rs b/silkd/tests/pty_e2e.rs index 3b329276..15b3f7f6 100644 --- a/silkd/tests/pty_e2e.rs +++ b/silkd/tests/pty_e2e.rs @@ -8,14 +8,8 @@ use std::time::Duration; use serde_json::{Value, json}; use silkd::server::State; -use tokio::io::AsyncWriteExt; -use common::{FrameLines, FrameWriter}; - -async fn send(cw: &mut FrameWriter, frame: Value) { - cw.write_all(frame.to_string().as_bytes()).await.unwrap(); - cw.write_all(b"\n").await.unwrap(); -} +use common::{FrameLines, b64, connect, decode, exchange, one, send, type_of}; async fn read_until(lines: &mut FrameLines, pred: impl Fn(&Value) -> bool) -> Value { tokio::time::timeout(Duration::from_secs(5), async { @@ -34,7 +28,7 @@ async fn read_until(lines: &mut FrameLines, pred: impl Fn(&Value) -> bool) -> Va #[tokio::test] async fn pty_runs_a_shell_and_echoes() { let state = Arc::new(State::new()); - let (mut cw, mut lines, handle) = common::connect(&state); + let (mut cw, mut lines, handle) = connect(&state); send(&mut cw, json!({"op":"pty_open","cols":80,"rows":24})).await; let started = read_until(&mut lines, |v| v["type"] == "started").await; @@ -43,17 +37,15 @@ async fn pty_runs_a_shell_and_echoes() { send( &mut cw, - json!({"op":"stdin","data":common::b64(b"echo silk-pty-marker\n")}), + json!({"op":"stdin","data":b64(b"echo silk-pty-marker\n")}), ) .await; - let out = read_until(&mut lines, |v| { - v["type"] == "stdout" - && String::from_utf8_lossy(&common::decode(v)).contains("silk-pty-marker") + read_until(&mut lines, |v| { + v["type"] == "stdout" && String::from_utf8_lossy(&decode(v)).contains("silk-pty-marker") }) .await; - assert_eq!(out["type"], "stdout"); - send(&mut cw, json!({"op":"stdin","data":common::b64(b"exit\n")})).await; + send(&mut cw, json!({"op":"stdin","data":b64(b"exit\n")})).await; let exit = read_until(&mut lines, |v| v["type"] == "exit").await; assert!(exit["code"].is_number()); @@ -64,13 +56,13 @@ async fn pty_runs_a_shell_and_echoes() { #[tokio::test] async fn pty_appears_in_ps_and_resizes() { let state = Arc::new(State::new()); - let (mut cw, mut lines, _) = common::connect(&state); + let (mut cw, mut lines, _) = connect(&state); send(&mut cw, json!({"op":"pty_open","cols":80,"rows":24})).await; let pid = read_until(&mut lines, |v| v["type"] == "started").await["pid"] .as_u64() .unwrap(); - let ps = common::one(&state, r#"{"op":"ps"}"#).await; + let ps = one(&state, r#"{"op":"ps"}"#).await; assert!( ps[0]["procs"] .as_array() @@ -79,12 +71,12 @@ async fn pty_appears_in_ps_and_resizes() { .any(|p| p["pid"].as_u64() == Some(pid)), "pty not in ps: {ps:?}" ); - let resized = common::one( + let resized = one( &state, &json!({"op":"pty_resize","pid":pid,"cols":120,"rows":40}).to_string(), ) .await; - assert_eq!(common::type_of(&resized[0]), "done"); + assert_eq!(type_of(&resized[0]), "done"); drop(cw); } @@ -92,18 +84,15 @@ async fn pty_appears_in_ps_and_resizes() { #[tokio::test] async fn resize_unknown_pid_is_not_found() { let frames = - common::exchange( - &[json!({"op":"pty_resize","pid":999999,"cols":80,"rows":24}).to_string()], - ) - .await; - assert_eq!(common::type_of(&frames[0]), "error"); + exchange(&[json!({"op":"pty_resize","pid":999999,"cols":80,"rows":24}).to_string()]).await; + assert_eq!(type_of(&frames[0]), "error"); assert_eq!(frames[0]["kind"], "not_found"); } #[tokio::test] async fn pty_disconnect_tears_down_without_spin() { let state = Arc::new(State::new()); - let (mut cw, mut lines, handle) = common::connect(&state); + let (mut cw, mut lines, handle) = connect(&state); send(&mut cw, json!({"op":"pty_open","cols":80,"rows":24})).await; let pid = read_until(&mut lines, |v| v["type"] == "started").await["pid"] .as_u64() @@ -116,7 +105,7 @@ async fn pty_disconnect_tears_down_without_spin() { let _ = tokio::time::timeout(Duration::from_secs(5), handle) .await .expect("pty did not tear down on disconnect"); - let ps = common::one(&state, r#"{"op":"ps"}"#).await; + let ps = one(&state, r#"{"op":"ps"}"#).await; assert!( !ps[0]["procs"] .as_array() diff --git a/silkd/tests/session_e2e.rs b/silkd/tests/session_e2e.rs index c3e30b2b..5ec58fe1 100644 --- a/silkd/tests/session_e2e.rs +++ b/silkd/tests/session_e2e.rs @@ -8,16 +8,15 @@ use std::time::Duration; use serde_json::{Value, json}; use silkd::server::State; +use tokio::task::JoinHandle; use common::{one, stdout_body, type_of}; -async fn create(state: &Arc, extra: Value) -> String { - let mut req = json!({"op":"session_create"}); - if let Value::Object(map) = extra { - for (k, v) in map { - req[k] = v; - } - } +async fn create(state: &Arc) -> String { + create_with(state, json!({"op":"session_create"})).await +} + +async fn create_with(state: &Arc, req: Value) -> String { let f = one(state, &req.to_string()).await; assert_eq!(type_of(&f[0]), "session_created", "create failed: {f:?}"); f[0]["id"].as_str().unwrap().to_string() @@ -32,10 +31,16 @@ async fn sh(state: &Arc, id: &str, cmd: &[&str]) -> Vec { .await } +fn spawn_sh(state: &Arc, id: &str, cmd: &'static [&'static str]) -> JoinHandle> { + let s = Arc::clone(state); + let sid = id.to_string(); + tokio::spawn(async move { sh(&s, &sid, cmd).await }) +} + #[tokio::test] async fn session_persists_env_across_calls() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let set = sh(&state, &id, &["export", "MARKER=silk123"]).await; assert_eq!(type_of(set.last().unwrap()), "exit"); @@ -52,9 +57,9 @@ async fn session_persists_env_across_calls() { async fn session_applies_cwd_and_env_at_create() { let dir = tempfile::tempdir().unwrap(); let state = Arc::new(State::new()); - let id = create( + let id = create_with( &state, - json!({"cwd": dir.path().to_str().unwrap(), "env": {"GREETING": "hello-silk"}}), + json!({"op":"session_create","cwd": dir.path().to_str().unwrap(), "env": {"GREETING": "hello-silk"}}), ) .await; @@ -66,10 +71,28 @@ async fn session_applies_cwd_and_env_at_create() { assert!(dir.path().join("marker").exists(), "cwd was not applied"); } +#[tokio::test] +async fn session_create_with_hostile_env_key_still_answers() { + let state = Arc::new(State::new()); + let created = tokio::time::timeout( + Duration::from_secs(8), + one( + &state, + &json!({"op":"session_create","env":{"BAD KEY'\n": "v", "GOOD": "ok"}}).to_string(), + ), + ) + .await + .expect("session_create hung on a hostile env key"); + assert_eq!(type_of(&created[0]), "session_created", "{created:?}"); + let id = created[0]["id"].as_str().unwrap(); + let got = sh(&state, id, &["sh", "-c", "echo $GOOD"]).await; + assert_eq!(stdout_body(&got).trim(), "ok", "{got:?}"); +} + #[tokio::test] async fn session_exit_code_propagates() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let ok = sh(&state, &id, &["true"]).await; assert_eq!(ok.last().unwrap()["code"], 0); @@ -80,7 +103,7 @@ async fn session_exit_code_propagates() { #[tokio::test] async fn session_list_and_rm() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let listed = one(&state, r#"{"op":"session_list"}"#).await; assert!( @@ -102,8 +125,8 @@ async fn session_list_and_rm() { #[tokio::test] async fn stdin_reading_command_does_not_wedge_the_session() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; - let r = tokio::time::timeout(std::time::Duration::from_secs(8), sh(&state, &id, &["cat"])) + let id = create(&state).await; + let r = tokio::time::timeout(Duration::from_secs(8), sh(&state, &id, &["cat"])) .await .expect("session wedged on a stdin-reading command"); assert_eq!(type_of(r.last().unwrap()), "exit"); @@ -114,7 +137,7 @@ async fn stdin_reading_command_does_not_wedge_the_session() { #[tokio::test] async fn shell_killing_command_removes_the_session() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let _ = sh(&state, &id, &["exit"]).await; let again = sh(&state, &id, &["echo", "hi"]).await; assert_eq!(type_of(&again[0]), "error"); @@ -124,7 +147,7 @@ async fn shell_killing_command_removes_the_session() { #[tokio::test] async fn forged_sentinel_in_output_does_not_desync() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let r = sh( &state, &id, @@ -146,7 +169,7 @@ async fn forged_sentinel_in_output_does_not_desync() { #[tokio::test] async fn reap_idle_removes_idle_sessions() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let reaped = state.sessions.reap_idle(Duration::ZERO); assert_eq!(reaped, 1); let gone = sh(&state, &id, &["echo", "hi"]).await; @@ -160,12 +183,8 @@ async fn reap_idle_removes_idle_sessions() { #[tokio::test] async fn reap_skips_a_session_running_a_command() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; - let busy = { - let s = Arc::clone(&state); - let sid = id.clone(); - tokio::spawn(async move { sh(&s, &sid, &["sleep", "1"]).await }) - }; + let id = create(&state).await; + let busy = spawn_sh(&state, &id, &["sleep", "1"]); tokio::time::sleep(Duration::from_millis(150)).await; let reaped = state.sessions.reap_idle(Duration::ZERO); assert_eq!(reaped, 0, "a running session must not be reaped"); @@ -177,7 +196,7 @@ async fn reap_skips_a_session_running_a_command() { #[tokio::test] async fn session_argv_with_metacharacters_is_one_literal_token() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let evil = "a b; echo INJECTED $(id) `whoami` 'q'"; let r = sh(&state, &id, &["printf", "%s", evil]).await; assert_eq!( @@ -194,16 +213,12 @@ async fn session_argv_with_metacharacters_is_one_literal_token() { #[tokio::test] async fn session_rm_unwedges_a_running_external_command() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; - let busy = { - let s = Arc::clone(&state); - let sid = id.clone(); - tokio::spawn(async move { sh(&s, &sid, &["sleep", "3600"]).await }) - }; - tokio::time::sleep(std::time::Duration::from_millis(200)).await; + let id = create(&state).await; + let busy = spawn_sh(&state, &id, &["sleep", "3600"]); + tokio::time::sleep(Duration::from_millis(200)).await; let rm = one(&state, &json!({"op":"session_rm","id":id}).to_string()).await; assert_eq!(type_of(&rm[0]), "done"); - let done = tokio::time::timeout(std::time::Duration::from_secs(5), busy) + let done = tokio::time::timeout(Duration::from_secs(5), busy) .await .expect("session_rm did not unwedge the external command"); let frames = done.unwrap(); @@ -213,7 +228,7 @@ async fn session_rm_unwedges_a_running_external_command() { #[tokio::test] async fn empty_argv_in_session_is_a_bad_request() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let f = one( &state, &json!({"op":"exec","argv":[],"session":id}).to_string(), @@ -226,7 +241,7 @@ async fn empty_argv_in_session_is_a_bad_request() { #[tokio::test] async fn detach_with_session_is_a_bad_request() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let f = one( &state, &json!({"op":"exec","argv":["true"],"session":id,"detach":true}).to_string(), @@ -247,7 +262,7 @@ async fn exec_in_unknown_session_is_not_found() { #[tokio::test] async fn session_merges_stderr_into_stdout() { let state = Arc::new(State::new()); - let id = create(&state, json!({})).await; + let id = create(&state).await; let frames = sh(&state, &id, &["sh", "-c", "echo oops >&2"]).await; assert_eq!( diff --git a/silkd/tests/tree_e2e.rs b/silkd/tests/tree_e2e.rs index 234cf7b4..9aa42eb4 100644 --- a/silkd/tests/tree_e2e.rs +++ b/silkd/tests/tree_e2e.rs @@ -3,6 +3,8 @@ #![allow(clippy::unwrap_used, clippy::expect_used)] mod common; +use std::io::Write; +use std::os::unix::fs::symlink; use std::path::Path; use std::process::Command; @@ -23,7 +25,6 @@ fn sys_tar_create(dir: &Path) -> Vec { } fn sys_tar_extract(archive: &[u8], into: &Path) { - use std::io::Write; let mut child = Command::new("tar") .arg("-x") .arg("-C") @@ -160,7 +161,6 @@ async fn pull_missing_parent_errors_not_found() { #[tokio::test] async fn pull_dangling_symlink_archives_the_link() { - use std::os::unix::fs::symlink; let src = tempfile::tempdir().unwrap(); let link = src.path().join("dangling"); symlink("/no/such/target", &link).unwrap(); @@ -202,13 +202,12 @@ async fn push_failure_leaves_dest_untouched() { .await; assert_eq!(type_of(frames.last().unwrap()), "error"); - let names: Vec = std::fs::read_dir(dest.path()) + let mut names: Vec = std::fs::read_dir(dest.path()) .unwrap() .map(|e| e.unwrap().file_name().to_string_lossy().into_owned()) .collect(); - let mut sorted = names.clone(); - sorted.sort(); - assert_eq!(sorted, ["keep.txt", "sub"], "dest mutated: {names:?}"); + names.sort(); + assert_eq!(names, ["keep.txt", "sub"], "dest mutated: {names:?}"); assert_eq!( std::fs::read(dest.path().join("keep.txt")).unwrap(), b"KEEP"