diff --git a/AGENTS.md b/AGENTS.md index 03b876e5..a5aee31b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -25,6 +25,7 @@ OpenAgentCore is protocol-first and modular. Core orchestrates operations that p | Provider–Sandbox I/O startup | `internal/sandboxbootstrap/bootstrap.go` | [Sandbox bootstrap](docs/sandbox-bootstrap.md) | | Runtime–file service | `internal/sandboxfs/protocol.go` | [File access protocol](docs/file-access-protocol.md) | | Runtime–process service | `internal/sandboxprocess/protocol.go` | [Process protocol](docs/process-protocol.md) | +| Runtime–network service | `internal/sandboxnet/protocol.go` | [Sandbox network protocol](docs/sandbox-network-protocol.md) | | Core–Runtime wire | `internal/agentdaemon/proto/` | [Core–Runtime protocol](docs/runtime-protocol.md) | | Runtime–Harness | `apps/daemon/internal/agent/harness.go` | [Harness onboarding](contracts/agents-api/harness-onboarding.md) | | Harness–Model provider | `internal/modelprovider/config.go` | [Model execution](contracts/agents-api/model-execution.md) | diff --git a/apps/sandboxio/internal/netservice/service.go b/apps/sandboxio/internal/netservice/service.go new file mode 100644 index 00000000..75a8a433 --- /dev/null +++ b/apps/sandboxio/internal/netservice/service.go @@ -0,0 +1,101 @@ +//go:build linux + +// Package netservice is the Linux network service of the Sandbox I/O service. +// It resolves names with the sandbox's resolver and dials TCP from the +// sandbox's network namespace. sandboxnet.Serve runs the Network protocol over +// it: the one Connect, the egress check, the answer and the splice. +package netservice + +import ( + "context" + "errors" + "net" + "net/netip" + "os" + "syscall" + + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxnet" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// Service resolves and dials for sandboxnet.Serve. +type Service struct { + resolver *net.Resolver + dialer net.Dialer +} + +var _ sandboxnet.Service = (*Service)(nil) + +// New returns a service that resolves with the sandbox's system resolver. +func New() *Service { return &Service{resolver: net.DefaultResolver} } + +// Handle serves one Network stream under the egress its Bind carries. It is +// the Serve function of the sandboxlink.ServiceNetwork handler. +func (s *Service) Handle(ctx context.Context, b sandboxlink.Bind, st sandboxlink.Stream) { + sandboxnet.Serve(ctx, st, b.Egress, s) +} + +// Resolve returns the IPv4 and IPv6 addresses of host. A name the resolver +// reports as nonexistent, or without addresses, is NameNotResolved. +func (s *Service) Resolve(ctx context.Context, host string) ([]netip.Addr, error) { + addrs, err := s.resolver.LookupNetIP(ctx, "ip", host) + var dnsErr *net.DNSError + if errors.As(err, &dnsErr) && dnsErr.IsNotFound { + return nil, &sandboxnet.Error{Code: sandboxnet.CodeNameNotResolved, Effect: sandboxwire.EffectNone, Cause: err} + } + return addrs, err +} + +// Dial connects to addr. The address is a literal and the network names its +// family, so nothing is resolved and no other address is tried. +func (s *Service) Dial(ctx context.Context, addr netip.AddrPort) (*net.TCPConn, error) { + network := "tcp4" + if addr.Addr().Is6() { + network = "tcp6" + } + c, err := s.dialer.DialContext(ctx, network, addr.String()) + if err != nil { + return nil, dialError(err) + } + return c.(*net.TCPConn), nil +} + +// dialError types a failed dial by its errno. Only an errno that socket or +// connect itself returned proves that no connection was made, so only that +// carries EffectNone. A kernel connect timeout, and any errno from a later +// step, such as registering the socket with the poller after connect was +// issued, carry EffectPossible. Without an errno, the socket deadline Go sets +// from the Connect's timeout is TimedOut: it can fire before the context +// reports its end. +func dialError(err error) error { + var errno syscall.Errno + if !errors.As(err, &errno) { + var ne net.Error + if errors.Is(err, os.ErrDeadlineExceeded) || errors.As(err, &ne) && ne.Timeout() { + return &sandboxnet.Error{Code: sandboxnet.CodeTimedOut, Effect: sandboxwire.EffectPossible, Cause: err} + } + return err + } + effect := sandboxwire.EffectPossible + var sys *os.SyscallError + if errors.As(err, &sys) && (sys.Syscall == "socket" || sys.Syscall == "connect") { + effect = sandboxwire.EffectNone + } + var code sandboxnet.Code + switch errno { + case syscall.ECONNREFUSED: + code = sandboxnet.CodeConnectionRefused + case syscall.ENETUNREACH, syscall.EHOSTUNREACH, syscall.ENETDOWN, syscall.EHOSTDOWN, syscall.EAFNOSUPPORT: + code = sandboxnet.CodeUnreachable + case syscall.EADDRNOTAVAIL, syscall.EMFILE, syscall.ENFILE, syscall.ENOBUFS, syscall.ENOMEM, syscall.EAGAIN: + code = sandboxnet.CodeResourceExhausted + case syscall.EACCES, syscall.EPERM: + code = sandboxnet.CodeDenied + case syscall.ETIMEDOUT: + code, effect = sandboxnet.CodeTimedOut, sandboxwire.EffectPossible + default: + return err + } + return &sandboxnet.Error{Code: code, Effect: effect, Cause: err} +} diff --git a/apps/sandboxio/internal/netservice/service_test.go b/apps/sandboxio/internal/netservice/service_test.go new file mode 100644 index 00000000..d7df57c3 --- /dev/null +++ b/apps/sandboxio/internal/netservice/service_test.go @@ -0,0 +1,408 @@ +//go:build linux + +package netservice + +import ( + "bytes" + "context" + "crypto/rand" + "errors" + "io" + "net" + "net/netip" + "os" + "strings" + "sync" + "testing" + "time" + + "golang.org/x/net/dns/dnsmessage" + + "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/sandboxnet" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +const wait = 5 * time.Second + +var ( + sessionID = sandboxwire.NewID() + assignmentID = sandboxwire.NewID() + resource = sandboxlink.ResourceRef{TenantID: sandboxwire.NewID(), EnvironmentID: sandboxwire.NewID(), + Kind: sandboxlink.ResourceAllocation, ID: sandboxwire.NewID(), Generation: 1} +) + +func recv[T any](t *testing.T, ch <-chan T) T { + t.Helper() + select { + case v := <-ch: + return v + case <-time.After(wait): + t.Fatalf("timed out waiting for %T", *new(T)) + panic("unreachable") + } +} + +// fixture runs this service as the network handler of a serve peer behind a +// test relay, with an attach link to open streams on. Its resolver asks a fake +// DNS server: echo.test is 127.0.0.1, blackhole.test never answers and every +// other name does not exist. +type fixture struct { + t *testing.T + auth *sandboxlinktest.Authority + srv *sandboxlinktest.Server + link *sandboxlink.AttachLink + runtime sandboxwire.ID + served chan error // sandboxnet.Serve's result for each stream + blackhole chan struct{} // a blackhole.test query arrived +} + +func newFixture(t *testing.T) *fixture { + auth := sandboxlinktest.NewAuthority() + f := &fixture{t: t, auth: auth, srv: sandboxlinktest.StartRelay(t, relay.Config{Authority: auth}), runtime: sandboxwire.NewID(), + served: make(chan error, 16), blackhole: make(chan struct{}, 16)} + dns := f.startDNS() + svc := &Service{resolver: &net.Resolver{PreferGo: true, Dial: func(ctx context.Context, _, _ string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, "udp", dns) + }}} + + auth.AddServe([]byte("serve credential"), sandboxlink.ServePeer{PeerID: sandboxwire.NewID(), Resource: resource}) + connected := make(chan struct{}, 1) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + sandboxlink.Serve(ctx, sandboxlink.ServeConfig{URL: f.srv.URL, TLS: f.srv.TLS, Credential: []byte("serve credential"), + Resource: resource, ServerInstanceID: sandboxwire.NewID(), + Services: []sandboxlink.ServiceHandler{{Service: sandboxlink.ServiceNetwork, Version: sandboxnet.Version, + Serve: func(ctx context.Context, b sandboxlink.Bind, s sandboxlink.Stream) { + f.served <- sandboxnet.Serve(ctx, s, b.Egress, svc) + }}}, + OnConnected: func(sandboxlink.HelloAccepted) { + select { + case connected <- struct{}{}: + default: + } + }}) + }() + t.Cleanup(func() { + cancel() + <-done + }) + recv(t, connected) + + auth.AddRuntime([]byte("runtime credential"), f.runtime) + link, err := sandboxlink.DialAttach(context.Background(), sandboxlink.AttachConfig{URL: f.srv.URL, TLS: f.srv.TLS, + RuntimeID: f.runtime, Credential: []byte("runtime credential")}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { link.Close() }) + f.link = link + return f +} + +func (f *fixture) startDNS() string { + pc, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + f.t.Fatal(err) + } + f.t.Cleanup(func() { pc.Close() }) + go func() { + buf := make([]byte, 1500) + for { + n, from, err := pc.ReadFrom(buf) + if err != nil { + return + } + if reply := f.answer(buf[:n]); reply != nil { + pc.WriteTo(reply, from) + } + } + }() + return pc.LocalAddr().String() +} + +func (f *fixture) answer(query []byte) []byte { + var p dnsmessage.Parser + h, err := p.Start(query) + if err != nil { + return nil + } + q, err := p.Question() + if err != nil { + return nil + } + if strings.HasPrefix(q.Name.String(), "blackhole.test.") { + select { + case f.blackhole <- struct{}{}: + default: + } + return nil + } + m := dnsmessage.Message{Header: dnsmessage.Header{ID: h.ID, Response: true, Authoritative: true, RecursionAvailable: true, + RCode: dnsmessage.RCodeNameError}, Questions: []dnsmessage.Question{q}} + if q.Name.String() == "echo.test." { + m.RCode = dnsmessage.RCodeSuccess + if q.Type == dnsmessage.TypeA { + m.Answers = []dnsmessage.Resource{{Header: dnsmessage.ResourceHeader{Name: q.Name, Type: dnsmessage.TypeA, Class: dnsmessage.ClassINET, TTL: 60}, + Body: &dnsmessage.AResource{A: [4]byte{127, 0, 0, 1}}}} + } + } + reply, err := m.Pack() + if err != nil { + return nil + } + return reply +} + +// open opens a Network stream of a new attachment whose egress is egress; nil +// allows everything. +func (f *fixture) open(egress []sandboxlink.EgressRule) (sandboxlink.Stream, sandboxwire.ID) { + f.t.Helper() + grant, attachment := sandboxwire.NewID(), sandboxwire.NewID() + f.auth.AddGrant(grant[:], sandboxlinktest.Grant{RuntimeID: f.runtime, Resource: resource, SessionID: sessionID, AssignmentID: assignmentID, + AssignmentEpoch: 1, Services: []sandboxlink.Service{sandboxlink.ServiceNetwork}, Lease: time.Minute, Egress: egress}) + ctx, cancel := context.WithTimeout(context.Background(), wait) + defer cancel() + s, _, err := f.link.OpenService(ctx, sandboxlink.Open{Service: sandboxlink.ServiceNetwork, Version: sandboxnet.Version, Resource: resource, + AttachmentID: attachment, SessionID: sessionID, AssignmentID: assignmentID, AssignmentEpoch: 1, AttachGrant: grant[:]}) + if err != nil { + f.t.Fatal(err) + } + return s, attachment +} + +func (f *fixture) connect(egress []sandboxlink.EgressRule, host string, port uint16, timeout time.Duration) (*sandboxnet.Conn, error) { + s, _ := f.open(egress) + ctx, cancel := context.WithTimeout(context.Background(), wait) + defer cancel() + return sandboxnet.Connect(ctx, s, host, port, timeout) +} + +func listen(t *testing.T) (*net.TCPListener, uint16) { + t.Helper() + ln, err := net.ListenTCP("tcp", &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + return ln, uint16(ln.Addr().(*net.TCPAddr).Port) +} + +func random(n int) []byte { + b := make([]byte, n) + rand.Read(b) + return b +} + +func TestHalfCloseFromEachSide(t *testing.T) { + f := newFixture(t) + ln, port := listen(t) + fromServer, fromClient := random(256<<10), random(1<<20) + received := make(chan []byte, 1) + go func() { + c, err := ln.AcceptTCP() + if err != nil { + received <- nil + return + } + defer c.Close() + c.Write(fromServer) + c.CloseWrite() + b, _ := io.ReadAll(c) + received <- b + }() + + conn, err := f.connect(nil, "echo.test", port, wait) + if err != nil { + t.Fatal(err) + } + written := make(chan error, 1) + go func() { + _, err := conn.Write(fromClient) + if err == nil { + err = conn.CloseWrite() + } + written <- err + }() + // The destination ended first; the other direction carries on. + got, err := io.ReadAll(conn) + if err != nil || !bytes.Equal(got, fromServer) { + t.Fatalf("read %d bytes, %v; want the destination's %d bytes and EOF", len(got), err, len(fromServer)) + } + if err := recv(t, written); err != nil { + t.Fatal(err) + } + if got := recv(t, received); !bytes.Equal(got, fromClient) { + t.Fatalf("destination read %d bytes, want %d", len(got), len(fromClient)) + } + if err := conn.Close(); err != nil { + t.Fatal(err) + } + if err := recv(t, f.served); err != nil { + t.Fatalf("Serve: %v, want an orderly end", err) + } +} + +func TestDestinationResetAborts(t *testing.T) { + f := newFixture(t) + ln, port := listen(t) + go func() { + c, err := ln.AcceptTCP() + if err != nil { + return + } + c.Read(make([]byte, 1)) + c.SetLinger(0) + c.Close() + }() + conn, err := f.connect(nil, "echo.test", port, wait) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if _, err := conn.Write([]byte{1}); err != nil { + t.Fatal(err) + } + if _, err := io.ReadAll(conn); err == nil { + t.Fatal("read ended with EOF, want an abort") + } + if err := recv(t, f.served); err == nil { + t.Fatal("Serve ended in order, want an abort") + } +} + +// Concurrent writes and reads each arrive whole, and a passed read deadline +// fails a Read even with input buffered. +func TestConcurrentCallsAndDeadline(t *testing.T) { + f := newFixture(t) + ln, port := listen(t) + go func() { + if c, err := ln.Accept(); err == nil { + io.Copy(c, c) + c.Close() + } + }() + conn, err := f.connect(nil, "127.0.0.1", port, wait) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + // Writers race one another while a reader drains the echo. + const writers, size = 4, 256 << 10 + var wg sync.WaitGroup + for range writers { + wg.Go(func() { + if _, err := conn.Write(random(size)); err != nil { + t.Error(err) + } + }) + } + if _, err := io.ReadFull(conn, make([]byte, writers*size)); err != nil { + t.Fatal(err) + } + wg.Wait() + + if _, err := conn.Write([]byte("ab")); err != nil { + t.Fatal(err) + } + b := make([]byte, 1) + if _, err := io.ReadFull(conn, b); err != nil { + t.Fatal(err) + } + conn.SetReadDeadline(time.Now().Add(-time.Second)) + if _, err := conn.Read(b); !errors.Is(err, os.ErrDeadlineExceeded) { + t.Fatalf("read after the deadline: %v, want os.ErrDeadlineExceeded", err) + } + conn.SetReadDeadline(time.Time{}) + if _, err := io.ReadFull(conn, b); err != nil || b[0] != 'b' { + t.Fatalf("read after clearing the deadline: %q, %v", b, err) + } +} + +func TestConnectFailures(t *testing.T) { + f := newFixture(t) + // The destination listens on every local address, so a dial to an + // unspecified address, which means this host, would reach it. + ln, err := net.ListenTCP("tcp", &net.TCPAddr{}) + if err != nil { + t.Fatal(err) + } + defer ln.Close() + open := uint16(ln.Addr().(*net.TCPAddr).Port) + refused, closedPort := listen(t) + refused.Close() + only := func(prefix string) []sandboxlink.EgressRule { + return []sandboxlink.EgressRule{{Prefix: netip.MustParsePrefix(prefix), PortFirst: 1, PortLast: 65535}} + } + for _, tc := range []struct { + name string + egress []sandboxlink.EgressRule + host string + port uint16 + timeout time.Duration + code sandboxnet.Code + }{ + {"denied", only("10.0.0.0/8"), "echo.test", open, wait, sandboxnet.CodeDenied}, + {"unspecified IPv6", only("::/0"), "::", open, wait, sandboxnet.CodeDenied}, + {"unspecified IPv4", only("0.0.0.0/8"), "0.0.0.0", open, wait, sandboxnet.CodeDenied}, + {"not resolved", nil, "missing.test", open, wait, sandboxnet.CodeNameNotResolved}, + {"refused", nil, "127.0.0.1", closedPort, wait, sandboxnet.CodeConnectionRefused}, + {"timed out", nil, "blackhole.test", open, 200 * time.Millisecond, sandboxnet.CodeTimedOut}, + } { + t.Run(tc.name, func(t *testing.T) { + conn, err := f.connect(tc.egress, tc.host, tc.port, tc.timeout) + var e *sandboxnet.Error + if !errors.As(err, &e) || e.Code != tc.code || e.Effect != sandboxwire.EffectNone { + t.Fatalf("Connect: %v, %v; want %s with no effect", conn, err, tc.code) + } + if err := recv(t, f.served); !errors.Is(err, tc.code) { + t.Fatalf("Serve: %v, want %s", err, tc.code) + } + }) + } + // No case may have dialed the listener. + ln.SetDeadline(time.Now().Add(100 * time.Millisecond)) + if c, err := ln.Accept(); err == nil { + c.Close() + t.Fatal("a refused Connect reached the destination") + } +} + +func TestSecondConnectIsViolation(t *testing.T) { + f := newFixture(t) + s, _ := f.open(nil) + var frames bytes.Buffer + req := sandboxnet.ConnectRequest{Network: sandboxnet.NetworkTCP, Host: "blackhole.test", Port: 80, TimeoutMillis: 10_000} + sandboxnet.WriteMessage(&frames, 1, req) + sandboxnet.WriteMessage(&frames, 2, req) + if _, err := s.Write(frames.Bytes()); err != nil { + t.Fatal(err) + } + if got, err := io.ReadAll(s); err == nil || len(got) != 0 { + t.Fatalf("read %d bytes, %v; want an abort without an answer", len(got), err) + } + if err := recv(t, f.served); !errors.Is(err, sandboxnet.ErrProtocolViolation) { + t.Fatalf("Serve: %v, want a protocol violation", err) + } +} + +func TestLostAnswerIsUncertain(t *testing.T) { + f := newFixture(t) + s, attachment := f.open(nil) + result := make(chan error, 1) + go func() { + _, err := sandboxnet.Connect(context.Background(), s, "blackhole.test", 80, 10*time.Second) + result <- err + }() + recv(t, f.blackhole) + f.srv.Relay.RevokeAttachment(attachment) + var e *sandboxnet.Error + if err := recv(t, result); !errors.As(err, &e) || e.Effect != sandboxwire.EffectPossible { + t.Fatalf("Connect: %v, want a failure with a possible effect", err) + } +} diff --git a/apps/sandboxio/internal/sandboxio/sandboxio.go b/apps/sandboxio/internal/sandboxio/sandboxio.go index 9a377e58..3a7dfe6f 100644 --- a/apps/sandboxio/internal/sandboxio/sandboxio.go +++ b/apps/sandboxio/internal/sandboxio/sandboxio.go @@ -2,7 +2,8 @@ // Package sandboxio assembles the Sandbox I/O service: it reads the // Provider's bootstrap, connects to the relay as the Link serve peer and -// serves the File and Process protocols on the streams the relay binds. +// serves the File, Process and Network protocols on the streams the relay +// binds. // docs/sandbox-bootstrap.md describes the launch. package sandboxio @@ -14,11 +15,13 @@ import ( "time" "github.com/MiniMax-AI/OpenAgentCore/apps/sandboxio/internal/fileservice" + "github.com/MiniMax-AI/OpenAgentCore/apps/sandboxio/internal/netservice" "github.com/MiniMax-AI/OpenAgentCore/apps/sandboxio/internal/processservice" "github.com/MiniMax-AI/OpenAgentCore/internal/runtimefs" "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/sandboxnet" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxprocess" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" ) @@ -101,6 +104,7 @@ func run(ctx context.Context, bootstrapPath string, opt options) error { {Service: sandboxlink.ServiceProcess, Version: sandboxprocess.Version, Serve: func(ctx context.Context, b sandboxlink.Bind, s sandboxlink.Stream) { sandboxprocess.Serve(ctx, s, sandboxprocess.Attachment{ID: b.AttachmentID}, procs) }}, + {Service: sandboxlink.ServiceNetwork, Version: sandboxnet.Version, Serve: netservice.New().Handle}, }, OnConnected: func(sandboxlink.HelloAccepted) { down.Store(false) }, OnDisconnected: func(err error) { diff --git a/apps/sandboxio/internal/sandboxio/sandboxio_test.go b/apps/sandboxio/internal/sandboxio/sandboxio_test.go index f7d6cfe5..b682b4e3 100644 --- a/apps/sandboxio/internal/sandboxio/sandboxio_test.go +++ b/apps/sandboxio/internal/sandboxio/sandboxio_test.go @@ -5,6 +5,9 @@ package sandboxio import ( "context" "errors" + "io" + "net" + "net/netip" "os" "path/filepath" "strconv" @@ -22,6 +25,7 @@ import ( "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/sandboxnet" sp "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxprocess" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" ) @@ -103,7 +107,7 @@ func next(t *testing.T, op *sp.Operation) sp.Event { } } -func TestServesFileAndProcessThroughTheRelay(t *testing.T) { +func TestServesEachProtocolThroughTheRelay(t *testing.T) { ctx := context.Background() auth := sandboxlinktest.NewAuthority() srv := sandboxlinktest.StartRelay(t, relay.Config{Authority: auth}) @@ -111,10 +115,24 @@ func TestServesFileAndProcessThroughTheRelay(t *testing.T) { auth.AddServe([]byte("serve-credential"), sandboxlink.ServePeer{PeerID: sandboxwire.NewID(), Resource: resource}) runtimeID := sandboxwire.NewID() auth.AddRuntime([]byte("runtime-credential"), runtimeID) + // The grant lets Network streams reach only an echo listener. + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + go func() { + if c, err := ln.Accept(); err == nil { + io.Copy(c, c) + c.Close() + } + }() + echo := ln.Addr().(*net.TCPAddr).AddrPort() o := sandboxlink.Open{Resource: resource, AttachmentID: sandboxwire.NewID(), SessionID: sandboxwire.NewID(), AssignmentID: sandboxwire.NewID(), AssignmentEpoch: 1, AttachGrant: []byte("grant")} auth.AddGrant(o.AttachGrant, sandboxlinktest.Grant{RuntimeID: runtimeID, Resource: resource, SessionID: o.SessionID, - AssignmentID: o.AssignmentID, AssignmentEpoch: 1, Services: []sandboxlink.Service{sandboxlink.ServiceFile, sandboxlink.ServiceProcess}, Lease: time.Minute}) + AssignmentID: o.AssignmentID, AssignmentEpoch: 1, Services: []sandboxlink.Service{sandboxlink.ServiceFile, sandboxlink.ServiceProcess, sandboxlink.ServiceNetwork}, Lease: time.Minute, + Egress: []sandboxlink.EgressRule{{Prefix: netip.PrefixFrom(echo.Addr(), 32), PortFirst: echo.Port(), PortLast: echo.Port()}}}) link, err := sandboxlink.DialAttach(ctx, sandboxlink.AttachConfig{URL: srv.URL, TLS: srv.TLS, RuntimeID: runtimeID, Credential: []byte("runtime-credential")}) if err != nil { t.Fatal(err) @@ -182,6 +200,26 @@ func TestServesFileAndProcessThroughTheRelay(t *testing.T) { t.Fatalf("output %q, exit %+v", output.String(), exited.Status) } + // A Network stream connects under the Bind's egress and carries bytes + // both ways. + o.Service, o.Version = sandboxlink.ServiceNetwork, sandboxnet.Version + stream, _, err = open(link, o) + if err != nil { + t.Fatal(err) + } + conn, err := sandboxnet.Connect(ctx, stream, echo.Addr().String(), echo.Port(), wait) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + if _, err := conn.Write([]byte("ping")); err != nil { + t.Fatal(err) + } + got := make([]byte, 4) + if _, err := io.ReadFull(conn, got); err != nil || string(got) != "ping" { + t.Fatalf("echo: %q, %v", got, err) + } + // Stopping the service terminates its live operations before it returns. op, _, err = procs.Start(ctx, described.ServerInstanceID, sandboxwire.NewID(), shell("echo $$; exec sleep 600")) if err != nil { diff --git a/docs/development.md b/docs/development.md index a6b676dc..b6dea0f6 100644 --- a/docs/development.md +++ b/docs/development.md @@ -69,7 +69,8 @@ For frontend development, run `pnpm dev:web` using the fixture or Core connectio | `internal/sandboxbootstrap` | Provider-to-Sandbox I/O service startup input | [Sandbox bootstrap](sandbox-bootstrap.md) | | `internal/sandboxfs` | Runtime–file service wire types, validators, client and server | [File access protocol](file-access-protocol.md) | | `internal/sandboxprocess` | Runtime–process service wire types, validators, client and server | [Process protocol](process-protocol.md) | -| `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) | +| `internal/sandboxnet` | Runtime–network service wire types, validators, client and server | [Sandbox network protocol](sandbox-network-protocol.md) | +| `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) | | `services/core/internal/sandbox` | Provider interfaces and managed compute lifecycle | [Provider onboarding](sandbox-provider.md) | diff --git a/docs/sandbox-bootstrap.md b/docs/sandbox-bootstrap.md index cd349b7c..472a0af4 100644 --- a/docs/sandbox-bootstrap.md +++ b/docs/sandbox-bootstrap.md @@ -36,7 +36,7 @@ The Provider creates the account and the sandbox, delivers this file and starts The service validates the input and owns the link: it connects as the serve peer, serves bound streams and reconnects while the credential stays valid. `resource`, including its generation, must be the resource the credential serves, or the relay refuses the link. -The File service serves the single export `world`, rooted at the sandbox's `/`, and the Provider's sandbox setup owns that topology's isolation. The [Process service](process-protocol.md#implement-a-service) runs processes as the service's account, and the service, a child subreaper, reaps their orphaned descendants. +The File service serves the single export `world`, rooted at the sandbox's `/`, and the Provider's sandbox setup owns that topology's isolation. The [Process service](process-protocol.md#implement-a-service) runs processes as the service's account, and the service, a child subreaper, reaps their orphaned descendants. The service also serves the [Network protocol](sandbox-network-protocol.md): it resolves names and dials TCP from the sandbox's network namespace, within the egress each stream's `Bind` carries. The service exits nonzero with a message naming the failed step when it cannot start, and with the relay's failure code when `Serve` returns a [refusal](sandbox-link-protocol.md#implement-a-serve-peer). Neither message includes the credential. On SIGTERM it stops accepting streams, cancels its live operations as [ownership cleanup](process-protocol.md#ownership) does, waits for them to end, at most the cancel grace limit plus five seconds, and exits 0. diff --git a/docs/sandbox-link-protocol.md b/docs/sandbox-link-protocol.md index b5f023c9..e533e840 100644 --- a/docs/sandbox-link-protocol.md +++ b/docs/sandbox-link-protocol.md @@ -85,7 +85,7 @@ Decoding rules: - Reject unknown tags, nonzero flags, unknown enum values, duplicate entries, overflow and trailing payload bytes. Every such error wraps `sandboxwire.ErrMalformed`. - There are no maps and no implicit defaults. - A file or process data chunk is at most 64 KiB (`sandboxwire.MaxChunk`). -- Each sender's request IDs on a stream strictly increase in wire order, so a receiver checks uniqueness in constant memory. A sender takes each ID from `sandboxwire.RequestSequence.Next` in the critical section that writes the frame. A receiver checks each ID with `Admit` and answers an ID that does not increase as a protocol violation, without dispatching the request. +- Each sender's request IDs on a stream strictly increase in wire order, so a receiver checks uniqueness in constant memory. A sender takes each ID from `sandboxwire.RequestSequence.Next` in the critical section that writes the frame. A receiver checks each ID with `Admit` and answers an ID that does not increase as a protocol violation, without dispatching the request. A service stream carries two sequences in turn: the Link exchange that opens it, `Open` and `Opened` on the attach side or `Bind` and `Bound` on the serve side, then the service protocol's own sequence, which starts after that exchange, so the first service request may use ID 1 again. - A failed request states whether it may have taken effect: `EffectNone` or `EffectPossible`. Transport loss after dispatch is `EffectPossible` unless the server later establishes the result. Each protocol keeps annotated golden frames under its package's `testdata` and a decoder fuzz target. @@ -221,7 +221,7 @@ An attachment's binding identity is its `AttachmentID`, `Resource`, `SessionID`, `Bind` carries the authorized binding and never the grant or a credential: - For a File stream, `Exports` lists 1 to 64 exports the stream may use, each by ID and read-write or `ReadOnly`. An export ID is 1 to 64 lowercase letters, digits, `_` and `-`, and IDs in a list are distinct. Process and Network binds carry no `Exports`. The [Sandbox bootstrap](sandbox-bootstrap.md#responsibilities-and-readiness) states which exports the File service serves. -- For a Network stream, `Egress` lists the destinations the stream may reach: an address inside a rule's prefix on a port from `PortFirst` to `PortLast`. An empty list denies everything. Each prefix has its host bits zero, `1 ≤ PortFirst ≤ PortLast`, and no rule appears twice. File and Process binds carry no `Egress`. +- For a Network stream, `Egress` lists the destinations the stream may reach: an address inside a rule's prefix on a port from `PortFirst` to `PortLast`. An empty list denies everything. Each prefix has its host bits zero, `1 ≤ PortFirst ≤ PortLast`, and no rule appears twice. File and Process binds carry no `Egress`. The [Network protocol](sandbox-network-protocol.md#egress-check) states how the service applies it. The exports and egress of a stream are fixed when it opens; a renewal changes only the lease. @@ -251,7 +251,7 @@ Losing a link resets the streams it carries and keeps its attachments until thei ## Stream ends -The streams handed to service handlers and returned by `OpenService` implement `sandboxlink.Stream`: `Read`, `Write`, `CloseWrite`, `Close`, which ends writing in order and discards further input, and `Reset`. +The streams handed to service handlers and returned by `OpenService` implement `sandboxlink.Stream`: `Read`, `Write`, `CloseWrite`, `Close`, which ends writing in order and discards further input, `Reset`, and `SetDeadline`, `SetReadDeadline` and `SetWriteDeadline`. The deadlines bound how long a `Read` or `Write` waits, as yamux's do: a call that would wait past its deadline fails with an error whose `Timeout` is true and leaves the stream usable, and a `Read` still returns input already buffered. One goroutine may read while another writes; concurrent reads, or concurrent writes, need the caller's own lock. The relay copies each direction through a 32 KiB buffer and holds at most one 256 KiB yamux window per stream: diff --git a/docs/sandbox-network-protocol.md b/docs/sandbox-network-protocol.md new file mode 100644 index 00000000..a98e2305 --- /dev/null +++ b/docs/sandbox-network-protocol.md @@ -0,0 +1,104 @@ +# Sandbox network protocol + +The Network protocol is how the agent host opens a TCP connection that originates in a sandbox. The agent host opens a Network stream through the Link and sends one `Connect`. The Sandbox I/O service resolves the name with the sandbox's resolver, checks the destination against the egress the Link bound to the stream, dials from the sandbox's network and answers. After `Connected` the stream carries the connection's raw bytes in both directions. Credentials and TLS stay with the agent host: the service carries the bytes the client writes and reads nothing in them. + +[`internal/sandboxnet/protocol.go`](../internal/sandboxnet/protocol.go) is the authored definition: message tags, payload layouts, validators, the egress rule and the `Service` interface. The same package has the client and the generic server. [`apps/sandboxio/internal/netservice`](../apps/sandboxio/internal/netservice) is the Linux service. Frames use the shared [framing](sandbox-link-protocol.md#framing). A stream's authority, including its `Egress`, comes from the Link when the stream [opens](sandbox-link-protocol.md#opening-a-stream), never from `Connect`. + +## How a connection works + +1. The client opens a Link stream for `ServiceNetwork` version 1. +2. The client sends one `Connect` with a request ID that follows the [request ID rule](sandbox-link-protocol.md#framing), and writes nothing else until the answer. +3. The service checks the request against the stream's egress, resolves the host, and dials the first permitted address within `TimeoutMillis`, as described in [Egress check](#egress-check). +4. The service answers `Connected` or `Failed` with the request's ID. After `Failed` it ends the stream in order. After `Connected` both directions carry raw bytes without framing. + +Each direction ends on its own. The client's orderly end (`CloseWrite`, a FIN) reaches the destination as a TCP half-close after every byte before it, and the destination's half-close reaches the client as EOF after every byte before it. A half-close starts no timeout, and the other direction carries on. + +Anything else aborts both directions: a reset from the destination, a failed read or write on either side, the end of the attachment or the loss of a link. The client then reads an error that is never EOF, and the service resets the socket. After `Connected` a failure is never reported as a frame. The service learns of a reset only when it next reads or writes the stream, as [Stream ends](sandbox-link-protocol.md#stream-ends) describes: if the serve link drops after the client's half-close while the destination stays silent, the socket stays open until the destination sends or ends, or the attachment closes. + +A stream carries one connection. There is no replay or resumption: a lost answer leaves the attempt uncertain, and nothing retries it automatically. + +## Implement a client + +The Go client is `sandboxnet.Connect(ctx, stream, host, port, timeout)`. It takes a newly opened Network stream, owns it from then on, and returns a `*sandboxnet.Conn`, which is a `net.Conn`. + +- A failure is a `*sandboxnet.Error` with a `Code` and an `Effect`, and `errors.Is(err, sandboxnet.CodeDenied)` matches its code. Arguments that fail validation fail with `InvalidArgument`, and a context that ended before the request was sent fails with `Cancelled` or `TimedOut`, both with `EffectNone`. Once the request may have been sent, a lost answer (`IO`), a malformed answer (`Unknown`) or the end of the context (`Cancelled` or `TimedOut`) fails with `EffectPossible`: the sandbox may have connected. `Connect` never retries. +- `timeout` is rounded up to whole milliseconds and must be from 1 ms to 60 s. The context bounds the wait for the answer. +- `Conn.CloseWrite` ends the write direction in order after any `Write` in progress. `Conn.Close` ends in order only after `Read` returned `io.EOF`, with no failed read or write and no `Write` in progress; otherwise it aborts, so unread input never holds the Link's flow-control window and a pending call returns. `Conn.Reset` aborts explicitly. +- Deadlines follow `net.Conn`: once a deadline has passed, `Read` or `Write` returns `os.ErrDeadlineExceeded`, even with input buffered, until the deadline is extended. Methods may be called concurrently. `RemoteAddr` is the requested host and port. `LocalAddr` is empty, because the protocol does not report the sandbox's local endpoint. + +A client in another language follows the same rules: one `Connect` per stream, no bytes before the answer, and no automatic retry of a failure with `EffectPossible`. + +## Implement a service + +Implement `sandboxnet.Service` and serve each Network stream with `sandboxnet.Serve(ctx, stream, bind.Egress, service)`, where `ctx` is the attachment's. `Serve` owns the protocol: + +- It reads one frame. A frame that is not a `Connect`, or whose request ID is zero, resets the stream. A `Connect` that fails validation is answered `InvalidArgument`. +- It keeps reading the stream while it resolves and dials. Any byte that arrives before the dial succeeds, including a second `Connect`, is a protocol violation: it cancels the dial and resets the stream. Bytes that arrive later wait until `Connected` is written, so nothing reaches the destination before the answer. +- It runs the [egress check](#egress-check) and calls `Resolve` and `Dial`. `Dial` connects to exactly the address it is given, in that address's family, and never resolves or tries another address. +- It answers an `*Error` from `Resolve` or `Dial` as is. Any other error becomes `TimedOut` when the timeout passed, `NameResolutionFailed` with `EffectNone` from `Resolve`, and `IO` with `EffectPossible` from `Dial`. +- After `Connected` it splices the stream and the socket. The end of `ctx` aborts both at any point. + +`netservice.New()` returns the Linux service. It resolves with the sandbox's system resolver, which reads the sandbox's `/etc/hosts` and `/etc/resolv.conf`, and dials from the sandbox's network namespace over `tcp4` or `tcp6`, matching the address. `Service.Handle` is the `Serve` function of the `sandboxlink.ServiceNetwork` handler. It types a failed dial by its errno as the [failure table](#failures) lists. + +## Reference + +### Messages + +```text +Version = 1 // the Link service version of ServiceNetwork +OpConnect = 1 +ConnectResponseTag = 0x8001 + +ConnectRequest payload, in order: + Network uint16 // NetworkTCP = 1 + HostLength uint32 + Host byte[HostLength] + Port uint16 + TimeoutMillis uint32 + +ConnectResponse payload, in order: + Result uint16 // ResultConnected = 1, ResultFailed = 2 + if ResultFailed: + Code uint16 + Effect uint16 // EffectNone = 1, EffectPossible = 2 +``` + +A payload is at most 265 bytes. Unknown enum values, a failure field after `Connected` and trailing bytes are malformed. + +### Arguments + +- `Host` is 1 to 253 bytes: an unbracketed IPv4 or IPv6 literal without a zone, or a DNS name. A name is labels separated by dots, with an optional final dot. Each label is 1 to 63 ASCII letters, digits, hyphens and underscores, and does not begin or end with a hyphen. The last label is not all digits. A name with non-ASCII characters is sent in IDNA A-label form (`xn--`). NUL, whitespace, URLs, brackets and embedded ports are rejected. +- `Port` is from 1 to 65535. +- `TimeoutMillis` is from 1 to 60,000 and covers resolution plus dialing. + +### Egress check + +The service checks a `Connect` against the stream's [`Egress`](sandbox-link-protocol.md#opening-a-stream) in this order: + +1. When no rule admits the port, the answer is `Denied` and nothing is resolved. +2. An IP literal is the only candidate address. Otherwise the service resolves the name, and each address it returns is a candidate. +3. Each candidate is checked with the port. An IPv4-mapped IPv6 address is checked and dialed as IPv4. An address with a zone is never permitted, and neither is an unspecified address (`0.0.0.0`, `::`), because a TCP stack connects it to the sandbox itself whatever the rules say. Every other address, multicast and broadcast included, is checked as written: TCP cannot connect to those, so permitting one only leads to a dial failure. +4. The service dials the first permitted candidate once, at that address, without resolving again. When no candidate is permitted, the answer is `Denied`. + +### Failures + +| Code | Name | Effect | Returned when | +| --- | --- | --- | --- | +| 1 | `InvalidArgument` | None | The `Connect` is malformed, or its host, port or timeout is out of range | +| 2 | `UnsupportedNetwork` | None | The service does not serve the requested network | +| 3 | `Denied` | None, except as below | The egress admits neither the port nor any candidate address, or the sandbox refused the connect (`EACCES`, `EPERM`) | +| 4 | `NameNotResolved` | None | The name does not exist or has no address | +| 5 | `NameResolutionFailed` | None | Resolution failed for another reason | +| 6 | `ConnectionRefused` | None, except as below | The destination refused the connection (`ECONNREFUSED`) | +| 7 | `Unreachable` | None, except as below | The sandbox has no route to the destination (`ENETUNREACH`, `EHOSTUNREACH`, `ENETDOWN`, `EHOSTDOWN`, `EAFNOSUPPORT`) | +| 8 | `TimedOut` | None during resolution or before the client sent the request, Possible after | `TimeoutMillis` passed or the kernel gave up the connect (`ETIMEDOUT`); for the client, its context's deadline passed | +| 9 | `ResourceExhausted` | None, except as below | The sandbox ran out of ports, descriptors or memory (`EADDRNOTAVAIL`, `EMFILE`, `ENFILE`, `ENOBUFS`, `ENOMEM`, `EAGAIN`) | +| 10 | `Cancelled` | None before the request is sent, Possible after | The client's context was cancelled, or the attachment ended | +| 11 | `IO` | Possible | The dial failed for another reason; for the client, the stream failed before the answer arrived | +| 12 | `Unknown` | Possible | For the client, the answer broke the protocol | + +A code typed from an errno has `EffectNone` only when `socket` or `connect` itself returned the errno. The same errno from a later step of the dial, such as registering the socket with the poller after `connect` was issued, has `EffectPossible`, because the connection may have been made. + +## Verification + +`go test ./internal/sandboxnet/ ./apps/sandboxio/internal/netservice/` covers the golden frames, decode rejection, the host grammar, the egress rule and bytes held until `Connected` is written, and, over a test relay, bytes in both directions, a half-close from each side, a destination reset, concurrent writes, a passed read deadline, `Denied` without a dial, including for unspecified addresses, `NameNotResolved`, `ConnectionRefused`, `TimedOut`, a second `Connect` and a lost answer. `go test -run '^$' -fuzz FuzzDecode ./internal/sandboxnet` fuzzes the decoder. diff --git a/internal/sandboxlink/transport.go b/internal/sandboxlink/transport.go index fabbabba..0cb71072 100644 --- a/internal/sandboxlink/transport.go +++ b/internal/sandboxlink/transport.go @@ -20,7 +20,9 @@ import ( // an abort: after the other end's CloseWrite, Read returns io.EOF once every // byte written before it has been read; after a Reset anywhere on the path, // including the relay's for lease expiry, revocation or link loss, Read and -// Write return an error that is not io.EOF. +// Write return an error that is not io.EOF. One goroutine may read while +// another writes; concurrent reads, or concurrent writes, need the caller's +// own serialization. type Stream interface { io.Reader io.Writer @@ -30,6 +32,14 @@ type Stream interface { Close() error // Reset aborts both directions. Reset() error + // SetDeadline, SetReadDeadline and SetWriteDeadline bound how long + // pending and future reads and writes wait, as yamux's do; a zero time + // removes the bound. A call that would wait past the deadline fails with + // an error whose Timeout is true and leaves the stream usable. A Read + // still returns input already buffered after its deadline has passed. + SetDeadline(t time.Time) error + SetReadDeadline(t time.Time) error + SetWriteDeadline(t time.Time) error } // Dialer opens the byte stream a peer runs yamux over. Production peers use diff --git a/internal/sandboxnet/client.go b/internal/sandboxnet/client.go new file mode 100644 index 00000000..ee5d6293 --- /dev/null +++ b/internal/sandboxnet/client.go @@ -0,0 +1,245 @@ +package sandboxnet + +import ( + "context" + "errors" + "io" + "net" + "os" + "strconv" + "sync" + "sync/atomic" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// Connect asks the network service on s, a freshly opened Network stream, to +// connect to host:port within timeout, and returns the connection. Connect +// owns s: on success the Conn owns it, and on failure Connect has ended it. +// +// A failure is an *Error. A Connect that cannot be sent, or whose ctx ended +// before it was sent, fails with EffectNone. Once it may have been sent, a +// lost answer, a broken one or the end of ctx fails with EffectPossible: the +// sandbox may have connected. Connect never retries; a caller that retries an +// uncertain Connect opens a second connection. +func Connect(ctx context.Context, s sandboxlink.Stream, host string, port uint16, timeout time.Duration) (*Conn, error) { + var seq sandboxwire.RequestSequence + id := seq.Next() + f, err := Encode(id, ConnectRequest{Network: NetworkTCP, Host: host, Port: port, TimeoutMillis: millis(timeout)}) + if err != nil { + s.Reset() + return nil, &Error{Code: CodeInvalidArgument, Effect: sandboxwire.EffectNone, Cause: err} + } + if err := ctx.Err(); err != nil { + s.Reset() + return nil, &Error{Code: contextCode(err), Effect: sandboxwire.EffectNone, Cause: err} + } + stop := context.AfterFunc(ctx, func() { s.Reset() }) + resp, err := exchange(s, f) + if !stop() { + return nil, &Error{Code: contextCode(ctx.Err()), Effect: sandboxwire.EffectPossible, Cause: ctx.Err()} + } + if err != nil { + s.Reset() + return nil, err + } + if err := resp.Err(); err != nil { + s.Close() + return nil, err + } + return &Conn{s: s, remote: Addr{Host: host, Port: port}}, nil +} + +// exchange writes the Connect frame and reads its answer. +func exchange(s sandboxlink.Stream, f sandboxwire.Frame) (ConnectResponse, error) { + if err := sandboxwire.WriteFrame(s, f); err != nil { + return ConnectResponse{}, &Error{Code: CodeIO, Effect: sandboxwire.EffectPossible, Cause: err} + } + answer, err := sandboxwire.ReadFrame(s, MaxMessageBytes) + if err != nil { + if errors.Is(err, sandboxwire.ErrMalformed) { + return ConnectResponse{}, &Error{Code: CodeUnknown, Effect: sandboxwire.EffectPossible, Cause: err} + } + return ConnectResponse{}, &Error{Code: CodeIO, Effect: sandboxwire.EffectPossible, Cause: err} + } + m, err := Decode(answer) + if err == nil && (answer.Type != ConnectResponseTag || answer.RequestID != f.RequestID) { + err = invalid("answer type %#04x request ID %d to request %d", answer.Type, answer.RequestID, f.RequestID) + } + if err != nil { + return ConnectResponse{}, &Error{Code: CodeUnknown, Effect: sandboxwire.EffectPossible, Cause: err} + } + return m.(ConnectResponse), nil +} + +// millis rounds timeout up to whole milliseconds. A timeout out of range +// yields a value the request validator rejects. +func millis(timeout time.Duration) uint32 { + if timeout <= 0 { + return 0 + } + if timeout > MaxTimeoutMillis*time.Millisecond { + return MaxTimeoutMillis + 1 + } + return uint32((timeout + time.Millisecond - 1) / time.Millisecond) +} + +func contextCode(err error) Code { + if errors.Is(err, context.DeadlineExceeded) { + return CodeTimedOut + } + return CodeCancelled +} + +// Conn is a connection the sandbox made. Read returns io.EOF only after the +// destination ended its write side and every byte before that arrived; an +// abort anywhere on the path, such as a reset from the destination or the end +// of the attachment, is an error that is not io.EOF. Its methods may be called +// concurrently, as net.Conn's may. +type Conn struct { + s sandboxlink.Stream + remote Addr + // rmu and wmu serialize reads and writes, which the stream does not. wmu + // also orders an orderly end after every write. + rmu, wmu sync.Mutex + // rdl and wdl hold the deadlines, so a passed one fails a call even when + // the stream could complete it from its buffer. dmu keeps them and the + // stream's deadlines in agreement under concurrent setters. + dmu sync.Mutex + rdl, wdl atomic.Pointer[time.Time] + closed atomic.Bool + failed atomic.Bool // a Read or Write failed for a reason other than a deadline + eof atomic.Bool // Read returned io.EOF +} + +var _ net.Conn = (*Conn)(nil) + +func (c *Conn) Read(b []byte) (int, error) { + c.rmu.Lock() + defer c.rmu.Unlock() + if c.closed.Load() { + return 0, net.ErrClosed + } + if passed(&c.rdl) { + return 0, os.ErrDeadlineExceeded + } + n, err := c.s.Read(b) + if err == io.EOF { + c.eof.Store(true) + return n, err + } + return n, c.note(err) +} + +func (c *Conn) Write(b []byte) (int, error) { + c.wmu.Lock() + defer c.wmu.Unlock() + if c.closed.Load() { + return 0, net.ErrClosed + } + if passed(&c.wdl) { + return 0, os.ErrDeadlineExceeded + } + n, err := c.s.Write(b) + return n, c.note(err) +} + +// note records a failure. A passed deadline is not one: it returns +// os.ErrDeadlineExceeded, as net.Conn requires, and the Conn stays usable. +func (c *Conn) note(err error) error { + if err == nil { + return nil + } + var ne net.Error + if errors.As(err, &ne) && ne.Timeout() { + return os.ErrDeadlineExceeded + } + c.failed.Store(true) + return err +} + +// CloseWrite ends the write direction in order: the destination reads EOF +// after every byte written before it. It waits for a Write in progress. +// Reading carries on. +func (c *Conn) CloseWrite() error { + c.wmu.Lock() + defer c.wmu.Unlock() + if c.closed.Load() { + return net.ErrClosed + } + return c.s.CloseWrite() +} + +// Close ends the connection. After Read returned io.EOF, with no failed Read +// or Write and no Write in progress, it ends the write direction in order. +// Otherwise it aborts the connection like Reset, because input may remain +// unread or a Write may be waiting; the abort ends any pending call. +func (c *Conn) Close() error { + if c.closed.Swap(true) { + return net.ErrClosed + } + if c.eof.Load() && !c.failed.Load() && c.wmu.TryLock() { + defer c.wmu.Unlock() + return c.s.Close() + } + return c.s.Reset() +} + +// Reset aborts both directions: the sandbox resets the destination +// connection. +func (c *Conn) Reset() error { + if c.closed.Swap(true) { + return net.ErrClosed + } + return c.s.Reset() +} + +// LocalAddr returns the zero Addr: the protocol does not report the sandbox's +// local endpoint. +func (c *Conn) LocalAddr() net.Addr { return Addr{} } + +// RemoteAddr returns the host and port the Conn was asked to reach. +func (c *Conn) RemoteAddr() net.Addr { return c.remote } + +// SetDeadline, SetReadDeadline and SetWriteDeadline follow net.Conn: once a +// deadline has passed, Read or Write fails with os.ErrDeadlineExceeded until +// the deadline is extended, and a zero time removes it. +func (c *Conn) SetDeadline(t time.Time) error { + c.dmu.Lock() + defer c.dmu.Unlock() + c.rdl.Store(&t) + c.wdl.Store(&t) + return c.s.SetDeadline(t) +} + +func (c *Conn) SetReadDeadline(t time.Time) error { + c.dmu.Lock() + defer c.dmu.Unlock() + c.rdl.Store(&t) + return c.s.SetReadDeadline(t) +} + +func (c *Conn) SetWriteDeadline(t time.Time) error { + c.dmu.Lock() + defer c.dmu.Unlock() + c.wdl.Store(&t) + return c.s.SetWriteDeadline(t) +} + +// passed reports whether the deadline in d is set and has passed. +func passed(d *atomic.Pointer[time.Time]) bool { + t := d.Load() + return t != nil && !t.IsZero() && !time.Now().Before(*t) +} + +// Addr is a destination as the Connect named it. +type Addr struct { + Host string + Port uint16 +} + +func (Addr) Network() string { return "tcp" } + +func (a Addr) String() string { return net.JoinHostPort(a.Host, strconv.Itoa(int(a.Port))) } diff --git a/internal/sandboxnet/protocol.go b/internal/sandboxnet/protocol.go new file mode 100644 index 00000000..0a88647d --- /dev/null +++ b/internal/sandboxnet/protocol.go @@ -0,0 +1,403 @@ +// Package sandboxnet is the Network protocol. On a Link stream of +// sandboxlink.ServiceNetwork, the agent-host Runtime asks the sandbox network +// service to connect to a TCP destination; once the service answers Connected, +// the stream carries the connection's raw bytes in both directions. Name +// resolution, the egress check and the dial happen in the sandbox, under the +// egress the Link bound to the stream. +// +// This file is the protocol's one authored definition: its vocabulary, message +// tags and payload layouts, validators, the egress rule and the Service a +// network service implements. The framing and primitive encoding come from +// sandboxwire. The protocol document is docs/sandbox-network-protocol.md. +package sandboxnet + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/netip" + + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// Version is the Network protocol version. It is matched exactly, through the +// Link's service version. +const Version uint16 = 1 + +// OpConnect is the only request: it opens the stream's one connection. +const OpConnect uint16 = 1 + +// ConnectResponseTag is the tag of the answer to OpConnect, OpConnect|0x8000. +const ConnectResponseTag uint16 = 0x8001 + +var tags = sandboxwire.Tags{Requests: OpConnect} + +const ( + // MaxHostBytes bounds a Connect host. + MaxHostBytes = 253 + // MaxTimeoutMillis bounds a Connect timeout. + MaxTimeoutMillis = 60_000 + // MaxMessageBytes bounds every Network message payload: a Connect with + // the longest host. + MaxMessageBytes = 2 + 4 + MaxHostBytes + 2 + 4 +) + +// ErrProtocolViolation ends a stream whose peer broke the protocol: a frame +// that is not a Connect, a request ID that does not increase, a second +// Connect, or bytes that arrive before the dial succeeds. +var ErrProtocolViolation = errors.New("sandbox network: protocol violation") + +// Network is the transport a Connect asks for. +type Network uint16 + +const NetworkTCP Network = 1 + +func (n Network) Valid() bool { return n == NetworkTCP } + +// Result says whether a Connect connected. +type Result uint16 + +const ( + ResultConnected Result = 1 + ResultFailed Result = 2 +) + +func (r Result) Valid() bool { return r == ResultConnected || r == ResultFailed } + +// Code is the typed outcome of a failed Connect. It is also an error value, so +// errors.Is(err, sandboxnet.CodeDenied) matches an *Error with that code. +type Code uint16 + +const ( + // CodeInvalidArgument: the Connect is malformed or its host, port or + // timeout is out of range. + CodeInvalidArgument Code = iota + 1 + // CodeUnsupportedNetwork: the service does not serve the Connect's + // network. + CodeUnsupportedNetwork + // CodeDenied: the binding's egress permits neither the port nor any + // address of the host, or the sandbox refused the connection. + CodeDenied + // CodeNameNotResolved: the host has no address in the sandbox. + CodeNameNotResolved + // CodeNameResolutionFailed: resolution failed for another reason. + CodeNameResolutionFailed + // CodeConnectionRefused: the destination refused the connection. + CodeConnectionRefused + // CodeUnreachable: the sandbox has no route to the destination. + CodeUnreachable + // CodeTimedOut: resolution and dialing did not finish within the timeout. + CodeTimedOut + // CodeResourceExhausted: the sandbox ran out of sockets, ports or memory. + CodeResourceExhausted + // CodeCancelled: the caller or the attachment gave up. + CodeCancelled + // CodeIO: the stream failed before an answer arrived. + CodeIO + // CodeUnknown: any other failure, including an answer that breaks the + // protocol. + CodeUnknown +) + +var codeNames = [...]string{"", "invalid argument", "unsupported network", "denied", "name not resolved", + "name resolution failed", "connection refused", "unreachable", "timed out", "resource exhausted", + "cancelled", "io", "unknown"} + +func (c Code) Valid() bool { return c >= CodeInvalidArgument && c <= CodeUnknown } + +func (c Code) String() string { + if c.Valid() { + return codeNames[c] + } + return fmt.Sprintf("code(%d)", uint16(c)) +} + +func (c Code) Error() string { return "sandbox network: " + c.String() } + +// Error is a failed Connect with its effect: the service's answer or a local +// failure of the exchange. Cause is the local error that produced it; it is +// never sent. +type Error struct { + Code Code + Effect sandboxwire.Effect + Cause error +} + +func (e *Error) Error() string { + effect := "no effect" + if e.Effect == sandboxwire.EffectPossible { + effect = "effect possible" + } + if e.Cause != nil { + return fmt.Sprintf("%s (%s): %v", e.Code.Error(), effect, e.Cause) + } + return fmt.Sprintf("%s (%s)", e.Code.Error(), effect) +} + +func (e *Error) Unwrap() error { return e.Cause } + +// Is matches a Code target. +func (e *Error) Is(target error) bool { + c, ok := target.(Code) + return ok && c == e.Code +} + +// ConnectRequest asks the service to connect to Host:Port. Host is an ASCII +// DNS name in A-label form or an unbracketed IP literal without a zone. +// TimeoutMillis bounds resolution plus dialing. +type ConnectRequest struct { + Network Network + Host string + Port uint16 + TimeoutMillis uint32 +} + +// ConnectResponse answers a ConnectRequest. Code and Effect are set exactly +// when Result is ResultFailed. +type ConnectResponse struct { + Result Result + Code Code + Effect sandboxwire.Effect +} + +// Err returns the failure an answer carries, or nil for Connected. +func (r ConnectResponse) Err() error { + if r.Result == ResultConnected { + return nil + } + return &Error{Code: r.Code, Effect: r.Effect} +} + +// Service is what a network service provides to Serve: resolution and a dial +// in the sandbox. Serve owns the protocol, the egress check and the splice. +type Service interface { + // Resolve returns the addresses of a DNS name. A typed outcome is an + // *Error; Serve answers any other error NameResolutionFailed, or + // TimedOut when the Connect's timeout passed. + Resolve(ctx context.Context, host string) ([]netip.Addr, error) + // Dial connects to exactly addr, in its address family, without + // resolving anything or falling back to another address. A typed outcome + // is an *Error; Serve answers any other error IO with EffectPossible, or + // TimedOut with EffectPossible when the Connect's timeout passed. + Dial(ctx context.Context, addr netip.AddrPort) (*net.TCPConn, error) +} + +// Message is a ConnectRequest or a ConnectResponse. +type Message interface { + frameType() uint16 + encode(*sandboxwire.Encoder) + validate() error +} + +func (ConnectRequest) frameType() uint16 { return OpConnect } +func (ConnectResponse) frameType() uint16 { return ConnectResponseTag } + +// Encode validates m and returns its frame. requestID must be nonzero. +func Encode(requestID uint64, m Message) (sandboxwire.Frame, error) { + if !sandboxwire.ValidRequestID(requestID) { + return sandboxwire.Frame{}, invalid("request ID %d", requestID) + } + if err := m.validate(); err != nil { + return sandboxwire.Frame{}, err + } + var e sandboxwire.Encoder + m.encode(&e) + return sandboxwire.Frame{Type: m.frameType(), RequestID: requestID, Payload: e.Payload()}, nil +} + +// Decode returns the message a frame carries. It rejects unknown tags, a zero +// request ID, invalid values and trailing bytes with errors wrapping +// sandboxwire.ErrMalformed. +func Decode(f sandboxwire.Frame) (Message, error) { + kind, err := tags.Classify(f.Type) + if err != nil { + return nil, err + } + if !sandboxwire.ValidRequestID(f.RequestID) { + return nil, invalid("request ID %d", f.RequestID) + } + r := &reader{d: sandboxwire.NewDecoder(f.Payload)} + var m Message + if kind == sandboxwire.KindRequest { + m = ConnectRequest{Network: Network(r.enum(func(v uint16) bool { return Network(v).Valid() })), + Host: string(r.bytes()), Port: r.u16(), TimeoutMillis: r.u32()} + } else { + resp := ConnectResponse{Result: Result(r.enum(func(v uint16) bool { return Result(v).Valid() }))} + if resp.Result == ResultFailed { + resp.Code = Code(r.enum(func(v uint16) bool { return Code(v).Valid() })) + resp.Effect = r.effect() + } + m = resp + } + if r.err == nil { + r.err = r.d.Finish() + } + if r.err != nil { + return nil, r.err + } + if err := m.validate(); err != nil { + return nil, err + } + return m, nil +} + +// WriteMessage encodes m and writes it as one frame. +func WriteMessage(w io.Writer, requestID uint64, m Message) error { + f, err := Encode(requestID, m) + if err != nil { + return err + } + return sandboxwire.WriteFrame(w, f) +} + +// ReadMessage reads and decodes one frame. The request ID is returned whenever +// a frame was read, even if it failed to decode. +func ReadMessage(r io.Reader) (uint64, Message, error) { + f, err := sandboxwire.ReadFrame(r, MaxMessageBytes) + if err != nil { + return 0, nil, err + } + m, err := Decode(f) + return f.RequestID, m, err +} + +func (c ConnectRequest) encode(e *sandboxwire.Encoder) { + e.Enum(uint16(c.Network)) + e.Bytes([]byte(c.Host)) + e.U16(c.Port) + e.U32(c.TimeoutMillis) +} + +func (r ConnectResponse) encode(e *sandboxwire.Encoder) { + e.Enum(uint16(r.Result)) + if r.Result == ResultFailed { + e.Enum(uint16(r.Code)) + e.Effect(r.Effect) + } +} + +func (c ConnectRequest) validate() error { + switch { + case !c.Network.Valid(): + return invalid("network %d", c.Network) + case c.Port == 0: + return invalid("port 0") + case c.TimeoutMillis == 0 || c.TimeoutMillis > MaxTimeoutMillis: + return invalid("timeout %d ms", c.TimeoutMillis) + } + return checkHost(c.Host) +} + +func (r ConnectResponse) validate() error { + switch { + case r.Result == ResultConnected && r.Code == 0 && r.Effect == 0: + return nil + case r.Result == ResultFailed && r.Code.Valid() && r.Effect.Valid(): + return nil + } + return invalid("response result %d code %d effect %d", r.Result, r.Code, r.Effect) +} + +// checkHost admits an unbracketed IP literal without a zone, or a DNS name: +// dot-separated labels of 1 to 63 ASCII letters, digits, hyphens and +// underscores, with no label starting or ending with a hyphen, an optional +// final dot, and a last label that is not all digits. Non-ASCII names arrive +// as IDNA A-labels. The host is 1 to MaxHostBytes bytes. +func checkHost(host string) error { + if len(host) == 0 || len(host) > MaxHostBytes { + return invalid("host of %d bytes", len(host)) + } + if a, err := netip.ParseAddr(host); err == nil { + if a.Zone() != "" { + return invalid("host %q has a zone", host) + } + return nil + } + name := host + if name[len(name)-1] == '.' { + name = name[:len(name)-1] + } + label, digits := 0, true + for i := 0; i <= len(name); i++ { + if i == len(name) || name[i] == '.' { + if label == 0 || label > 63 || name[i-1] == '-' { + return invalid("host %q", host) + } + label, digits = 0, true + continue + } + c := name[i] + switch { + case c >= '0' && c <= '9': + case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c == '_': + digits = false + case c == '-' && label > 0: + digits = false + default: + return invalid("host %q", host) + } + label++ + if i == len(name)-1 && digits { + return invalid("host %q ends in a numeric label", host) + } + } + return nil +} + +// permits reports whether egress admits a connection to addr: some rule's +// prefix contains the address and its port range contains the port. An +// IPv4-mapped IPv6 address is checked as IPv4. An address with a zone, and an +// unspecified address, which a TCP stack connects to this host, are never +// admitted. +func permits(egress []sandboxlink.EgressRule, addr netip.AddrPort) bool { + a := addr.Addr().Unmap() + if a.IsUnspecified() { + return false + } + for _, r := range egress { + if r.Prefix.Contains(a) && addr.Port() >= r.PortFirst && addr.Port() <= r.PortLast { + return true + } + } + return false +} + +// permitsPort reports whether some rule of egress admits port. +func permitsPort(egress []sandboxlink.EgressRule, port uint16) bool { + for _, r := range egress { + if port >= r.PortFirst && port <= r.PortLast { + return true + } + } + return false +} + +func invalid(format string, args ...any) error { + return fmt.Errorf("%w: "+format, append([]any{sandboxwire.ErrMalformed}, args...)...) +} + +// reader decodes fields in order, keeping the first error. +type reader struct { + d *sandboxwire.Decoder + err error +} + +func read[T any](r *reader, f func() (T, error)) T { + var v T + if r.err == nil { + v, r.err = f() + } + return v +} + +func (r *reader) u16() uint16 { return read(r, r.d.U16) } +func (r *reader) u32() uint32 { return read(r, r.d.U32) } +func (r *reader) bytes() []byte { return read(r, r.d.Bytes) } +func (r *reader) effect() sandboxwire.Effect { return read(r, r.d.Effect) } + +func (r *reader) enum(valid func(uint16) bool) uint16 { + return read(r, func() (uint16, error) { return r.d.Enum(valid) }) +} diff --git a/internal/sandboxnet/protocol_test.go b/internal/sandboxnet/protocol_test.go new file mode 100644 index 00000000..1ed814a6 --- /dev/null +++ b/internal/sandboxnet/protocol_test.go @@ -0,0 +1,188 @@ +package sandboxnet + +import ( + "bytes" + "encoding/hex" + "errors" + "io" + "net/netip" + "os" + "reflect" + "strings" + "testing" + + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +var goldenFrames = []struct { + requestID uint64 + m Message +}{ + {1, ConnectRequest{Network: NetworkTCP, Host: "example.com", Port: 443, TimeoutMillis: 10_000}}, + {1, ConnectResponse{Result: ResultConnected}}, + {1, ConnectResponse{Result: ResultFailed, Code: CodeDenied, Effect: sandboxwire.EffectNone}}, +} + +func readHexFixture(t testing.TB, name string) []byte { + t.Helper() + raw, err := os.ReadFile(name) + if err != nil { + t.Fatal(err) + } + var digits strings.Builder + for _, line := range strings.Split(string(raw), "\n") { + line, _, _ = strings.Cut(line, "#") + digits.WriteString(strings.Join(strings.Fields(line), "")) + } + b, err := hex.DecodeString(digits.String()) + if err != nil { + t.Fatal(err) + } + return b +} + +func TestGolden(t *testing.T) { + want := readHexFixture(t, "testdata/network_v1.hex") + var buf bytes.Buffer + for _, g := range goldenFrames { + if err := WriteMessage(&buf, g.requestID, g.m); err != nil { + t.Fatalf("encode %T: %v", g.m, err) + } + } + if !bytes.Equal(buf.Bytes(), want) { + t.Fatalf("encoded frames differ:\n got %x\nwant %x", buf.Bytes(), want) + } + r := bytes.NewReader(want) + for _, g := range goldenFrames { + id, m, err := ReadMessage(r) + if err != nil || id != g.requestID || !reflect.DeepEqual(m, g.m) { + t.Fatalf("decoded %d %#v %v, want %d %#v", id, m, err, g.requestID, g.m) + } + } + if r.Len() != 0 { + t.Fatalf("%d bytes left", r.Len()) + } +} + +func TestDecodeRejects(t *testing.T) { + frameOf := func(i int, edit func(p []byte) []byte) sandboxwire.Frame { + f, err := Encode(goldenFrames[i].requestID, goldenFrames[i].m) + if err != nil { + t.Fatal(err) + } + f.Payload = edit(bytes.Clone(f.Payload)) + return f + } + // Offsets into the Connect payload: Network at 0, host length at 2, host + // at 6, Port at 17 and TimeoutMillis at 19. + set := func(off int, v ...byte) func([]byte) []byte { + return func(p []byte) []byte { copy(p[off:], v); return p } + } + same := func(p []byte) []byte { return p } + cases := []struct { + name string + frame sandboxwire.Frame + }{ + {"trailing byte", frameOf(0, func(p []byte) []byte { return append(p, 0) })}, + {"unknown network", frameOf(0, set(0, 0, 2))}, + {"host past the payload", frameOf(0, set(2, 0, 0, 0, 0xff))}, + {"NUL in host", frameOf(0, set(6, 0))}, + {"zero port", frameOf(0, set(17, 0, 0))}, + {"zero timeout", frameOf(0, set(19, 0, 0, 0, 0))}, + {"timeout above the maximum", frameOf(0, set(19, 0, 0, 0xea, 0x61))}, + {"unknown result", frameOf(1, set(0, 0, 3))}, + {"connected with a code", frameOf(1, func(p []byte) []byte { return append(p, 0, 3, 0, 1) })}, + {"unknown code", frameOf(2, set(2, 0, 13))}, + {"unknown effect", frameOf(2, set(4, 0, 3))}, + {"zero request ID", func() sandboxwire.Frame { f := frameOf(0, same); f.RequestID = 0; return f }()}, + {"unknown tag", func() sandboxwire.Frame { f := frameOf(0, same); f.Type = 2; return f }()}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if _, err := Decode(c.frame); !errors.Is(err, sandboxwire.ErrMalformed) { + t.Fatalf("decode: %v, want ErrMalformed", err) + } + }) + } +} + +func TestHostGrammar(t *testing.T) { + label := strings.Repeat("a", 63) + longest := strings.Join([]string{label, label, label, label[:61]}, ".") + valid := []string{"example.com", "example.com.", "localhost", "xn--bcher-kva.example", "my_service.internal", "a-b.c1", + "10.0.0.1", "::1", "2001:db8::1", "::ffff:10.0.0.1", longest} + invalid := []string{"", ".", "a..b", "-a.com", "a-.com", "exa mple.com", "example.com\x00", "bücher.de", "http://example.com", + "example.com:443", "10.0.0.1:443", "[::1]", "fe80::1%eth0", "127.1", "1.2.3.4.5", "host.123", label + "a.com", longest + "a"} + for _, host := range valid { + if err := (ConnectRequest{Network: NetworkTCP, Host: host, Port: 1, TimeoutMillis: 1}).validate(); err != nil { + t.Errorf("host %q: %v, want valid", host, err) + } + } + for _, host := range invalid { + if err := (ConnectRequest{Network: NetworkTCP, Host: host, Port: 1, TimeoutMillis: 1}).validate(); err == nil { + t.Errorf("host %q is valid, want rejected", host) + } + } +} + +func TestEgress(t *testing.T) { + egress := []sandboxlink.EgressRule{ + {Prefix: netip.MustParsePrefix("10.0.0.0/8"), PortFirst: 443, PortLast: 443}, + {Prefix: netip.MustParsePrefix("2001:db8::/32"), PortFirst: 1, PortLast: 65535}, + {Prefix: netip.MustParsePrefix("::/0"), PortFirst: 22, PortLast: 22}, + {Prefix: netip.MustParsePrefix("0.0.0.0/8"), PortFirst: 443, PortLast: 443}, + } + for addr, want := range map[string]bool{ + "10.1.2.3:443": true, + "[::ffff:10.1.2.3]:443": true, + "[2001:db8::1]:22": true, + "10.1.2.3:80": false, + "11.0.0.1:443": false, + "[fe80::1%eth0]:443": false, + "[::1]:22": true, + "[::]:22": false, + "[::ffff:0.0.0.0]:443": false, + } { + if got := permits(egress, netip.MustParseAddrPort(addr)); got != want { + t.Errorf("permits %s = %v, want %v", addr, got, want) + } + } + if permits(nil, netip.MustParseAddrPort("10.1.2.3:443")) || permitsPort(nil, 443) { + t.Error("empty egress permits a connection") + } +} + +func FuzzDecode(f *testing.F) { + golden := readHexFixture(f, "testdata/network_v1.hex") + for r := bytes.NewReader(golden); r.Len() > 0; { + start := len(golden) - r.Len() + if _, err := sandboxwire.ReadFrame(r, MaxMessageBytes); err != nil { + f.Fatal(err) + } + f.Add(golden[start : len(golden)-r.Len()]) + } + f.Fuzz(func(t *testing.T, data []byte) { + fr, err := sandboxwire.ReadFrame(bytes.NewReader(data), MaxMessageBytes) + if err != nil { + if !errors.Is(err, sandboxwire.ErrMalformed) && err != io.EOF && err != io.ErrUnexpectedEOF { + t.Fatalf("frame error %v", err) + } + return + } + m, err := Decode(fr) + if err != nil { + if !errors.Is(err, sandboxwire.ErrMalformed) { + t.Fatalf("decode error %v", err) + } + return + } + var w bytes.Buffer + if err := WriteMessage(&w, fr.RequestID, m); err != nil { + t.Fatalf("re-encode %#v: %v", m, err) + } + if !bytes.HasPrefix(data, w.Bytes()) { + t.Fatalf("round trip changed the frame:\n got %x\nwant prefix of %x", w.Bytes(), data) + } + }) +} diff --git a/internal/sandboxnet/server.go b/internal/sandboxnet/server.go new file mode 100644 index 00000000..37aa0543 --- /dev/null +++ b/internal/sandboxnet/server.go @@ -0,0 +1,274 @@ +package sandboxnet + +import ( + "context" + "errors" + "fmt" + "io" + "net" + "net/netip" + "sync" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" +) + +// bufferSize is the copy buffer of each splice direction. +const bufferSize = 32 << 10 + +// Serve handles one Network stream for svc. It reads the stream's one Connect, +// checks the port, the host and every resolved address against egress, dials +// the first permitted address within the Connect's timeout and answers. After +// Connected it splices the stream and the connection: an end of either side's +// writing reaches the other as an orderly half-close after every byte before +// it, and any error, or the end of ctx, aborts both. No idle timeout applies. +// +// ctx is the attachment's. Serve owns s and returns when it is done with it: +// nil after both directions ended in order, ErrProtocolViolation when the +// attacher broke the protocol, the *Error it answered, or the error that +// aborted the stream. +func Serve(ctx context.Context, s sandboxlink.Stream, egress []sandboxlink.EgressRule, svc Service) error { + dialCtx, cancel := context.WithCancel(ctx) + defer cancel() + x := &splice{s: s, cancel: cancel, settled: make(chan struct{})} + stop := context.AfterFunc(ctx, func() { x.abort(ctx.Err()) }) + defer stop() + + f, err := sandboxwire.ReadFrame(s, MaxMessageBytes) + if err != nil { + s.Reset() + if errors.Is(err, sandboxwire.ErrMalformed) { + return fmt.Errorf("%w: %w", ErrProtocolViolation, err) + } + return err + } + var seq sandboxwire.RequestSequence + if kind, err := tags.Classify(f.Type); err != nil || kind != sandboxwire.KindRequest || !seq.Admit(f.RequestID) { + s.Reset() + return fmt.Errorf("%w: frame type %#04x request ID %d", ErrProtocolViolation, f.Type, f.RequestID) + } + m, err := Decode(f) + if err != nil { + failure := &Error{Code: CodeInvalidArgument, Effect: sandboxwire.EffectNone, Cause: err} + answerFailure(s, f.RequestID, failure) + return failure + } + req := m.(ConnectRequest) + + var wg sync.WaitGroup + wg.Go(x.pumpIn) + defer wg.Wait() + dialCtx, cancelTimeout := context.WithTimeout(dialCtx, time.Duration(req.TimeoutMillis)*time.Millisecond) + defer cancelTimeout() + conn, failure := connect(dialCtx, req, egress, svc) + if failure != nil { + if x.advance(dialing, answered, nil) { + answerFailure(s, f.RequestID, failure) + return failure + } + return x.result() + } + if !x.advance(dialing, connecting, conn) { + conn.SetLinger(0) + conn.Close() + return x.result() + } + if err := WriteMessage(s, f.RequestID, ConnectResponse{Result: ResultConnected}); err != nil { + x.abort(err) + return x.result() + } + if !x.advance(connecting, connected, nil) { + return x.result() + } + _, err = io.CopyBuffer(struct{ io.Writer }{s}, struct{ io.Reader }{conn}, make([]byte, bufferSize)) + if err == nil { + err = s.CloseWrite() + } + if err != nil { + x.abort(err) + } + wg.Wait() + if err := x.result(); err != nil { + return err + } + conn.Close() + return s.Close() +} + +// answerFailure writes a failure answer and ends the stream in order, so the +// attacher reads the answer before the end. +func answerFailure(s sandboxlink.Stream, requestID uint64, e *Error) { + if WriteMessage(s, requestID, ConnectResponse{Result: ResultFailed, Code: e.Code, Effect: e.Effect}) != nil { + s.Reset() + return + } + s.Close() +} + +// connect resolves and checks the destination and dials it. Every failure is +// an *Error to answer. +func connect(dialCtx context.Context, req ConnectRequest, egress []sandboxlink.EgressRule, svc Service) (*net.TCPConn, *Error) { + if !permitsPort(egress, req.Port) { + return nil, &Error{Code: CodeDenied, Effect: sandboxwire.EffectNone} + } + var addrs []netip.Addr + if a, err := netip.ParseAddr(req.Host); err == nil { + addrs = []netip.Addr{a} + } else if addrs, err = svc.Resolve(dialCtx, req.Host); err != nil { + return nil, outcome(dialCtx, err, CodeNameResolutionFailed, sandboxwire.EffectNone) + } else if len(addrs) == 0 { + return nil, &Error{Code: CodeNameNotResolved, Effect: sandboxwire.EffectNone} + } + for _, a := range addrs { + target := netip.AddrPortFrom(a.Unmap(), req.Port) + if !permits(egress, target) { + continue + } + conn, err := svc.Dial(dialCtx, target) + if err != nil { + return nil, outcome(dialCtx, err, CodeIO, sandboxwire.EffectPossible) + } + return conn, nil + } + return nil, &Error{Code: CodeDenied, Effect: sandboxwire.EffectNone} +} + +// outcome types a Service error: a valid *Error stands; otherwise a passed +// deadline or the end of dialCtx decides, and any other error becomes code. +// The deadline is read from the clock, because a socket deadline set from it +// can fire before dialCtx reports its end. +func outcome(dialCtx context.Context, err error, code Code, effect sandboxwire.Effect) *Error { + var e *Error + deadline, ok := dialCtx.Deadline() + switch { + case errors.As(err, &e) && e.Code.Valid() && e.Effect.Valid(): + return e + case ok && !time.Now().Before(deadline): + return &Error{Code: CodeTimedOut, Effect: effect, Cause: err} + case dialCtx.Err() != nil: + return &Error{Code: CodeCancelled, Effect: effect, Cause: err} + } + return &Error{Code: code, Effect: effect, Cause: err} +} + +// phase is where a splice stands. It moves from dialing to answered, or +// through connecting to connected, and to aborted from any phase. +type phase uint8 + +const ( + dialing phase = iota + connecting // the dial succeeded and Connected is being written + connected // Connected was written + answered // a failure is being or was answered + aborted +) + +// splice holds the state the two directions share. pumpIn reads the stream +// from the start, so it sees bytes or a second Connect that arrive before +// Connected. +type splice struct { + s sandboxlink.Stream + cancel context.CancelFunc // cancels the dial + settled chan struct{} // closed when the phase becomes connected, answered or aborted + mu sync.Mutex + phase phase + conn *net.TCPConn + err error +} + +// advance moves the splice from one phase to the next, recording conn when it +// is not nil. It fails when the splice is no longer in from, which only an +// abort causes. +func (x *splice) advance(from, to phase, conn *net.TCPConn) bool { + x.mu.Lock() + defer x.mu.Unlock() + if x.phase != from { + return false + } + x.phase = to + if conn != nil { + x.conn = conn + } + if to != connecting { + close(x.settled) + } + return true +} + +func (x *splice) state() (phase, *net.TCPConn) { + x.mu.Lock() + defer x.mu.Unlock() + return x.phase, x.conn +} + +func (x *splice) result() error { + x.mu.Lock() + defer x.mu.Unlock() + return x.err +} + +// abort resets the stream and the connection, and cancels a dial in progress. +// The first cause is kept. +func (x *splice) abort(cause error) { + x.mu.Lock() + if x.phase == aborted { + x.mu.Unlock() + return + } + if x.phase == dialing || x.phase == connecting { + close(x.settled) + } + x.phase, x.err = aborted, cause + conn := x.conn + x.mu.Unlock() + x.cancel() + x.s.Reset() + if conn != nil { + conn.SetLinger(0) + conn.Close() + } +} + +// pumpIn copies the stream to the connection. Bytes while dialing break the +// protocol; bytes after a successful dial wait until Connected is written, so +// nothing reaches the destination before the answer is out. The stream's +// orderly end becomes the connection's CloseWrite once Connected is written. +func (x *splice) pumpIn() { + buf := make([]byte, bufferSize) + for { + n, err := x.s.Read(buf) + if n > 0 { + if p, _ := x.state(); p == dialing { + x.abort(fmt.Errorf("%w: bytes before the dial succeeded", ErrProtocolViolation)) + return + } + <-x.settled + p, conn := x.state() + if p != connected { + return + } + if _, werr := conn.Write(buf[:n]); werr != nil { + x.abort(werr) + return + } + } + switch { + case err == io.EOF: + <-x.settled + if p, conn := x.state(); p == connected { + if err := conn.CloseWrite(); err != nil { + x.abort(err) + } + } + return + case err != nil: + // After a failure answer the stream is closed and reading it fails; + // resetting it then would discard the answer. + if p, _ := x.state(); p != answered { + x.abort(err) + } + return + } + } +} diff --git a/internal/sandboxnet/server_test.go b/internal/sandboxnet/server_test.go new file mode 100644 index 00000000..714ad35a --- /dev/null +++ b/internal/sandboxnet/server_test.go @@ -0,0 +1,99 @@ +package sandboxnet + +import ( + "context" + "errors" + "io" + "net" + "net/netip" + "sync" + "testing" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" +) + +// heldStream is a Stream whose writes wait until release is closed, so a test +// can hold Serve's answer on its way out. +type heldStream struct { + in *io.PipeReader + writing chan struct{} // closed when the first Write starts + release chan struct{} + once sync.Once +} + +func (h *heldStream) Read(b []byte) (int, error) { return h.in.Read(b) } +func (h *heldStream) Write(b []byte) (int, error) { + h.once.Do(func() { close(h.writing) }) + <-h.release + return len(b), nil +} +func (h *heldStream) CloseWrite() error { return nil } +func (h *heldStream) Close() error { return h.in.Close() } +func (h *heldStream) Reset() error { return h.in.CloseWithError(errors.New("reset")) } +func (h *heldStream) SetDeadline(time.Time) error { return nil } +func (h *heldStream) SetReadDeadline(time.Time) error { return nil } +func (h *heldStream) SetWriteDeadline(time.Time) error { return nil } + +// loopback dials IP literals and resolves nothing. +type loopback struct{} + +func (loopback) Resolve(context.Context, string) ([]netip.Addr, error) { + return nil, errors.New("no names") +} + +func (loopback) Dial(ctx context.Context, addr netip.AddrPort) (*net.TCPConn, error) { + var d net.Dialer + c, err := d.DialContext(ctx, "tcp4", addr.String()) + if err != nil { + return nil, err + } + return c.(*net.TCPConn), nil +} + +// Bytes that arrive while Connected is being written reach the destination +// only after the answer is out. +func TestForwardsOnlyAfterTheAnswer(t *testing.T) { + ln, err := net.Listen("tcp4", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + dst := ln.Addr().(*net.TCPAddr).AddrPort() + in, feed := io.Pipe() + s := &heldStream{in: in, writing: make(chan struct{}), release: make(chan struct{})} + served := make(chan error, 1) + egress := []sandboxlink.EgressRule{{Prefix: netip.PrefixFrom(dst.Addr(), 32), PortFirst: dst.Port(), PortLast: dst.Port()}} + go func() { served <- Serve(context.Background(), s, egress, loopback{}) }() + go WriteMessage(feed, 1, ConnectRequest{Network: NetworkTCP, Host: dst.Addr().String(), Port: dst.Port(), TimeoutMillis: 5000}) + + c, err := ln.Accept() + if err != nil { + t.Fatal(err) + } + defer c.Close() + select { + case <-s.writing: + case <-time.After(5 * time.Second): + t.Fatal("Connected was never written") + } + // The pipe returns once Serve has read the bytes. + if _, err := feed.Write([]byte("early")); err != nil { + t.Fatal(err) + } + c.SetReadDeadline(time.Now().Add(50 * time.Millisecond)) + if n, _ := c.Read(make([]byte, 8)); n != 0 { + t.Fatal("bytes reached the destination before Connected was written") + } + close(s.release) + c.SetReadDeadline(time.Now().Add(5 * time.Second)) + got := make([]byte, 5) + if _, err := io.ReadFull(c, got); err != nil || string(got) != "early" { + t.Fatalf("destination read %q, %v", got, err) + } + feed.Close() + c.Close() + if err := <-served; err != nil { + t.Fatalf("Serve: %v", err) + } +} diff --git a/internal/sandboxnet/testdata/network_v1.hex b/internal/sandboxnet/testdata/network_v1.hex new file mode 100644 index 00000000..93e6f3f2 --- /dev/null +++ b/internal/sandboxnet/testdata/network_v1.hex @@ -0,0 +1,18 @@ +# Network version 1 frames from goldenFrames in protocol_test.go, in order. Hex bytes; text after # is a comment. + +# ConnectRequest +00000017 0001 0000 0000000000000001 # header: length 23, OpConnect, flags, request 1 +0001 # Network NetworkTCP +0000000b 6578616d706c652e636f6d # Host "example.com" +01bb # Port 443 +00002710 # TimeoutMillis 10000 + +# ConnectResponse, connected +00000002 8001 0000 0000000000000001 # header: length 2, ConnectResponseTag, flags, request 1 +0001 # Result ResultConnected + +# ConnectResponse, failed +00000006 8001 0000 0000000000000001 # header: length 6, ConnectResponseTag, flags, request 1 +0002 # Result ResultFailed +0003 # Code CodeDenied +0001 # Effect EffectNone