diff --git a/apps/daemon/internal/agenthost/launch_linux.go b/apps/daemon/internal/agenthost/launch_linux.go index c48beb4a5..8f9d3bb78 100644 --- a/apps/daemon/internal/agenthost/launch_linux.go +++ b/apps/daemon/internal/agenthost/launch_linux.go @@ -38,6 +38,7 @@ type runningView interface { Wait() (sessionview.Exit, error) Close() error Relay() *os.File + RelayLost() <-chan struct{} Spawn(ctx context.Context, path string, args, env []string, dir string, stdin bool) (*sessionview.Spawned, error) } @@ -127,12 +128,8 @@ func (s *session) spawn(opts clirunner.StartOptions) (*clirunner.Process, error) } return nil, fmt.Errorf("%w: spawn: %w", ErrLaunch, err) } - var stdin io.WriteCloser - if p.Stdin != nil { - stdin = p.Stdin - } // FromHandle fails only without stdout and stderr, which a spawned process always has. - process, _ := clirunner.FromHandle(p, clirunner.HandleOptions{Parent: opts.Parent, Stdin: stdin, Stdout: p.Stdout, Stderr: p.Stderr, KillTimeout: opts.KillTimeout}) + process, _ := clirunner.FromHandle(p, clirunner.HandleOptions{Parent: opts.Parent, Stdin: writer(p.Stdin), Stdout: p.Stdout, Stderr: p.Stderr, KillTimeout: opts.KillTimeout}) return process, nil } @@ -169,13 +166,13 @@ func (s *session) start(lv *liveView, opts clirunner.StartOptions) (*clirunner.P s.release(lv) return nil, fmt.Errorf("%w: home: %w", ErrLaunch, err) } - ends, err := newStdio(opts.NeedStdin) + child, ends, err := sessionview.Stdio([3]*os.File{}, opts.NeedStdin, s.uid, s.uid) if err != nil { s.release(lv) return nil, fmt.Errorf("%w: stdio: %w", ErrLaunch, err) } world := worldfs.New(worldExport, s.openFile) - spec := s.spec(world, opts, ends) + spec := s.spec(world, opts, child) // The gateway serves from the view's network hook until the view has ended. var stopGateway func() spec.Network.Setup = func(netns *os.File) (err error) { @@ -183,12 +180,12 @@ func (s *session) start(lv *liveView, opts clirunner.StartOptions) (*clirunner.P return err } v, err := sessionview.Start(startCtx, spec) - ends.closeChild() + closeFiles(child[:]) if err != nil { if stopGateway != nil { stopGateway() } - ends.closeParent() + closeFiles(ends[:]) defer s.release(lv) // sessionview stops a world that served; Stop reports how that went. serr := world.Stop() @@ -241,7 +238,7 @@ func (s *session) brokerFailed(op string, err error) error { // own hands a started view to the clirunner.Process the adapter receives. A // view with a relay gets its own process broker, which serves the view's // shims in scope until the view ends. -func (s *session) own(lv *liveView, v runningView, world viewWorld, stopGateway func(), opts clirunner.StartOptions, ends *stdio, scope sandboxprocess.Scope) (*clirunner.Process, error) { +func (s *session) own(lv *liveView, v runningView, world viewWorld, stopGateway func(), opts clirunner.StartOptions, ends [3]*os.File, scope sandboxprocess.Scope) (*clirunner.Process, error) { h := &ownedView{s: s, lv: lv, v: v, world: world, stopGateway: stopGateway, ended: make(chan struct{}), watched: make(chan struct{})} var brokerErr error if relay := v.Relay(); relay != nil { @@ -269,8 +266,8 @@ func (s *session) own(lv *liveView, v runningView, world viewWorld, stopGateway } var process *clirunner.Process if err == nil { - process, err = clirunner.FromHandle(h, clirunner.HandleOptions{Parent: opts.Parent, Stdin: ends.stdin(), - Stdout: ends.parent[1], Stderr: ends.parent[2], KillTimeout: opts.KillTimeout}) + process, err = clirunner.FromHandle(h, clirunner.HandleOptions{Parent: opts.Parent, Stdin: writer(ends[0]), + Stdout: ends[1], Stderr: ends[2], KillTimeout: opts.KillTimeout}) if err != nil { err = fmt.Errorf("%w: %w", ErrLaunch, err) } @@ -278,7 +275,7 @@ func (s *session) own(lv *liveView, v runningView, world viewWorld, stopGateway if err != nil { v.Close() h.Wait() - ends.closeParent() + closeFiles(ends[:]) return nil, err } return process, nil @@ -306,33 +303,27 @@ type ownedView struct { // while the view runs. func (h *ownedView) watch() { defer close(h.watched) - var relayEnded <-chan struct{} + var relayLost <-chan struct{} if h.broker != nil { - relayEnded = h.broker.Done() - } - for { + relayLost = h.v.RelayLost() + } + select { + case <-h.world.Lost(): + h.s.fail(fmt.Errorf("%w: world: %w", ErrWorld, h.world.Err())) + return + case <-relayLost: + case <-h.ended: + // A lost relay is reported before the view ends. select { - case <-h.world.Lost(): - h.s.fail(fmt.Errorf("%w: world: %w", ErrWorld, h.world.Err())) - return - case <-relayEnded: - // The broker stops serving on Close, which end calls only after - // watch returns, or when its relay connection ends. sessionview - // ends that connection itself only in its teardown, which starts - // once the launcher has stopped answering, and from then on - // Signal fails with ErrExited, ErrClosed or a lost launcher. So a - // delivered signal means that the view still runs its process - // and the relay was lost while the Harness ran. A failed one - // means that the view is ending, and Wait reports how. - if h.v.Signal(0) == nil { - h.s.brokerFailed("relay", h.broker.Err()) - return - } - relayEnded = nil - case <-h.ended: + case <-relayLost: + default: return } } + // The relay's end ends its connection, so the broker stops. + <-h.broker.Done() + h.s.log.Error("process relay lost", "error", h.broker.Err()) + h.s.brokerFailed("relay", h.broker.Err()) } func (h *ownedView) Signal(sig syscall.Signal) error { return h.v.Signal(sig) } @@ -378,7 +369,7 @@ func (h *ownedView) end(waitErr error) { // spec builds the view: the closure and home directories, the agent // host's /etc files and CA directory, the adapter's overlays and masks, and // the shim. -func (s *session) spec(world *worldfs.World, opts clirunner.StartOptions, ends *stdio) sessionview.Spec { +func (s *session) spec(world *worldfs.World, opts clirunner.StartOptions, stdio [3]*os.File) sessionview.Spec { view := s.plan.view var private []sessionview.PrivateDir for _, m := range view.Closure { @@ -406,57 +397,22 @@ func (s *session) spec(world *worldfs.World, opts clirunner.StartOptions, ends * Overlays: overlays, Shim: sessionview.Shim{Binary: s.cfg.Shim, Names: view.Shims, Paths: view.ShimPaths}, Process: sessionview.Process{Path: opts.Binary, Args: append([]string{opts.Binary}, opts.Args...), Env: opts.Env, - Dir: opts.Dir, UID: s.uid, GID: s.uid, Stdin: ends.child[0], Stdout: ends.child[1], Stderr: ends.child[2], + Dir: opts.Dir, UID: s.uid, GID: s.uid, Stdin: stdio[0], Stdout: stdio[1], Stderr: stdio[2], Grace: opts.KillTimeout}, StagingParent: s.dir.entry(stagingEntry), CgroupParent: s.cfg.ViewCgroups, } } -// stdio holds the view process's stdio: the child ends sessionview passes to -// the process and the parent ends the clirunner.Process owns. Without a stdin -// pipe the child's stdin is /dev/null and there is no parent end. -type stdio struct { - child, parent [3]*os.File -} - -func newStdio(needStdin bool) (*stdio, error) { - e := &stdio{} - if needStdin { - r, w, err := os.Pipe() - if err != nil { - return nil, err - } - e.child[0], e.parent[0] = r, w - } else { - null, err := os.Open(os.DevNull) - if err != nil { - return nil, err - } - e.child[0] = null - } - for i := 1; i < 3; i++ { - r, w, err := os.Pipe() - if err != nil { - e.closeChild() - e.closeParent() - return nil, err - } - e.child[i], e.parent[i] = w, r - } - return e, nil -} - -func (e *stdio) stdin() io.WriteCloser { - if e.parent[0] == nil { +// writer returns f, or nil without f: a nil *os.File would be a writer +// that fails. +func writer(f *os.File) io.WriteCloser { + if f == nil { return nil } - return e.parent[0] + return f } -func (e *stdio) closeChild() { closeFiles(e.child[:]) } -func (e *stdio) closeParent() { closeFiles(e.parent[:]) } - func closeFiles(files []*os.File) { for _, f := range files { if f != nil { diff --git a/apps/daemon/internal/agenthost/session_linux_test.go b/apps/daemon/internal/agenthost/session_linux_test.go index 06bcb99dc..261859b75 100644 --- a/apps/daemon/internal/agenthost/session_linux_test.go +++ b/apps/daemon/internal/agenthost/session_linux_test.go @@ -54,11 +54,11 @@ func TestViewEndReleasesTheSlotBeforeTheProcessEnds(t *testing.T) { lv := &liveView{} s.live = lv s.views.Add(1) - ends, err := newStdio(false) + child, ends, err := sessionview.Stdio([3]*os.File{}, false, uint32(os.Getuid()), uint32(os.Getgid())) if err != nil { t.Fatal(err) } - ends.closeChild() + closeFiles(child[:]) v := &fakeView{exit: make(chan struct{})} p, err := s.own(lv, v, fakeWorld{stop: stop}, func() {}, clirunner.StartOptions{Parent: context.Background(), KillTimeout: time.Second}, ends, 0) if err != nil { @@ -108,11 +108,11 @@ func TestViewEndStopsTheGateway(t *testing.T) { lv := &liveView{} s.live = lv s.views.Add(1) - ends, err := newStdio(false) + child, ends, err := sessionview.Stdio([3]*os.File{}, false, uint32(os.Getuid()), uint32(os.Getgid())) if err != nil { t.Fatal(err) } - ends.closeChild() + closeFiles(child[:]) v := &fakeView{exit: make(chan struct{})} p, err := s.own(lv, v, fakeWorld{}, stop, clirunner.StartOptions{Parent: context.Background(), KillTimeout: time.Second}, ends, 0) if err != nil { @@ -417,6 +417,8 @@ func (v *fakeView) Signal(syscall.Signal) error { return nil } func (v *fakeView) Relay() *os.File { return nil } +func (v *fakeView) RelayLost() <-chan struct{} { return nil } + func (v *fakeView) Spawn(context.Context, string, []string, []string, string, bool) (*sessionview.Spawned, error) { return nil, v.spawnErr } diff --git a/apps/daemon/internal/processbroker/broker_linux.go b/apps/daemon/internal/processbroker/broker_linux.go index 332c49fda..58810905c 100644 --- a/apps/daemon/internal/processbroker/broker_linux.go +++ b/apps/daemon/internal/processbroker/broker_linux.go @@ -112,15 +112,12 @@ func (b *Broker) read() { } } b.mu.Lock() - lost := !b.closing - if lost { + if !b.closing { b.err = fmt.Errorf("%w: %w", ErrRelayLost, err) } b.mu.Unlock() - if lost { - b.log.Error("process relay lost", "error", err) - } b.cancel() + b.conn.CloseWrite() // a relay the broker stops serving ends too b.conn.Close() close(b.done) } diff --git a/apps/daemon/internal/processbroker/broker_linux_test.go b/apps/daemon/internal/processbroker/broker_linux_test.go index f7c076e9b..2ee9e164b 100644 --- a/apps/daemon/internal/processbroker/broker_linux_test.go +++ b/apps/daemon/internal/processbroker/broker_linux_test.go @@ -262,7 +262,13 @@ func TestEnvironmentIsDeclaredOnly(t *testing.T) { } func TestLinkLossKeepsOutputOrdered(t *testing.T) { - f := newFixture(t, first(func(c net.Conn) io.ReadWriteCloser { return &cutConn{Conn: c, left: 64 << 10} })) + var dials atomic.Int32 + f := newFixture(t, dialService(t, func(n int32, c net.Conn) io.ReadWriteCloser { + if dials.Store(n); n > 1 { + return c + } + return &cutConn{Conn: c, left: 64 << 10} + })) out, err := f.command("seq", "1", "2000000").Output() if err != nil { t.Fatalf("Output = %v", err) @@ -275,7 +281,7 @@ func TestLinkLossKeepsOutputOrdered(t *testing.T) { if !bytes.Equal(out, want) { t.Fatalf("output differs: %d bytes, want %d", len(out), len(want)) } - if n := f.dials.Load(); n < 2 { + if n := dials.Load(); n < 2 { t.Fatalf("%d dials; the link was never cut", n) } } @@ -325,7 +331,7 @@ func TestIncompleteRequestHoldsNothing(t *testing.T) { // A gone output reader must not hold the exit back while the stream that // would close the remote output is lost behind the settled operation. func TestGoneReaderDoesNotHoldExit(t *testing.T) { - f := newFixture(t, first(func(c net.Conn) io.ReadWriteCloser { return &holdEvents{Conn: c} })) + f := newFixture(t, dialService(t, first(func(c net.Conn) io.ReadWriteCloser { return &holdEvents{Conn: c} }))) r, w, err := os.Pipe() if err != nil { t.Fatal(err) @@ -555,7 +561,15 @@ func TestRelayDescriptorsAreDiscarded(t *testing.T) { // A Busy acknowledgement is retried, or output past the replay limit would // never arrive. func TestBusyAckIsRetried(t *testing.T) { - f := newFixture(t, first(func(c net.Conn) io.ReadWriteCloser { return &busyAck{Conn: c} })) + var acks atomic.Int32 + f := newFixture(t, dialService(t, first(func(c net.Conn) io.ReadWriteCloser { + return intercept(c, func(fr sandboxwire.Frame) verdict { + if fr.Type == sp.OpAckEvents && acks.Add(1) == 1 { + return refuseBusy + } + return pass + }) + }))) cmd := f.command("seq", "1", "2000000") var out countWriter cmd.Stdout = &out @@ -631,14 +645,14 @@ func TestSharedTerminalRestoredByLastUser(t *testing.T) { // accept, so the program reads every byte once. func TestBusyStdinIsRetried(t *testing.T) { var writes atomic.Int32 - f := newFixture(t, first(func(c net.Conn) io.ReadWriteCloser { + f := newFixture(t, dialService(t, first(func(c net.Conn) io.ReadWriteCloser { return intercept(c, func(fr sandboxwire.Frame) verdict { if fr.Type == sp.OpWriteStdin && writes.Add(1) == 2 { return refuseBusy } return pass }) - })) + }))) var in []byte for i := range 30000 { in = strconv.AppendInt(in, int64(i), 10) @@ -668,14 +682,14 @@ func TestBusyStdinIsRetried(t *testing.T) { // scope gets one TERM, and KILL after the grace. func TestUncertainCancelIsNotReplayed(t *testing.T) { var cancels atomic.Int32 - f := newFixture(t, func(_ int32, c net.Conn) io.ReadWriteCloser { + f := newFixture(t, dialService(t, func(_ int32, c net.Conn) io.ReadWriteCloser { return intercept(c, func(fr sandboxwire.Frame) verdict { if fr.Type == sp.OpCancel && cancels.Add(1) == 1 { return loseResponse } return pass }) - }) + })) r, w, err := os.Pipe() if err != nil { t.Fatal(err) @@ -713,7 +727,7 @@ func TestUncertainCancelIsNotReplayed(t *testing.T) { // invocation: the broker closes the shim's descriptors. func TestShimLossEndsWaitForStream(t *testing.T) { dialed := make(chan struct{}) - f := newFixture(t, func(n int32, c net.Conn) io.ReadWriteCloser { + f := newFixture(t, dialService(t, func(n int32, c net.Conn) io.ReadWriteCloser { if n > 1 { return c } @@ -723,7 +737,7 @@ func TestShimLossEndsWaitForStream(t *testing.T) { broker, svc := net.Pipe() go io.Copy(io.Discard, svc) return broker - }) + })) r, w, err := os.Pipe() if err != nil { t.Fatal(err) @@ -766,9 +780,9 @@ func TestShimLossEndsWaitForStream(t *testing.T) { func TestRetriedStartWaitsForStarted(t *testing.T) { for _, fails := range []bool{false, true} { t.Run(fmt.Sprintf("start fails %v", fails), func(t *testing.T) { - svc := newFakeService() + svc := newPeer() svc.startLate, svc.startFails = true, fails - f := newFixture(t, svc.serve(t.Context(), func(fr sandboxwire.Frame) bool { return fr.Type == sp.OpStart })) + f := newFixture(t, svc.dial(sp.OpStart)) // Once the retried Start's Attach found the operation starting, // it starts when the broker observes it without forwarding stdin. // A broker that forwards stdin first gets the refusal it counts. @@ -819,14 +833,14 @@ func awaitsStarted() bool { // resumes from the offset Inspect reports: the program reads every byte // once, and stdin closes at the end of the input. func TestUncertainStdinWriteResumes(t *testing.T) { - svc := newFakeService() + svc := newPeer() svc.partialFirst = true - f := newFixture(t, svc.serve(t.Context(), func(fr sandboxwire.Frame) bool { return fr.Type == sp.OpWriteStdin })) + f := newFixture(t, svc.dial(sp.OpWriteStdin)) const in = "stdin that crosses a lost stream\n" out, stderr, err := runFor(t, f.command("sh", "-c", "cat"), in) c := svc.counts() - if err != nil || out != in || string(c.stdin) != in || c.closedAt != uint64(len(in)) || c.writes < 2 { - t.Fatalf("Run = %v after %d writes; stdout %q, stderr %q; the service took %q and closed stdin at %d", err, c.writes, out, stderr, c.stdin, c.closedAt) + if err != nil || out != in || string(c.stdin) != in || !c.st.StdinClosed || c.writes < 2 { + t.Fatalf("Run = %v after %d writes; stdout %q, stderr %q; the service took %q, stdin closed %v", err, c.writes, out, stderr, c.stdin, c.st.StdinClosed) } } @@ -834,9 +848,9 @@ func TestUncertainStdinWriteResumes(t *testing.T) { // shim exits with 255 and the reason, and the program is cancelled, rather // than both waiting for input that never comes. func TestUnresolvedStdinEndsTheInvocation(t *testing.T) { - svc := newFakeService() + svc := newPeer() svc.partialFirst, svc.inspectFails = true, true - f := newFixture(t, svc.serve(t.Context(), func(fr sandboxwire.Frame) bool { return fr.Type == sp.OpWriteStdin })) + f := newFixture(t, svc.dial(sp.OpWriteStdin)) _, stderr, err := runFor(t, f.command("sh", "-c", "cat"), "stdin\n") if exitCode(err) != processshim.ExitLost || !strings.Contains(stderr, "stdin could not be resumed: inspect failed") { t.Fatalf("Run = %v; stderr %q", err, stderr) @@ -849,17 +863,17 @@ func TestUnresolvedStdinEndsTheInvocation(t *testing.T) { // before its Exited event arrives: stdin closes after the bytes the service // took, the background process gets end of file, and the operation settles. func TestBackgroundReaderGetsEOFAfterUncertainWrite(t *testing.T) { - svc := newFakeService() + svc := newPeer() svc.leaderExits = true - f := newFixture(t, svc.serve(t.Context(), func(fr sandboxwire.Frame) bool { return fr.Type == sp.OpWriteStdin })) + f := newFixture(t, svc.dial(sp.OpWriteStdin)) const in = "stdin for a background reader\n" out, stderr, err := runFor(t, f.command("sh", "-c", "cat"), in) if err != nil || out != in { t.Fatalf("Run = %v; stdout %q, stderr %q", err, out, stderr) } - await(t, "the operation's release", func() bool { return svc.counts().released }) - if c := svc.counts(); c.closedAt != uint64(len(in)) { - t.Fatalf("stdin closed at %d, not %d", c.closedAt, len(in)) + await(t, "the operation's release", func() bool { return svc.counts().st.Released }) + if st := svc.counts().st; !st.StdinClosed || st.StdinOffset != uint64(len(in)) { + t.Fatalf("stdin closed %v at %d, not %d", st.StdinClosed, st.StdinOffset, len(in)) } } @@ -890,19 +904,19 @@ func runFor(t *testing.T, cmd *exec.Cmd, in string) (string, string, error) { type fixture struct { dir, bin string uid, gid int // the relay's - service string - wrap func(int32, net.Conn) io.ReadWriteCloser - dials atomic.Int32 relay *exec.Cmd broker *Broker } // newFixture starts a relay and its broker, whose shims are this binary -// under the names bash, sh, env and seq. As root the relay runs as viewID. A -// non-nil wrap wraps each stream to the service, which it gets with the -// stream's number, counting from 1. -func newFixture(t *testing.T, wrap func(int32, net.Conn) io.ReadWriteCloser) *fixture { +// under the names bash, sh, env and seq. As root the relay runs as viewID. +// The broker reaches the process service through dial, or the shared one +// when dial is nil. +func newFixture(t *testing.T, dial func(context.Context) (io.ReadWriteCloser, error)) *fixture { t.Helper() + if dial == nil { + dial = dialService(t, nil) + } self, err := os.Executable() if err != nil { t.Fatal(err) @@ -915,7 +929,7 @@ func newFixture(t *testing.T, wrap func(int32, net.Conn) io.ReadWriteCloser) *fi if err := os.Chmod(dir, 0o755); err != nil { t.Fatal(err) } - f := &fixture{dir: dir, uid: os.Getuid(), gid: os.Getgid(), service: service.socket(t), wrap: wrap} + f := &fixture{dir: dir, uid: os.Getuid(), gid: os.Getgid()} if f.uid == 0 { f.uid, f.gid = viewID, viewID } @@ -941,7 +955,7 @@ func newFixture(t *testing.T, wrap func(int32, net.Conn) io.ReadWriteCloser) *fi Tool: map[string]string{"TOOL": "1"}, }, Scope: sp.ScopePOSIXSession, - Dial: f.dial, + Dial: dial, CancelGrace: time.Second, }) end.Close() @@ -1024,15 +1038,18 @@ func (f *fixture) command(name string, args ...string) *exec.Cmd { return cmd } -func (f *fixture) dial(ctx context.Context) (io.ReadWriteCloser, error) { - c, err := new(net.Dialer).DialContext(ctx, "unix", f.service) - if err != nil { - return nil, err - } - if n := f.dials.Add(1); f.wrap != nil { - return f.wrap(n, c), nil +// dialService dials the shared process service. A non-nil wrap wraps each +// stream, which it gets with the stream's number, counting from 1. +func dialService(t *testing.T, wrap func(int32, net.Conn) io.ReadWriteCloser) func(context.Context) (io.ReadWriteCloser, error) { + sock := service.socket(t) + var dials atomic.Int32 + return func(ctx context.Context) (io.ReadWriteCloser, error) { + c, err := new(net.Dialer).DialContext(ctx, "unix", sock) + if err != nil || wrap == nil { + return c, err + } + return wrap(dials.Add(1), c), nil } - return c, nil } // first applies wrap to the first stream alone. @@ -1155,31 +1172,6 @@ func (c *holdEvents) Read(p []byte) (int, error) { return c.out.Read(p) } -// busyAck answers the first acknowledgement with Busy after the service -// applied it. -type busyAck struct { - net.Conn - out bytes.Buffer - sent bool -} - -func (c *busyAck) Read(p []byte) (int, error) { - for c.out.Len() == 0 { - fr, err := sandboxwire.ReadFrame(c.Conn, sandboxwire.MaxPayload) - if err != nil { - return 0, err - } - if !c.sent && fr.Type == sandboxwire.ResponseType(sp.OpAckEvents) { - c.sent = true - fr.Payload = sp.Encode(sp.ResponseFailure{Request: sp.OpAckEvents, Failure: *sp.Fail(sp.CodeBusy, sandboxwire.EffectNone, "busy")}) - } - if err := sandboxwire.WriteFrame(&c.out, fr); err != nil { - return 0, err - } - } - return c.out.Read(p) -} - type countWriter struct{ n int } func (w *countWriter) Write(p []byte) (int, error) { diff --git a/apps/daemon/internal/processbroker/broker_other.go b/apps/daemon/internal/processbroker/broker_other.go deleted file mode 100644 index 7c4971c86..000000000 --- a/apps/daemon/internal/processbroker/broker_other.go +++ /dev/null @@ -1,23 +0,0 @@ -//go:build !linux - -package processbroker - -import "errors" - -// ErrUnsupported is returned on platforms without the process broker. -var ErrUnsupported = errors.New("processbroker: unsupported on this platform") - -// Broker is unavailable on this platform. -type Broker struct{} - -// Start returns ErrUnsupported. -func Start(Config) (*Broker, error) { return nil, ErrUnsupported } - -// Close does nothing. -func (*Broker) Close() error { return nil } - -// Done returns nil. -func (*Broker) Done() <-chan struct{} { return nil } - -// Err returns ErrUnsupported. -func (*Broker) Err() error { return ErrUnsupported } diff --git a/apps/daemon/internal/processbroker/doc.go b/apps/daemon/internal/processbroker/doc.go index a511ea96b..faa566931 100644 --- a/apps/daemon/internal/processbroker/doc.go +++ b/apps/daemon/internal/processbroker/doc.go @@ -1,20 +1,14 @@ // Package processbroker runs a Session's shim invocations in its sandbox. // -// A Harness in the Session view executes oac-process-shim -// (apps/daemon/internal/processshim), which hands its invocation and its -// descriptors 0, 1 and 2 to the Session's process relay. The relay is the -// same binary in relay mode, which sessionview starts in the view as the -// Session user, with no capabilities, before the Harness. It is the only -// process that holds a descriptor from the view: it does every read, write -// and terminal ioctl on them with the Session's own authority. It hands the -// broker the request and the terminal's mode over a socketpair that -// sessionview creates. The broker does all process protocol work: it -// resolves the invocation against the declared executable table and -// environment policy, starts the program with the process protocol -// (internal/sandboxprocess), sends the relay the program's output and asks -// it for stdin, forwards the signals the shim reports, and has the relay -// send the shim the remote exit. Neither the shim nor the relay holds -// credentials. +// Package processshim (apps/daemon/internal/processshim) describes the shim, +// the Session's process relay and their IPC with the broker. The broker does +// all process protocol work: it resolves each invocation against the +// declared executable table and environment policy, starts the program with +// the process protocol (internal/sandboxprocess), sends the relay the +// program's output and asks it for stdin, forwards the signals the shim +// reports, and has the relay send the shim the remote exit. Neither the shim +// nor the relay holds credentials. The broker runs only on Linux; elsewhere +// the package declares only its configuration. // // The broker treats the relay as untrusted Session input. It reads the // socketpair without a control buffer, so the kernel closes any descriptor @@ -24,23 +18,11 @@ // ErrRelayLost. The broker never restarts the relay and never replays output // whose delivery is uncertain. Close never waits on the relay. // -// The relay pumps each descriptor on its own, and its dispatch never waits -// on one, so a descriptor nobody reads holds back only its own stream. The -// broker sends the relay at most processshim.OutputWindow bytes of a stream -// that the relay has not reported written, and acknowledges output to the -// process service only after that report, so the service in turn holds back -// the program. -// -// The relay never changes the flags of a passed descriptor, whose open file -// description the Harness shares. It reopens a pipe, FIFO or pty slave -// through /proc/self/fd as its own non-blocking description, uses a socket -// with MSG_DONTWAIT and a regular file or block device as it is, and polls -// any other descriptor before each call, so it waits on a peer only in a -// poll that ending the invocation interrupts. Output on an AF_UNIX socket -// carries the relay's own credentials. -// -// Invocations on one terminal share its saved mode in the relay: the first -// saves it, each runs the terminal raw, and the last to finish restores it. +// The relay pumps each descriptor on its own, so a descriptor nobody reads +// holds back only its own stream. The broker sends the relay at most +// processshim.OutputWindow bytes of a stream that the relay has not reported +// written, and acknowledges output to the process service only after that +// report, so the service in turn holds back the program. // // The broker forwards the exit without waiting for output, and the relay // answers the shim once the output the program wrote before exiting is diff --git a/apps/daemon/internal/processbroker/fakeservice_linux_test.go b/apps/daemon/internal/processbroker/fakeservice_linux_test.go deleted file mode 100644 index 0e1e1a95b..000000000 --- a/apps/daemon/internal/processbroker/fakeservice_linux_test.go +++ /dev/null @@ -1,410 +0,0 @@ -//go:build linux - -package processbroker - -import ( - "context" - "io" - "net" - "slices" - "sync" - - sp "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxprocess" - "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" -) - -// fakeService is a process service whose program copies stdin to stdout and -// exits 0 when stdin closes. Its knobs reproduce one service behavior each. -type fakeService struct { - instance sandboxwire.ID - caps sp.Capabilities - - // startLate keeps an operation Starting until runStarting, or until - // stdin is written, which it refuses. attachedStarting closes once an - // Attach finds the operation Starting. - startLate bool - attachedStarting chan struct{} - // startFails ends the start with StartFailed instead of Started. - startFails bool - // partialFirst accepts half of the first stdin write. - partialFirst bool - // inspectFails refuses Inspect for good. - inspectFails bool - // leaderExits exits the leader with 0 as it takes the first stdin - // write. A background process it leaves copies the rest of stdin to - // stdout until end of file. - leaderExits bool - - mu sync.Mutex - ops map[sandboxwire.ID]*fakeOp - writes int // stdin writes the service took - cancels int - refusedIn int // stdin requests refused while Starting -} - -type fakeOp struct { - id sandboxwire.ID - state sp.OperationState - events []sp.Event // events[i] has sequence i+1 - changed chan struct{} - stdin []byte - stdinClosed bool - closedAt uint64 // the CloseStdin offset - stdout uint64 // the stdout offset - exit *sp.ExitStatus - failure *sp.Failure - background bool // a background process outlives the leader - released bool -} - -func newFakeService() *fakeService { - return &fakeService{ - instance: sandboxwire.NewID(), - caps: sp.Capabilities{ - Platform: sp.PlatformLinux, - Scopes: []sp.Scope{sp.ScopePOSIXSession}, - IOModes: []sp.IOMode{sp.IOPipes}, - Signals: []sp.Signal{1, 2, 15}, - SignalTargets: []sp.SignalTarget{sp.TargetInitialProcessGroup}, - MaxStartBytes: sandboxwire.MaxPayload, - MaxDataBytes: sandboxwire.MaxChunk, - MaxActiveOperations: 8, - MaxOperationRecords: 8, - MaxReplayBytesPerOperation: 1 << 20, - OwnerLossGraceMillis: 60000, - CancelGraceLimitMillis: 60000, - }, - ops: map[sandboxwire.ID]*fakeOp{}, - attachedStarting: make(chan struct{}), - } -} - -// serve replaces each stream to the real service with one to s, and loses -// the first stream in place of the response to the first request that lose -// selects. -func (s *fakeService) serve(ctx context.Context, lose func(sandboxwire.Frame) bool) func(int32, net.Conn) io.ReadWriteCloser { - return func(n int32, c net.Conn) io.ReadWriteCloser { - c.Close() - broker, svc := net.Pipe() - go sp.Serve(ctx, svc, sp.Attachment{ID: s.instance}, s) - if n > 1 || lose == nil { - return broker - } - return intercept(broker, func(fr sandboxwire.Frame) verdict { - if lose(fr) { - return loseResponse - } - return pass - }) - } -} - -// emit appends an event; s.mu is held. -func (s *fakeService) emit(op *fakeOp, ev func(sp.EventHeader) sp.Event) { - op.events = append(op.events, ev(sp.EventHeader{OperationID: op.id, Sequence: uint64(len(op.events) + 1)})) - close(op.changed) - op.changed = make(chan struct{}) -} - -// subscribe sends op's events after seq on conn until the stream ends. -func (s *fakeService) subscribe(conn *sp.Conn, op *fakeOp, after uint64) { - go func() { - next := after - for { - s.mu.Lock() - evs := slices.Clone(op.events[min(next, uint64(len(op.events))):]) - changed := op.changed - s.mu.Unlock() - for _, ev := range evs { - if conn.Send(ev) != nil { - return - } - next++ - } - select { - case <-changed: - case <-conn.Context().Done(): - return - } - } - }() -} - -// run ends the start; s.mu is held. -func (s *fakeService) run(op *fakeOp) { - if op.state != sp.StateStarting { - return - } - if s.startFails { - op.state = sp.StateStartFailed - op.failure = sp.Fail(sp.CodeNotFound, sandboxwire.EffectNone, "no such file") - f := *op.failure - s.emit(op, func(h sp.EventHeader) sp.Event { return sp.StartFailedEvent{EventHeader: h, Failure: f} }) - return - } - op.state = sp.StateRunning - s.emit(op, func(h sp.EventHeader) sp.Event { return sp.StartedEvent{EventHeader: h} }) -} - -// runStarting ends every start in progress. -func (s *fakeService) runStarting() { - s.mu.Lock() - defer s.mu.Unlock() - for _, op := range s.ops { - s.run(op) - } -} - -// exited ends the program with status, or only its background process -// when the leader already exited; s.mu is held. -func (s *fakeService) exited(op *fakeOp, status sp.ExitStatus) { - switch { - case op.background: - op.background = false - s.drained(op, nil) - case op.state == sp.StateRunning: - op.state, op.exit = sp.StateExited, &status - s.drained(op, &status) - } -} - -// drained closes the output and the scope, with the leader's exit between -// them when it exits now; s.mu is held. -func (s *fakeService) drained(op *fakeOp, exit *sp.ExitStatus) { - out := op.stdout - s.emit(op, func(h sp.EventHeader) sp.Event { - return sp.StreamClosedEvent{EventHeader: h, Stream: sp.StreamStdout, Offset: out, Disposition: sp.OutputDrained} - }) - s.emit(op, func(h sp.EventHeader) sp.Event { - return sp.StreamClosedEvent{EventHeader: h, Stream: sp.StreamStderr, Disposition: sp.OutputDrained} - }) - if exit != nil { - s.emit(op, func(h sp.EventHeader) sp.Event { return sp.ExitedEvent{EventHeader: h, Status: *exit} }) - } - s.emit(op, func(h sp.EventHeader) sp.Event { - return sp.OutputClosedEvent{EventHeader: h, Disposition: sp.OutputDrained} - }) - s.emit(op, func(h sp.EventHeader) sp.Event { return sp.ScopeClosedEvent{EventHeader: h} }) -} - -func (s *fakeService) status(op *fakeOp) sp.OperationStatus { - st := sp.OperationStatus{ - State: op.state, Exit: op.exit, StartFailure: op.failure, - StdinOffset: uint64(len(op.stdin)), StdinClosed: op.stdinClosed, - Scope: sp.ScopeStateActive, Released: op.released, - FirstRetained: 1, LastSequence: uint64(len(op.events)), - } - if op.state == sp.StateExited && !op.background || op.state == sp.StateStartFailed { - d := sp.OutputDrained - st.Output, st.Scope = &d, sp.ScopeStateClosed - } - return st -} - -// lookup returns the operation with s.mu held. -func (s *fakeService) lookup(ref sp.OperationRef) (*fakeOp, error) { - if ref.ServerInstanceID != s.instance { - return nil, sp.Fail(sp.CodeInstanceChanged, sandboxwire.EffectNone, "instance changed") - } - s.mu.Lock() - op := s.ops[ref.OperationID] - if op == nil { - s.mu.Unlock() - return nil, sp.Fail(sp.CodeNotFound, sandboxwire.EffectNone, "no operation") - } - if op.released { - s.mu.Unlock() - return nil, sp.Fail(sp.CodeReleased, sandboxwire.EffectNone, "released") - } - return op, nil -} - -func (s *fakeService) Describe(context.Context, *sp.Conn, sp.DescribeRequest) (sp.DescribeResponse, error) { - return sp.DescribeResponse{ServerInstanceID: s.instance, Capabilities: s.caps}, nil -} - -func (s *fakeService) Start(_ context.Context, conn *sp.Conn, req sp.StartRequest) (sp.StartResponse, error) { - if req.ServerInstanceID != s.instance { - return sp.StartResponse{}, sp.Fail(sp.CodeInstanceChanged, sandboxwire.EffectNone, "instance changed") - } - s.mu.Lock() - defer s.mu.Unlock() - if s.ops[req.OperationID] != nil { - return sp.StartResponse{Disposition: sp.StartExisting}, nil - } - op := &fakeOp{id: req.OperationID, state: sp.StateStarting, changed: make(chan struct{})} - s.ops[op.id] = op - s.subscribe(conn, op, 0) - if !s.startLate { - s.run(op) - } - return sp.StartResponse{Disposition: sp.StartCreated}, nil -} - -func (s *fakeService) Attach(_ context.Context, conn *sp.Conn, req sp.AttachRequest) (sp.AttachResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.AttachResponse{}, err - } - defer s.mu.Unlock() - s.subscribe(conn, op, req.AfterSequence) - if op.state == sp.StateStarting { - select { - case <-s.attachedStarting: - default: - close(s.attachedStarting) - } - } - return sp.AttachResponse{Status: s.status(op)}, nil -} - -func (s *fakeService) Inspect(_ context.Context, _ *sp.Conn, req sp.InspectRequest) (sp.InspectResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.InspectResponse{}, err - } - defer s.mu.Unlock() - if s.inspectFails { - return sp.InspectResponse{}, sp.Fail(sp.CodeIO, sandboxwire.EffectNone, "inspect failed") - } - return sp.InspectResponse{Status: s.status(op)}, nil -} - -// stdinRefused refuses stdin while the operation starts, which also ends a -// late start; s.mu is held. -func (s *fakeService) stdinRefused(op *fakeOp) error { - switch { - case op.state == sp.StateStarting: - s.refusedIn++ - s.run(op) - return sp.Fail(sp.CodeNotRunning, sandboxwire.EffectNone, "the operation is starting") - case op.state != sp.StateRunning && !op.background: - return sp.Fail(sp.CodeNotRunning, sandboxwire.EffectNone, "the operation is not running") - case op.stdinClosed: - return sp.Fail(sp.CodeStdinClosed, sandboxwire.EffectNone, "stdin is closed") - } - return nil -} - -func (s *fakeService) WriteStdin(_ context.Context, _ *sp.Conn, req sp.WriteStdinRequest) (sp.WriteStdinResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.WriteStdinResponse{}, err - } - defer s.mu.Unlock() - if err := s.stdinRefused(op); err != nil { - return sp.WriteStdinResponse{}, err - } - if req.Offset != uint64(len(op.stdin)) { - return sp.WriteStdinResponse{}, sp.Fail(sp.CodeInputOffsetConflict, sandboxwire.EffectNone, "offset %d is not %d", req.Offset, len(op.stdin)) - } - data := req.Data - if s.writes++; s.writes == 1 && s.partialFirst { - data = data[:len(data)/2] - } - op.stdin = append(op.stdin, data...) - if len(data) > 0 { - out, echo := op.stdout, slices.Clone(data) - op.stdout += uint64(len(data)) - s.emit(op, func(h sp.EventHeader) sp.Event { - return sp.OutputEvent{EventHeader: h, Stream: sp.StreamStdout, Offset: out, Data: echo} - }) - } - if s.leaderExits && op.state == sp.StateRunning { - status := sp.ExitStatus{Kind: sp.ExitCode} - op.state, op.exit, op.background = sp.StateExited, &status, true - s.emit(op, func(h sp.EventHeader) sp.Event { return sp.ExitedEvent{EventHeader: h, Status: status} }) - } - return sp.WriteStdinResponse{Accepted: uint32(len(data))}, nil -} - -func (s *fakeService) CloseStdin(_ context.Context, _ *sp.Conn, req sp.CloseStdinRequest) (sp.CloseStdinResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.CloseStdinResponse{}, err - } - defer s.mu.Unlock() - if op.stdinClosed && req.Offset == op.closedAt { - return sp.CloseStdinResponse{}, nil - } - if err := s.stdinRefused(op); err != nil { - return sp.CloseStdinResponse{}, err - } - if req.Offset != uint64(len(op.stdin)) { - return sp.CloseStdinResponse{}, sp.Fail(sp.CodeInputOffsetConflict, sandboxwire.EffectNone, "offset %d is not %d", req.Offset, len(op.stdin)) - } - op.stdinClosed, op.closedAt = true, req.Offset - s.exited(op, sp.ExitStatus{Kind: sp.ExitCode}) - return sp.CloseStdinResponse{}, nil -} - -func (s *fakeService) CloseOutput(_ context.Context, _ *sp.Conn, req sp.CloseOutputRequest) (sp.CloseOutputResponse, error) { - if _, err := s.lookup(req.OperationRef); err != nil { - return sp.CloseOutputResponse{}, err - } - s.mu.Unlock() - return sp.CloseOutputResponse{}, nil -} - -func (s *fakeService) ResizePTY(context.Context, *sp.Conn, sp.ResizePTYRequest) (sp.ResizePTYResponse, error) { - return sp.ResizePTYResponse{}, sp.Fail(sp.CodeUnsupported, sandboxwire.EffectNone, "no terminal") -} - -func (s *fakeService) Signal(_ context.Context, _ *sp.Conn, req sp.SignalRequest) (sp.SignalResponse, error) { - if _, err := s.lookup(req.OperationRef); err != nil { - return sp.SignalResponse{}, err - } - s.mu.Unlock() - return sp.SignalResponse{}, nil -} - -// Cancel ends the program as TERM does. -func (s *fakeService) Cancel(_ context.Context, _ *sp.Conn, req sp.CancelRequest) (sp.CancelResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.CancelResponse{}, err - } - defer s.mu.Unlock() - s.cancels++ - s.exited(op, sp.ExitStatus{Kind: sp.ExitSignal, Signal: 15}) - return sp.CancelResponse{}, nil -} - -func (s *fakeService) AckEvents(_ context.Context, _ *sp.Conn, req sp.AckEventsRequest) (sp.AckEventsResponse, error) { - if _, err := s.lookup(req.OperationRef); err != nil { - return sp.AckEventsResponse{}, err - } - s.mu.Unlock() - return sp.AckEventsResponse{}, nil -} - -func (s *fakeService) Release(_ context.Context, _ *sp.Conn, req sp.ReleaseRequest) (sp.ReleaseResponse, error) { - op, err := s.lookup(req.OperationRef) - if err != nil { - return sp.ReleaseResponse{}, err - } - defer s.mu.Unlock() - if op.state != sp.StateExited && op.state != sp.StateStartFailed || op.background { - return sp.ReleaseResponse{}, sp.Fail(sp.CodeBusy, sandboxwire.EffectNone, "not settled") - } - op.released = true - return sp.ReleaseResponse{}, nil -} - -// fakeCounts is what a fakeService saw. -type fakeCounts struct { - writes, cancels, refusedIn int - stdin []byte // the only operation's accepted stdin - closedAt uint64 // and its CloseStdin offset - released bool // and whether it was released -} - -func (s *fakeService) counts() fakeCounts { - s.mu.Lock() - defer s.mu.Unlock() - c := fakeCounts{writes: s.writes, cancels: s.cancels, refusedIn: s.refusedIn} - for _, op := range s.ops { - c.stdin, c.closedAt, c.released = slices.Clone(op.stdin), op.closedAt, op.released - } - return c -} diff --git a/apps/daemon/internal/processbroker/invocation_linux.go b/apps/daemon/internal/processbroker/invocation_linux.go index eea43903a..667004205 100644 --- a/apps/daemon/internal/processbroker/invocation_linux.go +++ b/apps/daemon/internal/processbroker/invocation_linux.go @@ -585,7 +585,7 @@ func (inv *invocation) releaseOp(h handle) bool { if f.Code == sp.CodeBusy && inv.sleep(backoff) { continue } - if f.Code != sp.CodeReleased && f.Code != sp.CodeBusy { + if f.Code != sp.CodeReleased && f.Code != sp.CodeBusy && inv.b.ctx.Err() == nil { inv.log.Warn("operation release failed", "error", err) } h.op.Detach() @@ -653,7 +653,9 @@ func (inv *invocation) answer(r processshim.Result, marks []processshim.Mark, be // reason is written even after the shim exited, because the Harness may still // read the output the failure cut short. func (inv *invocation) fail(reason string) { - inv.log.Warn("process invocation failed", "reason", reason) + if inv.b.ctx.Err() == nil { // a stopped broker fails every invocation + inv.log.Warn("process invocation failed", "reason", reason) + } inv.halted() if !inv.reply(processshim.Result{Code: processshim.ExitLost, Message: message(reason)}, nil) { inv.send(processshim.Notice{ID: inv.rid, Message: message(reason)}) @@ -673,7 +675,7 @@ func (inv *invocation) shimGone() { inv.loseShim() cancel := !inv.exited inv.mu.Unlock() - inv.log.Info("process shim lost", "cancel", cancel) + inv.log.Debug("process shim lost", "cancel", cancel) inv.stopInput() if cancel { inv.cancelRemote() @@ -776,7 +778,12 @@ func (inv *invocation) receive(m processshim.RelayMessage) error { for _, seq := range sent { inv.acks.deliver(seq) } - inv.log.Info("output reader gone", "stream", w.stream, "error", unix.Errno(m.Errno)) + // A reader that closes its end, as on a Cancel, is expected. + if errno := unix.Errno(m.Errno); errno == unix.EPIPE || errno == unix.ECONNRESET { + inv.log.Debug("output reader gone", "stream", w.stream, "error", errno) + } else { + inv.log.Info("output write failed", "stream", w.stream, "error", errno) + } inv.helpers.Add(1) go func() { defer inv.helpers.Done() diff --git a/apps/daemon/internal/processbroker/link_linux.go b/apps/daemon/internal/processbroker/link_linux.go index b4abd56fe..652fd4738 100644 --- a/apps/daemon/internal/processbroker/link_linux.go +++ b/apps/daemon/internal/processbroker/link_linux.go @@ -67,7 +67,9 @@ func (l *link) get(ctx context.Context) (*stream, error) { l.mu.Unlock() return s, nil } - l.log.Warn("process stream unavailable", "error", err, "retry", backoff) + if ctx.Err() == nil { + l.log.Warn("process stream unavailable", "error", err, "retry", backoff) + } select { case <-time.After(backoff): case <-ctx.Done(): diff --git a/apps/daemon/internal/processbroker/peer_linux_test.go b/apps/daemon/internal/processbroker/peer_linux_test.go new file mode 100644 index 000000000..dd710444e --- /dev/null +++ b/apps/daemon/internal/processbroker/peer_linux_test.go @@ -0,0 +1,281 @@ +//go:build linux + +package processbroker + +import ( + "context" + "io" + "net" + "slices" + "sync" + "sync/atomic" + + sp "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxprocess" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// peer is a process service with a single operation, scripted for the +// faults the uncertain-request tests need. Its program copies stdin to +// stdout and exits 0 when stdin closes. +type peer struct { + // startLate keeps the operation Starting until runStarting, or until + // stdin is written, which it refuses. attachedStarting closes once an + // Attach finds the operation Starting. + startLate bool + attachedStarting chan struct{} + startFails bool // the start ends with StartFailed + partialFirst bool // the first stdin write is half accepted + inspectFails bool // Inspect fails for good + // leaderExits exits the leader with 0 as it takes the first stdin + // write. A background process it leaves copies the rest of stdin to + // stdout until end of file. + leaderExits bool + + instance sandboxwire.ID + + mu sync.Mutex + seen + id sandboxwire.ID + events []sp.Event // events[i] has sequence i+1 + out net.Conn // the stream subscribed to the events + background bool // a background process outlives the leader +} + +// seen is what the peer saw. +type seen struct { + st sp.OperationStatus + stdin []byte // the stdin the service took + writes, cancels, refusedIn int // refusedIn counts stdin refused while Starting +} + +func newPeer() *peer { + return &peer{instance: sandboxwire.NewID(), attachedStarting: make(chan struct{}), + seen: seen{st: sp.OperationStatus{State: sp.StateStarting, Scope: sp.ScopeStateActive, FirstRetained: 1}}} +} + +func (p *peer) counts() seen { + p.mu.Lock() + defer p.mu.Unlock() + s := p.seen + s.stdin = slices.Clone(s.stdin) + return s +} + +// dial serves each stream the broker dials. The first stream is lost in +// place of the response to its first request of type lose. +func (p *peer) dial(lose uint16) func(context.Context) (io.ReadWriteCloser, error) { + var dials atomic.Int32 + return func(context.Context) (io.ReadWriteCloser, error) { + broker, svc := net.Pipe() + go p.serve(svc, dials.Add(1) == 1, lose) + return broker, nil + } +} + +func (p *peer) serve(c net.Conn, first bool, lose uint16) { + defer c.Close() + for { + f, err := sandboxwire.ReadFrame(c, sandboxwire.MaxPayload) + if err != nil { + return + } + m, err := sp.Decode(f.Type, f.Payload) + if err != nil { + return + } + p.mu.Lock() + resp, failure := p.handle(m) + if first && f.Type == lose { + p.mu.Unlock() + return + } + if failure != nil { + resp = sp.ResponseFailure{Request: f.Type, Failure: *failure} + } + send(c, f.RequestID, resp) + switch m := m.(type) { + case sp.StartRequest: + if resp == (sp.StartResponse{Disposition: sp.StartCreated}) { + p.subscribe(c, 0) + } + case sp.AttachRequest: + p.subscribe(c, m.AfterSequence) + } + p.mu.Unlock() + } +} + +// subscribe sends c the events after seq, and every later one; p.mu is +// held. +func (p *peer) subscribe(c net.Conn, seq uint64) { + p.out = c + for _, ev := range p.events[seq:] { + send(c, 0, ev) + } +} + +func send(c net.Conn, requestID uint64, m sp.Message) { + sandboxwire.WriteFrame(c, sandboxwire.Frame{Type: m.MessageType(), RequestID: requestID, Payload: sp.Encode(m)}) +} + +// handle applies one request; p.mu is held. +func (p *peer) handle(m sp.Message) (sp.Message, *sp.Failure) { + switch m := m.(type) { + case sp.DescribeRequest: + return sp.DescribeResponse{ServerInstanceID: p.instance, Capabilities: sp.Capabilities{ + Platform: sp.PlatformLinux, + Scopes: []sp.Scope{sp.ScopePOSIXSession}, + IOModes: []sp.IOMode{sp.IOPipes}, + Signals: []sp.Signal{1, 2, 15}, + SignalTargets: []sp.SignalTarget{sp.TargetInitialProcessGroup}, + MaxStartBytes: sandboxwire.MaxPayload, + MaxDataBytes: sandboxwire.MaxChunk, + MaxActiveOperations: 8, + MaxOperationRecords: 8, + MaxReplayBytesPerOperation: 1 << 20, + OwnerLossGraceMillis: 60000, + CancelGraceLimitMillis: 60000, + }}, nil + case sp.StartRequest: + if !p.id.IsZero() { + return sp.StartResponse{Disposition: sp.StartExisting}, nil + } + p.id = m.OperationID + if !p.startLate { + p.run() + } + return sp.StartResponse{Disposition: sp.StartCreated}, nil + case sp.AttachRequest: + if p.st.State == sp.StateStarting { + select { + case <-p.attachedStarting: + default: + close(p.attachedStarting) + } + } + return sp.AttachResponse{Status: p.st}, nil + case sp.InspectRequest: + if p.inspectFails { + return nil, sp.Fail(sp.CodeIO, sandboxwire.EffectNone, "inspect failed") + } + return sp.InspectResponse{Status: p.st}, nil + case sp.WriteStdinRequest: + if f := p.refuseStdin(m.Offset); f != nil { + return nil, f + } + data := m.Data + if p.writes++; p.writes == 1 && p.partialFirst { + data = data[:len(data)/2] + } + p.stdin = append(p.stdin, data...) + if len(data) > 0 { + out, echo := p.st.StdinOffset, slices.Clone(data) + p.emit(func(h sp.EventHeader) sp.Event { + return sp.OutputEvent{EventHeader: h, Stream: sp.StreamStdout, Offset: out, Data: echo} + }) + } + p.st.StdinOffset += uint64(len(data)) + if p.leaderExits && p.st.State == sp.StateRunning { + status := sp.ExitStatus{Kind: sp.ExitCode} + p.st.State, p.st.Exit, p.background = sp.StateExited, &status, true + p.emit(func(h sp.EventHeader) sp.Event { return sp.ExitedEvent{EventHeader: h, Status: status} }) + } + return sp.WriteStdinResponse{Accepted: uint32(len(data))}, nil + case sp.CloseStdinRequest: + if f := p.refuseStdin(m.Offset); f != nil { + return nil, f + } + p.st.StdinClosed = true + p.exit(sp.ExitStatus{Kind: sp.ExitCode}) + return sp.CloseStdinResponse{}, nil + case sp.CancelRequest: + p.cancels++ + p.exit(sp.ExitStatus{Kind: sp.ExitSignal, Signal: 15}) + return sp.CancelResponse{}, nil + case sp.AckEventsRequest: + return sp.AckEventsResponse{}, nil + case sp.ReleaseRequest: + if p.st.Scope != sp.ScopeStateClosed { + return nil, sp.Fail(sp.CodeBusy, sandboxwire.EffectNone, "not settled") + } + p.st.Released = true + return sp.ReleaseResponse{}, nil + } + return nil, sp.Fail(sp.CodeUnsupported, sandboxwire.EffectNone, "not scripted") +} + +// refuseStdin refuses stdin while the operation starts, which also ends a +// late start; p.mu is held. +func (p *peer) refuseStdin(offset uint64) *sp.Failure { + switch { + case p.st.State == sp.StateStarting: + p.refusedIn++ + p.run() + return sp.Fail(sp.CodeNotRunning, sandboxwire.EffectNone, "the operation is starting") + case p.st.State != sp.StateRunning && !p.background: + return sp.Fail(sp.CodeNotRunning, sandboxwire.EffectNone, "the operation is not running") + case p.st.StdinClosed: + return sp.Fail(sp.CodeStdinClosed, sandboxwire.EffectNone, "stdin is closed") + case offset != p.st.StdinOffset: + return sp.Fail(sp.CodeInputOffsetConflict, sandboxwire.EffectNone, "offset %d is not %d", offset, p.st.StdinOffset) + } + return nil +} + +// runStarting ends a start in progress. +func (p *peer) runStarting() { + p.mu.Lock() + defer p.mu.Unlock() + p.run() +} + +// run ends the start; p.mu is held. +func (p *peer) run() { + switch { + case p.st.State != sp.StateStarting: + case p.startFails: + f := *sp.Fail(sp.CodeNotFound, sandboxwire.EffectNone, "no such file") + d := sp.OutputDrained + p.st.State, p.st.StartFailure, p.st.Output, p.st.Scope = sp.StateStartFailed, &f, &d, sp.ScopeStateClosed + p.emit(func(h sp.EventHeader) sp.Event { return sp.StartFailedEvent{EventHeader: h, Failure: f} }) + default: + p.st.State = sp.StateRunning + p.emit(func(h sp.EventHeader) sp.Event { return sp.StartedEvent{EventHeader: h} }) + } +} + +// exit ends the program with status, or only its background process when +// the leader already exited: the output and then the scope close, with +// the leader's exit between them when it exits now; p.mu is held. +func (p *peer) exit(status sp.ExitStatus) { + leader := p.st.State == sp.StateRunning + if !leader && !p.background { + return + } + p.background = false + out, d := p.st.StdinOffset, sp.OutputDrained + p.emit(func(h sp.EventHeader) sp.Event { + return sp.StreamClosedEvent{EventHeader: h, Stream: sp.StreamStdout, Offset: out, Disposition: d} + }) + p.emit(func(h sp.EventHeader) sp.Event { + return sp.StreamClosedEvent{EventHeader: h, Stream: sp.StreamStderr, Disposition: d} + }) + if leader { + p.st.State, p.st.Exit = sp.StateExited, &status + p.emit(func(h sp.EventHeader) sp.Event { return sp.ExitedEvent{EventHeader: h, Status: status} }) + } + p.emit(func(h sp.EventHeader) sp.Event { return sp.OutputClosedEvent{EventHeader: h, Disposition: d} }) + p.emit(func(h sp.EventHeader) sp.Event { return sp.ScopeClosedEvent{EventHeader: h} }) + p.st.Output, p.st.Scope = &d, sp.ScopeStateClosed +} + +// emit records an event and sends it to the subscribed stream; p.mu is +// held. +func (p *peer) emit(ev func(sp.EventHeader) sp.Event) { + e := ev(sp.EventHeader{OperationID: p.id, Sequence: uint64(len(p.events) + 1)}) + p.events = append(p.events, e) + p.st.LastSequence++ + if p.out != nil { + send(p.out, 0, e) + } +} diff --git a/apps/daemon/internal/processbroker/pumps_linux.go b/apps/daemon/internal/processbroker/pumps_linux.go index 16e85cf94..3df3b61e3 100644 --- a/apps/daemon/internal/processbroker/pumps_linux.go +++ b/apps/daemon/internal/processbroker/pumps_linux.go @@ -436,7 +436,7 @@ func (inv *invocation) signal(s processshim.Signaled) { if err == nil { return } - if f := asFailure(err); f.Code != sp.CodeNotRunning && f.Code != sp.CodeReleased { + if f := asFailure(err); f.Code != sp.CodeNotRunning && f.Code != sp.CodeReleased && inv.b.ctx.Err() == nil { inv.log.Info("signal not delivered", "signal", n, "error", err) } } diff --git a/apps/daemon/internal/processbroker/view_linux_test.go b/apps/daemon/internal/processbroker/view_linux_test.go index 294a41b58..f884a4b34 100644 --- a/apps/daemon/internal/processbroker/view_linux_test.go +++ b/apps/daemon/internal/processbroker/view_linux_test.go @@ -6,7 +6,6 @@ import ( "context" "fmt" "io" - "net" "os" "path/filepath" "sync" @@ -88,7 +87,6 @@ func startView(t *testing.T, w *hangWorld, p sessionview.Process) (*sessionview. if err := sessionview.Probe(); err != nil { t.Fatalf("Probe: %v", err) } - sock := service.socket(t) harness := t.TempDir() self, err := os.Executable() if err != nil { @@ -119,9 +117,7 @@ func startView(t *testing.T, w *hangWorld, p sessionview.Process) (*sessionview. Executables: Executables{Paths: map[string]string{"/bin/sh": "/bin/sh"}}, Environment: Environment{Sandbox: map[string]string{"PATH": "/usr/bin:/bin"}}, Scope: sp.ScopePOSIXSession, - Dial: func(ctx context.Context) (io.ReadWriteCloser, error) { - return new(net.Dialer).DialContext(ctx, "unix", sock) - }, + Dial: dialService(t, nil), CancelGrace: time.Second, }) if err != nil { diff --git a/apps/daemon/internal/processshim/conn_linux.go b/apps/daemon/internal/processshim/conn_linux.go index a83d1a855..99e57f9d8 100644 --- a/apps/daemon/internal/processshim/conn_linux.go +++ b/apps/daemon/internal/processshim/conn_linux.go @@ -3,6 +3,7 @@ package processshim import ( + "bytes" "errors" "fmt" "io" @@ -29,26 +30,23 @@ func NewConn(c *net.UnixConn) *Conn { return &Conn{c: c, first: true} } // Close closes the socket. func (c *Conn) Close() error { return c.c.Close() } -// Unix returns the socket. -func (c *Conn) Unix() *net.UnixConn { return c.c } - // SendRequest sends r with fds attached to its first byte. func (c *Conn) SendRequest(r Request, fds [3]int) error { f := Frame(r) if len(f.Payload) > MaxRequestBytes { return fmt.Errorf("%w: request of %d bytes exceeds %d", ErrProtocol, len(f.Payload), MaxRequestBytes) } - var b frameBuffer + var b bytes.Buffer if err := sandboxwire.WriteFrame(&b, f); err != nil { return err } - n, _, err := c.c.WriteMsgUnix(b, unix.UnixRights(fds[:]...), nil) + n, _, err := c.c.WriteMsgUnix(b.Bytes(), unix.UnixRights(fds[:]...), nil) // An empty write fails with EPIPE once the relay has run the invocation // and closed the connection, so write only what remains. - if err != nil || n == len(b) { + if err != nil || n == b.Len() { return err } - _, err = c.c.Write(b[n:]) + _, err = c.c.Write(b.Bytes()[n:]) return err } @@ -149,10 +147,3 @@ func closeAll(fds []int) { unix.Close(fd) } } - -type frameBuffer []byte - -func (b *frameBuffer) Write(p []byte) (int, error) { - *b = append(*b, p...) - return len(p), nil -} diff --git a/apps/daemon/internal/processshim/conn_linux_test.go b/apps/daemon/internal/processshim/conn_linux_test.go index b2e654dee..3bf892fce 100644 --- a/apps/daemon/internal/processshim/conn_linux_test.go +++ b/apps/daemon/internal/processshim/conn_linux_test.go @@ -1,6 +1,7 @@ package processshim import ( + "bytes" "errors" "net" "os" @@ -95,9 +96,9 @@ func TestUnexpectedDescriptorsAreRejected(t *testing.T) { } closeAll(recv[:]) frame := Frame(Signal{Number: 2}) - var buf frameBuffer + var buf bytes.Buffer _ = sandboxwire.WriteFrame(&buf, frame) - if _, _, err := shim.c.WriteMsgUnix(buf, unix.UnixRights(fds[0]), nil); err != nil { + if _, _, err := shim.c.WriteMsgUnix(buf.Bytes(), unix.UnixRights(fds[0]), nil); err != nil { t.Fatal(err) } if _, err := broker.ReadMessage(); !errors.Is(err, ErrProtocol) { diff --git a/apps/daemon/internal/processshim/endpoint_linux.go b/apps/daemon/internal/processshim/endpoint_linux.go index 284ec3690..df713477a 100644 --- a/apps/daemon/internal/processshim/endpoint_linux.go +++ b/apps/daemon/internal/processshim/endpoint_linux.go @@ -172,13 +172,13 @@ func (e endpoint) write(data []byte) (int, error) { } // readFD reads once. It returns 0 and nil at end of file. -func readFD(e endpoint, buf []byte, stops ...*stopFlag) (int, error) { +func readFD(e endpoint, buf []byte, stop *stopFlag) (int, error) { for { - if stopped(stops) { + if stop.isSet() { return 0, errStopped } if e.kind == ioShared { - if err := waitFD(e.io, unix.POLLIN, stops...); err != nil { + if err := waitFD(e.io, unix.POLLIN, stop); err != nil { return 0, err } } @@ -187,7 +187,7 @@ func readFD(e endpoint, buf []byte, stops ...*stopFlag) (int, error) { case unix.EINTR: continue case unix.EAGAIN: - if err := waitFD(e.io, unix.POLLIN, stops...); err != nil { + if err := waitFD(e.io, unix.POLLIN, stop); err != nil { return 0, err } continue @@ -197,13 +197,13 @@ func readFD(e endpoint, buf []byte, stops ...*stopFlag) (int, error) { } // writeFD writes all of data, handling short writes. -func writeFD(e endpoint, data []byte, stops ...*stopFlag) error { +func writeFD(e endpoint, data []byte, stop *stopFlag) error { for len(data) > 0 { - if stopped(stops) { + if stop.isSet() { return errStopped } if e.kind == ioShared { - if err := waitFD(e.io, unix.POLLOUT, stops...); err != nil { + if err := waitFD(e.io, unix.POLLOUT, stop); err != nil { return err } } @@ -212,7 +212,7 @@ func writeFD(e endpoint, data []byte, stops ...*stopFlag) error { case err == unix.EINTR: continue case err == unix.EAGAIN: - if err := waitFD(e.io, unix.POLLOUT, stops...); err != nil { + if err := waitFD(e.io, unix.POLLOUT, stop); err != nil { return err } continue @@ -235,21 +235,9 @@ func tryWrite(e endpoint, data []byte) { e.write(data[:min(len(data), pipeBuf)]) } -func stopped(stops []*stopFlag) bool { - for _, s := range stops { - if s.isSet() { - return true - } - } - return false -} - -// waitFD waits until fd has events or a stop flag is set. -func waitFD(fd int, events int16, stops ...*stopFlag) error { - pfds := []unix.PollFd{{Fd: int32(fd), Events: events}} - for _, s := range stops { - pfds = append(pfds, unix.PollFd{Fd: int32(s.fd), Events: unix.POLLIN}) - } +// waitFD waits until fd has events or stop is set. +func waitFD(fd int, events int16, stop *stopFlag) error { + pfds := []unix.PollFd{{Fd: int32(fd), Events: events}, {Fd: int32(stop.fd), Events: unix.POLLIN}} for { if _, err := unix.Poll(pfds, -1); err != nil { if err == unix.EINTR { @@ -257,12 +245,9 @@ func waitFD(fd int, events int16, stops ...*stopFlag) error { } return err } - for _, p := range pfds[1:] { - if p.Revents != 0 { - return errStopped - } - } switch r := pfds[0].Revents; { + case pfds[1].Revents != 0: + return errStopped case r&unix.POLLNVAL != 0: return unix.EBADF case r != 0: diff --git a/apps/daemon/internal/processshim/relay_linux.go b/apps/daemon/internal/processshim/relay_linux.go index 0bb37296b..1bb642a47 100644 --- a/apps/daemon/internal/processshim/relay_linux.go +++ b/apps/daemon/internal/processshim/relay_linux.go @@ -340,6 +340,8 @@ func (inv *invocation) receive(m BrokerMessage) { inv.out[2].push(item{hold: e.written}) } inv.control(e) + case Notice: + inv.out[2].push(item{notice: m.Message}) case End: inv.end.set() inv.control(m) @@ -393,8 +395,6 @@ func (inv *invocation) run() { inv.rawOnce.Do(func() { close(inv.raw) }) case exiting: inv.exit(m) - case Notice: - inv.out[2].message(m.Message) case End: inv.finish("") return @@ -553,15 +553,16 @@ func (in *input) run() { } } -// item is an Output or Close for an output pump, or a hold. A local Close -// closes the descriptor without a report. A hold keeps the items behind it -// until it closes. +// item is an Output, Close or Notice for an output pump, or a hold. A local +// Close closes the descriptor without a report. A hold keeps the items +// behind it until it closes. type item struct { - seq uint64 - data []byte - close bool - local bool - hold <-chan struct{} + seq uint64 + data []byte + close bool + local bool + hold <-chan struct{} + notice []byte } // output writes one descriptor's Output in order and reports each write. @@ -616,6 +617,7 @@ func (o *output) next() (item, bool) { func (o *output) run() { defer o.inv.pumps.Done() + defer o.notices() id := o.inv.id for { it, ok := o.next() @@ -630,6 +632,10 @@ func (o *output) run() { } continue } + if it.notice != nil { + o.message(it.notice) + continue + } if it.close { o.closeEP() if it.local { @@ -670,6 +676,20 @@ func (o *output) run() { } } +// notices writes the Notices still queued once the pump has stopped, each +// as far as the descriptor takes it now, so End drops none it can take. +func (o *output) notices() { + o.mu.Lock() + queue := o.queue + o.queue = nil + o.mu.Unlock() + for _, it := range queue { + if it.notice != nil { + o.message(it.notice) + } + } +} + // progress records seq as done, or the descriptor as broken, and reports // whether the broker is told. func (o *output) progress(seq uint64, failed bool) bool { diff --git a/apps/daemon/internal/processshim/relay_linux_test.go b/apps/daemon/internal/processshim/relay_linux_test.go index 16288c006..bae42ca7b 100644 --- a/apps/daemon/internal/processshim/relay_linux_test.go +++ b/apps/daemon/internal/processshim/relay_linux_test.go @@ -5,6 +5,7 @@ package processshim import ( "bufio" "fmt" + "io" "net" "os" "path/filepath" @@ -299,6 +300,134 @@ func TestTerminalOutputWaitsForRawMode(t *testing.T) { } } +// A Notice reaches fd 2 after the stderr queued before it, even while that +// stderr waits in its pump and the control goroutine has moved on. +func TestNoticeFollowsQueuedStderr(t *testing.T) { + answered := make(chan struct{}) + r := startTestRelay(t, func(r *relay) { + r.gating = func() { + select { + case <-answered: + case <-time.After(10 * time.Second): + t.Error("the shim never got its Result") + } + } + }) + go func() { + for { + m, err := r.read() + if err != nil { + return + } + switch m := m.(type) { + case Open: + id := m.ID + // The Exit has no Marks, so its Result shows that the + // control goroutine is past the Notice. + r.send(Accept{ID: id}, Started{ID: id}, Output{ID: id, FD: 2, Seq: 1, Data: []byte("err\n")}, + Notice{ID: id, Message: []byte("note")}, Exit{ID: id, Result: Result{Code: 0}}, Close{ID: id, FD: 2, Seq: 2}) + case Written: + if m.Seq == 2 { + r.send(End{ID: m.ID}) + } + } + } + }() + pr, pw, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + defer pr.Close() + fds := devNull(t) + fds[2] = int(pw.Fd()) + conn := r.shim(t, "sh", fds) + pw.Close() + if res, err := finished(conn); err != nil || res.Code != 0 { + t.Fatalf("Result %+v, %v", res, err) + } + close(answered) + pr.SetDeadline(time.Now().Add(10 * time.Second)) + got, err := io.ReadAll(pr) + if want := "err\noac-process-shim: note\n"; err != nil || string(got) != want { + t.Fatalf("fd 2 got %q, %v; want %q", got, err, want) + } +} + +// A Notice queued behind stderr still reaches fd 2 when End stops that +// stderr first, and a stderr nobody reads holds neither from ending. +func TestNoticeOutlivesEnd(t *testing.T) { + for _, full := range []bool{false, true} { + t.Run(fmt.Sprintf("full=%v", full), func(t *testing.T) { + var rl *relay + release := make(chan struct{}) + r := startTestRelay(t, func(r *relay) { + rl = r + if !full { + r.gating = func() { <-release } // the stderr waits until End has stopped it + } + }) + opened := make(chan uint64, 1) + go func() { + for { + m, err := r.read() + if err != nil { + return + } + if m, ok := m.(Open); ok { + id := m.ID + r.send(Accept{ID: id}, Started{ID: id}, Output{ID: id, FD: 2, Seq: 1, Data: []byte("err\n")}, + Notice{ID: id, Message: []byte("note")}, Exit{ID: id, Result: Result{Code: 0}}) + opened <- id + } + } + }() + pr, pw, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + defer pr.Close() + filled := 0 + if full { + fd := int(pw.Fd()) + unix.SetNonblock(fd, true) + for _, chunk := range []int{pipeBuf, 1} { + for n, err := 0, error(nil); err == nil; n, err = unix.Write(fd, make([]byte, chunk)) { + filled += n + } + } + } + fds := devNull(t) + fds[2] = int(pw.Fd()) + conn := r.shim(t, "sh", fds) + pw.Close() + if res, err := finished(conn); err != nil || res.Code != 0 { + t.Fatalf("Result %+v, %v", res, err) + } + r.send(End{ID: <-opened}) + for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(time.Millisecond) { + rl.mu.Lock() + ended := len(rl.invs) == 0 + rl.mu.Unlock() + if ended { + break + } + if time.Now().After(deadline) { + t.Fatal("the invocation never took End") + } + } + close(release) + pr.SetDeadline(time.Now().Add(10 * time.Second)) + got, err := io.ReadAll(pr) + const note = "oac-process-shim: note\n" + rest, ok := strings.CutPrefix(string(got), string(make([]byte, filled))) + // A full stderr may still be full when the Notice is tried. + if err != nil || !ok || (rest != note && !(full && rest == "")) { + t.Fatalf("fd 2 got %d bytes ending in %q, %v; want %d filler bytes and %q", len(got), got[max(0, len(got)-32):], err, filled, note) + } + }) + } +} + // readN reads n bytes from the terminal's master side. It fails the test // when they do not arrive within 10s. func readN(t *testing.T, fd, n int) []byte { diff --git a/apps/daemon/internal/sessionview/control_linux.go b/apps/daemon/internal/sessionview/control_linux.go index 2e28bf14f..34dc420a8 100644 --- a/apps/daemon/internal/sessionview/control_linux.go +++ b/apps/daemon/internal/sessionview/control_linux.go @@ -36,7 +36,6 @@ type launchSpec struct { Command command UID uint32 GID uint32 - Groups []uint32 Grace time.Duration } @@ -60,6 +59,7 @@ const ( msgSignaled // launcher: whether msgSignal reached it msgSpawn // daemon: start another process; carries a pipe with its command, then its stdin, stdout and stderr msgSpawned // launcher: the spawned process's Pid, or why none started + msgNoRelay // launcher: the relay ended before the launcher ) type message struct { diff --git a/apps/daemon/internal/sessionview/doc.go b/apps/daemon/internal/sessionview/doc.go index 8e1c63099..d198a111c 100644 --- a/apps/daemon/internal/sessionview/doc.go +++ b/apps/daemon/internal/sessionview/doc.go @@ -14,5 +14,5 @@ // // Teardown never waits unconditionally. Once the launcher has exited, the view shuts its end of the relay connection down, ends every process in its cgroup with cgroup.kill and stops the world server, which ends the requests still pending on the view's FUSE connection so that a process blocked on the world can exit. Only when the world server and the launcher have ended and the cgroup is empty, all within 30 seconds, does it remove the cgroup. Otherwise Wait and Close report [ErrCleanup] and the cgroup stays for Recover; nothing touches it after the teardown. When Start's context ends during the build, Start kills the launcher, stops waiting for it and tears the view down the same way, so a launcher blocked on the world cannot hold Start past this bound. // -// The package works only on Linux. Elsewhere [Start] returns [ErrUnsupported]. +// Views work only on Linux; elsewhere the package offers only [Init] and its types. package sessionview diff --git a/apps/daemon/internal/sessionview/errors.go b/apps/daemon/internal/sessionview/errors.go index 8c0466de9..a0f0ba559 100644 --- a/apps/daemon/internal/sessionview/errors.go +++ b/apps/daemon/internal/sessionview/errors.go @@ -8,7 +8,6 @@ import ( // Error kinds. Every error the package returns matches one of them with errors.Is. var ( - ErrUnsupported = errors.New("sessionview: unsupported platform") ErrInvalidSpec = errors.New("sessionview: invalid spec") ErrCapability = errors.New("sessionview: missing capability") ErrNoFUSE = errors.New("sessionview: /dev/fuse unavailable") diff --git a/apps/daemon/internal/sessionview/launcher_linux.go b/apps/daemon/internal/sessionview/launcher_linux.go index 440be1d62..2d2d26a60 100644 --- a/apps/daemon/internal/sessionview/launcher_linux.go +++ b/apps/daemon/internal/sessionview/launcher_linux.go @@ -345,8 +345,12 @@ func (l *launcher) reapChildren() bool { } } -// reaped retires the ID of pid, if it has one, and reports its exit. mu is held. +// reaped retires the ID of pid, if it has one, and reports its exit, or reports the relay's loss: the relay ignores the view's signals, so it ends before the launcher only when it is lost. mu is held. func (l *launcher) reaped(pid int, ws unix.WaitStatus) { + if pid == l.relay { + l.post(message{Kind: msgNoRelay}) + return + } id, ok := l.spawned[pid] if !ok { return @@ -362,7 +366,7 @@ func (l *launcher) reaped(pid int, ws unix.WaitStatus) { // exitOf describes how a process ended, with the code the launcher exits with for it. func exitOf(ws unix.WaitStatus) (Exit, int) { if ws.Signaled() { - return Exit{Signal: ws.Signal(), CoreDumped: ws.CoreDump()}, 128 + int(ws.Signal()) + return Exit{Signal: ws.Signal()}, 128 + int(ws.Signal()) } return Exit{Code: ws.ExitStatus()}, ws.ExitStatus() } @@ -417,7 +421,7 @@ func startRelay(spec *launchSpec, listener int) (int, error) { Files: files, Sys: &syscall.SysProcAttr{ Setsid: true, - Credential: &syscall.Credential{Uid: spec.UID, Gid: spec.GID, Groups: spec.Groups}, + Credential: &syscall.Credential{Uid: spec.UID, Gid: spec.GID}, // and no supplementary groups }, }) if err != nil { @@ -426,12 +430,8 @@ func startRelay(spec *launchSpec, listener int) (int, error) { return pid, nil } -// takeIdentity gives this thread a working directory of its own and the process's file system identity, with no capability but the two a fork needs to set the child's user, so that it enters a directory as the process would. The other threads keep theirs. +// takeIdentity gives this thread a working directory of its own and the process's file system identity, with no supplementary groups and no capability but the two a fork needs to set the child's user, so that it enters a directory as the process would. The other threads keep theirs. func takeIdentity(spec *launchSpec) error { - groups := make([]int, len(spec.Groups)) - for i, g := range spec.Groups { - groups[i] = int(g) - } const setID = 1<