diff --git a/go/internal/runner/gateway/socket.go b/go/internal/runner/gateway/socket.go index 193ff785..9b5bae33 100644 --- a/go/internal/runner/gateway/socket.go +++ b/go/internal/runner/gateway/socket.go @@ -46,6 +46,31 @@ const socketDirMode os.FileMode = 0o700 // teardown (design SocketListener.Close). const shutdownGrace = 5 * time.Second +// sunPathMax is the longest AF_UNIX path the kernel accepts: sockaddr_un's +// sun_path holds the path plus a NUL terminator. Exceeded, bind(2) returns a +// bare EINVAL naming neither the limit nor the length, so the path is checked +// here instead (see listenAgentSocket). +// +// Derived from the platform's own sockaddr_un rather than hard-coded, because +// the array is not one size across the platforms //go:build unix admits, and +// the spread is wider than the two sizes anyone reaches for: measured over all +// nine, it is 108 bytes on linux, solaris and illumos, 104 on darwin and the +// BSDs including dragonfly, and 1023 on aix. No pair of hand-written numbers is +// correct here, and an author who enumerates the ones they thought of ships the +// bug on the rest — an under-sized constant merely refuses a path that would +// have worked, but an over-sized one passes a path the kernel rejects at bind, +// which is the exact failure this check exists to prevent. +// +// The -1 is the NUL terminator, which is justification enough on its own — no +// appeal to Go's internals is needed, and citing them by line would pin this +// comment to a toolchain version nothing here asserts. Worth knowing that the +// bound is deliberately conservative rather than exact at two edges: AIX's +// wrapper permits a name of len(Path), and a Linux ABSTRACT name carries no +// terminator, so both admit one byte more than this constant. Neither reaches +// the Runner, which only ever binds a filesystem path, and refusing a byte that +// would have worked is the safe direction. +const sunPathMax = len(syscall.RawSockaddrUnix{}.Path) - 1 + // runnerUID reports the uid a reclaimable stale socket must be owned by. It is a // package var over os.Getuid so a hermetic test can drive the wrong-owner // fail-closed branch (which cannot be forged on disk without root). @@ -71,7 +96,41 @@ type SocketListener struct { // (regular file, dir, symlink, or a socket owned by another uid — a path // collision or partial op, never an abandoned Runner socket) is rejected fail- // closed and never deleted. +// +// Path length is checked FIRST, before any directory is created. Not only for +// the diagnostic: a bind that fails EINVAL has already run MkdirAll, and the +// dir survives — measured, 0700 left on disk. Nothing reclaims it on any path, +// because nothing in this socket's lifecycle removes the directory at all: +// Close removes the socket file and never filepath.Dir(path), so the +// Launch-failure teardown Provision does run (host.go:120-125) leaks it too. +// Refusing before the mkdir is what makes the question moot, rather than a +// teardown that would have to be added. func listenAgentSocket(ctx context.Context, path string, h http.Handler) (*SocketListener, error) { + // The socket path is RuntimeDir + /containers//agent.sock + // (host.go), where is NamePrefix + the agent account id + // (spec.go). RuntimeDir is the operator-supplied variable: --runtime-dir is + // unbounded on every hop and inflates the path for every agent at once. + // + // The account id's SHAPE is validated where it enters, not here — see + // validAccountID in spec.go, which refuses an id that is not a single safe + // path element. That is a separate property from length and cannot be + // checked here: a traversing id SHORTENS the path, so it would sail past + // this guard. + // + // With the minted id the tail is 69 bytes, so the default /run/compass lands + // at 81 and the --runtime-dir budget is sunPathMax-69: 38 on Linux, 34 on + // darwin/BSD. That is the figure a startup precheck should assert, so an + // operator learns it at boot rather than at first provision (SEA-1443). + // + // The bind's own error is "bind: invalid argument", an EINVAL naming neither + // the limit nor the actual length, which reads as a permissions or path + // problem. Check first, before any directory is created, so a misconfigured + // deployment is self-diagnosing at Provision and leaves nothing behind. The + // message names both knobs: the path alone does not say which one to shrink. + if len(path) > sunPathMax { + return nil, fmt.Errorf("agent socket path %q is %d bytes, over the %d-byte AF_UNIX limit: shorten the Runner's --runtime-dir or the agent account id", path, len(path), sunPathMax) + } + dir := filepath.Dir(path) if err := os.MkdirAll(dir, socketDirMode); err != nil { return nil, fmt.Errorf("creating agent socket dir %q: %w", dir, err) diff --git a/go/internal/runner/gateway/socket_test.go b/go/internal/runner/gateway/socket_test.go index 2ce12ac7..92d10ba8 100644 --- a/go/internal/runner/gateway/socket_test.go +++ b/go/internal/runner/gateway/socket_test.go @@ -19,10 +19,12 @@ package gateway import ( "context" "errors" + "fmt" "net" "net/http" "os" "path/filepath" + "strings" "testing" "time" @@ -361,6 +363,112 @@ func TestRejectsWrongOwnerSocketNeverDeletes(t *testing.T) { } } +// Case 7b. A path over the AF_UNIX sun_path limit is refused with an error that +// names the limit and the actual length, and nothing is created on disk; a path +// exactly at the limit still binds. The kernel enforces the cap at bind(2) with +// a bare EINVAL ("invalid argument"), which names neither number and reads as a +// permissions or path fault; the Runner's socket path is RuntimeDir plus a +// 69-byte tail at the store-minted 32-hex account id — a width fixed at its +// minting site, not validated at this hop — and RuntimeDir is operator-supplied +// with no bound anywhere, so this is a reachable misconfiguration, not a +// defensive check. +// +// Mutations, both RED: dropping the pre-check lets the over-long path reach bind, +// failing the message assertions and the never-created-dir assertion; tightening +// the bound to >= refuses the at-cap path the kernel accepts, failing the first +// subtest. +func TestRejectsPathOverSunPathLimit(t *testing.T) { + // padTo returns a path under a fresh temp root of exactly n bytes, or fails + // the subtest when the root alone already exceeds the budget (a very deep + // TMPDIR). + padTo := func(t *testing.T, n int) (dir, path string) { + t.Helper() + // A short fixed root, not t.TempDir(): the latter embeds the test name, + // which lands these subtests at ~135 bytes under a darwin-shaped TMPDIR + // (/var/folders/<2>//T, ~49 bytes) and skips them — on the one + // platform where sunPathMax differs from Linux's, and so the one platform + // this test most needs to run. MkdirTemp("", "g") is bounded instead: the + // random suffix is a uint32 in decimal, so at most 10 digits, giving <=16 + // bytes on Linux and <=61 under that darwin TMPDIR — both well under the + // cap, and a bound rather than a measurement that could drift. + root, err := os.MkdirTemp("", "g") //nolint:usetesting // t.TempDir embeds the test name and blows this test's own byte budget on darwin — see above + if err != nil { + t.Fatalf("MkdirTemp: %v", err) + } + t.Cleanup(func() { + if err := os.RemoveAll(root); err != nil { + t.Errorf("removing %q: %v", root, err) + } + }) + dir = filepath.Join(root, "run") + base := len(dir) + 1 // + separator + // These two subtests are the only coverage of the boundary the guard + // exists to enforce, so a deep TMPDIR must redden the run rather than + // skip it — a skip here reports `ok` for a package that asserted + // nothing, which is the failure mode this whole change is about. + if base >= n { + t.Fatalf("temp root %q is %d bytes, leaving no room for a %d-byte path; set TMPDIR to a short path", dir, base, n) + } + path = filepath.Join(dir, strings.Repeat("s", n-base)) + if len(path) != n { + t.Fatalf("constructed path is %d bytes, want exactly %d", len(path), n) + } + return dir, path + } + + t.Run("exactly at the cap binds", func(t *testing.T) { + // The boundary case: the check must reject only what the kernel would. + // An off-by-one (>= instead of >) refuses a path that binds fine, so this + // is what keeps the guard from being over-tight. + _, path := padTo(t, sunPathMax) + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + l, err := listenAgentSocket(ctx, path, stubHandler(t)) + if err != nil { + t.Fatalf("listen on a %d-byte path (exactly the cap) = %v, want success", len(path), err) + } + if err := l.Close(context.Background()); err != nil { + t.Errorf("Close: %v", err) + } + }) + + t.Run("one byte over is refused with a diagnostic", func(t *testing.T) { + dir, path := padTo(t, sunPathMax+1) + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + l, err := listenAgentSocket(ctx, path, stubHandler(t)) + if err == nil { + _ = l.Close(context.Background()) + t.Fatal("listen on an over-long path must fail, got nil error") + } + // The diagnostic is the whole point: the message must carry both numbers + // and the flag to change, none of which the kernel's EINVAL does. Match + // the surrounding phrases, not the bare digits — a temp path carries + // random digits that could contain the cap by chance and pass a + // substring check against a message that never mentioned it. + msg := err.Error() + for _, want := range []string{ + fmt.Sprintf("is %d bytes", len(path)), + fmt.Sprintf("over the %d-byte", sunPathMax), + "--runtime-dir", + } { + if !strings.Contains(msg, want) { + t.Errorf("error %q does not contain %q", msg, want) + } + } + // A bare EINVAL is exactly what the check exists to replace, so the + // message must not be one: that would mean the path reached bind. + if strings.Contains(msg, "invalid argument") { + t.Errorf("error %q reached bind; the pre-check must refuse it first", msg) + } + + // Refused before anything was created: no directory left behind. + if _, err := os.Lstat(dir); !os.IsNotExist(err) { + t.Errorf("Lstat(%q) = %v, want the dir never to have been created", dir, err) + } + }) +} + // Case 8. Close is idempotent and tolerates the socket file already being gone: // a second Close returns nil (the os.Remove ErrNotExist is swallowed), and a // Close after the socket was removed out from under the listener does not error diff --git a/go/internal/runner/spec.go b/go/internal/runner/spec.go index 83c19c8c..3f6ea0e2 100644 --- a/go/internal/runner/spec.go +++ b/go/internal/runner/spec.go @@ -11,8 +11,10 @@ package runner import ( "errors" "fmt" + "io/fs" "net/url" "strings" + "unicode" compassv1 "github.com/sealedsecurity/compass/go/gen/compass/v1" "github.com/sealedsecurity/compass/go/internal/runtime" @@ -51,6 +53,13 @@ func NewConfigSpecBuilder(defaults SpecDefaults) (SpecBuilder, error) { if defaults.CheckoutDir == "" || defaults.HomeDir == "" { return nil, errors.New("spec defaults require checkout and home dirs") } + // The other operand of the container name. validAccountID constrains the + // request-derived half; this constrains the operator-derived half, so both + // inputs to a path segment are checked and a separator here cannot escape + // RuntimeDir through the same filepath.Join clean. + if strings.Contains(defaults.NamePrefix, "/") { + return nil, errors.New("spec defaults name prefix must not contain a path separator") + } return &configSpecBuilder{defaults: defaults}, nil } @@ -67,8 +76,8 @@ func (b *configSpecBuilder) BuildSpec(req *compassv1.ProvisionAgentWorkspaceRequ d := b.defaults accountID := req.GetAgentAccountId() - if accountID == "" { - return runtime.AgentSpec{}, errors.New("provision request requires an agent account id") + if err := validAccountID(accountID); err != nil { + return runtime.AgentSpec{}, err } name := d.NamePrefix + accountID return runtime.AgentSpec{ @@ -86,6 +95,49 @@ func (b *configSpecBuilder) BuildSpec(req *compassv1.ProvisionAgentWorkspaceRequ }, nil } +// validAccountID refuses an agent account id that is not a single safe path +// element. The id is not merely a label: it is concatenated into the container +// name (below) and that name becomes a path segment of the agent socket, +// RuntimeDir/containers//agent.sock (host.go). filepath.Join CLEANS, +// so a "../" in the id escapes RuntimeDir entirely — "../../../../tmp/pwned" +// resolves to /run/tmp/pwned/agent.sock — and the socket's own length guard +// cannot catch it, because traversal SHORTENS the path. +// +// Nothing upstream makes this check redundant. The id is minted 32-hex, but the +// width is a property of the minting site and is never re-checked here; the +// foreign key that ties the id to a real account is enforced by +// RecordAgentContainer, which runs after hub.Provision has already created the +// 0700 directory and bound the socket; and an admin-only RPC narrows who calls +// it, not what they may pass. So this is the hop that has to reject the shape. +func validAccountID(id string) error { + if id == "" { + return errors.New("provision request requires an agent account id") + } + // fs.ValidPath rejects "..", any empty segment, a leading or trailing "/", + // and invalid UTF-8, but returns true for "." (documented) and for a legal + // multi-element path like "abc/def" — hence the other two conjuncts. (The + // empty id returned above, with its own message.) + if !fs.ValidPath(id) || id == "." || strings.Contains(id, "/") { + return fmt.Errorf("agent account id %q is not a valid path element", id) + } + // A control or format character clears every check above — it is valid UTF-8 + // and carries no separator — and nothing downstream stops it either. + // Measured on Linux against listenAgentSocket's own ordering: a newline, a + // DEL, a C1 control and a bidi override all pass MkdirAll AND bind, so the + // id reaches the container name and every log line that quotes it intact. + // (A NUL is the lone exception, failing at MkdirAll — one hop before the + // listener — as a bare EINVAL naming nothing.) The name is what an operator + // reads back out of `podman ps` and the Runner's logs to identify an agent, + // so a character that can forge a line break or reorder the display makes + // that identification unreliable. unicode.IsControl covers C0, DEL and C1; + // the format class (Cf) is checked with it because a bidi override is not a + // control character but spoofs a rendered name just as effectively. + if strings.ContainsFunc(id, func(r rune) bool { return unicode.IsControl(r) || unicode.Is(unicode.Cf, r) }) { + return fmt.Errorf("agent account id %q contains a control or format character", id) + } + return nil +} + // repoSourceFromRequest maps the request's repo oneof to a runtime.RepoSource, // enforcing that exactly one variant is set. func repoSourceFromRequest(req *compassv1.ProvisionAgentWorkspaceRequest) (runtime.RepoSource, error) { diff --git a/go/internal/runner/spec_test.go b/go/internal/runner/spec_test.go index cd05ce2d..f78d2d2d 100644 --- a/go/internal/runner/spec_test.go +++ b/go/internal/runner/spec_test.go @@ -41,6 +41,11 @@ func TestNewConfigSpecBuilderRejectsIncompleteDefaults(t *testing.T) { {"no image", func(d *SpecDefaults) { d.Image = "" }}, {"no checkout dir", func(d *SpecDefaults) { d.CheckoutDir = "" }}, {"no home dir", func(d *SpecDefaults) { d.HomeDir = "" }}, + // NamePrefix is the operator-derived half of the container name, which + // becomes a path segment of the agent socket. A separator here escapes + // RuntimeDir through the same filepath.Join clean that validAccountID + // guards on the request-derived half, so both operands are checked. + {"a name prefix containing a path separator", func(d *SpecDefaults) { d.NamePrefix = "a/../../" }}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -170,6 +175,74 @@ func TestBuildSpecRejectsEmptyAgentAccountID(t *testing.T) { } } +// The agent account id becomes a path segment of the agent socket +// (RuntimeDir/containers//agent.sock), and filepath.Join cleans, so +// a "../" in the id escapes RuntimeDir: "../../../../tmp/pwned" resolves to +// /run/tmp/pwned/agent.sock, where the Runner would MkdirAll a 0700 directory +// and bind. The socket's length guard cannot catch this, because traversal +// SHORTENS the path — every row here is well under the AF_UNIX cap. Each reject +// must surface a non-nil error AND the zero AgentSpec, so no half-built spec +// carries an escaping name to Launch. Uses a valid repo so the id is the only +// failing dimension. +func TestBuildSpecRejectsAgentAccountIDThatEscapesItsPathElement(t *testing.T) { + builder, err := NewConfigSpecBuilder(goodDefaults()) + if err != nil { + t.Fatalf("NewConfigSpecBuilder: %v", err) + } + rejected := []struct { + name string + accountID string + }{ + {"parent traversal escaping the runtime dir", "../../../../tmp/pwned"}, + {"shallow parent traversal", "../../etc"}, + {"a single parent segment", ".."}, + {"the current directory", "."}, + {"an embedded separator", "abc/def"}, + {"a leading separator (absolute)", "/etc/passwd"}, + {"a trailing separator", "abc/"}, + // Control and format characters clear every check above and, measured, + // pass MkdirAll and bind too (a NUL is the exception, failing at + // MkdirAll) — so without this guard they reach the container name and + // the logs that quote it. The C1 and bidi rows are the ones a predicate + // written as `r < 0x20 || r == 0x7f` would silently admit. + {"an embedded NUL", "abc\x00def"}, + {"an embedded newline", "abc\ndef"}, + {"an embedded DEL", "abc\x7fdef"}, + {"an embedded C1 control", "abc\u0085def"}, + {"an embedded bidi override", "abc\u202edef"}, + } + for _, tc := range rejected { + t.Run(tc.name, func(t *testing.T) { + spec, err := builder.BuildSpec(&compassv1.ProvisionAgentWorkspaceRequest{ + AgentAccountId: tc.accountID, + Repo: &compassv1.ProvisionAgentWorkspaceRequest_RemoteUrl{RemoteUrl: "https://example.com/r.git"}, + }) + if err == nil { + t.Fatalf("BuildSpec with agent_account_id %q = nil error, want a path-element rejection", tc.accountID) + } + if !strings.Contains(err.Error(), "agent account id") { + t.Fatalf("BuildSpec error = %v, want the agent-account-id validation error (not a repo failure)", err) + } + if spec.Name != "" { + t.Fatalf("BuildSpec on rejection returned spec with name %q, want the zero AgentSpec", spec.Name) + } + }) + } + + // The ordinary minted shape still builds, so the guard refuses traversal + // rather than every id. + spec, err := builder.BuildSpec(&compassv1.ProvisionAgentWorkspaceRequest{ + AgentAccountId: strings.Repeat("f", 32), + Repo: &compassv1.ProvisionAgentWorkspaceRequest_RemoteUrl{RemoteUrl: "https://example.com/r.git"}, + }) + if err != nil { + t.Fatalf("BuildSpec with a 32-hex account id: %v", err) + } + if want := "compass-agent-" + strings.Repeat("f", 32); spec.Name != want { + t.Fatalf("BuildSpec name = %q, want %q", spec.Name, want) + } +} + // BuildSpec must constrain the caller-supplied repo value before it is forwarded // verbatim as the `git clone` target: an unconstrained remote_url reaches git's // `ext::` (arbitrary command) / `file://` (read-outside-boundary) transports, diff --git a/go/internal/runnerhub/integration_pgtest_test.go b/go/internal/runnerhub/integration_pgtest_test.go index 380e50d1..79ae5ec6 100644 --- a/go/internal/runnerhub/integration_pgtest_test.go +++ b/go/internal/runnerhub/integration_pgtest_test.go @@ -22,6 +22,7 @@ package runnerhub_test import ( "context" + "errors" "io" "log/slog" "net" @@ -30,6 +31,8 @@ import ( "os" "path/filepath" stdruntime "runtime" + "strings" + "syscall" "testing" "time" @@ -50,35 +53,27 @@ const integrationTimeout = 30 * time.Second func TestIntegrationProvisionStartRelayToStoreAndBus(t *testing.T) { dsn := pgtest.RequireDSN(t) // SKIPs when no podman/DSN — never fails hard. + // First cleanup registered, so LIFO removes the tree LAST — after + // runSessionsLoop's drain has confirmed the loop left dispatch (see there). + runtimeDir := shortRuntimeDir(t) ctx, cancel := context.WithCancel(context.Background()) + // Registered adjacent to WithCancel so the ~60 lines of fixture setup below + // (openStoreFixture, runner.Dial, NewConfigSpecBuilder) cannot t.Fatalf out + // with the context never cancelled. runSessionsLoop registers cancel again, + // later, so LIFO still runs its copy FIRST and the drain ordering the loop + // documents is unchanged; context.CancelFunc is idempotent, so the second + // call here is a no-op. t.Cleanup(cancel) - st, err := store.Open(ctx, dsn) - if err != nil { - t.Fatalf("store.Open: %v", err) - } - t.Cleanup(st.Close) - - // A real agent account + its home channel: the write-through target for a - // relayed conversation frame. The agent is a member of its home channel, so - // AppendMessage authorizes. - admin, err := st.BootstrapAdmin(ctx, store.NewUser{Handle: "admin", DisplayName: "Administrator"}) - if err != nil { - t.Fatalf("BootstrapAdmin: %v", err) - } - agent, err := st.CreateAgent(ctx, admin.ID, store.NewAgent{Handle: "atlas", DisplayName: "Atlas"}) - if err != nil { - t.Fatalf("CreateAgent: %v", err) - } + st, agent, bus, sub := openStoreFixture(t, ctx, dsn) homeChannel := agent.Agent.HomeChannelID - - // The real event bus (the SubscribeEvents surface) + a live subscription to - // observe the lifecycle status the seam fans out. - bus := events.NewBus[*compassv1.SubscribeEventsResponse]() - t.Cleanup(bus.Close) - sub, err := bus.Subscribe(0, 0) - if err != nil { - t.Fatalf("bus.Subscribe: %v", err) + // shortRuntimeDir budgeted the path against a MODEL of the account id + // (accountIDHexLen "f"s), because it runs before an account exists. Tie the + // model to the real minted value now that it does: widen store ids and this + // reddens here, rather than silently invalidating that budget and letting + // the real socket path overrun. + if got := len(agent.ID); got != accountIDHexLen { + t.Fatalf("minted account id is %d chars, but shortRuntimeDir budgeted for %d; update accountIDHexLen", got, accountIDHexLen) } // A store-backed ConversationSink: a relayed conversation frame is committed @@ -120,51 +115,17 @@ func TestIntegrationProvisionStartRelayToStoreAndBus(t *testing.T) { CheckoutDir: "/work/repo", HomeDir: "/home/agent", UID: 1000, - NamePrefix: "compass-agent-", + NamePrefix: agentNamePrefix, }) if err != nil { t.Fatalf("NewConfigSpecBuilder: %v", err) } registry := runtime.NewAgentRegistry() rt := runtime.NewAgentRuntimeWithRegistry(engine, registry) - host := runner.NewSessionHost(link, rt, registry, engine, specs, t.TempDir(), discardLog(), nil) - loopDone := make(chan error, 1) - go func() { loopDone <- link.RunSessions(ctx, host) }() - - // Provision the workspace through the public hub path → the Runner launches - // the (fake) container and returns its name. The Sessions stream is - // server-speaks-first: RunSessions' bootstrap Send flushes the headers that - // run the server handler's router.attach, and that round-trip is async to the - // goroutine spawned above. A command dispatched into the pre-attach window - // gets a retriable Unavailable ("no live runner sessions stream") — the same - // transient a production client rides out. Gate on the seam being live by - // retrying Provision (idempotent on its stable request id "prov-1", so no - // double-provision), yielding to the handler goroutine between probes; a - // deadline bounds it so a genuinely wedged seam fails fast, never a sleep. - var provResp *compassv1.ProvisionAgentWorkspaceResponse - provDeadline := time.After(integrationTimeout) - for { - provResp, err = hub.Provision(ctx, "prov-1", &compassv1.ProvisionAgentWorkspaceRequest{ - AgentAccountId: string(agent.ID), - Repo: &compassv1.ProvisionAgentWorkspaceRequest_LocalPath{LocalPath: "/mirror/repo.git"}, - }) - if err == nil { - break - } - if connect.CodeOf(err) != connect.CodeUnavailable { - t.Fatalf("hub.Provision over the seam = %v", err) - } - select { - case <-provDeadline: - t.Fatalf("hub.Provision never reached a live Sessions stream: %v", err) - default: - } - stdruntime.Gosched() - } - containerName := provResp.GetContainerName() - if containerName == "" { - t.Fatal("Provision returned an empty container name") - } + host := runner.NewSessionHost(link, rt, registry, engine, specs, runtimeDir, discardLog(), nil) + loopDone := runSessionsLoop(t, ctx, cancel, link, host) + + containerName := provisionWhenSeamLive(t, ctx, hub, agent.ID) // Start the session → the Runner spawns the agent relay over the pipe engine. startResp, err := hub.Start(ctx, "start-1", &compassv1.StartAgentSessionRequest{ContainerName: containerName}) @@ -176,23 +137,7 @@ func TestIntegrationProvisionStartRelayToStoreAndBus(t *testing.T) { t.Fatal("Start returned an empty session id") } - // The relay goroutine is now reading the agent's stdout pipe. Write one - // conversation frame (→ committed to the store) and one lifecycle frame (→ - // fanned onto the bus). - writeAgentFrame(t, engine.stdoutW, &compassv1internal.AgentFrame{ - Frame: &compassv1internal.AgentFrame_ConversationPosted{ - ConversationPosted: &compassv1.MessagePosted{ - Message: &compassv1.Message{ - Blocks: []*compassv1.MessageBlock{{Block: &compassv1.MessageBlock_Text{Text: "e2e relayed reply"}}}, - }, - }, - }, - }) - writeAgentFrame(t, engine.stdoutW, &compassv1internal.AgentFrame{ - Frame: &compassv1internal.AgentFrame_Session{ - Session: &compassv1internal.SessionFrame{State: compassv1.AgentSessionState_AGENT_SESSION_STATE_WORKING}, - }, - }) + writeRelayFrames(t, engine) // The conversation frame was written THROUGH Deliver to the real store: gate // on the commit signal, then read the message back. @@ -221,22 +166,131 @@ func TestIntegrationProvisionStartRelayToStoreAndBus(t *testing.T) { t.Fatalf("lifecycle status state = %v, want WORKING (the relayed transition)", status.GetState()) } - // Clean teardown mirrors production shutdown (run.go): stop the session, then - // cancel the loop. closeStdout models the container's stdout closing on death - // (the fake's pipe is not ctx-bound the way a real exec's is), so the relay's - // scan ends; cancel then unwinds RunSessions' blocking Receive. + assertCleanShutdown(t, ctx, cancel, host, engine, sessionID, loopDone) +} + +// assertCleanShutdown mirrors production shutdown (run.go): stop the session, +// then cancel the loop. closeStdout models the container's stdout closing on +// death (the fake's pipe is not ctx-bound the way a real exec's is), so the +// relay's scan ends; cancel then unwinds RunSessions' blocking Receive. +// +// This is the CLEAN path, asserted as its own property. The cleanup registered +// by runSessionsLoop covers the failing paths, where a t.Fatalf skips this. +func assertCleanShutdown(t *testing.T, ctx context.Context, cancel context.CancelFunc, host runner.SessionHost, engine *integPipeRuntime, sessionID string, loopDone <-chan error) { + t.Helper() if err := host.Stop(ctx, sessionID); err != nil { t.Fatalf("host.Stop = %v", err) } engine.closeStdout() cancel() select { - case <-loopDone: + case err := <-loopDone: + // A cancelled ctx is the expected end; anything else is a real stream + // failure, and discarding it here would let the loop die of a genuine + // error while this assertion still reads as a clean shutdown. + if err != nil && !errors.Is(err, context.Canceled) { + t.Fatalf("RunSessions = %v, want a clean end after ctx cancel", err) + } case <-time.After(integrationTimeout): t.Fatal("RunSessions loop did not end after ctx cancel") } } +// provisionWhenSeamLive provisions the workspace through the public hub path → +// the Runner launches the (fake) container and returns its name. +// +// The Sessions stream is server-speaks-first: RunSessions' bootstrap Send +// flushes the headers that run the server handler's router.attach, and that +// round-trip is async to the loop goroutine. A command dispatched into the +// pre-attach window gets a retriable Unavailable ("no live runner sessions +// stream") — the same transient a production client rides out. So gate on the +// seam being live by retrying, idempotent on the stable request id "prov-1" so +// there is no double-provision, yielding to the handler goroutine between +// probes. A deadline bounds it, so a genuinely wedged seam fails fast rather +// than spinning, and nothing here is a sleep. +func provisionWhenSeamLive(t *testing.T, ctx context.Context, hub *runnerhub.Hub, agentID store.AccountID) string { + t.Helper() + deadline := time.After(integrationTimeout) + for { + resp, err := hub.Provision(ctx, "prov-1", &compassv1.ProvisionAgentWorkspaceRequest{ + AgentAccountId: string(agentID), + Repo: &compassv1.ProvisionAgentWorkspaceRequest_LocalPath{LocalPath: "/mirror/repo.git"}, + }) + if err == nil { + name := resp.GetContainerName() + if name == "" { + t.Fatal("Provision returned an empty container name") + } + return name + } + if connect.CodeOf(err) != connect.CodeUnavailable { + t.Fatalf("hub.Provision over the seam = %v", err) + } + select { + case <-deadline: + t.Fatalf("hub.Provision never reached a live Sessions stream: %v", err) + default: + } + stdruntime.Gosched() + } +} + +// writeRelayFrames drives the agent side of the seam: the relay goroutine is +// reading the agent's stdout pipe, so write one conversation frame (committed +// through to the store) and one lifecycle frame (fanned onto the bus). Together +// they exercise both write-through paths the integration exists to cover. +func writeRelayFrames(t *testing.T, engine *integPipeRuntime) { + t.Helper() + writeAgentFrame(t, engine.stdoutW, &compassv1internal.AgentFrame{ + Frame: &compassv1internal.AgentFrame_ConversationPosted{ + ConversationPosted: &compassv1.MessagePosted{ + Message: &compassv1.Message{ + Blocks: []*compassv1.MessageBlock{{Block: &compassv1.MessageBlock_Text{Text: "e2e relayed reply"}}}, + }, + }, + }, + }) + writeAgentFrame(t, engine.stdoutW, &compassv1internal.AgentFrame{ + Frame: &compassv1internal.AgentFrame_Session{ + Session: &compassv1internal.SessionFrame{State: compassv1.AgentSessionState_AGENT_SESSION_STATE_WORKING}, + }, + }) +} + +// openStoreFixture opens the store and builds the account + bus fixture the seam +// writes through: a real agent account and its home channel (the write-through +// target for a relayed conversation frame — the agent is a member of its home +// channel, so AppendMessage authorizes), plus the real event bus and a live +// subscription observing the lifecycle status the seam fans out. +// +// Its cleanups register here, so under LIFO they run after everything the caller +// registers later: the store and bus outlive the Runner loop that talks to them. +func openStoreFixture(t *testing.T, ctx context.Context, dsn string) (*store.Store, store.Account, *events.Bus[*compassv1.SubscribeEventsResponse], events.Subscription[*compassv1.SubscribeEventsResponse]) { + t.Helper() + st, err := store.Open(ctx, dsn) + if err != nil { + t.Fatalf("store.Open: %v", err) + } + t.Cleanup(st.Close) + + admin, err := st.BootstrapAdmin(ctx, store.NewUser{Handle: "admin", DisplayName: "Administrator"}) + if err != nil { + t.Fatalf("BootstrapAdmin: %v", err) + } + agent, err := st.CreateAgent(ctx, admin.ID, store.NewAgent{Handle: "atlas", DisplayName: "Atlas"}) + if err != nil { + t.Fatalf("CreateAgent: %v", err) + } + + bus := events.NewBus[*compassv1.SubscribeEventsResponse]() + t.Cleanup(bus.Close) + sub, err := bus.Subscribe(0, 0) + if err != nil { + t.Fatalf("bus.Subscribe: %v", err) + } + return st, agent, bus, sub +} + // storeConversationSink commits a relayed conversation frame to a fixed channel // as a fixed author (the seam's session→channel mapping T5 will own; fixed here // so the test drives a real store write-through). It signals each commit so the @@ -413,6 +467,120 @@ func textOf(m store.Message) string { func discardLog() *slog.Logger { return slog.New(slog.NewTextHandler(io.Discard, nil)) } +// agentNamePrefix is the container-name prefix this test wires into its +// SpecDefaults, hoisted so shortRuntimeDir models the same name the Runner +// actually builds (BuildSpec in spec.go joins it with the account id). Editing +// the prefix in one place would otherwise silently shrink the modelled path and +// turn the budget assertion into a false negative. +const agentNamePrefix = "compass-agent-" + +// accountIDHexLen is the width of a store account id: 16 random bytes +// hex-encoded (internal/store/ids.go:21-28). Fixed at that minting site — the +// only one — rather than validated where the path is built, so it is the right +// width to model the tail with and the wrong thing to call a guarantee. +const accountIDHexLen = 32 + +// runSessionsLoop starts the Runner's dispatch loop and registers the teardown +// that must bracket it. The ordering is load-bearing in both directions, which +// is why it lives beside the goroutine rather than beside context.WithCancel. +// +// It must run BEFORE httptest's srv.Close (registered inside mountRunnerServer, +// therefore earlier, therefore later under LIFO): Close waits on its handlers, +// and the live Sessions handler returns only once ctx is cancelled, so +// cancelling after Close deadlocks the entire cleanup stack — every later +// cleanup, the runtime dir removal included, never runs. +// +// It must run AFTER the runtime dir removal registered at the top of the test, +// so LIFO reclaims that tree only once the loop has left dispatch. Cancel alone +// would not do it: cancel signals and returns, so the WAIT is what makes the +// ordering mean anything. +// +// The wait is what orders the two; it is not a gate on the removal. If the +// drain times out, the later cleanups still run — a cleanup cannot cancel the +// ones registered before it. Neither `return` (it exits only its own closure) +// nor t.Fatalf (it marks the test and runs the rest of the stack anyway) skips +// them, so the timeout arm reports the collision rather than averting it. That +// is the right trade at this point: the test has already failed, and the +// alternative — a flag threaded into shortRuntimeDir to skip RemoveAll — leaks +// the tree and couples two independent helpers to buy nothing a red test needs. +func runSessionsLoop(t *testing.T, ctx context.Context, cancel context.CancelFunc, link *runner.ServerLink, host runner.SessionHost) <-chan error { + t.Helper() + loopDone := make(chan error, 1) + go func() { + loopDone <- link.RunSessions(ctx, host) + // Closed as well as sent: the clean-teardown drain at the end of the test + // takes the value, and the cleanup below must still observe the exit. + close(loopDone) + }() + t.Cleanup(func() { + cancel() + select { + case err := <-loopDone: + // On the clean path assertCleanShutdown already took the value and + // this reads the zero from a closed channel. On a failure path it + // takes the real one, and a stream error that killed the loop must + // surface here — otherwise the test reports only whatever Fatalf'd + // first, with the actual cause discarded. + if err != nil && !errors.Is(err, context.Canceled) { + t.Errorf("RunSessions ended with %v", err) + } + case <-time.After(integrationTimeout): + t.Errorf("RunSessions still running %s after cancel; the runtime dir removal below runs anyway and will race an in-flight Provision", integrationTimeout) + } + }) + return loopDone +} + +// shortRuntimeDir is a Runner RuntimeDir bounded to fit the AF_UNIX sun_path +// limit, replacing t.TempDir() for the one test here that builds a real agent +// socket. The Runner appends a 69-byte tail to its RuntimeDir +// (/containers/compass-agent-<32hex>/agent.sock, host.go:291) at the +// store-minted 32-hex id, leaving 38 bytes of a 107-byte cap on Linux. +// t.TempDir() derives its path from the TEST NAME, and every one of this +// package's tests exceeds that budget: the ceiling is a 19-character name and +// the shortest here is 26. Only this test fails today because only this one +// opens a socket, so the name-length dependency is a trap for the next test +// that wires a SessionHost. +// +// A fixed short root removes the TEST-NAME dependency. It does not make the +// budget unconditional: the root still comes from TMPDIR, and a deep one (a CI +// work dir, or macOS's ~49-byte /var/folders/<2>//T) re-inflates it. So +// the resulting path is asserted rather than assumed. +// +// This site FAILS rather than skips on an over-budget root: a skip would +// silently drop the only end-to-end coverage of the socket path, reporting `ok` +// for a test that asserted nothing. The gateway's padTo fails closed for the +// same reason. The cap is derived the way the production guard derives it +// (gateway.sunPathMax) rather than written down — sun_path is not one size +// across the platforms //go:build unix admits (108 on linux/solaris/illumos, +// 104 on darwin and the BSDs, 1023 on aix), and this file is //go:build unix. +// The derivation is duplicated below because that constant is unexported; the +// code below derives the cap rather than writing a literal. (The sizes quoted +// just above are for the reader, not values anything computes against.) +func shortRuntimeDir(t *testing.T) string { + t.Helper() + dir, err := os.MkdirTemp("", "cr") //nolint:usetesting // t.TempDir embeds the test name, which is what put this path over the sun_path cap — the bug this helper exists to prevent + if err != nil { + t.Fatalf("MkdirTemp for runner runtime dir: %v", err) + } + // Longest path the Runner builds under dir. sun_path holds the path plus a + // NUL, so the usable cap is one less than the platform's array. + const sunPathMax = len(syscall.RawSockaddrUnix{}.Path) - 1 + longest := filepath.Join(dir, "containers", agentNamePrefix+strings.Repeat("f", accountIDHexLen), "agent.sock") + if len(longest) > sunPathMax { + if rmErr := os.RemoveAll(dir); rmErr != nil { + t.Errorf("removing over-budget runner runtime dir %q: %v", dir, rmErr) + } + t.Fatalf("runner runtime dir %q yields a %d-byte agent socket path, over the %d-byte sun_path cap (TMPDIR too deep)", dir, len(longest), sunPathMax) + } + t.Cleanup(func() { + if err := os.RemoveAll(dir); err != nil { + t.Errorf("removing runner runtime dir %q: %v", dir, err) + } + }) + return dir +} + func cleartextH2() *http.Protocols { p := new(http.Protocols) p.SetHTTP1(true)