diff --git a/apps/daemon/internal/agent/clirunner/handle.go b/apps/daemon/internal/agent/clirunner/handle.go index fd952579..c676efef 100644 --- a/apps/daemon/internal/agent/clirunner/handle.go +++ b/apps/daemon/internal/agent/clirunner/handle.go @@ -41,7 +41,7 @@ func FromHandle(h Handle, opts HandleOptions) (*Process, error) { opts.Parent = context.Background() } if opts.KillTimeout <= 0 { - opts.KillTimeout = 3 * time.Second + opts.KillTimeout = DefaultKillTimeout } ctx, cancel := context.WithCancel(opts.Parent) p := &Process{Stdin: opts.Stdin, Stdout: opts.Stdout, Stderr: opts.Stderr, ctx: ctx, cancel: cancel, done: make(chan struct{}), killAfter: opts.KillTimeout} diff --git a/apps/daemon/internal/agent/clirunner/process.go b/apps/daemon/internal/agent/clirunner/process.go index d9d40efc..fa406401 100644 --- a/apps/daemon/internal/agent/clirunner/process.go +++ b/apps/daemon/internal/agent/clirunner/process.go @@ -11,6 +11,10 @@ import ( "time" ) +// DefaultKillTimeout is the KillTimeout that a zero or negative StartOptions +// or HandleOptions KillTimeout selects. +const DefaultKillTimeout = 3 * time.Second + type StartOptions struct { Parent context.Context Binary string @@ -52,7 +56,7 @@ func Start(opts StartOptions) (*Process, error) { return nil, fmt.Errorf("clirunner: binary required") } if opts.KillTimeout <= 0 { - opts.KillTimeout = 3 * time.Second + opts.KillTimeout = DefaultKillTimeout } if opts.OwnProcessGroup || runtime.GOOS == "windows" { diff --git a/apps/daemon/internal/agenthost/admit.go b/apps/daemon/internal/agenthost/admit.go new file mode 100644 index 00000000..e4a320f9 --- /dev/null +++ b/apps/daemon/internal/agenthost/admit.go @@ -0,0 +1,254 @@ +package agenthost + +import ( + "context" + "crypto/x509" + "encoding/json" + "fmt" + "maps" + "math" + "os" + "path" + "path/filepath" + "slices" + "strings" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/gateway" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/modelprovider" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxfs" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// modelName is the gateway's name for the request's model provider. +const modelName = "model_provider" + +// etcFiles are the files the agent host writes for each Session and presents +// at /etc/. +var etcFiles = []string{"passwd", "group", "hosts", "resolv.conf", "nsswitch.conf"} + +// plan is an admitted Session: everything Run derives before any effect. +type plan struct { + view agent.View + gateway gateway.Config + // request is the request the view Executor factory receives. + request proto.PromptRequestPayload + // mcp and proxy are ViewSession.MCP and ViewSession.Proxy. + mcp []agent.MCPBinding + proxy string +} + +// checkConfig validates cfg and loads the roots in its CA directory. +func checkConfig(cfg Config) (*x509.CertPool, error) { + switch { + case !isHostPath(cfg.StateDir): + return nil, invalidConfig("state directory %q is not absolute and clean", cfg.StateDir) + case !cfg.UIDs.valid(): + return nil, invalidConfig("uid range %d+%d", cfg.UIDs.First, cfg.UIDs.Count) + case sandboxlink.CheckRelayURL(cfg.RelayURL) != nil: + return nil, invalidConfig("relay URL") + case cfg.RuntimeID.IsZero() || len(cfg.Credential) == 0: + return nil, invalidConfig("no Runtime ID or credential") + case cfg.Harnesses == nil: + return nil, invalidConfig("no Harness declarations") + case !isHostPath(cfg.Shim): + return nil, invalidConfig("shim %q is not absolute and clean", cfg.Shim) + case !isHostPath(cfg.CADir) || cfg.CADir == "/" || agent.ViewReserved(cfg.CADir): + return nil, invalidConfig("CA directory %q", cfg.CADir) + } + return loadRoots(cfg.CADir) +} + +// loadRoots reads every certificate in dir. Each entry is a regular file of +// PEM certificates, so the view presents exactly what the gateway trusts. +func loadRoots(dir string) (*x509.CertPool, error) { + entries, err := os.ReadDir(dir) + if err != nil { + return nil, &Error{Kind: ErrInvalidConfig, Op: "CA directory", Err: err} + } + if len(entries) == 0 { + return nil, invalidConfig("CA directory %s is empty", dir) + } + roots := x509.NewCertPool() + for _, e := range entries { + if !e.Type().IsRegular() { + return nil, invalidConfig("CA entry %s is not a regular file", e.Name()) + } + data, err := os.ReadFile(filepath.Join(dir, e.Name())) + if err != nil { + return nil, &Error{Kind: ErrInvalidConfig, Op: "CA directory", Err: err} + } + if !roots.AppendCertsFromPEM(data) { + return nil, invalidConfig("CA entry %s holds no PEM certificate", e.Name()) + } + } + return roots, nil +} + +// admit checks the Session and derives its plan without touching anything. +// openNetwork is the Session's Network dial for the gateway. +func admit(cfg Config, roots *x509.CertPool, s Session, openNetwork func(context.Context) (sandboxlink.Stream, error)) (*plan, error) { + if err := checkSession(s); err != nil { + return nil, err + } + req := s.Request + view, err := cfg.Harnesses.ResolveView(req.AgentKind) + if err != nil { + return nil, &Error{Kind: ErrUnsupported, Op: "admit", Err: err} + } + local := req.LocalEnvironment + switch { + case req.DisableExecutionEnvironment: + return nil, unsupported("a Session without an execution environment") + case local == nil: + return nil, unsupported("a Session without a workspace") + case !isViewPath(local.WorkspaceRoot): + return nil, invalidSession("workspace %q is not absolute and clean", local.WorkspaceRoot) + case !req.StrictResume: + return nil, unsupported("a Session without strict resume") + case local.Capabilities || len(local.Skills) > 0 || local.CapabilityRoot != "": + return nil, unsupported("installed Capabilities and skills") + case local.NetworkAccess != "enabled" || len(local.AllowedDomains) > 0: + return nil, unsupported("a restricted workspace network") + case len(req.FunctionTools) > 0 || req.ToolSearch: + // A function call waits for a result that Input cannot deliver. + return nil, unsupported("function tools and their discovery") + } + raw, ok := req.AgentOptions["model_provider"] + if !ok { + return nil, unsupported("a Session without a frozen model provider") + } + provider, err := modelprovider.ParseProvider(raw) + if err != nil { + return nil, invalidSession("model provider: %v", err) + } + bindings, err := agent.ResolveMCPBindings(req) + if err != nil { + return nil, invalidSession("MCP: %v", err) + } + for _, b := range bindings { + if b.Transport != "http" { + return nil, unsupported("%s MCP server %q", b.Transport, b.ServerLabel) + } + } + if err := checkLayout(cfg, view); err != nil { + return nil, err + } + gw := gateway.Config{ + Models: []gateway.Model{{Name: modelName, Provider: provider}}, + MCP: bindings, + Prompt: req, + OpenNetwork: openNetwork, + RootCAs: roots, + Proxy: view.Proxy == agent.ViewProxyEnv, + } + endpoints, err := gateway.Plan(gw) + if err != nil { + return nil, &Error{Kind: ErrInvalidSession, Op: "gateway", Err: err} + } + p := &plan{view: view, gateway: gw, proxy: endpoints.Proxy} + if p.request, err = handoff(req, provider, endpoints); err != nil { + return nil, err + } + for _, b := range bindings { + b.ServerURL, b.BearerToken, b.HTTPHeaders = endpoints.MCP[b.ServerLabel], nil, nil + if b.AllowedTools != nil { + tools := slices.Clone(*b.AllowedTools) + b.AllowedTools = &tools + } + p.mcp = append(p.mcp, b) + } + return p, nil +} + +// handoff rewrites the request as a view Executor receives it: the model +// provider is the gateway's listener with the placeholder key, and MCP is +// only in ViewSession.MCP. +func handoff(req proto.PromptRequestPayload, provider modelprovider.Provider, endpoints gateway.Endpoints) (proto.PromptRequestPayload, error) { + provider.BaseURL, provider.APIKey = endpoints.Models[modelName], modelprovider.Placeholder + encoded, err := json.Marshal(provider) + if err != nil { + return req, invalidSession("model provider: %v", err) + } + var option map[string]any + if err := json.Unmarshal(encoded, &option); err != nil { + return req, invalidSession("model provider: %v", err) + } + req.AgentOptions = maps.Clone(req.AgentOptions) + req.AgentOptions["model_provider"] = option + req.MCPHTTPServers = nil + local := *req.LocalEnvironment + local.MCP = nil + req.LocalEnvironment = &local + return req, nil +} + +// checkLayout rejects a view whose overlays, masks or shim paths meet the +// agent host's own overlays: the /etc files and the CA directory. +func checkLayout(cfg Config, view agent.View) error { + own := []string{cfg.CADir} + for _, name := range etcFiles { + own = append(own, "/etc/"+name) + } + claimed := slices.Clone(view.ShimPaths) + for _, o := range view.Overlays { + claimed = append(claimed, o.Path) + } + for _, m := range view.Masks { + claimed = append(claimed, m.Path) + } + for _, p := range claimed { + for _, q := range own { + if p == q || strings.HasPrefix(p, q+"/") || strings.HasPrefix(q, p+"/") { + return &Error{Kind: ErrUnsupported, Op: "admit", Err: fmt.Errorf("%w: view path %s meets the agent host's %s", agent.ErrInvalidView, p, q)} + } + } + } + return nil +} + +// checkSession checks the Session's own fields. Its binding is valid when the +// Open the Session sends is, as Link encoding checks it. +func checkSession(s Session) error { + if _, err := sandboxlink.Encode(1, s.Binding.open(sandboxlink.ServiceFile, sandboxfs.Version, sandboxwire.ID{})); err != nil { + return invalidSession("binding: %v", err) + } + if s.Input == nil || s.Output == nil { + return invalidSession("no input or output channel") + } + for _, env := range []map[string]string{s.Environment.Sandbox, s.Environment.Tool} { + for name, value := range env { + if name == "" || strings.ContainsAny(name, "=\x00") || strings.ContainsRune(value, 0) { + return invalidSession("environment variable %q", name) + } + } + } + return nil +} + +// valid reports whether r is a nonempty range of nonzero uids. +func (r UIDRange) valid() bool { + return r.First != 0 && r.Count != 0 && uint64(r.First)+uint64(r.Count) <= math.MaxUint32 +} + +func isHostPath(p string) bool { + return filepath.IsAbs(p) && filepath.Clean(p) == p && !strings.ContainsRune(p, 0) +} + +func isViewPath(p string) bool { + return strings.HasPrefix(p, "/") && path.Clean(p) == p && !strings.ContainsRune(p, 0) +} + +func unsupported(format string, args ...any) error { + return &Error{Kind: ErrUnsupported, Op: "admit", Err: fmt.Errorf("%w: %s", agent.ErrUnsupportedOperation, fmt.Sprintf(format, args...))} +} + +func invalidSession(format string, args ...any) error { + return &Error{Kind: ErrInvalidSession, Op: "admit", Err: fmt.Errorf(format, args...)} +} + +func invalidConfig(format string, args ...any) error { + return &Error{Kind: ErrInvalidConfig, Err: fmt.Errorf(format, args...)} +} diff --git a/apps/daemon/internal/agenthost/admit_linux_test.go b/apps/daemon/internal/agenthost/admit_linux_test.go new file mode 100644 index 00000000..a4a3cbba --- /dev/null +++ b/apps/daemon/internal/agenthost/admit_linux_test.go @@ -0,0 +1,169 @@ +//go:build linux + +package agenthost + +import ( + "context" + "errors" + "maps" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "strings" + "sync/atomic" + "testing" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentplugin" + "github.com/MiniMax-AI/OpenAgentCore/internal/modelprovider" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" +) + +var errFactory = errors.New("factory reached") + +// viewFixture registers "viewed", whose factory records what it receives, +// "masked", whose view masks an /etc file the agent host writes, and +// "plain", which declares no view. +type viewFixture struct { + cfg Config + req proto.PromptRequestPayload + session agent.ViewSession + homeSet bool +} + +func newViewFixture(t *testing.T) *viewFixture { + upstream := httptest.NewTLSServer(http.NotFoundHandler()) + upstream.Close() + f := &viewFixture{} + reg := agent.NewRegistry() + view := agent.View{ + Closure: []agent.ViewMount{{Name: "harness", HostDir: t.TempDir()}}, + LocalExec: []string{"/.oac/harness/harness"}, + Proxy: agent.ViewProxyEnv, + Executor: func(_ context.Context, req proto.PromptRequestPayload, s agent.ViewSession) (agent.Executor, error) { + f.req, f.session = req, s + info, err := os.Stat(s.Home.Host) + f.homeSet = err == nil && info.IsDir() + return nil, errFactory + }, + } + register(reg, "viewed", &view, "mcp_servers") + masked := view + masked.Masks = []agent.ViewMask{{Path: "/etc/passwd"}} + register(reg, "masked", &masked) + register(reg, "plain", nil) + f.cfg = newConfig(t, reg, upstream.Certificate()) + return f +} + +func TestAdmissionRejectsBeforeAnyEffect(t *testing.T) { + f := newViewFixture(t) + for name, c := range map[string]struct { + change func(*proto.PromptRequestPayload) + want []error + }{ + "kind without a view": {func(r *proto.PromptRequestPayload) { r.AgentKind = "plain" }, []error{ErrUnsupported, agent.ErrUnsupportedOperation}}, + "view meeting the agent host's /etc": {func(r *proto.PromptRequestPayload) { r.AgentKind = "masked" }, []error{ErrUnsupported, agent.ErrInvalidView}}, + "environment none": {func(r *proto.PromptRequestPayload) { + r.DisableExecutionEnvironment, r.LocalEnvironment = true, nil + }, []error{ErrUnsupported, agent.ErrUnsupportedOperation}}, + "relative workspace": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.WorkspaceRoot = "workspace" }, []error{ErrInvalidSession}}, + "no model provider": {func(r *proto.PromptRequestPayload) { delete(r.AgentOptions, "model_provider") }, []error{ErrUnsupported}}, + "no strict resume": {func(r *proto.PromptRequestPayload) { r.StrictResume = false }, []error{ErrUnsupported}}, + "capabilities": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.Capabilities = true }, []error{ErrUnsupported}}, + "restricted network": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.NetworkAccess = "disabled" }, []error{ErrUnsupported}}, + "allowed domains only": {func(r *proto.PromptRequestPayload) { r.LocalEnvironment.AllowedDomains = []string{"example.com"} }, []error{ErrUnsupported}}, + "function tools": {func(r *proto.PromptRequestPayload) { r.FunctionTools = []proto.FunctionTool{{Name: "lookup"}} }, []error{ErrUnsupported, agent.ErrUnsupportedOperation}}, + "stdio MCP": {func(r *proto.PromptRequestPayload) { + r.LocalEnvironment.MCP = []proto.EnvironmentMCP{{Server: agentplugin.MCPServer{Name: "tools", Type: "stdio", Command: "tools"}}} + }, []error{ErrUnsupported, agent.ErrUnsupportedOperation}}, + } { + req := request("viewed", "/workspace", "https://model.test", "sk-test") + c.change(&req) + s, _, _ := newSession(newResource(), req) + var dials atomic.Int32 + err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), broker: func() processBroker { return noBroker{} }, procs: &fakeProcesses{}}) + for _, want := range c.want { + if !errors.Is(err, want) { + t.Errorf("%s: Run = %v, want %v", name, err, want) + } + } + if dials.Load() != 0 || len(leftSessions(t, f.cfg)) != 0 { + t.Errorf("%s: %d dials and %d Session directories", name, dials.Load(), len(leftSessions(t, f.cfg))) + } + if _, err := os.Stat(sessionsDir(f.cfg.StateDir)); !errors.Is(err, os.ErrNotExist) { + t.Errorf("%s: the sessions directory exists", name) + } + } + // A binding that Link encoding refuses is refused before any effect. + for name, change := range map[string]func(*Binding){ + "zero assignment epoch": func(b *Binding) { b.AssignmentEpoch = 0 }, + "invalid resource kind": func(b *Binding) { b.Resource.Kind = 0 }, + "oversized attach grant": func(b *Binding) { b.AttachGrant = make([]byte, sandboxlink.MaxGrantBytes+1) }, + } { + s, _, _ := newSession(newResource(), request("viewed", "/workspace", "https://model.test", "sk-test")) + change(&s.Binding) + var dials atomic.Int32 + err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), broker: func() processBroker { return noBroker{} }, procs: &fakeProcesses{}}) + if !errors.Is(err, ErrInvalidSession) || dials.Load() != 0 { + t.Errorf("%s: Run = %v after %d dials, want ErrInvalidSession", name, err, dials.Load()) + } + if _, err := os.Stat(sessionsDir(f.cfg.StateDir)); !errors.Is(err, os.ErrNotExist) { + t.Errorf("%s: the sessions directory exists", name) + } + } +} + +func TestViewExecutorReceivesTheGatewayRequest(t *testing.T) { + f := newViewFixture(t) + bearer := "mcp-secret" + req := request("viewed", "/workspace", "https://model.test", "sk-test") + req.MCPHTTPServers = &[]proto.MCPHTTPServer{{ConnectionOrigin: "environment", ServerLabel: "docs", ServerURL: "https://mcp.test/docs?tenant=a", BearerToken: &bearer}} + original := maps.Clone(req.AgentOptions) + s, _, _ := newSession(newResource(), req) + var dials atomic.Int32 + err := run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), broker: func() processBroker { return noBroker{} }, procs: &fakeProcesses{}}) + if !errors.Is(err, ErrExecutor) || !errors.Is(err, errFactory) { + t.Fatalf("Run = %v, want the factory's error as ErrExecutor", err) + } + provider, err := modelprovider.ParseProvider(f.req.AgentOptions["model_provider"]) + if err != nil || provider.BaseURL != "http://127.0.0.1:17101" || provider.APIKey != modelprovider.Placeholder || provider.Protocol != modelprovider.Anthropic { + t.Errorf("model provider %+v, %v; want the gateway with the placeholder", provider.BaseURL, err) + } + if !reflect.DeepEqual(req.AgentOptions, original) { + t.Error("the Session's request changed") + } + if f.req.MCPHTTPServers != nil || f.req.LocalEnvironment == nil || f.req.LocalEnvironment.MCP != nil || f.req.LocalEnvironment.WorkspaceRoot != "/workspace" { + t.Error("the request still carries MCP or lost its workspace") + } + mcp := f.session.MCP + if len(mcp) != 1 || mcp[0].ServerLabel != "docs" || mcp[0].ServerURL != "http://127.0.0.1:17102/docs" || mcp[0].BearerToken != nil || mcp[0].HTTPHeaders != nil { + t.Errorf("ViewSession.MCP = %+v, want one credential-free gateway binding", mcp) + } + if f.session.Proxy != "http://127.0.0.1:17100" || f.session.Home.View != "/.oac/home" || !f.homeSet || f.session.Launch == nil { + t.Errorf("ViewSession proxy %q, home %+v (present %v)", f.session.Proxy, f.session.Home, f.homeSet) + } + if !strings.HasPrefix(f.session.Home.Host, sessionsDir(f.cfg.StateDir)+string(filepath.Separator)) { + t.Errorf("home %s is outside the Session directories", f.session.Home.Host) + } + if dials.Load() != 0 || len(leftSessions(t, f.cfg)) != 0 { + t.Errorf("%d dials and %d Session directories after Run", dials.Load(), len(leftSessions(t, f.cfg))) + } + + // A connection option reaches the factory's handoff check, which rejects + // it before the adapter runs. + req = request("viewed", "/workspace", "https://model.test", "sk-test") + req.AgentOptions["mcp_servers"] = map[string]any{} + s, _, _ = newSession(newResource(), req) + f.req = proto.PromptRequestPayload{} + err = run(context.Background(), f.cfg, s, deps{dial: countingDial(&dials), broker: func() processBroker { return noBroker{} }, procs: &fakeProcesses{}}) + if !errors.Is(err, ErrUnsupported) || !errors.Is(err, agent.ErrViewHandoff) || f.req.AgentKind != "" { + t.Errorf("Run with a connection option = %v, want ErrUnsupported and ErrViewHandoff before the adapter", err) + } + if dials.Load() != 0 || len(leftSessions(t, f.cfg)) != 0 { + t.Errorf("%d dials and %d Session directories after Run", dials.Load(), len(leftSessions(t, f.cfg))) + } +} diff --git a/apps/daemon/internal/agenthost/agenthost.go b/apps/daemon/internal/agenthost/agenthost.go new file mode 100644 index 00000000..7ad0558f --- /dev/null +++ b/apps/daemon/internal/agenthost/agenthost.go @@ -0,0 +1,156 @@ +package agenthost + +import ( + "crypto/tls" + "errors" + "log/slog" + "strings" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// Config is the agent host's own configuration, shared by its Sessions. It +// holds a credential: keep it in memory and never log it. +// +// The agent host does not yet recover Sessions that an earlier agent-host +// process left. Until it does, an agent-host process must not reuse the +// StateDir or the UIDs of an earlier one. +type Config struct { + // StateDir is an absolute host directory private to the agent host. Each + // Session's directory is StateDir/sessions/. + StateDir string + // UIDs is the range Session uids are allocated from; each Session's gid + // equals its uid. Only one agent host runs per kernel, and nothing else + // uses the range or starts session views. + UIDs UIDRange + // RelayURL and TLS reach the Link relay, as sandboxlink.DialAttach takes + // them. A nil TLS uses the system roots. + RelayURL string + TLS *tls.Config + // RuntimeID and Credential authenticate the agent host to the relay. + RuntimeID sandboxwire.ID + Credential []byte + // Harnesses holds the Harness declarations. A kind runs only when it + // declares an agent.View. + Harnesses *agent.Registry + // Shim is the absolute host path of the static oac-process-shim binary. + Shim string + // CADir is an absolute host directory of regular PEM files: the roots the + // agent host trusts. The gateway trusts exactly these for upstream TLS, + // and the view presents the directory read-only at the same path. + CADir string + // Log receives each view's presentation report. Nil discards it. + Log *slog.Logger +} + +// UIDRange is Count ids from First. First is nonzero. +type UIDRange struct { + First, Count uint32 +} + +// Session is one Session the agent host runs. +type Session struct { + // Binding is the Session's Link attachment. + Binding Binding + // Environment is what processes forwarded to the sandbox receive. + Environment Environment + // Request is the Session's frozen request. Its Input is ignored; each + // Turn's input arrives on Input. + Request proto.PromptRequestPayload + // Input carries one Turn each. Run runs them in order and ends the + // Session once Input is closed and the last Turn has settled. + Input <-chan Input + // Output receives every Turn's envelopes. Run never closes it. + Output chan<- proto.Envelope +} + +// Binding is the identity of the Session's Link attachment, as each Open +// carries it. +type Binding struct { + Resource sandboxlink.ResourceRef + AttachmentID sandboxwire.ID + SessionID sandboxwire.ID + AssignmentID sandboxwire.ID + AssignmentEpoch uint64 + // AttachGrant authorizes each Open and renewal. It is secret. + AttachGrant []byte +} + +// Environment is the remote environment policy of processes forwarded to the +// sandbox. +type Environment struct { + // Sandbox holds the Environment's fixed values, such as HOME, PATH, + // TMPDIR and LANG in the sandbox. + Sandbox map[string]string + // Tool is the Environment's tool environment. + Tool map[string]string +} + +// Input is one Turn: its run ID and its input. +type Input struct { + RunID string + Message proto.MessageInput +} + +// Error kinds. Every error Run returns matches one of them with errors.Is. +var ( + // ErrUnsupported is a platform other than Linux, or a Session that asks + // for what the agent host does not run. A Session's error also matches + // agent.ErrUnsupportedKind, agent.ErrUnsupportedOperation, + // agent.ErrViewHandoff, or agent.ErrInvalidView for a view whose paths + // meet the agent host's own overlays. + ErrUnsupported = errors.New("agenthost: unsupported") + // ErrInvalidConfig is a Config that Run rejects. + ErrInvalidConfig = errors.New("agenthost: invalid configuration") + // ErrInvalidSession is a malformed Session. + ErrInvalidSession = errors.New("agenthost: invalid session") + // ErrCapacity means every Session uid is in use. + ErrCapacity = errors.New("agenthost: no free session uid") + // ErrSessionExists means the Session's directory already exists. + ErrSessionExists = errors.New("agenthost: session directory exists") + // ErrExecutor is a view Executor factory that failed. + ErrExecutor = errors.New("agenthost: view executor failed") + // ErrLink is a Link attachment that failed or ended. + ErrLink = errors.New("agenthost: link attachment failed") + // ErrWorld is a world that no longer shows the sandbox faithfully, or + // that cannot show that its attachment holds nothing. + ErrWorld = errors.New("agenthost: world lost") + // ErrLaunch is a view that could not be launched. + ErrLaunch = errors.New("agenthost: launch failed") + // ErrProcessBroker is a process broker that could not start. + ErrProcessBroker = errors.New("agenthost: process broker failed") + // ErrTurn is a Turn that failed or left its Executor unusable. + ErrTurn = errors.New("agenthost: turn failed") + // ErrTeardown is a Session resource that could not be released. + ErrTeardown = errors.New("agenthost: teardown incomplete") +) + +// Error is a typed agent host failure. It matches Kind and, when present, +// Err. Its message never includes a credential. +type Error struct { + Kind error + Op string + Err error +} + +func (e *Error) Error() string { + var b strings.Builder + b.WriteString(e.Kind.Error()) + if e.Op != "" { + b.WriteString(": " + e.Op) + } + if e.Err != nil { + b.WriteString(": " + e.Err.Error()) + } + return b.String() +} + +func (e *Error) Unwrap() []error { + if e.Err == nil { + return []error{e.Kind} + } + return []error{e.Kind, e.Err} +} diff --git a/apps/daemon/internal/agenthost/agenthost_linux_test.go b/apps/daemon/internal/agenthost/agenthost_linux_test.go new file mode 100644 index 00000000..8c7243f6 --- /dev/null +++ b/apps/daemon/internal/agenthost/agenthost_linux_test.go @@ -0,0 +1,121 @@ +//go:build linux + +package agenthost + +import ( + "context" + "crypto/x509" + "encoding/pem" + "errors" + "os" + "path/filepath" + "sync/atomic" + "testing" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/sessionview" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto/prototest" + "github.com/MiniMax-AI/OpenAgentCore/internal/harnessconfig" + "github.com/MiniMax-AI/OpenAgentCore/internal/modelprovider" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// The test binary is also the privileged suite's Harness inside the view and +// its process with a zombie leader. +func TestMain(m *testing.M) { + sessionview.Init() + if os.Getenv(harnessEnv) != "" { + os.Exit(runHarness(os.Args[1:])) + } + if os.Getenv(zombieLeaderEnv) != "" { + runZombieLeader() + } + os.Exit(m.Run()) +} + +// newConfig returns a Config whose CA directory holds ca. +func newConfig(t *testing.T, reg *agent.Registry, ca *x509.Certificate) Config { + t.Helper() + dir := t.TempDir() + if err := os.Chmod(dir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "ca.pem"), pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: ca.Raw}), 0o644); err != nil { + t.Fatal(err) + } + exe, err := os.Executable() + if err != nil { + t.Fatal(err) + } + return Config{StateDir: t.TempDir(), UIDs: UIDRange{First: 70000, Count: 8}, RelayURL: "ws://127.0.0.1:9", RuntimeID: sandboxwire.NewID(), + Credential: []byte("runtime-credential"), Harnesses: reg, Shim: exe, CADir: dir} +} + +// register declares kind with view, or without one when view is nil. +func register(reg *agent.Registry, kind string, view *agent.View, connection ...string) { + info := proto.SupportedAgentKind{Kind: kind, Available: true, Capabilities: prototest.Capabilities(proto.AgentKindCapabilities{})} + declaration := agent.Declaration{Info: info, ConnectionOptions: connection, + Configuration: harnessconfig.Configuration{Providers: []harnessconfig.Provider{{Protocol: string(modelprovider.Anthropic)}}}} + reg.Register(declaration, agent.Runtime{Info: info, View: view, + Session: func(context.Context, proto.PromptRequestPayload, chan<- proto.Envelope) (agent.Session, error) { + return nil, errors.New("not used") + }}) +} + +// request is a Session request the agent host admits. +func request(kind, workspace, baseURL, key string) proto.PromptRequestPayload { + return proto.PromptRequestPayload{ + AgentKind: kind, + StrictResume: true, + AgentOptions: map[string]any{"model": "m", "model_provider": map[string]any{"protocol": "anthropic", "base_url": baseURL, "api_key": key}}, + LocalEnvironment: &proto.LocalEnvironment{WorkspaceRoot: workspace, NetworkAccess: "enabled"}, + } +} + +// newSession returns a Session for req on a fresh attachment of resource. +func newSession(resource sandboxlink.ResourceRef, req proto.PromptRequestPayload) (Session, chan Input, chan proto.Envelope) { + in, out := make(chan Input), make(chan proto.Envelope, 16) + return Session{ + Binding: Binding{Resource: resource, AttachmentID: sandboxwire.NewID(), SessionID: sandboxwire.NewID(), + AssignmentID: sandboxwire.NewID(), AssignmentEpoch: 1, AttachGrant: []byte("grant-" + sandboxwire.NewID().String())}, + Request: req, Input: in, Output: out, + }, in, out +} + +func newResource() sandboxlink.ResourceRef { + return sandboxlink.ResourceRef{TenantID: sandboxwire.NewID(), EnvironmentID: sandboxwire.NewID(), Kind: sandboxlink.ResourceAllocation, + ID: sandboxwire.NewID(), Generation: 1} +} + +// countingDial counts dials and connects nothing. +func countingDial(n *atomic.Int32) dialFunc { + return func(context.Context, func(sandboxlink.AttachmentClosed)) (attachLink, error) { + n.Add(1) + return nil, errors.New("no relay in this test") + } +} + +// noBroker is a process broker that serves nothing. +type noBroker struct{} + +func (noBroker) Start(brokerConfig) error { return nil } +func (noBroker) Close() error { return nil } + +// leftSessions lists what remains under the state directory's sessions. +func leftSessions(t *testing.T, cfg Config) []os.DirEntry { + t.Helper() + entries, err := os.ReadDir(sessionsDir(cfg.StateDir)) + if err != nil && !errors.Is(err, os.ErrNotExist) { + t.Fatal(err) + } + return entries +} + +// fakeProcesses is a process table that lists fixed tasks. +type fakeProcesses struct { + list []task +} + +func (f *fakeProcesses) tasks() ([]task, error) { return f.list, nil } diff --git a/apps/daemon/internal/agenthost/broker.go b/apps/daemon/internal/agenthost/broker.go new file mode 100644 index 00000000..de5d0866 --- /dev/null +++ b/apps/daemon/internal/agenthost/broker.go @@ -0,0 +1,52 @@ +package agenthost + +import ( + "context" + "errors" + "io" + "time" +) + +// processBroker runs the commands the view's shims forward to the sandbox +// over the Process service. One broker serves a Session from its first launch +// until teardown. +type processBroker interface { + // Start begins serving the Session's run directory. + Start(brokerConfig) error + // Close cancels and releases the remote operations that remain and stops + // serving. + Close() error +} + +// brokerConfig is what a Session's broker serves. +type brokerConfig struct { + // RunDir is the host directory the view presents read-only at + // agent.ViewPrivateRoot/agent.ViewRunName. + RunDir string + // UID and GID are the Session's. + UID, GID uint32 + // Names maps each shim name to the program it runs in the sandbox, found + // on the remote PATH; Paths maps each view path the shim is bound over to + // the same sandbox path. + Names, Paths map[string]string + // Pass names the Harness variables a forwarded process keeps + // (agent.View.ForwardEnv). + Pass []string + // Sandbox and Tool are the Session's Environment. + Sandbox, Tool map[string]string + // Dial opens a Process stream on the Session's attachment. + Dial func(context.Context) (io.ReadWriteCloser, error) + // CancelGrace is the grace of a Cancel the broker sends on its own. + CancelGrace time.Duration +} + +// errNoBroker is unavailableBroker's Start error. +var errNoBroker = errors.New("no process broker in this build") + +// unavailableBroker is the broker this build has. Its Start fails, so a +// Session admits, prepares its Executor and fails its first launch with +// ErrProcessBroker. +type unavailableBroker struct{} + +func (unavailableBroker) Start(brokerConfig) error { return errNoBroker } +func (unavailableBroker) Close() error { return nil } diff --git a/apps/daemon/internal/agenthost/doc.go b/apps/daemon/internal/agenthost/doc.go new file mode 100644 index 00000000..26014d6b --- /dev/null +++ b/apps/daemon/internal/agenthost/doc.go @@ -0,0 +1,53 @@ +// Package agenthost runs Sessions whose Harness runs on the agent host, next to +// Core, while its tools, files and network act in the sandbox through the +// Session's Link attachment. It needs Linux; elsewhere Run returns +// ErrUnsupported. +// +// Run runs one Session. It admits the Session before any effect: the kind +// must declare an agent.View, and the request must use only what a view runs +// and no function tools, whose results Input cannot carry. It then allocates +// the Session uid, skipping each uid that a running thread holds as its real, +// effective, saved or file-system uid; this check only detects a conflict and +// never ends a process. It creates the Session directory under +// Config.StateDir, rewrites the request so the model provider and HTTP MCP +// reach the network only through the Session's gateway, and calls the view's +// Executor factory. The first ViewSession.Launch starts the process broker; +// each Launch builds one sessionview view, of which one at a time is live, +// over the world that worldfs serves from the attachment's File service, with +// the gateway listening in the view's network namespace. +// +// Each view presents the closure directories read-only and executable, the +// Session home read-write and noexec, the broker's run directory read-only, +// the agent host's /etc/passwd, group, hosts, resolv.conf and nsswitch.conf, +// the agent host's CA directory at its host path, then the adapter's overlays +// and masks and the process shim. Everything else is the world. +// +// The agent host owns the Session's Link attachment: it opens each stream +// with the Session's binding, renews the lease and fails the Session when the +// relay closes the attachment, a Link request fails in a way that is not +// retryable, or the world is lost or did not stop cleanly, which leaves what +// the attachment holds uncertain. A failure cancels the running Turn and +// closes the live view. A view's end is settled before its clirunner.Process +// reports it: the gateway has stopped, the world's end is recorded and the +// view slot is free. Teardown releases, in order, the Executor, the view, the +// process broker, the Link attachment, the Session directory and the uid; +// Run decides its result only afterwards, so a failure recorded during +// teardown counts, and from the close of the attachment on, what the Link +// reports changes nothing. When Executor.Close fails, teardown ends the +// views, which kills their processes, and retries Close once. If Close +// fails again, the Executor may still use the Session directory: Run returns +// ErrTeardown and keeps the directory, and the uid stays in use until the +// agent host exits. +// +// Run drives each Turn as the daemon's dispatch drives a prepared execution. +// One output consumer starts before StartTurn and forwards the Turn's +// envelopes to Output in order. The Turn's Done waits until the Turn has +// settled and, when the Turn leaves the Executor unusable, until the +// Executor has closed; a failed Turn publishes an Error envelope before it. +// When Close fails, nothing more is published. A Turn that fails or leaves +// the Executor unusable ends the Session, and nothing is sent to Output +// after Run returns. +// +// The Harness view protocol is in contracts/agents-api/harness-onboarding.md +// and the gateway's in contracts/agents-api/model-execution.md. +package agenthost diff --git a/apps/daemon/internal/agenthost/launch_linux.go b/apps/daemon/internal/agenthost/launch_linux.go new file mode 100644 index 00000000..a87ebb1c --- /dev/null +++ b/apps/daemon/internal/agenthost/launch_linux.go @@ -0,0 +1,351 @@ +//go:build linux + +package agenthost + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "slices" + "sync" + "syscall" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent/clirunner" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/gateway" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/sessionview" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/worldfs" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" +) + +// worldExport is the File service export that holds the sandbox's world, as +// docs/sandbox-bootstrap.md defines it. +const worldExport sandboxlink.ExportID = "world" + +// liveView is the Session's one live view slot. +type liveView struct { + view runningView // nil while the view is being built + closed bool // the Session ended while the view was being built +} + +// runningView is the part of *sessionview.View the Session owns. +type runningView interface { + Signal(syscall.Signal) error + Wait() (sessionview.Exit, error) + Close() error +} + +// viewWorld is the part of *worldfs.World the Session watches. +type viewWorld interface { + Stop() error + Lost() <-chan struct{} + Err() error +} + +// closeLive closes the live view. It runs when the Session's context ends. +func (s *session) closeLive() { + s.mu.Lock() + lv := s.live + var v runningView + if lv != nil { + lv.closed, v = true, lv.view + } + s.mu.Unlock() + if v != nil { + v.Close() + } +} + +// launch is ViewSession.Launch: it builds one view and runs opts.Binary in it. +func (s *session) launch(opts clirunner.StartOptions) (*clirunner.Process, error) { + switch { + case !slices.Contains(s.plan.view.LocalExec, opts.Binary): + return nil, &Error{Kind: ErrLaunch, Err: fmt.Errorf("%q is not a LocalExec path", opts.Binary)} + case !isViewPath(opts.Dir): + return nil, &Error{Kind: ErrLaunch, Err: fmt.Errorf("directory %q is not absolute and clean", opts.Dir)} + case !opts.OwnProcessGroup: + return nil, &Error{Kind: ErrLaunch, Err: errors.New("a view process runs in its own process group")} + } + if opts.Parent == nil { + opts.Parent = context.Background() + } + if opts.KillTimeout <= 0 { + opts.KillTimeout = clirunner.DefaultKillTimeout + } + s.mu.Lock() + switch { + case s.ctx.Err() != nil: + s.mu.Unlock() + return nil, &Error{Kind: ErrLaunch, Err: errors.New("the Session is ending")} + case s.live != nil: + s.mu.Unlock() + return nil, &Error{Kind: ErrLaunch, Err: errors.New("the Session already has a live view")} + } + lv := &liveView{} + s.live = lv + s.views.Add(1) + s.mu.Unlock() + return s.start(lv, opts) +} + +// release frees the view slot and ends the launch's count. +func (s *session) release(lv *liveView) { + s.mu.Lock() + if s.live == lv { + s.live = nil + } + s.mu.Unlock() + s.views.Done() +} + +// start builds the view for lv. Until the view runs, each failure releases +// lv; from then on the view's owner does. +func (s *session) start(lv *liveView, opts clirunner.StartOptions) (*clirunner.Process, error) { + if err := s.startBroker(opts.KillTimeout); err != nil { + s.release(lv) + return nil, err + } + if err := s.dir.chownHome(s.uid); err != nil { + s.release(lv) + return nil, &Error{Kind: ErrLaunch, Op: "home", Err: err} + } + ends, err := newStdio(opts.NeedStdin) + if err != nil { + s.release(lv) + return nil, &Error{Kind: ErrLaunch, Op: "stdio", Err: err} + } + // The gateway serves until the view has ended. + viewCtx, stopGateway := context.WithCancel(context.Background()) + world := worldfs.New(worldExport, s.openFile) + spec := s.spec(viewCtx, world, opts, ends) + + // Construction ends with the Session or with the caller. + startCtx, cancel := context.WithCancel(s.ctx) + stop := context.AfterFunc(opts.Parent, cancel) + v, err := sessionview.Start(startCtx, spec) + stop() + cancel() + ends.closeChild() + if err != nil { + stopGateway() + ends.closeParent() + defer s.release(lv) + // sessionview stops a world that served; Stop reports how that went. + if serr := world.Stop(); serr != nil || errors.Is(err, worldfs.ErrAttachmentDirty) { + return nil, s.worldEnded("launch", errors.Join(err, serr)) + } + return nil, &Error{Kind: ErrLaunch, Err: err} + } + p := v.Presentation() + s.log.Info("agent host view started", "binary", opts.Binary, "targets", p.Targets, "links", p.Links, "synthesized", p.Synthesized) + return s.own(lv, v, world, stopGateway, opts, ends) +} + +// own hands a started view to the clirunner.Process the adapter receives. +func (s *session) own(lv *liveView, v runningView, world viewWorld, stopGateway func(), opts clirunner.StartOptions, ends *stdio) (*clirunner.Process, error) { + h := &ownedView{s: s, lv: lv, v: v, world: world, stopGateway: stopGateway, ended: make(chan struct{}), watched: make(chan struct{})} + go h.watch() + s.mu.Lock() + lv.view = v + closed := lv.closed + s.mu.Unlock() + if closed { + v.Close() + h.Wait() + ends.closeParent() + return nil, &Error{Kind: ErrLaunch, Err: errors.New("the Session is ending")} + } + process, err := clirunner.FromHandle(h, clirunner.HandleOptions{Parent: opts.Parent, Stdin: ends.stdin(), + Stdout: ends.parent[1], Stderr: ends.parent[2], KillTimeout: opts.KillTimeout}) + if err != nil { + v.Close() + h.Wait() + ends.closeParent() + return nil, &Error{Kind: ErrLaunch, Err: err} + } + return process, nil +} + +// ownedView is a running view as a clirunner.Handle. Its Wait ends the +// Session's ownership of the view before it returns, so the end the adapter +// observes through the Process comes after it: the gateway has stopped, a +// lost world or one that did not stop cleanly has failed the Session, and +// the view slot is free for the next Launch. +type ownedView struct { + s *session + lv *liveView + v runningView + world viewWorld + stopGateway func() + ended chan struct{} // closed once the view has ended + watched chan struct{} // closed when watch returns + once sync.Once +} + +// watch fails the Session as soon as the world is lost while the view runs. +func (h *ownedView) watch() { + defer close(h.watched) + select { + case <-h.world.Lost(): + h.s.fail(&Error{Kind: ErrWorld, Op: "world", Err: h.world.Err()}) + case <-h.ended: + } +} + +func (h *ownedView) Signal(sig syscall.Signal) error { return h.v.Signal(sig) } + +func (h *ownedView) Close() error { return h.v.Close() } + +func (h *ownedView) Wait() (int, error) { + exit, err := h.v.Wait() + h.once.Do(h.end) + switch { + case err != nil: + return -1, err + case exit.Signal != 0: + return -1, nil + } + return exit.Code, nil +} + +// end releases the view once it has ended and its world has stopped. +func (h *ownedView) end() { + h.stopGateway() + close(h.ended) + <-h.watched + if lost := h.world.Err(); lost != nil { + h.s.fail(&Error{Kind: ErrWorld, Op: "world", Err: lost}) + } + // The view has stopped its world; Stop reports how that went. + if err := h.world.Stop(); err != nil { + h.s.worldEnded("stop world", err) + } + h.s.release(h.lv) +} + +// startBroker starts the Session's process broker at its first launch. +func (s *session) startBroker(grace time.Duration) error { + s.brokerMu.Lock() + defer s.brokerMu.Unlock() + if s.broker != nil { + return nil + } + view := s.plan.view + names := make(map[string]string, len(view.Shims)) + for _, n := range view.Shims { + names[n] = n + } + paths := make(map[string]string, len(view.ShimPaths)) + for _, p := range view.ShimPaths { + paths[p] = p + } + b := s.deps.broker() + err := b.Start(brokerConfig{RunDir: s.dir.entry(runEntry), UID: s.uid, GID: s.uid, Names: names, Paths: paths, + Pass: slices.Clone(view.ForwardEnv), Sandbox: s.in.Environment.Sandbox, Tool: s.in.Environment.Tool, + Dial: s.openProcess, CancelGrace: grace}) + if err != nil { + err = &Error{Kind: ErrProcessBroker, Err: err} + s.fail(err) + return err + } + s.broker = b + return nil +} + +// spec builds the view: the closure, home and run directories, the agent +// host's /etc files and CA directory, the adapter's overlays and masks, the +// shim and the gateway in the view's network namespace. +func (s *session) spec(viewCtx context.Context, world *worldfs.World, opts clirunner.StartOptions, ends *stdio) sessionview.Spec { + view := s.plan.view + var private []sessionview.PrivateDir + for _, m := range view.Closure { + private = append(private, sessionview.PrivateDir{Name: m.Name, HostDir: m.HostDir, Exec: true}) + } + private = append(private, + sessionview.PrivateDir{Name: agent.ViewHomeName, HostDir: s.dir.entry(homeEntry), Writable: true}, + sessionview.PrivateDir{Name: agent.ViewRunName, HostDir: s.dir.entry(runEntry)}) + var overlays []sessionview.Overlay + for _, name := range etcFiles { + overlays = append(overlays, sessionview.Overlay{Path: "/etc/" + name, Source: s.dir.entry(etcEntry, name)}) + } + overlays = append(overlays, sessionview.Overlay{Path: s.cfg.CADir, Source: s.cfg.CADir}) + for _, o := range view.Overlays { + overlays = append(overlays, sessionview.Overlay{Path: o.Path, Source: o.Source, Exec: o.Exec}) + } + for _, m := range view.Masks { + source := s.dir.entry(maskEntry, "file") + if m.Dir { + source = s.dir.entry(maskEntry, "dir") + } + overlays = append(overlays, sessionview.Overlay{Path: m.Path, Source: source}) + } + return sessionview.Spec{ + World: world.Serve, + Private: private, + 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], + Grace: opts.KillTimeout}, + Network: sessionview.Network{Setup: func(netns *os.File) error { + _, err := gateway.Start(viewCtx, gateway.SessionNetwork{Namespace: netns}, s.plan.gateway) + return err + }}, + StagingParent: s.dir.entry(stagingEntry), + } +} + +// 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 { + return nil + } + return e.parent[0] +} + +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 { + f.Close() + } + } +} diff --git a/apps/daemon/internal/agenthost/link.go b/apps/daemon/internal/agenthost/link.go new file mode 100644 index 00000000..51ca9396 --- /dev/null +++ b/apps/daemon/internal/agenthost/link.go @@ -0,0 +1,274 @@ +package agenthost + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// attachLink is the part of *sandboxlink.AttachLink the Session uses. +type attachLink interface { + OpenService(context.Context, sandboxlink.Open) (sandboxlink.Stream, sandboxlink.Opened, error) + Renew(context.Context, sandboxlink.RenewAttachment) (sandboxlink.AttachmentRenewed, error) + CloseAttachment(context.Context, sandboxwire.ID) error + Done() <-chan struct{} + Close() error +} + +// dialFunc connects an attach link; onClosed is its OnAttachmentClosed. +type dialFunc func(ctx context.Context, onClosed func(sandboxlink.AttachmentClosed)) (attachLink, error) + +func relayDial(cfg Config) dialFunc { + return func(ctx context.Context, onClosed func(sandboxlink.AttachmentClosed)) (attachLink, error) { + return sandboxlink.DialAttach(ctx, sandboxlink.AttachConfig{URL: cfg.RelayURL, TLS: cfg.TLS, RuntimeID: cfg.RuntimeID, + Credential: cfg.Credential, OnAttachmentClosed: onClosed}) + } +} + +const ( + // retryWait is the pause between attempts of a renewal or close that + // failed with a retryable error. + retryWait = 250 * time.Millisecond + // closeBound bounds CloseAttachment at teardown, redials included. + closeBound = 10 * time.Second +) + +// linkOwner owns the Session's Link attachment. It dials the relay when a +// stream is first needed and again after the link drops, pins the service +// instance the first Opened reports, renews the lease before it passes, and +// fails the Session on any Link failure that is not retryable and on the +// relay closing the attachment. +type linkOwner struct { + dial dialFunc + binding Binding + fail func(error) + + dialMu sync.Mutex // serializes dials + mu sync.Mutex + link attachLink + // instance is the service instance of the first Opened; zero before. + instance sandboxwire.ID + lease time.Time + // opened records that an Open was sent, so the attachment may exist. + opened bool + closing bool + renewer chan struct{} // closed when the renewal loop returns; nil before it starts + stop context.CancelFunc +} + +func newLinkOwner(dial dialFunc, b Binding, fail func(error)) *linkOwner { + return &linkOwner{dial: dial, binding: b, fail: fail} +} + +// open is the Open that carries b for service. +func (b Binding) open(service sandboxlink.Service, version uint16, expected sandboxwire.ID) sandboxlink.Open { + return sandboxlink.Open{Service: service, Version: version, Resource: b.Resource, ExpectedServerInstanceID: expected, + AttachmentID: b.AttachmentID, SessionID: b.SessionID, AssignmentID: b.AssignmentID, AssignmentEpoch: b.AssignmentEpoch, + AttachGrant: b.AttachGrant} +} + +// current returns the live link, dialing a new one when there is none. +func (l *linkOwner) current(ctx context.Context) (attachLink, error) { + l.dialMu.Lock() + defer l.dialMu.Unlock() + l.mu.Lock() + link := l.link + l.mu.Unlock() + if link != nil { + select { + case <-link.Done(): + link.Close() + default: + return link, nil + } + } + link, err := l.dial(ctx, l.closed) + if err != nil { + return nil, err + } + l.mu.Lock() + l.link = link + l.mu.Unlock() + return link, nil +} + +// open opens a stream of service on the Session's attachment. +func (l *linkOwner) open(ctx context.Context, service sandboxlink.Service, version uint16) (sandboxlink.Stream, error) { + l.mu.Lock() + closing := l.closing + l.mu.Unlock() + if closing { + return nil, &Error{Kind: ErrLink, Op: "open " + service.String(), Err: errors.New("the Session is ending")} + } + link, err := l.current(ctx) + if err != nil { + return nil, l.observe("dial", err) + } + l.mu.Lock() + if l.closing { + l.mu.Unlock() + return nil, &Error{Kind: ErrLink, Op: "open " + service.String(), Err: errors.New("the Session is ending")} + } + l.opened = true + expected := l.instance + l.mu.Unlock() + st, opened, err := link.OpenService(ctx, l.binding.open(service, version, expected)) + if err != nil { + return nil, l.observe("open "+service.String(), err) + } + l.mu.Lock() + defer l.mu.Unlock() + if l.instance.IsZero() { + l.instance = opened.ServerInstanceID + } + if opened.LeaseExpiresAt.After(l.lease) { + l.lease = opened.LeaseExpiresAt + } + if l.renewer == nil && !l.closing { + ctx, stop := context.WithCancel(context.Background()) + l.renewer, l.stop = make(chan struct{}), stop + go l.renew(ctx) + } + return st, nil +} + +// observe fails the Session on a Link failure that is not retryable and +// returns err as a typed error. +func (l *linkOwner) observe(op string, err error) error { + err = &Error{Kind: ErrLink, Op: op, Err: err} + if !retryable(err) { + l.report(err) + } + return err +} + +// report fails the Session with err unless close has begun: from then on the +// Session's own close of the attachment explains whatever the Link reports. +func (l *linkOwner) report(err error) { + l.mu.Lock() + defer l.mu.Unlock() + if !l.closing { + l.fail(err) + } +} + +// retryable reports whether a failed Link request may succeed later: a Link +// failure whose code says so, or a transport failure. +func retryable(err error) bool { + var le *sandboxlink.Error + return !errors.As(err, &le) || le.Code.Retryable() +} + +// closed is the link's OnAttachmentClosed. It never blocks. +func (l *linkOwner) closed(c sandboxlink.AttachmentClosed) { + if c.AttachmentID == l.binding.AttachmentID { + l.report(&Error{Kind: ErrLink, Op: "attachment", Err: fmt.Errorf("the relay closed the attachment (reason %d)", c.Reason)}) + } +} + +// renew extends the lease at half its remaining time until ctx ends. A +// renewal that fails retryably is tried again until the lease passes. +func (l *linkOwner) renew(ctx context.Context) { + defer close(l.renewer) + for { + l.mu.Lock() + lease := l.lease + l.mu.Unlock() + timer := time.NewTimer(time.Until(lease) / 2) + select { + case <-ctx.Done(): + timer.Stop() + return + case <-timer.C: + } + for { + if !time.Now().Before(lease) { + l.report(&Error{Kind: ErrLink, Op: "renew", Err: sandboxlink.LeaseExpired}) + return + } + attempt, cancel := context.WithDeadline(ctx, lease) + renewed, err := l.renewOnce(attempt) + cancel() + if ctx.Err() != nil { + return + } + if err == nil { + l.mu.Lock() + if renewed.LeaseExpiresAt.After(l.lease) { + l.lease = renewed.LeaseExpiresAt + } + l.mu.Unlock() + break + } + if err = l.observe("renew", err); !retryable(err) { + return + } + if !sleep(ctx, retryWait) { + return + } + } + } +} + +func (l *linkOwner) renewOnce(ctx context.Context) (sandboxlink.AttachmentRenewed, error) { + link, err := l.current(ctx) + if err != nil { + return sandboxlink.AttachmentRenewed{}, err + } + return link.Renew(ctx, sandboxlink.RenewAttachment{AttachmentID: l.binding.AttachmentID, AttachGrant: l.binding.AttachGrant}) +} + +// close stops renewal, closes the attachment when an Open may have created +// it, and closes the link. Later opens fail. +func (l *linkOwner) close() error { + l.mu.Lock() + l.closing = true + opened, renewer, stop := l.opened, l.renewer, l.stop + l.mu.Unlock() + if stop != nil { + stop() + <-renewer + } + var err error + if opened { + ctx, cancel := context.WithTimeout(context.Background(), closeBound) + for { + var link attachLink + if link, err = l.current(ctx); err == nil { + err = link.CloseAttachment(ctx, l.binding.AttachmentID) + } + if err == nil || !retryable(err) || !sleep(ctx, retryWait) { + break + } + } + cancel() + } + l.mu.Lock() + link := l.link + l.link = nil + l.mu.Unlock() + if link != nil { + link.Close() + } + if err != nil { + return &Error{Kind: ErrTeardown, Op: "close attachment", Err: err} + } + return nil +} + +// sleep waits d and reports whether ctx is still live. +func sleep(ctx context.Context, d time.Duration) bool { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} diff --git a/apps/daemon/internal/agenthost/procs_linux.go b/apps/daemon/internal/agenthost/procs_linux.go new file mode 100644 index 00000000..fe7ba350 --- /dev/null +++ b/apps/daemon/internal/agenthost/procs_linux.go @@ -0,0 +1,141 @@ +//go:build linux + +package agenthost + +import ( + "bufio" + "bytes" + "errors" + "fmt" + "io/fs" + "os" + "strconv" + "strings" + + "golang.org/x/sys/unix" +) + +// task is one running task, a thread of a process, as /proc lists it. +type task struct { + tgid, tid int + uids [4]uint32 // real, effective, saved and file-system +} + +func (r UIDRange) has(id uint32) bool { return id >= r.First && id-r.First < r.Count } + +// processTable lists the host's tasks. Tests replace it. +type processTable interface { + // tasks returns every task that runs, each thread of each process; a + // zombie runs nothing and is left out. + tasks() ([]task, error) +} + +// heldUIDs returns the uids in r that a running task holds. +func heldUIDs(procs processTable, r UIDRange) (map[uint32]bool, error) { + tasks, err := procs.tasks() + if err != nil { + return nil, err + } + held := map[uint32]bool{} + for _, t := range tasks { + for _, id := range t.uids { + if r.has(id) { + held[id] = true + } + } + } + return held, nil +} + +// procfs is the host's /proc. +type procfs struct{} + +// errGone is a process or task that has ended. +var errGone = errors.New("ended") + +func (procfs) tasks() ([]task, error) { + pids, err := os.ReadDir("/proc") + if err != nil { + return nil, err + } + var list []task + for _, p := range pids { + pid, err := strconv.Atoi(p.Name()) + if err != nil || pid <= 0 { + continue + } + tids, err := os.ReadDir(procPath(pid, "task")) + if errors.Is(err, fs.ErrNotExist) || errors.Is(err, unix.ESRCH) { + continue + } + if err != nil { + return nil, err + } + for _, e := range tids { + tid, err := strconv.Atoi(e.Name()) + if err != nil { + continue + } + s, err := readStatus(procPath(pid, "task", e.Name(), "status")) + if errors.Is(err, errGone) { + continue + } + if err != nil { + return nil, err + } + if s.running() { + list = append(list, task{tgid: pid, tid: tid, uids: s.uids}) + } + } + } + return list, nil +} + +func procPath(pid int, name ...string) string { + return "/proc/" + strconv.Itoa(pid) + "/" + strings.Join(name, "/") +} + +// status is what a status file in /proc reports. +type status struct { + state string + uids [4]uint32 +} + +func (s status) running() bool { + return !strings.HasPrefix(s.state, "Z") && !strings.HasPrefix(s.state, "X") +} + +// readStatus reads a status file in /proc. It returns errGone when the +// process or task has ended. +func readStatus(path string) (status, error) { + var s status + data, err := os.ReadFile(path) + if errors.Is(err, fs.ErrNotExist) || errors.Is(err, unix.ESRCH) { + return s, errGone + } + if err != nil { + return s, err + } + var uids []string + sc := bufio.NewScanner(bytes.NewReader(data)) + for sc.Scan() { + key, value, _ := strings.Cut(sc.Text(), ":") + switch key { + case "State": + s.state = strings.TrimSpace(value) + case "Uid": + uids = strings.Fields(value) + } + } + if len(uids) != 4 { + return s, fmt.Errorf("%s has no uids", path) + } + for i, f := range uids { + id, err := strconv.ParseUint(f, 10, 32) + if err != nil { + return s, fmt.Errorf("%s: uid %q", path, f) + } + s.uids[i] = uint32(id) + } + return s, nil +} diff --git a/apps/daemon/internal/agenthost/run_linux.go b/apps/daemon/internal/agenthost/run_linux.go new file mode 100644 index 00000000..0cb07bfb --- /dev/null +++ b/apps/daemon/internal/agenthost/run_linux.go @@ -0,0 +1,485 @@ +//go:build linux + +package agenthost + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "os" + "sync" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxfs" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxnet" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxprocess" +) + +const ( + // turnBuffer is how many envelopes a Turn may emit ahead of Output. + turnBuffer = 64 + // nativeBound bounds a Turn's settlement and each Executor.Close, as + // dispatch's preparedCancelTimeout does. + nativeBound = 10 * time.Second +) + +// deps are the parts tests replace. +type deps struct { + dial dialFunc + broker func() processBroker + procs processTable +} + +// Run runs one Session until Input is closed, ctx ends or the Session fails, +// then tears it down. It returns nil after Input closed and every Turn +// settled, ctx's error when ctx ended the Session, and otherwise the error +// that ended it, joined with any view cleanup and teardown failure. A failure +// recorded during teardown counts. +func Run(ctx context.Context, cfg Config, s Session) error { + return run(ctx, cfg, s, deps{dial: relayDial(cfg), broker: func() processBroker { return unavailableBroker{} }, procs: procfs{}}) +} + +// session is one running Session. +type session struct { + cfg Config + in Session + deps deps + plan *plan + link *linkOwner + log *slog.Logger + uid uint32 + dir sessionDir + + // ctx ends when the Session fails or tears down. + ctx context.Context + cancel context.CancelFunc + + failMu sync.Mutex + failure error // the first failure, which ended the Session + cleanup []error // each world that did not stop cleanly + + mu sync.Mutex + // live is the one view that may run; nil when none does. + live *liveView + // views counts launches and their views until each is torn down. + views sync.WaitGroup + + brokerMu sync.Mutex + broker processBroker // started at the first launch + + // The goroutine that runs drive and then teardown owns these. + fwd *forwarder // the last Turn's forwarder + execClosed bool // a Close of the Executor succeeded + execErr error // the last Close's error +} + +func run(ctx context.Context, cfg Config, in Session, d deps) error { + roots, err := checkConfig(cfg) + if err != nil { + return err + } + s := &session{cfg: cfg, in: in, deps: d, log: cfg.Log} + if s.log == nil { + s.log = slog.New(slog.DiscardHandler) + } + s.ctx, s.cancel = context.WithCancel(ctx) + defer s.cancel() + s.link = newLinkOwner(d.dial, in.Binding, s.fail) + if s.plan, err = admit(cfg, roots, in, s.openNetwork); err != nil { + return err + } + if s.uid, err = allocUID(cfg.UIDs, d.procs); err != nil { + return err + } + if s.dir, err = createSessionDir(cfg.StateDir, in.Binding.SessionID, s.uid); err != nil { + freeUID(s.uid) + return err + } + context.AfterFunc(s.ctx, s.closeLive) + + exec, err := s.plan.view.Executor(s.ctx, s.plan.request, agent.ViewSession{ + Home: agent.ViewDir{Host: s.dir.entry(homeEntry), View: agent.ViewPrivateRoot + "/" + agent.ViewHomeName}, + Proxy: s.plan.proxy, + MCP: s.plan.mcp, + Launch: s.launch, + }) + if err != nil { + err = executorError(err) + } else { + err = s.drive(exec) + } + return s.finish(exec, err, ctx.Err()) +} + +// finish tears the Session down and only then decides its result, so a +// failure recorded during teardown counts: the Session's first failure, else +// ended, the end of Run's ctx, else err. Teardown has joined every view and +// stopped the link owner's reports, so nothing changes the result later. +func (s *session) finish(exec agent.Executor, err, ended error) error { + terr := s.teardown(exec) + s.failMu.Lock() + failure, cleanup := s.failure, s.cleanup + s.failMu.Unlock() + switch { + case failure != nil: + err = failure + case ended != nil: + err = ended + } + errs := []error{err} + for _, c := range cleanup { + if c != failure { + errs = append(errs, c) + } + } + return errors.Join(append(errs, terr)...) +} + +func executorError(err error) error { + if errors.Is(err, agent.ErrUnsupportedOperation) || errors.Is(err, agent.ErrViewHandoff) { + return &Error{Kind: ErrUnsupported, Op: "executor", Err: err} + } + return &Error{Kind: ErrExecutor, Err: err} +} + +// fail records the Session's first failure and ends the Session: its live +// view closes and its Turn is cancelled. +func (s *session) fail(err error) { + s.failMu.Lock() + if s.failure == nil { + s.failure = err + } + s.failMu.Unlock() + s.cancel() +} + +// worldEnded records a world that did not stop cleanly, or that cannot show +// that its attachment holds nothing. Only ending the attachment settles its +// state, so the Session fails, and Run reports the error even after another +// failure. +func (s *session) worldEnded(op string, err error) error { + e := &Error{Kind: ErrWorld, Op: op, Err: err} + s.failMu.Lock() + s.cleanup = append(s.cleanup, e) + s.failMu.Unlock() + s.fail(e) + return e +} + +// drive runs each Turn from Input in order until Input is closed, a Turn +// fails or the Session ends. +func (s *session) drive(exec agent.Executor) error { + for { + select { + case <-s.ctx.Done(): + return nil + case in, ok := <-s.in.Input: + if !ok { + return nil + } + if err := s.turn(exec, in); err != nil { + return err + } + } + } +} + +// The Turn driving below mirrors the daemon's prepared execution in +// apps/daemon/internal/dispatch: startPreparedExecution +// (preparation_start.go), forwardPreparedOutput, runPreparedRelease and +// forwardPreparedTerminal (prepared_handoff.go), as harness-onboarding.md's +// "What the Runtime does around a Turn" describes them. A Session has no +// steering, functions or interactions, so no admitted operation joins the +// release; the end of the Session stands in for the connection's shutdown. + +// forwarder is a Turn's one output consumer. It starts before StartTurn and +// drains out: it forwards each envelope to Output in order until the Session +// ends, and keeps the Turn's Done for turn to publish after settlement. +type forwarder struct { + runID string + out chan proto.Envelope + // ended closes at the Turn's terminal observation: its Done, a protocol + // error or the close of out. + ended chan struct{} + // abort closes when the Turn is to be cancelled: a protocol error, a + // failed start or the end of the Session. + abort chan struct{} + // stop makes the forwarder return without draining further. + stop chan struct{} + // done closes when the forwarder has returned and sends nothing more. + done chan struct{} + + endOnce, abortOnce, stopOnce sync.Once + + mu sync.Mutex + terminal *proto.Envelope + protocolErr error +} + +func newForwarder(runID string) *forwarder { + return &forwarder{runID: runID, out: make(chan proto.Envelope, turnBuffer), + ended: make(chan struct{}), abort: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{})} +} + +func (f *forwarder) end() { f.endOnce.Do(func() { close(f.ended) }) } +func (f *forwarder) cancel() { f.abortOnce.Do(func() { close(f.abort) }) } +func (f *forwarder) halt() { f.stopOnce.Do(func() { close(f.stop) }) } + +func (f *forwarder) aborted() bool { + select { + case <-f.abort: + return true + default: + return false + } +} + +// forward runs f until out closes or f is halted. +func (s *session) forward(f *forwarder) { + defer close(f.done) + for { + select { + case <-f.stop: + return + case e, ok := <-f.out: + if !ok { + f.end() + return + } + f.mu.Lock() + if e.ID != f.runID || f.terminal != nil { + if f.protocolErr == nil { + f.protocolErr = errors.New("executor output crossed the Turn boundary") + } + f.mu.Unlock() + f.end() + f.cancel() + continue + } + if e.Type == proto.TypeDone { + f.terminal = &e + f.mu.Unlock() + f.end() + continue + } + f.mu.Unlock() + if s.ctx.Err() == nil { + select { + case s.in.Output <- e: + case <-s.ctx.Done(): + } + } + } + } +} + +// turn runs one Turn. Its forwarder starts before StartTurn. Once the Turn's +// output ends, or the Turn is to be cancelled, turn awaits its settlement, +// closes the Executor when the Turn leaves it unusable and only then +// publishes the Turn's Done, after an Error envelope when the Turn failed. +// When Close fails, the Executor keeps the Turn and nothing is published. +// When the Session has ended, nothing is published and turn returns nil. +func (s *session) turn(exec agent.Executor, in Input) error { + f := newForwarder(in.RunID) + s.fwd = f + go s.forward(f) + defer context.AfterFunc(s.ctx, f.cancel)() + turn, startErr := exec.StartTurn(s.ctx, in.RunID, in.Message, f.out) + if turn == nil { + // out stays with the caller. The failed Turn ends the Session, and + // teardown closes the Executor. + close(f.out) + <-f.done + if startErr == nil { + startErr = errors.New("no Turn") + } + return &Error{Kind: ErrTurn, Op: "start", Err: startErr} + } + if startErr != nil { + f.cancel() + } + select { + case <-f.ended: + case <-f.abort: + } + settlement, nativeErr := settle(turn, f.abort) + if nativeErr == nil { + // Settlement confirms that out is closed. + select { + case <-f.done: + case <-s.ctx.Done(): + } + } + f.mu.Lock() + terminal, protocolErr := f.terminal, f.protocolErr + if terminal == nil && protocolErr == nil && !f.aborted() { + protocolErr = errors.New("executor output ended without a terminal result") + } + f.mu.Unlock() + if nativeErr != nil || !settlement.Reusable || startErr != nil || protocolErr != nil || s.ctx.Err() != nil { + if err := s.closeExecutor(exec); err != nil { + return &Error{Kind: ErrTurn, Op: "close executor", Err: errors.Join(nativeErr, err)} + } + } + // A confirmed Close confirms that out is closed too. What arrived during + // Close counts: Close may deliver the Turn's Done. + select { + case <-f.done: + case <-s.ctx.Done(): + return nil + } + terminal = f.terminal + + var failure string + var result error + switch { + case nativeErr != nil: + failure, result = "executor Turn settlement failed", &Error{Kind: ErrTurn, Op: "settle", Err: nativeErr} + case protocolErr != nil: + failure, result = protocolErr.Error(), &Error{Kind: ErrTurn, Op: "output", Err: protocolErr} + case startErr != nil: + failure, result = "executor Turn could not start", &Error{Kind: ErrTurn, Op: "start", Err: startErr} + case !settlement.Reusable: + result = &Error{Kind: ErrTurn, Op: "settle", Err: fmt.Errorf("the Executor is not reusable: %s", settlement.Reason)} + } + if failure != "" { + e, err := proto.NewEnvelope(proto.TypeError, in.RunID, proto.ErrorPayload{Error: failure}) + if err != nil { + return errors.Join(result, err) + } + s.publish(e) + } + if terminal == nil { + e, err := proto.NewEnvelope(proto.TypeDone, in.RunID, proto.DonePayload{}) + if err != nil { + return errors.Join(result, err) + } + terminal = &e + } + s.publish(*terminal) + if s.ctx.Err() != nil { + return nil + } + return result +} + +// settle awaits turn's settlement for at most nativeBound. When abort closes +// first it cancels the Turn, and a failed Cancel ends the wait; natural +// completion never calls Cancel. +func settle(turn agent.Turn, abort <-chan struct{}) (agent.TurnSettlement, error) { + ctx, cancel := context.WithTimeout(context.Background(), nativeBound) + defer cancel() + cancelled := make(chan error, 1) + settled := make(chan struct{}) + go func() { + select { + case <-abort: + err := turn.Cancel(ctx) + if err != nil { + cancel() + } + cancelled <- err + case <-settled: + cancelled <- nil + } + }() + settlement, err := turn.AwaitSettlement(ctx) + close(settled) + return settlement, errors.Join(err, <-cancelled) +} + +// publish sends e to Output while the Session runs. +func (s *session) publish(e proto.Envelope) { + if s.ctx.Err() != nil { + return + } + select { + case s.in.Output <- e: + case <-s.ctx.Done(): + } +} + +// closeExecutor closes exec for at most nativeBound, until a Close succeeds. +// A failed Close retains the Executor's resources, and a later call retries +// it. +func (s *session) closeExecutor(exec agent.Executor) error { + if exec == nil || s.execClosed { + return nil + } + ctx, cancel := context.WithTimeout(context.Background(), nativeBound) + defer cancel() + s.execErr = exec.Close(ctx) + s.execClosed = s.execErr == nil + return s.execErr +} + +// teardown releases the Session in order: the Executor, the view, the +// process broker, the Link attachment, the Session directory and the uid. +// When Close fails, teardown ends the views, which kills each view's +// processes, and retries Close once. If that fails too, the Executor may +// still use the Session directory: teardown returns ErrTeardown and keeps +// the directory and the uid, which stays in use until the agent host exits. +func (s *session) teardown(exec agent.Executor) error { + var errs []error + closeErr := s.execErr + if closeErr == nil { + closeErr = s.closeExecutor(exec) + } + // Ending the Session closes the live view and refuses new launches. Under + // mu, every launch that passed its check has already counted itself. + s.mu.Lock() + s.cancel() + s.mu.Unlock() + s.views.Wait() + if closeErr != nil { + closeErr = s.closeExecutor(exec) + } + // The Session has ended, so the forwarder sends nothing more. + if f := s.fwd; f != nil { + f.halt() + <-f.done + } + s.brokerMu.Lock() + broker := s.broker + s.brokerMu.Unlock() + if broker != nil { + if err := broker.Close(); err != nil { + errs = append(errs, &Error{Kind: ErrTeardown, Op: "close process broker", Err: err}) + } + } + errs = append(errs, s.link.close()) + if closeErr != nil { + errs = append(errs, &Error{Kind: ErrTeardown, Op: "close executor", Err: closeErr}) + return errors.Join(errs...) + } + if err := os.RemoveAll(string(s.dir)); err != nil { + errs = append(errs, &Error{Kind: ErrTeardown, Op: "remove session directory", Err: err}) + } + freeUID(s.uid) + return errors.Join(errs...) +} + +func (s *session) openFile(ctx context.Context) (io.ReadWriteCloser, error) { + st, err := s.link.open(ctx, sandboxlink.ServiceFile, sandboxfs.Version) + if err != nil { + return nil, err + } + return st, nil +} + +func (s *session) openProcess(ctx context.Context) (io.ReadWriteCloser, error) { + st, err := s.link.open(ctx, sandboxlink.ServiceProcess, sandboxprocess.Version) + if err != nil { + return nil, err + } + return st, nil +} + +func (s *session) openNetwork(ctx context.Context) (sandboxlink.Stream, error) { + return s.link.open(ctx, sandboxlink.ServiceNetwork, sandboxnet.Version) +} diff --git a/apps/daemon/internal/agenthost/run_other.go b/apps/daemon/internal/agenthost/run_other.go new file mode 100644 index 00000000..f6758ae5 --- /dev/null +++ b/apps/daemon/internal/agenthost/run_other.go @@ -0,0 +1,10 @@ +//go:build !linux + +package agenthost + +import "context" + +// Run reports that the agent host needs Linux. +func Run(context.Context, Config, Session) error { + return &Error{Kind: ErrUnsupported, Op: "run"} +} diff --git a/apps/daemon/internal/agenthost/session_linux_test.go b/apps/daemon/internal/agenthost/session_linux_test.go new file mode 100644 index 00000000..daa5e3c3 --- /dev/null +++ b/apps/daemon/internal/agenthost/session_linux_test.go @@ -0,0 +1,372 @@ +//go:build linux + +package agenthost + +import ( + "context" + "errors" + "log/slog" + "os" + "reflect" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent/clirunner" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/sessionview" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +func TestAllocationSkipsUIDsThatProcessesHold(t *testing.T) { + r := UIDRange{First: 71000, Count: 2} + // A thread holds the uid; its process's leader does not. + procs := &fakeProcesses{list: []task{{tgid: 10, tid: 10, uids: [4]uint32{1000, 1000, 1000, 1000}}, {tgid: 10, tid: 11, uids: [4]uint32{1000, 71000, 1000, 1000}}}} + id, err := allocUID(r, procs) + if err != nil || id != 71001 { + t.Fatalf("allocUID = %d, %v; want 71001", id, err) + } + defer freeUID(id) + if _, err := allocUID(r, procs); !errors.Is(err, ErrCapacity) { + t.Fatalf("allocUID with every uid taken = %v", err) + } + // The host's table shows this process with its own uids. + self := uint32(os.Getuid()) + if held, err := heldUIDs(procfs{}, UIDRange{First: self, Count: 1}); err != nil || !held[self] { + t.Fatalf("/proc shows uid %d held: %v, %v", self, held[self], err) + } +} + +func TestViewEndReleasesTheSlotBeforeTheProcessEnds(t *testing.T) { + errDetach := errors.New("detach failed") + for name, stop := range map[string]error{"clean world": nil, "failed detach": errDetach} { + s := newOwnerSession(t) + lv := &liveView{} + s.live = lv + s.views.Add(1) + ends, err := newStdio(false) + if err != nil { + t.Fatal(err) + } + ends.closeChild() + 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) + if err != nil { + t.Fatal(err) + } + v.Close() + _ = p.Wait() + s.mu.Lock() + live := s.live + s.mu.Unlock() + if live != nil { + t.Errorf("%s: the view slot is taken after Process.Wait", name) + } + failed := s.ctx.Err() != nil + err = s.finish(nil, nil, nil) + switch { + case stop == nil && (err != nil || failed): + t.Errorf("%s: Run = %v, Session failed %v", name, err, failed) + case stop != nil && (!errors.Is(err, ErrWorld) || !errors.Is(err, errDetach) || !failed): + t.Errorf("%s: Run = %v, Session failed %v; want ErrWorld with the detach error", name, err, failed) + } + } +} + +func TestFailureDuringTeardownCounts(t *testing.T) { + s := newOwnerSession(t) + // The relay revokes the attachment while Executor.Close waits. + revoke := func() { s.link.closed(sandboxlink.AttachmentClosed{AttachmentID: s.link.binding.AttachmentID}) } + if err := s.finish(closingExecutor(revoke), nil, nil); !errors.Is(err, ErrLink) { + t.Fatalf("Run = %v, want the revocation", err) + } + // Once the link owner closes the attachment, the Link's reports explain nothing. + s = newOwnerSession(t) + if err := s.finish(nil, nil, nil); err != nil { + t.Fatal(err) + } + s.link.closed(sandboxlink.AttachmentClosed{AttachmentID: s.link.binding.AttachmentID}) + if s.ctx.Err() == nil { + t.Fatal("teardown left the Session's context live") + } + s.failMu.Lock() + defer s.failMu.Unlock() + if s.failure != nil { + t.Fatalf("a report after close failed the Session: %v", s.failure) + } +} + +func TestTurnDrainsOutputFromStart(t *testing.T) { + s := newOwnerSession(t) + output := make(chan proto.Envelope, 2*turnBuffer) + s.in.Output = output + // The adapter emits more than out holds before StartTurn returns. + turn := &fakeTurn{} + exec := &fakeExecutor{start: func(runID string, out chan<- proto.Envelope) (agent.Turn, error) { + for range turnBuffer + 1 { + out <- proto.Envelope{Type: proto.TypeOutputMessage, ID: runID} + } + out <- doneEnvelope(runID) + close(out) + return turn, nil + }} + if err := within(t, func() error { return s.turn(exec, Input{RunID: "r"}) }); err != nil { + t.Fatalf("turn = %v", err) + } + if got := drainAll(output); len(got) != turnBuffer+2 || got[len(got)-1].Type != proto.TypeDone || turn.cancelled.Load() || exec.closes != 0 { + t.Fatalf("Output got %d envelopes; Turn cancelled %v; Executor closed %d times", len(got), turn.cancelled.Load(), exec.closes) + } + // A nil Turn leaves out with its caller, which closes it. + var kept chan<- proto.Envelope + exec = &fakeExecutor{start: func(_ string, out chan<- proto.Envelope) (agent.Turn, error) { + kept = out + return nil, errors.New("refused") + }} + if err := within(t, func() error { return s.turn(exec, Input{RunID: "r"}) }); !errors.Is(err, ErrTurn) || !isClosed(kept) { + t.Fatalf("turn without a Turn = %v; out closed %v", err, isClosed(kept)) + } +} + +func TestTurnPublishesDoneAfterSettlementAndClose(t *testing.T) { + s := newOwnerSession(t) + output := make(chan proto.Envelope, 4) + s.in.Output = output + published := -1 + exec := &fakeExecutor{ + start: func(runID string, out chan<- proto.Envelope) (agent.Turn, error) { + out <- doneEnvelope(runID) + close(out) + return &fakeTurn{settleErr: errors.New("settlement lost")}, nil + }, + close: func() error { + published = len(output) + return nil + }, + } + err := within(t, func() error { return s.turn(exec, Input{RunID: "r"}) }) + got := drainAll(output) + if !errors.Is(err, ErrTurn) || published != 0 || len(got) != 2 || got[0].Type != proto.TypeError || got[1].Type != proto.TypeDone { + t.Fatalf("turn = %v; %d envelopes published before Close; Output got %v, want Error then Done", err, published, got) + } + // A Turn that failed to start delivers its Done during Close. + s = newOwnerSession(t) + s.in.Output = output + native, err := proto.NewEnvelope(proto.TypeDone, "r", proto.DonePayload{Content: "native"}) + if err != nil { + t.Fatal(err) + } + var kept chan<- proto.Envelope + exec = &fakeExecutor{ + start: func(_ string, out chan<- proto.Envelope) (agent.Turn, error) { + kept = out + return &fakeTurn{settleErr: errors.New("settlement lost")}, errors.New("start failed") + }, + close: func() error { + kept <- native + close(kept) + return nil + }, + } + err = within(t, func() error { return s.turn(exec, Input{RunID: "r"}) }) + if got := drainAll(output); !errors.Is(err, ErrTurn) || len(got) != 2 || got[0].Type != proto.TypeError || !reflect.DeepEqual(got[1], native) { + t.Fatalf("turn = %v; Output got %v, want Error then the Done from Close", err, got) + } +} + +func TestFailedCloseKeepsTheSessionDirectoryAndUID(t *testing.T) { + errStuck := errors.New("close stuck") + for name, recovers := range map[string]bool{"Close fails until the view ends": true, "Close keeps failing": false} { + s := newOwnerSession(t) + output := make(chan proto.Envelope, 4) + s.in.Output = output + // Each case takes its own uid. + uid, err := allocUID(UIDRange{First: 72000, Count: 2}, &fakeProcesses{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { freeUID(uid) }) + s.uid = uid + if s.dir, err = createSessionDir(t.TempDir(), sandboxwire.NewID(), uid); err != nil { + t.Fatal(err) + } + v := &fakeView{exit: make(chan struct{})} + s.live = &liveView{view: v} + s.views.Add(1) + go func() { + v.Wait() + s.views.Done() + }() + // The Turn's settlement fails, and its adapter never closes out. + var kept chan<- proto.Envelope + exec := &fakeExecutor{ + start: func(runID string, out chan<- proto.Envelope) (agent.Turn, error) { + kept = out + out <- proto.Envelope{Type: proto.TypeOutputMessage, ID: runID} + out <- doneEnvelope(runID) + return &fakeTurn{settleErr: errors.New("settlement lost")}, nil + }, + close: func() error { + select { + case <-v.exit: + if recovers { + return nil + } + default: + } + return errStuck + }, + } + err = within(t, func() error { return s.finish(exec, s.turn(exec, Input{RunID: "r"}), nil) }) + _, statErr := os.Stat(string(s.dir)) + uids.Lock() + used := uids.used[uid] + uids.Unlock() + switch { + case !errors.Is(err, ErrTurn) || exec.closes != 2 || len(output) != 1: + t.Errorf("%s: Run = %v; Executor closed %d times; Output got %d envelopes, want only the output message", name, err, exec.closes, len(output)) + case recovers && (errors.Is(err, ErrTeardown) || statErr == nil || used): + t.Errorf("%s: Run = %v; directory kept %v; uid in use %v", name, err, statErr == nil, used) + case !recovers && (!errors.Is(err, ErrTeardown) || !errors.Is(err, errStuck) || statErr != nil || !used): + t.Errorf("%s: Run = %v; directory kept %v; uid in use %v; want ErrTeardown keeping both", name, err, statErr == nil, used) + } + // The forwarder has returned, so nothing reaches Output any more. + select { + case <-s.fwd.done: + default: + t.Errorf("%s: the forwarder outlives Run", name) + } + kept <- proto.Envelope{Type: proto.TypeOutputMessage, ID: "r"} + } +} + +// newOwnerSession is a Session with no directory, uid or link. +func newOwnerSession(t *testing.T) *session { + s := &session{log: slog.New(slog.DiscardHandler)} + s.ctx, s.cancel = context.WithCancel(context.Background()) + t.Cleanup(s.cancel) + context.AfterFunc(s.ctx, s.closeLive) + s.link = newLinkOwner(nil, Binding{AttachmentID: sandboxwire.NewID()}, s.fail) + return s +} + +// within runs f and fails t unless f returns within a bound. +func within(t *testing.T, f func() error) error { + t.Helper() + done := make(chan error, 1) + go func() { done <- f() }() + select { + case err := <-done: + return err + case <-time.After(5 * time.Second): + t.Fatal("blocked") + return nil + } +} + +// drainAll returns what ch holds. +func drainAll(ch chan proto.Envelope) []proto.Envelope { + var list []proto.Envelope + for len(ch) > 0 { + list = append(list, <-ch) + } + return list +} + +// isClosed reports whether ch is closed. +func isClosed(ch chan<- proto.Envelope) (closed bool) { + defer func() { closed = recover() != nil }() + select { + case ch <- proto.Envelope{}: + default: + } + return false +} + +func doneEnvelope(runID string) proto.Envelope { + e, err := proto.NewEnvelope(proto.TypeDone, runID, proto.DonePayload{}) + if err != nil { + panic(err) + } + return e +} + +// fakeExecutor starts each Turn with start; Close returns close's result. +type fakeExecutor struct { + start func(runID string, out chan<- proto.Envelope) (agent.Turn, error) + close func() error + closes int +} + +func (e *fakeExecutor) StartTurn(_ context.Context, runID string, _ proto.MessageInput, out chan<- proto.Envelope) (agent.Turn, error) { + return e.start(runID, out) +} + +func (e *fakeExecutor) Close(context.Context) error { + e.closes++ + if e.close == nil { + return nil + } + return e.close() +} + +// fakeTurn settles at once: reusable, or with settleErr. +type fakeTurn struct { + settleErr error + cancelled atomic.Bool +} + +func (t *fakeTurn) Cancel(context.Context) error { + t.cancelled.Store(true) + return nil +} + +func (t *fakeTurn) CancellationOutcome() proto.DonePayload { return proto.DonePayload{} } + +func (t *fakeTurn) SteerWithReceipt(context.Context, proto.PromptSteerPayload, func()) error { + return agent.ErrUnsupportedOperation +} + +func (t *fakeTurn) AwaitSettlement(context.Context) (agent.TurnSettlement, error) { + return agent.TurnSettlement{Reusable: t.settleErr == nil}, t.settleErr +} + +// fakeView is a view that ends when closed. +type fakeView struct { + exit chan struct{} + once sync.Once +} + +func (v *fakeView) Signal(syscall.Signal) error { return nil } + +func (v *fakeView) Wait() (sessionview.Exit, error) { + <-v.exit + return sessionview.Exit{}, nil +} + +func (v *fakeView) Close() error { + v.once.Do(func() { close(v.exit) }) + return nil +} + +// fakeWorld is a world that is never lost and whose Stop returns stop. +type fakeWorld struct{ stop error } + +func (w fakeWorld) Stop() error { return w.stop } +func (fakeWorld) Lost() <-chan struct{} { return nil } +func (fakeWorld) Err() error { return nil } + +// closingExecutor runs itself when closed. +type closingExecutor func() + +func (closingExecutor) StartTurn(context.Context, string, proto.MessageInput, chan<- proto.Envelope) (agent.Turn, error) { + return nil, errors.New("no Turns") +} + +func (e closingExecutor) Close(context.Context) error { + e() + return nil +} diff --git a/apps/daemon/internal/agenthost/sessiondir_linux.go b/apps/daemon/internal/agenthost/sessiondir_linux.go new file mode 100644 index 00000000..42d1ae8c --- /dev/null +++ b/apps/daemon/internal/agenthost/sessiondir_linux.go @@ -0,0 +1,129 @@ +//go:build linux + +package agenthost + +import ( + "errors" + "fmt" + "io/fs" + "os" + "path/filepath" + "sync" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// uids holds the Session uids in use. Only one agent host runs per kernel, so +// the set is process-wide. +var uids = struct { + sync.Mutex + used map[uint32]bool +}{used: map[uint32]bool{}} + +// allocUID returns a uid in r that no Session uses and no process holds. +func allocUID(r UIDRange, procs processTable) (uint32, error) { + held, err := heldUIDs(procs, r) + if err != nil { + return 0, &Error{Kind: ErrInvalidConfig, Op: "processes", Err: err} + } + uids.Lock() + defer uids.Unlock() + for i := range r.Count { + if id := r.First + i; !uids.used[id] && !held[id] { + uids.used[id] = true + return id, nil + } + } + return 0, &Error{Kind: ErrCapacity} +} + +func freeUID(id uint32) { + uids.Lock() + defer uids.Unlock() + delete(uids.used, id) +} + +// The entries of a Session directory. +const ( + homeEntry = agent.ViewHomeName // the Session home, owned by the Session uid + runEntry = agent.ViewRunName // the process broker's run directory + etcEntry = "etc" // the /etc files + maskEntry = "mask" // an empty file and an empty directory that masks present + stagingEntry = "staging" // sessionview's staging parent +) + +func sessionsDir(stateDir string) string { return filepath.Join(stateDir, "sessions") } + +// sessionDir is one Session's host directory, private to the agent host. +type sessionDir string + +func (d sessionDir) entry(name ...string) string { + return filepath.Join(append([]string{string(d)}, name...)...) +} + +// createSessionDir creates the Session directory with its home, run, etc, +// mask and staging entries. +func createSessionDir(stateDir string, id sandboxwire.ID, uid uint32) (sessionDir, error) { + parent := sessionsDir(stateDir) + if err := os.MkdirAll(parent, 0o700); err != nil { + return "", &Error{Kind: ErrInvalidConfig, Op: "state directory", Err: err} + } + d := sessionDir(filepath.Join(parent, id.String())) + if err := os.Mkdir(string(d), 0o700); errors.Is(err, fs.ErrExist) { + return "", &Error{Kind: ErrSessionExists, Err: err} + } else if err != nil { + return "", &Error{Kind: ErrInvalidConfig, Op: "session directory", Err: err} + } + if err := d.populate(uid); err != nil { + os.RemoveAll(string(d)) + return "", &Error{Kind: ErrInvalidConfig, Op: "session directory", Err: err} + } + return d, nil +} + +func (d sessionDir) populate(uid uint32) error { + dirs := []struct { + name string + mode fs.FileMode + }{{homeEntry, 0o700}, {runEntry, 0o755}, {etcEntry, 0o755}, {maskEntry, 0o755}, {filepath.Join(maskEntry, "dir"), 0o555}, {stagingEntry, 0o700}} + for _, e := range dirs { + if err := os.Mkdir(d.entry(e.name), e.mode); err != nil { + return err + } + } + home := agent.ViewPrivateRoot + "/" + agent.ViewHomeName + files := map[string]string{ + filepath.Join(maskEntry, "file"): "", + filepath.Join(etcEntry, "passwd"): fmt.Sprintf("root:x:0:0:root:/root:/usr/sbin/nologin\noac:x:%d:%d:oac:%s:/bin/bash\nnobody:x:65534:65534:nobody:/nonexistent:/usr/sbin/nologin\n", + uid, uid, home), + filepath.Join(etcEntry, "group"): fmt.Sprintf("root:x:0:\noac:x:%d:\nnogroup:x:65534:\n", uid), + filepath.Join(etcEntry, "hosts"): "127.0.0.1 localhost\n::1 localhost\n", + filepath.Join(etcEntry, "resolv.conf"): "", + filepath.Join(etcEntry, "nsswitch.conf"): "passwd: files\ngroup: files\nshadow: files\nhosts: files\n", + } + for name, content := range files { + if err := os.WriteFile(d.entry(name), []byte(content), 0o444); err != nil { + return err + } + } + return nil +} + +// chownHome gives the Session uid the home tree, including what the view +// Executor factory wrote there. It runs while no view is live, so no Session +// process changes the tree meanwhile, and os.Root keeps every change inside +// it. +func (d sessionDir) chownHome(uid uint32) error { + root, err := os.OpenRoot(d.entry(homeEntry)) + if err != nil { + return err + } + defer root.Close() + return fs.WalkDir(root.FS(), ".", func(name string, _ fs.DirEntry, err error) error { + if err != nil { + return err + } + return root.Lchown(name, int(uid), int(uid)) + }) +} diff --git a/apps/daemon/internal/agenthost/view_linux_test.go b/apps/daemon/internal/agenthost/view_linux_test.go new file mode 100644 index 00000000..3331d495 --- /dev/null +++ b/apps/daemon/internal/agenthost/view_linux_test.go @@ -0,0 +1,679 @@ +//go:build linux + +package agenthost + +import ( + "bytes" + "context" + "encoding/json" + "encoding/pem" + "errors" + "fmt" + "io" + "io/fs" + "net" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "runtime" + "strconv" + "strings" + "syscall" + "testing" + "time" + + "github.com/google/uuid" + "golang.org/x/sys/unix" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent/clirunner" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/sessionview" + "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/modelprovider" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxbootstrap" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxfs" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink/relay" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink/sandboxlinktest" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// The view suite needs root with CAP_SYS_ADMIN and CAP_NET_ADMIN, /dev/fuse, +// no AppArmor confinement and a static oac-sandbox-io, whose world is the +// container's /. Run it in a throwaway container: +// +// CGO_ENABLED=0 go build -o /tmp/oac-sandbox-io ./apps/sandboxio/cmd/oac-sandbox-io +// CGO_ENABLED=0 go test -c -o /tmp/agenthost.test ./apps/daemon/internal/agenthost +// docker run --rm --cap-add SYS_ADMIN --cap-add NET_ADMIN --device /dev/fuse --security-opt apparmor=unconfined \ +// -e OAC_TEST_AGENTHOST=1 -e OAC_TEST_SANDBOXIO=/sandboxio -v /tmp/oac-sandbox-io:/sandboxio:ro \ +// -v /tmp/agenthost.test:/t.test:ro debian:bookworm-slim /t.test -test.v +const ( + gateEnv = "OAC_TEST_AGENTHOST" + sandboxIOEnv = "OAC_TEST_SANDBOXIO" + harnessEnv = "OAC_AGENTHOST_HARNESS" + modelEnv = "OAC_AGENTHOST_MODEL" + caEnv = "OAC_AGENTHOST_CA" + harnessPath = "/.oac/harness/harness" + upstreamKey = "sk-agenthost-upstream" + wait = 20 * time.Second +) + +func TestSessionRunsInAViewOverItsAttachment(t *testing.T) { + if os.Getenv(gateEnv) != "1" { + t.Skipf("set %s=1 and run the test binary as root in a privileged container; see the comment above", gateEnv) + } + if err := sessionview.Probe(); err != nil { + t.Fatalf("Probe: %v", err) + } + sb := startSandbox(t, os.Getenv(sandboxIOEnv)) + // The model upstream reports whether each request carried the real key. + keyed := make(chan bool, 4) + upstream := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + keyed <- r.Header.Get("X-Api-Key") == upstreamKey + io.WriteString(w, "answer") + })) + defer upstream.Close() + + reg := agent.NewRegistry() + cfg := newConfig(t, reg, upstream.Certificate()) + cfg.RelayURL = sb.url + closure := t.TempDir() + if err := os.Chmod(closure, 0o755); err != nil { + t.Fatal(err) + } + copyExecutable(t, filepath.Join(closure, "harness")) + register(reg, "test", &agent.View{ + Closure: []agent.ViewMount{{Name: "harness", HostDir: closure}}, + Masks: []agent.ViewMask{{Path: "/etc/ld.so.preload"}, {Path: "/etc/hostname"}, {Path: "/etc/apt", Dir: true}}, + LocalExec: []string{harnessPath}, + Proxy: agent.ViewProxyNone, + Executor: func(_ context.Context, req proto.PromptRequestPayload, s agent.ViewSession) (agent.Executor, error) { + provider, err := modelprovider.ParseProvider(req.AgentOptions["model_provider"]) + if err != nil { + return nil, err + } + return &testExecutor{session: s, dir: req.LocalEnvironment.WorkspaceRoot, + env: []string{harnessEnv + "=1", modelEnv + "=" + provider.BaseURL, caEnv + "=" + cfg.CADir}}, nil + }, + }) + sb.auth.AddRuntime(cfg.Credential, cfg.RuntimeID) + sb.ready(t, cfg) + workspace, err := os.MkdirTemp("/tmp", "agenthost-workspace-") + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(workspace) + if err := os.Chmod(workspace, 0o777); err != nil { + t.Fatal(err) + } + + t.Run("one Session", func(t *testing.T) { + s := startSession(t, cfg, sb, request("test", workspace, upstream.URL, upstreamKey), 2*time.Second) + r := s.turn(t, "check") + for _, name := range harnessChecks { + if msg, ok := r.Checks[name]; !ok || msg != "" { + t.Errorf("%s: %q", name, msg) + } + } + if r.Exit != "" { + t.Errorf("Harness: %s; stderr %s", r.Exit, r.Stderr) + } + select { + case ok := <-keyed: + if !ok { + t.Error("the upstream did not receive the configured key") + } + default: + t.Error("no request reached the upstream") + } + if data, err := os.ReadFile(filepath.Join(workspace, "renamed")); err != nil || string(data) != "world" { + t.Errorf("the renamed world file holds %q, %v", data, err) + } + // Only renewal keeps the attachment past its 2 second lease. + time.Sleep(3 * time.Second) + if r := s.turn(t, "touch"); r.Checks["touch"] != "" || r.Exit != "" { + t.Errorf("touch after the first lease: %+v", r) + } + if data, err := os.ReadFile(filepath.Join(workspace, "touched")); err != nil || string(data) != "renewed" { + t.Errorf("the file written after the first lease holds %q, %v", data, err) + } + close(s.in) + if err := s.wait(t); err != nil { + t.Fatalf("Run = %v", err) + } + if err := sb.renew(t, cfg, s.binding); !errors.Is(err, sandboxlink.LeaseExpired) { + t.Errorf("Renew after teardown = %v, want LeaseExpired for a closed attachment", err) + } + checkReleased(t, cfg) + }) + + t.Run("a restarted sandbox service fails the Session", func(t *testing.T) { + s := startSession(t, cfg, sb, request("test", workspace, upstream.URL, upstreamKey), time.Minute) + s.send(t, "wait") + beat := filepath.Join(workspace, "beat") + for deadline := time.Now().Add(wait); ; time.Sleep(50 * time.Millisecond) { + if _, err := os.Stat(beat); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("the Harness never wrote to the world") + } + } + sb.stop() + sb.start(t) + err := s.wait(t) + t.Logf("Run = %v", err) + if !errors.Is(err, ErrLink) && !errors.Is(err, ErrWorld) { + t.Errorf("Run = %v, want ErrLink or ErrWorld", err) + } + if errors.Is(err, ErrTeardown) { + t.Errorf("teardown incomplete: %v", err) + } + checkReleased(t, cfg) + }) + + t.Run("allocation skips a uid that a thread holds under an exited leader", func(t *testing.T) { + id := cfg.UIDs.First + 1 + exe, err := os.Executable() + if err != nil { + t.Fatal(err) + } + cmd := exec.Command(exe) + cmd.Env = append(os.Environ(), zombieLeaderEnv+"=1") + cmd.SysProcAttr = &syscall.SysProcAttr{Credential: &syscall.Credential{Uid: id, Gid: id}} + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + defer func() { + cmd.Process.Kill() + cmd.Wait() + }() + until(t, "a zombie leader with a running thread", func() bool { return zombieLeaderHolds(id) }) + if got, err := allocUID(UIDRange{First: id, Count: 1}, procfs{}); !errors.Is(err, ErrCapacity) { + freeUID(got) + t.Errorf("allocUID beside a running thread = %d, %v", got, err) + } + }) +} + +func until(t *testing.T, what string, ok func() bool) { + t.Helper() + deadline := time.Now().Add(wait) + for !ok() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + time.Sleep(10 * time.Millisecond) + } +} + +// zombieLeaderHolds reports whether a process whose leader thread is a zombie +// runs a thread whose real uid is id. +func zombieLeaderHolds(id uint32) bool { + zombie, holding := map[int]bool{}, map[int]bool{} + for _, t := range threads() { + switch { + case t.tid == t.tgid && !t.running: + zombie[t.tgid] = true + case t.running && t.uids[0] == id: + holding[t.tgid] = true + } + } + for tgid := range holding { + if zombie[tgid] { + return true + } + } + return false +} + +type thread struct { + tgid, tid int + uids [4]uint32 + running bool +} + +// threads lists every thread in /proc, zombies included. +func threads() []thread { + var list []thread + pids, _ := os.ReadDir("/proc") + for _, p := range pids { + tgid, err := strconv.Atoi(p.Name()) + if err != nil { + continue + } + tids, _ := os.ReadDir(procPath(tgid, "task")) + for _, e := range tids { + tid, _ := strconv.Atoi(e.Name()) + if s, err := readStatus(procPath(tgid, "task", e.Name(), "status")); err == nil { + list = append(list, thread{tgid: tgid, tid: tid, uids: s.uids, running: s.running()}) + } + } + } + return list +} + +// zombieLeaderEnv makes the test binary a process whose leader thread exits +// while another thread runs on. +const zombieLeaderEnv = "OAC_AGENTHOST_ZOMBIE_LEADER" + +func runZombieLeader() { + runtime.LockOSThread() + started := make(chan struct{}) + go func() { + runtime.LockOSThread() + close(started) + time.Sleep(time.Hour) + }() + <-started + unix.RawSyscall(unix.SYS_EXIT, 0, 0, 0) // ends this thread only +} + +// sandbox is a relay and the oac-sandbox-io serving its one resource. +type sandbox struct { + auth *sandboxlinktest.Authority + url string + bin string + bootstrap string + resource sandboxlink.ResourceRef + cmd *exec.Cmd +} + +func startSandbox(t *testing.T, bin string) *sandbox { + if bin == "" { + t.Fatalf("set %s to a static oac-sandbox-io", sandboxIOEnv) + } + auth := sandboxlinktest.NewAuthority() + rl, err := relay.New(relay.Config{Authority: auth}) + if err != nil { + t.Fatal(err) + } + srv := httptest.NewServer(rl) + t.Cleanup(srv.Close) + t.Cleanup(func() { rl.Close() }) + url := "ws://" + strings.TrimPrefix(srv.URL, "http://") + in := sandboxbootstrap.Input{Version: sandboxbootstrap.Version, LinkURL: url, Credential: "serve-credential", Resource: sandboxbootstrap.Resource{ + TenantID: uuid.NewString(), EnvironmentID: uuid.NewString(), Kind: "allocation", ID: uuid.NewString(), Generation: 1}} + raw, err := in.Marshal() + if err != nil { + t.Fatal(err) + } + bootstrap := filepath.Join(t.TempDir(), "bootstrap.json") + if err := os.WriteFile(bootstrap, raw, 0o600); err != nil { + t.Fatal(err) + } + sb := &sandbox{auth: auth, url: url, bin: bin, bootstrap: bootstrap, resource: in.Resource.Ref()} + auth.AddServe([]byte(in.Credential), sandboxlink.ServePeer{PeerID: sandboxwire.NewID(), Resource: sb.resource}) + sb.start(t) + t.Cleanup(sb.stop) + return sb +} + +func (sb *sandbox) start(t *testing.T) { + cmd := exec.Command(sb.bin, "--bootstrap-file", sb.bootstrap) + cmd.Stdout, cmd.Stderr = os.Stderr, os.Stderr + if err := cmd.Start(); err != nil { + t.Fatal(err) + } + sb.cmd = cmd +} + +func (sb *sandbox) stop() { + if sb.cmd != nil { + sb.cmd.Process.Signal(syscall.SIGTERM) + sb.cmd.Wait() + sb.cmd = nil + } +} + +func (sb *sandbox) grant(b Binding, runtimeID sandboxwire.ID, lease time.Duration) { + sb.auth.AddGrant(b.AttachGrant, sandboxlinktest.Grant{RuntimeID: runtimeID, Resource: b.Resource, SessionID: b.SessionID, + AssignmentID: b.AssignmentID, AssignmentEpoch: b.AssignmentEpoch, Lease: lease, + Services: []sandboxlink.Service{sandboxlink.ServiceFile, sandboxlink.ServiceProcess, sandboxlink.ServiceNetwork}}) +} + +func (sb *sandbox) dial(t *testing.T, cfg Config) *sandboxlink.AttachLink { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), wait) + defer cancel() + link, err := sandboxlink.DialAttach(ctx, sandboxlink.AttachConfig{URL: sb.url, RuntimeID: cfg.RuntimeID, Credential: cfg.Credential}) + if err != nil { + t.Fatal(err) + } + return link +} + +// ready waits until the relay holds oac-sandbox-io as the resource's serve +// peer, so that an Open reaches it. +func (sb *sandbox) ready(t *testing.T, cfg Config) { + t.Helper() + probe, _, _ := newSession(sb.resource, proto.PromptRequestPayload{}) + b := probe.Binding + sb.grant(b, cfg.RuntimeID, time.Minute) + link := sb.dial(t, cfg) + defer link.Close() + ctx, cancel := context.WithTimeout(context.Background(), wait) + defer cancel() + for { + st, _, err := link.OpenService(ctx, sandboxlink.Open{Service: sandboxlink.ServiceFile, Version: sandboxfs.Version, Resource: b.Resource, + AttachmentID: b.AttachmentID, SessionID: b.SessionID, AssignmentID: b.AssignmentID, AssignmentEpoch: b.AssignmentEpoch, AttachGrant: b.AttachGrant}) + if err == nil { + st.Close() + break + } + if !errors.Is(err, sandboxlink.ServiceUnavailable) || ctx.Err() != nil { + t.Fatalf("oac-sandbox-io is not serving: %v", err) + } + time.Sleep(50 * time.Millisecond) + } + if err := link.CloseAttachment(ctx, b.AttachmentID); err != nil { + t.Fatal(err) + } +} + +// renew renews b's attachment from a fresh link. +func (sb *sandbox) renew(t *testing.T, cfg Config, b Binding) error { + link := sb.dial(t, cfg) + defer link.Close() + ctx, cancel := context.WithTimeout(context.Background(), wait) + defer cancel() + _, err := link.Renew(ctx, sandboxlink.RenewAttachment{AttachmentID: b.AttachmentID, AttachGrant: b.AttachGrant}) + return err +} + +// sessionRun is a running Session under test. +type sessionRun struct { + binding Binding + in chan Input + out chan proto.Envelope + done chan error +} + +func startSession(t *testing.T, cfg Config, sb *sandbox, req proto.PromptRequestPayload, lease time.Duration) *sessionRun { + s, in, out := newSession(sb.resource, req) + sb.grant(s.Binding, cfg.RuntimeID, lease) + r := &sessionRun{binding: s.Binding, in: in, out: out, done: make(chan error, 1)} + go func() { + r.done <- run(context.Background(), cfg, s, deps{dial: relayDial(cfg), broker: func() processBroker { return noBroker{} }, procs: procfs{}}) + }() + return r +} + +func (r *sessionRun) send(t *testing.T, mode string) { + t.Helper() + select { + case r.in <- Input{RunID: mode, Message: proto.TextInput(mode)}: + case err := <-r.done: + t.Fatalf("Run ended before the %s Turn: %v", mode, err) + case <-time.After(wait): + t.Fatalf("Run took no %s Turn", mode) + } +} + +// turn runs a Turn in mode and returns its report, which its Done follows. +func (r *sessionRun) turn(t *testing.T, mode string) report { + t.Helper() + r.send(t, mode) + var rep report + if err := json.Unmarshal(r.next(t, mode).Payload, &rep); err != nil { + t.Fatal(err) + } + if e := r.next(t, mode); e.Type != proto.TypeDone { + t.Fatalf("the %s Turn sent %s after its report, want its Done", mode, e.Type) + } + return rep +} + +func (r *sessionRun) next(t *testing.T, mode string) proto.Envelope { + t.Helper() + select { + case e := <-r.out: + return e + case err := <-r.done: + t.Fatalf("Run ended during the %s Turn: %v", mode, err) + case <-time.After(wait): + t.Fatalf("the %s Turn sent nothing", mode) + } + return proto.Envelope{} +} + +func (r *sessionRun) wait(t *testing.T) error { + t.Helper() + select { + case err := <-r.done: + return err + case <-time.After(3 * wait): + t.Fatal("Run did not return") + return nil + } +} + +// checkReleased checks that no Session directory, mount or process remains. +func checkReleased(t *testing.T, cfg Config) { + t.Helper() + if left := leftSessions(t, cfg); len(left) != 0 { + t.Errorf("%d Session directories remain", len(left)) + } + mounts, err := os.ReadFile("/proc/self/mountinfo") + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(mounts), cfg.StateDir) { + t.Error("a mount under the state directory remains") + } + procs, err := os.ReadDir("/proc") + if err != nil { + t.Fatal(err) + } + for _, p := range procs { + if _, err := strconv.Atoi(p.Name()); err != nil { + continue + } + status, err := os.ReadFile(filepath.Join("/proc", p.Name(), "status")) + if err != nil { + continue + } + for line := range strings.Lines(string(status)) { + if fields := strings.Fields(line); len(fields) > 1 && fields[0] == "Uid:" { + if uid, _ := strconv.ParseUint(fields[1], 10, 32); uid >= uint64(cfg.UIDs.First) && uid < uint64(cfg.UIDs.First+cfg.UIDs.Count) { + t.Errorf("process %s runs with Session uid %d", p.Name(), uid) + } + } + } + } +} + +func copyExecutable(t *testing.T, dst string) { + t.Helper() + exe, err := os.Executable() + if err != nil { + t.Fatal(err) + } + data, err := os.ReadFile(exe) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(dst, data, 0o755); err != nil { + t.Fatal(err) + } +} + +// testExecutor runs one Harness view per Turn; the Turn's input text is the +// Harness mode. +type testExecutor struct { + session agent.ViewSession + dir string + env []string +} + +func (e *testExecutor) StartTurn(_ context.Context, runID string, input proto.MessageInput, out chan<- proto.Envelope) (agent.Turn, error) { + mode := *input[0].Content[0].Text + p, err := e.session.Launch(clirunner.StartOptions{Binary: harnessPath, Args: []string{mode}, Dir: e.dir, Env: e.env, + OwnProcessGroup: true, KillTimeout: time.Second}) + if err != nil { + return nil, err + } + turn := &testTurn{p: p, settled: make(chan struct{})} + go turn.run(runID, out) + return turn, nil +} + +func (e *testExecutor) Close(context.Context) error { return nil } + +// report is a Turn's report envelope: the Harness's checks, its stderr and how +// it exited. +type report struct { + Checks map[string]string `json:"checks"` + Stderr string `json:"stderr"` + Exit string `json:"exit"` +} + +type testTurn struct { + p *clirunner.Process + settled chan struct{} +} + +func (t *testTurn) run(runID string, out chan<- proto.Envelope) { + defer close(t.settled) + defer close(out) + var stderr bytes.Buffer + copied := make(chan struct{}) + go func() { + io.Copy(&stderr, t.p.Stderr) + close(copied) + }() + stdout, _ := io.ReadAll(t.p.Stdout) + <-copied + var r report + json.Unmarshal(stdout, &r.Checks) + if err := t.p.Wait(); err != nil { + r.Exit = err.Error() + } + r.Stderr = stderr.String() + payload, _ := json.Marshal(r) + out <- proto.Envelope{Type: proto.TypeOutputMessage, ID: runID, Payload: payload} + done, _ := proto.NewEnvelope(proto.TypeDone, runID, proto.DonePayload{}) + out <- done +} + +func (t *testTurn) Cancel(context.Context) error { + t.p.Cancel() + return nil +} + +func (t *testTurn) CancellationOutcome() proto.DonePayload { return proto.DonePayload{} } + +func (t *testTurn) SteerWithReceipt(context.Context, proto.PromptSteerPayload, func()) error { + return agent.ErrUnsupportedOperation +} + +func (t *testTurn) AwaitSettlement(ctx context.Context) (agent.TurnSettlement, error) { + select { + case <-t.settled: + return agent.TurnSettlement{Reusable: true}, nil + case <-ctx.Done(): + return agent.TurnSettlement{}, ctx.Err() + } +} + +var harnessChecks = []string{"world rename", "model through the gateway", "no direct route", "world is noexec", "masks", "home", "passwd", "CA directory"} + +// runHarness runs inside the view, in the workspace, and prints a JSON map +// from each check to its failure, empty when it passed. +func runHarness(args []string) int { + if len(args) != 1 { + return 2 + } + checks := map[string]func() error{} + switch args[0] { + case "check": + checks = map[string]func() error{ + "world rename": func() error { + if err := os.WriteFile("staged", []byte("world"), 0o644); err != nil { + return err + } + return os.Rename("staged", "renamed") + }, + "model through the gateway": func() error { + req, _ := http.NewRequest("POST", os.Getenv(modelEnv)+"/v1/messages", strings.NewReader("{}")) + req.Header.Set("X-Api-Key", modelprovider.Placeholder) + resp, err := (&http.Client{Timeout: wait}).Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if body, _ := io.ReadAll(resp.Body); resp.StatusCode != http.StatusOK || string(body) != "answer" { + return fmt.Errorf("answered %d %q", resp.StatusCode, body) + } + return nil + }, + "no direct route": func() error { + c, err := net.DialTimeout("tcp", "192.0.2.1:80", 2*time.Second) + if err == nil { + c.Close() + return errors.New("connected outside the gateway") + } + if !errors.Is(err, syscall.ENETUNREACH) { + return fmt.Errorf("dial: %v, want ENETUNREACH", err) + } + return nil + }, + "world is noexec": func() error { + if err := exec.Command("/bin/true").Run(); !errors.Is(err, fs.ErrPermission) { + return fmt.Errorf("exec of a world binary: %v, want a permission error", err) + } + return nil + }, + "masks": func() error { + for _, p := range []string{"/etc/ld.so.preload", "/etc/hostname"} { + if data, err := os.ReadFile(p); err != nil || len(data) != 0 { + return fmt.Errorf("%s holds %d bytes, %v", p, len(data), err) + } + } + if entries, err := os.ReadDir("/etc/apt"); err != nil || len(entries) != 0 { + return fmt.Errorf("/etc/apt holds %d entries, %v", len(entries), err) + } + return nil + }, + "home": func() error { + return os.WriteFile(agent.ViewPrivateRoot+"/"+agent.ViewHomeName+"/probe", []byte("x"), 0o600) + }, + "passwd": func() error { + data, err := os.ReadFile("/etc/passwd") + if want := fmt.Sprintf("oac:x:%d:%d:oac:/.oac/home:/bin/bash\n", os.Getuid(), os.Getgid()); err != nil || !strings.Contains(string(data), want) { + return fmt.Errorf("/etc/passwd lacks %q: %v", want, err) + } + return nil + }, + "CA directory": func() error { + data, err := os.ReadFile(filepath.Join(os.Getenv(caEnv), "ca.pem")) + if block, _ := pem.Decode(data); err != nil || block == nil { + return fmt.Errorf("no CA certificate: %v", err) + } + return nil + }, + } + case "touch": + checks["touch"] = func() error { return os.WriteFile("touched", []byte("renewed"), 0o644) } + case "wait": + // Beat in the world until the view ends. + for i := 0; i < 600; i++ { + os.WriteFile("beat", []byte(strconv.Itoa(i)), 0o644) + os.ReadDir(".") + time.Sleep(100 * time.Millisecond) + } + default: + return 2 + } + report := map[string]string{} + for name, check := range checks { + report[name] = "" + if err := check(); err != nil { + report[name] = err.Error() + } + } + json.NewEncoder(os.Stdout).Encode(report) + return 0 +} diff --git a/apps/daemon/internal/gateway/gateway.go b/apps/daemon/internal/gateway/gateway.go index 1f906fc8..702cb553 100644 --- a/apps/daemon/internal/gateway/gateway.go +++ b/apps/daemon/internal/gateway/gateway.go @@ -1,22 +1,25 @@ // Package gateway is the Session gateway on the agent host. Inside the // Session's loopback-only network namespace it serves one listener per frozen -// model upstream, one per MCP HTTP binding and one generic proxy, so the -// Harness never holds an upstream credential and has no network route of its -// own. +// model upstream, one per MCP HTTP binding and, when the view has one, a +// generic proxy, so the Harness never holds an upstream credential and has no +// network route of its own. // // A listener's identity selects its upstream and credential; nothing is routed // by hostname. A model listener relays the declared native routes of its // protocol (internal/modelprovider) to the upstream from the agent host and // injects the credential. An MCP listener relays to its binding's server and -// injects the bearer token: an environment-origin binding connects through the -// sandbox's Network service, a service-origin binding from the agent host, and -// the gateway does the TLS either way. The generic proxy carries HTTP CONNECT -// tunnels and plain-HTTP forward requests, and connects only through the -// sandbox's Network service. Redirects reach the Harness unchanged and are -// never followed. Response headers and trailers that carry an injected -// credential are withheld; bodies pass unchanged. The end of the Session -// closes every connection, tunnels and upgraded ones included. The gateway -// logs nothing. +// injects the binding's bearer token and HTTP headers: an environment-origin +// binding connects through the sandbox's Network service, a service-origin +// binding from the agent host, and the gateway does the TLS either way. An MCP +// listener serves only its server URL's path, without a query, and relays to +// exactly the server URL. The server URL's query is a credential: a binding +// with a query or an injected value needs https. The generic proxy carries +// HTTP CONNECT tunnels and plain-HTTP forward requests, and connects only +// through the sandbox's Network service. Redirects reach the Harness unchanged +// and are never followed. Response header and trailer values that contain an +// injected credential, header value or MCP query value are withheld; bodies +// pass unchanged. The end of the Session closes every connection, tunnels and +// upgraded ones included. The gateway logs nothing. package gateway import ( @@ -32,6 +35,7 @@ import ( "strconv" "time" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" "github.com/MiniMax-AI/OpenAgentCore/internal/modelprovider" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" @@ -43,11 +47,13 @@ type Config struct { // Models are the frozen model upstreams, each under the adapter's name for // it. Names are unique. Models []Model - // MCP are the Session's MCP HTTP bindings. Server labels are unique. - MCP []proto.MCPHTTPServer - // Prompt is the request that declared MCP. Each binding's origin is - // admitted against its placement with ValidateConnectionOrigin before - // anything else. + // MCP are the Session's effective MCP bindings as + // agent.ResolveMCPBindings returns them, bearer tokens and HTTP headers + // included. Each is an HTTP binding, and server labels are unique. + MCP []agent.MCPBinding + // Prompt is the request the bindings were resolved from. Each binding's + // origin is admitted against its placement with + // proto.MCPHTTPServer.ValidateConnectionOrigin before anything else. Prompt proto.PromptRequestPayload // OpenNetwork opens a new Network stream to the Session's sandbox, as // sandboxnet.Connect takes it. Nil means the Session has no sandbox @@ -57,6 +63,10 @@ type Config struct { // RootCAs are the roots the gateway trusts for upstream TLS. Nil means // the system roots. The server name is always the destination's hostname. RootCAs *x509.CertPool + // Proxy serves the generic proxy at ProxyPort. Without it nothing listens + // there and Endpoints.Proxy is empty; the other listeners keep their + // ports. + Proxy bool } // Model is one frozen model upstream. @@ -76,9 +86,10 @@ type Endpoints struct { // base URL's path. Models map[string]string // MCP maps each binding's server label to the URL the Harness uses: its - // listener with the server URL's path and query. + // listener with the server URL's path and no query. MCP map[string]string - // Proxy is the generic proxy's URL, for HTTP and HTTPS proxy settings. + // Proxy is the generic proxy's URL, for HTTP and HTTPS proxy settings, + // or empty when Config.Proxy is unset. Proxy string } @@ -89,7 +100,8 @@ type SessionNetwork struct { } // ProxyPort is the generic proxy's port in the Session's namespace. The model -// listeners take the following ports in Config order, then the MCP listeners. +// listeners take the following ports in Config order, then the MCP listeners, +// whether or not the proxy is served. // The namespace is the Session's own and the gateway listens before the // Harness starts, so the ports are free; fixing them lets the Harness's // environment be built before the namespace exists. @@ -115,7 +127,7 @@ func Plan(cfg Config) (Endpoints, error) { if err != nil { return Endpoints{}, err } - return g.endpoints(fixedPorts(len(g.listeners))), nil + return g.endpoints(g.fixedPorts()), nil } // Start validates cfg, opens its listeners inside the Session's network @@ -131,7 +143,7 @@ func Start(ctx context.Context, n SessionNetwork, cfg Config) (Endpoints, error) if n.Namespace == nil { return Endpoints{}, fmt.Errorf("%w: no namespace", ErrNetwork) } - ports := fixedPorts(len(g.listeners)) + ports := g.fixedPorts() lns, err := listen(n.Namespace, ports) if err != nil { return Endpoints{}, err @@ -140,10 +152,16 @@ func Start(ctx context.Context, n SessionNetwork, cfg Config) (Endpoints, error) return g.endpoints(ports), nil } -func fixedPorts(n int) []int { - ports := make([]int, n) +// fixedPorts returns each listener's port: ProxyPort for the proxy, then the +// following ports in listener order. +func (g *gateway) fixedPorts() []int { + first := ProxyPort + 1 + if len(g.listeners) > 0 && g.listeners[0].role == roleProxy { + first = ProxyPort + } + ports := make([]int, len(g.listeners)) for i := range ports { - ports[i] = ProxyPort + i + ports[i] = first + i } return ports } @@ -161,12 +179,12 @@ const ( type listener struct { role role name string // model name or MCP server label - suffix string // MCP: the server URL's path and query + suffix string // MCP: the server URL's path handler http.Handler } type gateway struct { - listeners []listener // the proxy, then models, then MCP, in Config order + listeners []listener // the proxy when served, then models, then MCP, in Config order transports []*http.Transport } @@ -190,7 +208,9 @@ func build(session context.Context, cfg Config) (*gateway, error) { forward, sandbox = newTransport(cfg.RootCAs, dial), relayTransport(cfg.RootCAs, dial) g.transports = append(g.transports, forward, sandbox) } - g.listeners = append(g.listeners, listener{role: roleProxy, handler: newProxy(cfg.OpenNetwork, forward)}) + if cfg.Proxy { + g.listeners = append(g.listeners, listener{role: roleProxy, handler: newProxy(cfg.OpenNetwork, forward)}) + } names := map[string]bool{} for _, m := range cfg.Models { @@ -206,26 +226,29 @@ func build(session context.Context, cfg Config) (*gateway, error) { } labels := map[string]bool{} - for _, s := range cfg.MCP { - if err := s.ValidateConnectionOrigin(cfg.Prompt); err != nil { - return nil, invalid("MCP server %q: %v", s.ServerLabel, err) + for _, b := range cfg.MCP { + if err := (proto.MCPHTTPServer{ConnectionOrigin: b.ConnectionOrigin}).ValidateConnectionOrigin(cfg.Prompt); err != nil { + return nil, invalid("MCP server %q: %v", b.ServerLabel, err) + } + if b.ServerLabel == "" || labels[b.ServerLabel] { + return nil, invalid("MCP server label %q is empty or repeated", b.ServerLabel) } - if s.ServerLabel == "" || labels[s.ServerLabel] { - return nil, invalid("MCP server label %q is empty or repeated", s.ServerLabel) + labels[b.ServerLabel] = true + if b.Transport != "http" { + return nil, invalid("MCP server %q has transport %q, not http", b.ServerLabel, b.Transport) } - labels[s.ServerLabel] = true transport := host - if s.ConnectionOrigin == "environment" { + if b.ConnectionOrigin == "environment" { if sandbox == nil { - return nil, invalid("MCP server %q has environment origin and the Session has no sandbox network", s.ServerLabel) + return nil, invalid("MCP server %q has environment origin and the Session has no sandbox network", b.ServerLabel) } transport = sandbox } - h, suffix, err := newMCPRelay(s, transport) + h, suffix, err := newMCPRelay(b, transport) if err != nil { - return nil, invalid("MCP server %q: %v", s.ServerLabel, err) + return nil, invalid("MCP server %q: %v", b.ServerLabel, err) } - g.listeners = append(g.listeners, listener{role: roleMCP, name: s.ServerLabel, suffix: suffix, handler: h}) + g.listeners = append(g.listeners, listener{role: roleMCP, name: b.ServerLabel, suffix: suffix, handler: h}) } return g, nil } diff --git a/apps/daemon/internal/gateway/gateway_test.go b/apps/daemon/internal/gateway/gateway_test.go index 266b978b..4372f388 100644 --- a/apps/daemon/internal/gateway/gateway_test.go +++ b/apps/daemon/internal/gateway/gateway_test.go @@ -16,7 +16,9 @@ import ( "testing" "time" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "github.com/MiniMax-AI/OpenAgentCore/internal/modelprovider" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink/relay" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink/sandboxlinktest" @@ -157,6 +159,28 @@ func startSandbox(t *testing.T) *sandbox { return sb } +func TestPlanKeepsPortsWithoutTheProxy(t *testing.T) { + cfg := Config{ + Models: []Model{{Name: "main", Provider: modelprovider.Provider{Protocol: modelprovider.Anthropic, BaseURL: "https://upstream.test", APIKey: upstreamKey}}}, + MCP: []agent.MCPBinding{{ConnectionOrigin: "service", ServerLabel: "tools", Transport: "http", ServerURL: "https://tools.test/mcp"}}, + Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, + } + for _, proxy := range []bool{false, true} { + cfg.Proxy = proxy + eps, err := Plan(cfg) + if err != nil { + t.Fatal(err) + } + want := Endpoints{Placeholder: modelprovider.Placeholder, Models: map[string]string{"main": "http://127.0.0.1:17101"}, MCP: map[string]string{"tools": "http://127.0.0.1:17102/mcp"}} + if proxy { + want.Proxy = "http://127.0.0.1:17100" + } + if fmt.Sprint(eps) != fmt.Sprint(want) { + t.Errorf("proxy %v: %+v", proxy, eps) + } + } +} + func TestSessionEndEndsBlockedRelays(t *testing.T) { // The upstream upgrades the connection and writes until the Harness's // side is full, then reads until the gateway closes its side. @@ -183,7 +207,7 @@ func TestSessionEndEndsBlockedRelays(t *testing.T) { })) defer srv.Close() gw := serveOnLoopback(t, Config{ - MCP: []proto.MCPHTTPServer{{ConnectionOrigin: "service", ServerLabel: "tools", ServerURL: srv.URL + "/mcp"}}, + MCP: []agent.MCPBinding{{ConnectionOrigin: "service", ServerLabel: "tools", Transport: "http", ServerURL: srv.URL + "/mcp"}}, Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, }) @@ -235,7 +259,7 @@ func TestRejectedUpgradeClosesTheUpstream(t *testing.T) { })) defer srv.Close() gw := serveOnLoopback(t, Config{ - MCP: []proto.MCPHTTPServer{{ConnectionOrigin: "service", ServerLabel: "tools", ServerURL: srv.URL + "/mcp"}}, + MCP: []agent.MCPBinding{{ConnectionOrigin: "service", ServerLabel: "tools", Transport: "http", ServerURL: srv.URL + "/mcp"}}, Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, }) diff --git a/apps/daemon/internal/gateway/mcp.go b/apps/daemon/internal/gateway/mcp.go index 16064e10..49369407 100644 --- a/apps/daemon/internal/gateway/mcp.go +++ b/apps/daemon/internal/gateway/mcp.go @@ -2,61 +2,124 @@ package gateway import ( "errors" + "fmt" "net/http" "net/http/httputil" "net/url" + "slices" + "strings" - "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" + "golang.org/x/net/http/httpguts" + + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" ) -// mcpRelay serves one MCP HTTP binding: it relays each request to the -// server's origin with the same path and query. When the binding has a bearer -// token, it injects the token and withholds it from response headers and -// trailers. +// mcpRelay serves one MCP HTTP binding at its server URL's path and relays +// each request to exactly the server URL, query included. The server URL's +// query is credential material: the Harness's URL carries none, so a request +// with a query or for another path is refused, and response headers and +// trailers never carry it. When the binding has a bearer token or HTTP +// headers, it replaces the Harness's credential headers and same-named +// headers with them, and withholds each injected value from response headers +// and trailers. type mcpRelay struct { - scheme string - host string - token *string + upstream url.URL // the binding's server URL + path string // the escaped path the Harness requests + inject http.Header // canonical names, values as sent; empty when the binding has none transport http.RoundTripper } -// newMCPRelay returns the binding's handler and the path and query the -// Harness appends to the listener's address. -func newMCPRelay(s proto.MCPHTTPServer, t http.RoundTripper) (*mcpRelay, string, error) { - u, err := url.Parse(s.ServerURL) - if err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Hostname() == "" || u.User != nil || u.Opaque != "" || u.Fragment != "" { +// newMCPRelay returns the binding's handler and the path the Harness appends +// to the listener's address. A binding with a bearer token, HTTP headers or a +// query needs an https server URL, so no credential crosses a network in +// plaintext, and a server URL with userinfo is rejected. Errors name a header +// but never carry a value. +func newMCPRelay(b agent.MCPBinding, t http.RoundTripper) (*mcpRelay, string, error) { + u, err := url.Parse(b.ServerURL) + switch { + case err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Hostname() == "" || u.Opaque != "" || u.Fragment != "": return nil, "", errors.New("server URL is not an absolute http or https URL") + case u.User != nil: + return nil, "", errors.New("server URL carries userinfo; use a bearer token or HTTP headers") + } + m := &mcpRelay{upstream: *u, path: u.EscapedPath(), inject: http.Header{}} + if m.path == "" { + m.path = "/" } - m := &mcpRelay{scheme: u.Scheme, host: u.Host, transport: t} - if s.BearerToken != nil { - if u.Scheme != "https" { - return nil, "", errors.New("a bearer token needs an https server URL") + var secrets []string + if b.BearerToken != nil { + token := sentValue(*b.BearerToken) + m.inject.Set("Authorization", "Bearer "+token) + secrets = append(secrets, token) + } + for name, value := range b.HTTPHeaders { + key := http.CanonicalHeaderKey(name) + if !httpguts.ValidHeaderFieldName(name) || !httpguts.ValidHeaderFieldValue(value) { + return nil, "", fmt.Errorf("HTTP header %q is not a valid header", name) + } + if m.inject[key] != nil { + return nil, "", fmt.Errorf("HTTP header %q is repeated or replaces the bearer token", name) } - token := sentValue(*s.BearerToken) - m.token = &token - m.transport = withhold(t, token) + v := sentValue(value) + m.inject[key] = []string{v} + secrets = append(secrets, v) } - suffix := u.EscapedPath() - if u.RawQuery != "" || u.ForceQuery { - suffix += "?" + u.RawQuery + if u.RawQuery != "" { + secrets = append(secrets, queryValues(u.RawQuery)...) + } + if len(secrets) > 0 && u.Scheme != "https" { + return nil, "", errors.New("a bearer token, HTTP headers or a query need an https server URL") + } + m.transport = withhold(t, secrets...) + return m, u.EscapedPath(), nil +} + +// queryValues returns what a raw query discloses: the query itself and each +// parameter's value, as sent and decoded. A parameter without "=" is its own +// value. +func queryValues(raw string) []string { + values := []string{raw} + for _, param := range strings.Split(raw, "&") { + _, v, ok := strings.Cut(param, "=") + if !ok { + v = param + } + values = append(values, v) + if d, err := url.QueryUnescape(v); err == nil && d != v { + values = append(values, d) + } } - return m, suffix, nil + return values } func (m *mcpRelay) ServeHTTP(w http.ResponseWriter, r *http.Request) { - path, query, hasQuery, ok := requestTarget(r) - if !ok { + path, _, hasQuery, ok := requestTarget(r) + switch { + case !ok: http.Error(w, "origin-form request target required", http.StatusBadRequest) return + case hasQuery: + http.Error(w, "the MCP endpoint takes no query", http.StatusBadRequest) + return + case path != m.path: + http.NotFound(w, r) + return } - upstream := &url.URL{Scheme: m.scheme, Host: m.host, Path: r.URL.Path, RawPath: path, - RawQuery: query, ForceQuery: hasQuery && query == ""} reverseProxy(m.transport, func(pr *httputil.ProxyRequest) { - pr.Out.URL = upstream + upstream := m.upstream + pr.Out.URL = &upstream pr.Out.Host = "" - if m.token != nil { - stripCredentials(pr.Out.Header) - pr.Out.Header.Set("Authorization", "Bearer "+*m.token) + if len(m.inject) == 0 { + return + } + stripCredentials(pr.Out.Header) + for name := range pr.Out.Header { + if m.inject[http.CanonicalHeaderKey(name)] != nil { + delete(pr.Out.Header, name) + } + } + for name, values := range m.inject { + pr.Out.Header[name] = slices.Clone(values) } }).ServeHTTP(w, r) } diff --git a/apps/daemon/internal/gateway/mcp_test.go b/apps/daemon/internal/gateway/mcp_test.go index 196d213e..3c3c14de 100644 --- a/apps/daemon/internal/gateway/mcp_test.go +++ b/apps/daemon/internal/gateway/mcp_test.go @@ -8,6 +8,7 @@ import ( "strings" "testing" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" ) @@ -15,13 +16,16 @@ func TestMCPBrokersBothOrigins(t *testing.T) { sb := startSandbox(t) seen := make(chan string, 1) srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - seen <- r.Header.Get("Authorization") + " " + r.URL.RequestURI() + seen <- r.Header.Get("Authorization") + " " + strings.Join(r.Header.Values("X-Tenant"), ",") + " " + r.URL.RequestURI() + w.Header().Set("X-Echo", "tenant-secret") + w.Header().Set("X-Plain", "visible") io.WriteString(w, "tools") })) defer srv.Close() token := "vault-token" - binding := func(origin string) proto.MCPHTTPServer { - return proto.MCPHTTPServer{ConnectionOrigin: origin, ServerLabel: "tools", ServerURL: srv.URL + "/mcp", BearerToken: &token} + binding := func(origin string) agent.MCPBinding { + return agent.MCPBinding{ConnectionOrigin: origin, ServerLabel: "tools", Transport: "http", ServerURL: srv.URL + "/mcp", + BearerToken: &token, HTTPHeaders: map[string]string{"x-tenant": " tenant-secret "}} } service := proto.PromptRequestPayload{DisableExecutionEnvironment: true} environment := proto.PromptRequestPayload{LocalEnvironment: &proto.LocalEnvironment{NetworkAccess: "enabled"}} @@ -32,12 +36,13 @@ func TestMCPBrokersBothOrigins(t *testing.T) { dials int32 }{{"service", service, 0}, {"environment", environment, 1}} { before := sb.dials.Load() - eps := serveOnLoopback(t, Config{MCP: []proto.MCPHTTPServer{binding(c.origin)}, Prompt: c.prompt, OpenNetwork: sb.open, RootCAs: trust(srv)}) + eps := serveOnLoopback(t, Config{MCP: []agent.MCPBinding{binding(c.origin)}, Prompt: c.prompt, OpenNetwork: sb.open, RootCAs: trust(srv)}) if !strings.HasPrefix(eps.MCP["tools"], "http://127.0.0.1:") || !strings.HasSuffix(eps.MCP["tools"], "/mcp") { t.Fatalf("%s: Harness URL %q", c.origin, eps.MCP["tools"]) } req, _ := http.NewRequest("POST", eps.MCP["tools"], strings.NewReader(`{"jsonrpc":"2.0"}`)) req.Header.Set("Authorization", "Bearer harness-value") + req.Header.Set("X-TENANT", "harness-value") resp, err := noRedirects.Do(req) if err != nil { t.Fatal(err) @@ -47,9 +52,12 @@ func TestMCPBrokersBothOrigins(t *testing.T) { if resp.StatusCode != 200 || string(body) != "tools" { t.Fatalf("%s: %d %q", c.origin, resp.StatusCode, body) } - if got := <-seen; got != "Bearer vault-token /mcp" { + if got := <-seen; got != "Bearer vault-token tenant-secret /mcp" { t.Errorf("%s: server saw %q", c.origin, got) } + if resp.Header.Get("X-Echo") != "" || resp.Header.Get("X-Plain") != "visible" { + t.Errorf("%s: an injected header value reached the Harness, or a plain one did not", c.origin) + } if n := sb.dials.Load() - before; n != c.dials { t.Errorf("%s: the sandbox made %d connections, want %d", c.origin, n, c.dials) } @@ -57,7 +65,72 @@ func TestMCPBrokersBothOrigins(t *testing.T) { // Origin admission runs first: an environment binding needs an enabled // workspace network. - if _, err := Plan(Config{MCP: []proto.MCPHTTPServer{binding("environment")}, Prompt: service, OpenNetwork: sb.open}); !errors.Is(err, ErrInvalidConfig) { + if _, err := Plan(Config{MCP: []agent.MCPBinding{binding("environment")}, Prompt: service, OpenNetwork: sb.open}); !errors.Is(err, ErrInvalidConfig) { t.Errorf("Plan with a relocated binding: %v", err) } + // The gateway relays HTTP only, and the bearer token owns Authorization. + stdio := binding("service") + stdio.Transport, stdio.ServerURL, stdio.BearerToken, stdio.HTTPHeaders = "stdio", "", nil, nil + twice := binding("service") + twice.HTTPHeaders = map[string]string{"Authorization": "Basic other"} + // No credential crosses a network in plaintext, and userinfo is none. + plain := binding("environment") + plain.ServerURL, plain.BearerToken, plain.HTTPHeaders = "http://mcp.test/mcp", nil, map[string]string{"X-Api-Key": "header-secret"} + query := binding("environment") + query.ServerURL, query.BearerToken, query.HTTPHeaders = "http://mcp.test/mcp?api_key=query-secret", nil, nil + userinfo := binding("service") + userinfo.ServerURL = "https://user:info-secret@mcp.test/mcp" + for name, c := range map[string]struct { + b agent.MCPBinding + prompt proto.PromptRequestPayload + }{"stdio": {stdio, service}, "Authorization twice": {twice, service}, "headers over http": {plain, environment}, "a query over http": {query, environment}, "userinfo": {userinfo, service}} { + _, err := Plan(Config{MCP: []agent.MCPBinding{c.b}, Prompt: c.prompt, OpenNetwork: sb.open}) + if !errors.Is(err, ErrInvalidConfig) || strings.Contains(err.Error(), "secret") { + t.Errorf("Plan with %s: %v", name, err) + } + } +} + +// The Harness's URL carries no query; the listener relays to exactly the +// server URL, refuses any other target and keeps the query out of response +// headers. +func TestMCPServesOnlyItsServerURL(t *testing.T) { + seen := make(chan string, 8) + srv := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case seen <- r.URL.RequestURI(): + default: + } + w.Header().Set("Location", r.URL.RequestURI()) + w.Header().Set("X-Key", r.URL.Query().Get("key")) + w.Header().Set("X-Plain", "visible") + })) + defer srv.Close() + const query = "tenant=acme&key=query%2Bsecret" + b := agent.MCPBinding{ConnectionOrigin: "service", ServerLabel: "tools", Transport: "http", ServerURL: srv.URL + "/mcp?" + query} + eps := serveOnLoopback(t, Config{MCP: []agent.MCPBinding{b}, Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, RootCAs: trust(srv)}) + harness := eps.MCP["tools"] + if !strings.HasPrefix(harness, "http://127.0.0.1:") || !strings.HasSuffix(harness, "/mcp") || strings.Contains(harness, "?") { + t.Fatalf("Harness URL %q", harness) + } + base := strings.TrimSuffix(harness, "/mcp") + for _, c := range []struct { + url string + status int + }{{harness, 200}, {harness + "?tenant=b", 400}, {harness + "?", 400}, {base + "/other", 404}, {base + "/mcp/", 404}} { + resp, err := noRedirects.Get(c.url) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + if resp.StatusCode != c.status { + t.Errorf("GET %s: %d, want %d", strings.TrimPrefix(c.url, base), resp.StatusCode, c.status) + } + if c.status == 200 && (resp.Header.Get("Location") != "" || resp.Header.Get("X-Key") != "" || resp.Header.Get("X-Plain") != "visible") { + t.Errorf("response headers %v", resp.Header) + } + } + if got := <-seen; got != "/mcp?"+query || len(seen) != 0 { + t.Errorf("the server saw %q and %d more requests", got, len(seen)) + } } diff --git a/apps/daemon/internal/gateway/proxy_test.go b/apps/daemon/internal/gateway/proxy_test.go index c7d9c85a..f6538aa2 100644 --- a/apps/daemon/internal/gateway/proxy_test.go +++ b/apps/daemon/internal/gateway/proxy_test.go @@ -11,7 +11,7 @@ import ( func TestProxyConnectsOnlyThroughTheSandbox(t *testing.T) { sb := startSandbox(t) - eps := serveOnLoopback(t, Config{OpenNetwork: sb.open}) + eps := serveOnLoopback(t, Config{OpenNetwork: sb.open, Proxy: true}) hello := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "hello "+r.RequestURI) }) secure := httptest.NewTLSServer(hello) defer secure.Close() diff --git a/apps/daemon/internal/gateway/relay.go b/apps/daemon/internal/gateway/relay.go index d0bc5e2e..c9f26547 100644 --- a/apps/daemon/internal/gateway/relay.go +++ b/apps/daemon/internal/gateway/relay.go @@ -202,23 +202,23 @@ func sentValue(v string) string { return textproto.TrimString(strings.NewReplacer("\n", " ", "\r", " ").Replace(v)) } -// withhold returns t, or, when secret is set, a transport that keeps secret -// out of the response headers the Harness receives. secret is the value as -// sent. -func withhold(t http.RoundTripper, secret string) http.RoundTripper { - if secret == "" { +// withhold returns t, or, when a secret is set, a transport that keeps each +// secret out of the response headers the Harness receives. Each secret is the +// value as sent; empty ones are ignored. +func withhold(t http.RoundTripper, secrets ...string) http.RoundTripper { + secrets = slices.DeleteFunc(slices.Clone(secrets), func(s string) bool { return s == "" }) + if len(secrets) == 0 { return t } - return withholding{next: t, secret: secret} + return withholding{next: t, secrets: secrets} } -// withholding removes every header and trailer value that contains secret +// withholding removes every header and trailer value that contains a secret // from each response, informational ones included, so an upstream that echoes -// the injected credential in a header does not disclose it. Bodies pass -// unchanged. +// an injected value in a header does not disclose it. Bodies pass unchanged. type withholding struct { - next http.RoundTripper - secret string + next http.RoundTripper + secrets []string } func (t withholding) RoundTrip(r *http.Request) (*http.Response, error) { @@ -243,11 +243,13 @@ func (t withholding) RoundTrip(r *http.Request) (*http.Response, error) { return resp, nil } -// remove deletes each value that contains the secret. A name whose values -// are all removed stays with none, so an announced trailer stays announced. +// remove deletes each value that contains a secret. A name whose values are +// all removed stays with none, so an announced trailer stays announced. func (t withholding) remove(h http.Header) { for name, values := range h { - h[name] = slices.DeleteFunc(values, func(v string) bool { return strings.Contains(v, t.secret) }) + h[name] = slices.DeleteFunc(values, func(v string) bool { + return slices.ContainsFunc(t.secrets, func(s string) bool { return strings.Contains(v, s) }) + }) } } diff --git a/apps/daemon/internal/gateway/view_linux_test.go b/apps/daemon/internal/gateway/view_linux_test.go index efa544cd..bfdf052a 100644 --- a/apps/daemon/internal/gateway/view_linux_test.go +++ b/apps/daemon/internal/gateway/view_linux_test.go @@ -21,6 +21,7 @@ import ( "github.com/hanwen/go-fuse/v2/fuse" "golang.org/x/sys/unix" + "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/agent" "github.com/MiniMax-AI/OpenAgentCore/apps/daemon/internal/sessionview" "github.com/MiniMax-AI/OpenAgentCore/internal/agentdaemon/proto" "github.com/MiniMax-AI/OpenAgentCore/internal/modelprovider" @@ -68,9 +69,10 @@ func TestListenersExistOnlyInTheSession(t *testing.T) { cfg := Config{ Models: []Model{{Name: "main", Provider: modelprovider.Provider{Protocol: modelprovider.Anthropic, BaseURL: "https://127.0.0.1:1", APIKey: upstreamKey}}}, - MCP: []proto.MCPHTTPServer{{ConnectionOrigin: "service", ServerLabel: "tools", ServerURL: "http://127.0.0.1:" + port + "/mcp"}}, + MCP: []agent.MCPBinding{{ConnectionOrigin: "service", ServerLabel: "tools", Transport: "http", ServerURL: "http://127.0.0.1:" + port + "/mcp"}}, Prompt: proto.PromptRequestPayload{DisableExecutionEnvironment: true}, OpenNetwork: startSandbox(t).open, + Proxy: true, } // The Harness's environment is built before the view exists. eps, err := Plan(cfg) diff --git a/contracts/agents-api/environments.md b/contracts/agents-api/environments.md index a20e10e8..f27733d2 100644 --- a/contracts/agents-api/environments.md +++ b/contracts/agents-api/environments.md @@ -367,7 +367,7 @@ Environment MCP needs enabled network. Duplicate server identities are rejected. ### Effective bindings -The Runtime resolves public HTTP declarations and installed Plugin MCP through `agent.ResolveMCPBindings` before adapter projection. Each transient binding keeps its connection origin, transport, nullable tool allowlist, required flag, credential authority and installed stdio identity. Bindings are never persisted or logged; duplicate identities and unavailable selected credentials are rejected. MiniMax reads its session-private native runtime-name registry for exact first-frame identities and cross-checks completed native results for both transports; adapters never fabricate a delayed start event or guess identities. +The Runtime resolves public HTTP declarations and installed Plugin MCP through `agent.ResolveMCPBindings` before adapter projection. Each transient binding keeps its connection origin, transport, nullable tool allowlist, required flag, credential authority and installed stdio identity. Bindings are never persisted or logged; duplicate identities and unavailable selected credentials are rejected. MiniMax reads its session-private native runtime-name registry for exact first-frame identities and cross-checks completed native results for both transports; adapters never fabricate a delayed start event or guess identities. In an agent-host view, the Session's [credential gateway](model-execution.md#credential-gateway) holds each HTTP binding's bearer token and headers, and the Harness receives credential-free loopback URLs ([Endpoints and proxy](harness-onboarding.md#endpoints-and-proxy)). ### Public MCP connection origin diff --git a/contracts/agents-api/model-execution.md b/contracts/agents-api/model-execution.md index 53a35c80..29ff3fa4 100644 --- a/contracts/agents-api/model-execution.md +++ b/contracts/agents-api/model-execution.md @@ -96,9 +96,11 @@ The Harness reaches its frozen upstream through a Session-local credential gatew - The gateway relays only the declared native routes of the provider's protocol, with the request and response unchanged apart from the credential rules below. An undeclared path or method, or an undeclared WebSocket upgrade, is rejected and never reaches the upstream. - It removes every value of each stripped header, matching names case-insensitively, then injects the upstream credential in the protocol's declared header, with the key's surrounding whitespace removed. -- It removes every response header and trailer value that contains the key, in informational responses too. Response bodies pass unchanged, so an upstream that echoes the key in a body discloses it to the Harness. The gateway applies the same rule to the bearer token it injects for an HTTP MCP server. +- It removes every response header and trailer value that contains the key, in informational responses too. Response bodies pass unchanged, so an upstream that echoes the key in a body discloses it to the Harness. For an HTTP MCP server, the gateway injects the binding's bearer token and HTTP headers in place of the Harness's credential headers and same-named headers, and applies the same rule to each injected value. - It never follows a redirect with the credential. - It never converts between protocols. +- An HTTP MCP server URL's query is a credential too. A binding with a bearer token, HTTP headers or a query needs an `https` server URL, and a server URL with userinfo is rejected. The gateway rejects any other before the Session starts, so no credential crosses a network in plaintext. +- The Harness's URL for an HTTP MCP binding is its listener with the server URL's path and no query. The listener relays each request to exactly the server URL, query included. It refuses a request that carries a query with 400 and one for another path with 404. It removes every response header and trailer value that contains the query or one of its parameter values, as sent or decoded, so a short value also withholds any header value that contains it. [`internal/modelprovider/config.go`](../../internal/modelprovider/config.go) declares each protocol's routes and credential header, the stripped headers and the placeholder. Its `LookupRoute` matches a request against the routes, and `UpstreamPath` joins a matched route to the upstream base URL. A Harness that calls a route the table does not declare needs a protocol change, not a gateway exception. diff --git a/docs/development.md b/docs/development.md index b6dea0f6..a4b2e1a7 100644 --- a/docs/development.md +++ b/docs/development.md @@ -73,6 +73,7 @@ For frontend development, run `pnpm dev:web` using the fixture or Core connectio | `apps/sandboxio` | Sandbox I/O service binary `oac-sandbox-io` and its Linux protocol services | [Sandbox bootstrap](sandbox-bootstrap.md#responsibilities-and-readiness), [File access protocol](file-access-protocol.md#the-linux-service), [Process protocol](process-protocol.md#implement-a-service), [Network protocol](sandbox-network-protocol.md#implement-a-service) | | `apps/daemon/internal/dispatch` | Runtime preparation, Executor reuse, Turn and cleanup ownership | [Harness lifecycle](../contracts/agents-api/harness-onboarding.md#required-adapter-interfaces) | | `apps/daemon/internal/agent` | Native harness adapters | [Native references](../contracts/agents-api/harness-onboarding.md#native-references) | +| `apps/daemon/internal/agenthost` | Agent-host Sessions: admission, Session directory, Link attachment, views and teardown | [Run in an agent-host view](../contracts/agents-api/harness-onboarding.md#run-in-an-agent-host-view) | | `services/core/internal/sandbox` | Provider interfaces and managed compute lifecycle | [Provider onboarding](sandbox-provider.md) | | `services/web` | Console login and the server-side management proxy | [Console server](web/console-server.md) | | `apps/web` and `packages/agents-client` | Console UI and typed clients | [Web guide](../apps/web/README.md) |