diff --git a/apps/daemon/internal/agenthost/view_linux_test.go b/apps/daemon/internal/agenthost/view_linux_test.go index 04a88b1b0..306e14c5f 100644 --- a/apps/daemon/internal/agenthost/view_linux_test.go +++ b/apps/daemon/internal/agenthost/view_linux_test.go @@ -363,10 +363,7 @@ func startSandbox(t *testing.T, bin string) *sandbox { 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) - } + rl := relay.New(auth) srv := httptest.NewServer(rl) t.Cleanup(srv.Close) t.Cleanup(func() { rl.Close() }) diff --git a/apps/daemon/internal/agenthostqualify/qualify_linux_test.go b/apps/daemon/internal/agenthostqualify/qualify_linux_test.go index 3a6906d0c..7dbc297d4 100644 --- a/apps/daemon/internal/agenthostqualify/qualify_linux_test.go +++ b/apps/daemon/internal/agenthostqualify/qualify_linux_test.go @@ -227,10 +227,7 @@ type sandbox struct { func startSandbox(t *testing.T) *sandbox { auth := sandboxlinktest.NewAuthority() - rl, err := relay.New(relay.Config{Authority: auth}) - if err != nil { - t.Fatal(err) - } + rl := relay.New(auth) srv := httptest.NewServer(rl) t.Cleanup(srv.Close) t.Cleanup(func() { rl.Close() }) diff --git a/apps/daemon/internal/gateway/gateway_test.go b/apps/daemon/internal/gateway/gateway_test.go index 4372f3880..2925e0d20 100644 --- a/apps/daemon/internal/gateway/gateway_test.go +++ b/apps/daemon/internal/gateway/gateway_test.go @@ -20,7 +20,6 @@ import ( "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" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxnet" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxwire" @@ -105,7 +104,7 @@ func (s *sandbox) Dial(ctx context.Context, addr netip.AddrPort) (*net.TCPConn, func startSandbox(t *testing.T) *sandbox { t.Helper() auth := sandboxlinktest.NewAuthority() - srv := sandboxlinktest.StartRelay(t, relay.Config{Authority: auth}) + srv := sandboxlinktest.StartRelay(t, auth) resource := sandboxlink.ResourceRef{TenantID: sandboxwire.NewID(), EnvironmentID: sandboxwire.NewID(), Kind: sandboxlink.ResourceAllocation, ID: sandboxwire.NewID(), Generation: 1} auth.AddServe([]byte("serve credential"), sandboxlink.ServePeer{PeerID: sandboxwire.NewID(), Resource: resource}) @@ -122,7 +121,7 @@ func startSandbox(t *testing.T) *sandbox { Serve: func(ctx context.Context, b sandboxlink.Bind, _ uint64, s sandboxlink.Stream) { sandboxnet.Serve(ctx, s, b.Egress, sb) }}}, - OnConnected: func(sandboxlink.HelloAccepted) { + OnConnected: func() { select { case connected <- struct{}{}: default: diff --git a/apps/sandboxio/internal/netservice/service_test.go b/apps/sandboxio/internal/netservice/service_test.go index 1925315cb..a59d85dd6 100644 --- a/apps/sandboxio/internal/netservice/service_test.go +++ b/apps/sandboxio/internal/netservice/service_test.go @@ -19,7 +19,6 @@ import ( "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" @@ -61,7 +60,7 @@ type fixture struct { 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(), + f := &fixture{t: t, auth: auth, srv: sandboxlinktest.StartRelay(t, 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) { @@ -81,7 +80,7 @@ func newFixture(t *testing.T) *fixture { Serve: func(ctx context.Context, b sandboxlink.Bind, _ uint64, s sandboxlink.Stream) { f.served <- sandboxnet.Serve(ctx, s, b.Egress, svc) }}}, - OnConnected: func(sandboxlink.HelloAccepted) { + OnConnected: func() { select { case connected <- struct{}{}: default: diff --git a/apps/sandboxio/internal/sandboxio/sandboxio.go b/apps/sandboxio/internal/sandboxio/sandboxio.go index 50ce77242..de6122c96 100644 --- a/apps/sandboxio/internal/sandboxio/sandboxio.go +++ b/apps/sandboxio/internal/sandboxio/sandboxio.go @@ -89,7 +89,7 @@ func run(ctx context.Context, bootstrapPath string, opt options) error { }}, {Service: sandboxlink.ServiceNetwork, Version: sandboxnet.Version, Serve: netservice.New().Handle}, }, - OnConnected: func(sandboxlink.HelloAccepted) { down.Store(false) }, + OnConnected: func() { down.Store(false) }, OnDisconnected: func(err error) { if !down.Swap(true) { log.Printf("oac-sandbox-io: relay link ended, reconnecting: %v", err) diff --git a/apps/sandboxio/internal/sandboxio/sandboxio_test.go b/apps/sandboxio/internal/sandboxio/sandboxio_test.go index 4afafd42d..d827dcb41 100644 --- a/apps/sandboxio/internal/sandboxio/sandboxio_test.go +++ b/apps/sandboxio/internal/sandboxio/sandboxio_test.go @@ -23,7 +23,6 @@ import ( "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/sandboxnet" sp "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxprocess" @@ -110,7 +109,7 @@ func next(t *testing.T, op *sp.Operation) sp.Event { func TestServesEachProtocolThroughTheRelay(t *testing.T) { ctx := context.Background() auth := sandboxlinktest.NewAuthority() - srv := sandboxlinktest.StartRelay(t, relay.Config{Authority: auth}) + srv := sandboxlinktest.StartRelay(t, auth) bootstrap, resource := writeBootstrap(t, srv.URL, "serve-credential") auth.AddServe([]byte("serve-credential"), sandboxlink.ServePeer{PeerID: sandboxwire.NewID(), Resource: resource}) runtimeID := sandboxwire.NewID() @@ -254,7 +253,7 @@ func TestServesEachProtocolThroughTheRelay(t *testing.T) { // A refused serve credential ends the service with the relay's typed failure, // and the message never carries the credential. func TestRefusedCredentialEndsTheService(t *testing.T) { - srv := sandboxlinktest.StartRelay(t, relay.Config{Authority: sandboxlinktest.NewAuthority()}) + srv := sandboxlinktest.StartRelay(t, sandboxlinktest.NewAuthority()) bootstrap, _ := writeBootstrap(t, srv.URL, "unknown-credential") ctx, cancel := context.WithTimeout(context.Background(), wait) defer cancel() diff --git a/docs/sandbox-link-protocol.md b/docs/sandbox-link-protocol.md index 097e0f825..5ed25afb4 100644 --- a/docs/sandbox-link-protocol.md +++ b/docs/sandbox-link-protocol.md @@ -41,7 +41,7 @@ An attachment outlives its link. After reconnecting, the Runtime opens a stream ## Run a relay -`relay.New` takes a `relay.Config` with an `Authority` and returns a `*relay.Relay`, which is an `http.Handler`. The relay endpoint is served behind the installation's HTTPS ingress, which terminates TLS, so the handler accepts the upgrade on the ingress's plain HTTP hop; peers enforce TLS when they dial. `MaxStreams` (default 256) bounds each link's concurrent service streams and `MaxFrameBytes` (default 1 MiB) is the frame limit the relay advertises. +`relay.New` takes an `Authority` and returns a `*relay.Relay`, which is an `http.Handler`. The relay endpoint is served behind the installation's HTTPS ingress, which terminates TLS, so the handler accepts the upgrade on the ingress's plain HTTP hop; peers enforce TLS when they dial. Each link carries at most 256 concurrent service streams. The owner of the relay implements `Authority` from its durable records, and the relay consults it for every Hello, Open and renewal. To revoke, withdraw the authority first, then call `RevokeAttachment` or `RevokeResource` so the relay closes what it holds. @@ -55,7 +55,7 @@ A frame is a 16-byte header followed by the payload. Integers are big-endian. | Offset | Field | Type | Rule | | --- | --- | --- | --- | -| 0 | `PayloadLength` | uint32 | At most 1 MiB (`sandboxwire.MaxPayload`) and at most the limit advertised to the sender; checked before the payload is read | +| 0 | `PayloadLength` | uint32 | At most 1 MiB (`sandboxwire.MaxPayload`); checked before the payload is read | | 4 | `MessageType` | uint16 | A tag the protocol defines | | 6 | `Flags` | uint16 | Zero | | 8 | `RequestID` | uint64 | For a request, nonzero and greater than the sender's previous request ID on the stream; the request's ID in its response; zero for an event | @@ -124,10 +124,7 @@ Hello RuntimeID ID Credential bytes // 1..4096 bytes -HelloAccepted - LinkID ID - MaxStreams u32 // at least 1 - MaxFrameBytes u32 // 16 KiB..1 MiB +HelloAccepted (no fields) Open Service enum @@ -144,7 +141,6 @@ Opened AttachmentID ID ServerInstanceID ID LeaseExpiresAt i64 ms - MaxFrameBytes u32 Bind AttachmentID ID @@ -155,7 +151,6 @@ Bind AssignmentEpoch u64 LeaseExpiresAt i64 ms ExpectedServerInstanceID ID // the serve peer's ServerInstanceID as the relay knows it - MaxFrameBytes u32 Exports optional, present exactly when Service is ServiceFile: count 1..64 of ExportGrant, no ID twice Egress optional, present exactly when Service is ServiceNetwork: @@ -203,8 +198,6 @@ AttachmentClosed Later control requests continue the Hello's request IDs. The relay ends an attach link whose request ID does not increase with `ProtocolViolation`. `Open` and `Bind` are each the only request on their stream and use request ID 1. -`HelloAccepted.MaxStreams` bounds the link's concurrent service streams. `MaxFrameBytes` bounds the payload of every frame on the link's service streams. - For a serve peer, the Authority returns the peer's identity and the resource, including generation, that the credential serves. The resource must equal the Hello's, otherwise the answer is `PermissionDenied`. The relay then applies the [generation rule](#authority-and-staleness) and makes the link the resource's current serve peer. The relay and the serve peer bound each handshake step, the WebSocket upgrade, the Hello, reading an `Open`, a `Bind` and its answer and each Authority call, by `sandboxlink.HandshakeTimeout` (10 seconds). The attach peer bounds an Open with its context. @@ -281,6 +274,8 @@ The relay copies each direction through a 32 KiB buffer and holds at most one 25 `ServiceUnavailable` and `LimitExceeded` are transient: the same request may succeed later, and `Code.Retryable` reports them. Every other code is final: repeating the request with the same credential, attachment and generation fails again. +An answer that is malformed, or that carries another request ID or operation than its request, fails the request with `ProtocolViolation` and `EffectPossible`, since the request may have taken effect. + ## Verification `go test ./internal/sandboxlink/...` covers the golden frames, decode rejection, and the relay's authorization, generation, lease, revocation, renewal bound and reconnect behavior, including orderly end and abort propagation. `go test -run '^$' -fuzz FuzzDecode ./internal/sandboxlink` fuzzes the decoder. diff --git a/docs/zh/sandbox-link-protocol.md b/docs/zh/sandbox-link-protocol.md index 1395a1f5d..af83e8973 100644 --- a/docs/zh/sandbox-link-protocol.md +++ b/docs/zh/sandbox-link-protocol.md @@ -1,7 +1,7 @@ --- title: "沙箱 Link 协议" source: docs/sandbox-link-protocol.md -source_hash: 0d6793b12de0dcb11c3355855c514538f78653fa90008f9865407d1792743859 +source_hash: d753586c244650329796b8abd17994c2e57aa433b55ebd78c499c86368e8aa6f --- Link 协议通过 relay 连接沙箱 I/O 的两端。Sandbox I/O 服务运行在沙箱内并为其提供服务,是 serve peer。agent host 上的 Runtime 在沙箱外运行 Harness,并通过该服务使用沙箱,是 attach peer。每个 peer 各自向 relay 认证自己的 link。relay 授权 attach peer 打开的每个服务 stream,将其绑定到该资源当前的 serve peer,然后在两个 stream 之间复制字节而不读取内容。服务帧从不携带凭据或 grant。 @@ -43,7 +43,7 @@ attachment 的生命周期长于其 link。重连后,Runtime 使用相同的 b ## 运行 relay {#run-a-relay} -`relay.New` 接收带 `Authority` 的 `relay.Config`,返回 `*relay.Relay`,它是一个 `http.Handler`。relay endpoint 位于安装实例的 HTTPS ingress 之后,由 ingress 终止 TLS,因此 handler 在 ingress 的明文 HTTP 一跳上接受 upgrade;peer 在拨号时强制 TLS。`MaxStreams`(默认 256)限制每条 link 的并发服务 stream,`MaxFrameBytes`(默认 1 MiB)是 relay 通告的帧上限。 +`relay.New` 接收 `Authority`,返回 `*relay.Relay`,它是一个 `http.Handler`。relay endpoint 位于安装实例的 HTTPS ingress 之后,由 ingress 终止 TLS,因此 handler 在 ingress 的明文 HTTP 一跳上接受 upgrade;peer 在拨号时强制 TLS。每条 link 最多承载 256 个并发服务 stream。 relay 的 owner 基于其持久记录实现 `Authority`,relay 对每个 Hello、Open 和续期都咨询它。撤销时,先撤回授权,再调用 `RevokeAttachment` 或 `RevokeResource`,让 relay 关闭其持有的对象。 @@ -57,7 +57,7 @@ relay 的 owner 基于其持久记录实现 `Authority`,relay 对每个 Hello | 偏移 | 字段 | 类型 | 规则 | | --- | --- | --- | --- | -| 0 | `PayloadLength` | uint32 | 不超过 1 MiB(`sandboxwire.MaxPayload`),且不超过向发送方通告的上限;在读取 payload 前检查 | +| 0 | `PayloadLength` | uint32 | 不超过 1 MiB(`sandboxwire.MaxPayload`);在读取 payload 前检查 | | 4 | `MessageType` | uint16 | 协议定义的 tag | | 6 | `Flags` | uint16 | 零 | | 8 | `RequestID` | uint64 | 请求中为非零值,且大于发送方在该 stream 上的上一个请求 ID;响应中为对应请求的 ID;事件中为零 | @@ -126,10 +126,7 @@ Hello RuntimeID ID Credential bytes // 1..4096 bytes -HelloAccepted - LinkID ID - MaxStreams u32 // at least 1 - MaxFrameBytes u32 // 16 KiB..1 MiB +HelloAccepted (no fields) Open Service enum @@ -146,7 +143,6 @@ Opened AttachmentID ID ServerInstanceID ID LeaseExpiresAt i64 ms - MaxFrameBytes u32 Bind AttachmentID ID @@ -157,7 +153,6 @@ Bind AssignmentEpoch u64 LeaseExpiresAt i64 ms ExpectedServerInstanceID ID // the serve peer's ServerInstanceID as the relay knows it - MaxFrameBytes u32 Exports optional, present exactly when Service is ServiceFile: count 1..64 of ExportGrant, no ID twice Egress optional, present exactly when Service is ServiceNetwork: @@ -205,8 +200,6 @@ AttachmentClosed 后续控制请求延续 Hello 的请求 ID。attach link 的请求 ID 未递增时,relay 以 `ProtocolViolation` 结束该 link。`Open` 和 `Bind` 各自是其 stream 上唯一的请求,使用请求 ID 1。 -`HelloAccepted.MaxStreams` 限制该 link 的并发服务 stream。`MaxFrameBytes` 限制该 link 服务 stream 上每个帧的 payload。 - 对于 serve peer,Authority 返回 peer 的身份及该凭据所服务的资源(包括 generation)。该资源必须与 Hello 中的一致,否则回复 `PermissionDenied`。随后 relay 应用 [generation 规则](#authority-and-staleness),并将该 link 设为资源当前的 serve peer。 relay 和 serve peer 以 `sandboxlink.HandshakeTimeout`(10 秒)限制每个握手步骤:WebSocket upgrade、Hello、读取 `Open`、`Bind` 及其回复,以及每次 Authority 调用。attach peer 用其 context 限制 Open。 @@ -283,6 +276,8 @@ relay 通过 32 KiB 缓冲区复制每个方向的数据,每个 stream 最多 `ServiceUnavailable` 和 `LimitExceeded` 是临时失败:相同请求稍后可能成功,`Code.Retryable` 将它们报告为可重试。其他 code 都是最终失败:以相同凭据、attachment 和 generation 重复请求会再次失败。 +格式错误的响应,或其请求 ID、操作与请求不符的响应,会使该请求以 `ProtocolViolation` 和 `EffectPossible` 失败,因为请求可能已经生效。 + ## 验证 {#verification} `go test ./internal/sandboxlink/...` 覆盖 golden 帧、解码拒绝,以及 relay 的授权、generation、lease、撤销、续期上限和重连行为,包括有序结束与中止的传播。`go test -run '^$' -fuzz FuzzDecode ./internal/sandboxlink` 对解码器进行 fuzz 测试。 diff --git a/internal/sandboxlink/attach.go b/internal/sandboxlink/attach.go index 44bbbf430..072274767 100644 --- a/internal/sandboxlink/attach.go +++ b/internal/sandboxlink/attach.go @@ -27,7 +27,6 @@ type AttachConfig struct { type AttachLink struct { sess *yamux.Session ctl *yamux.Stream - accepted HelloAccepted onClosed func(AttachmentClosed) done chan struct{} @@ -38,7 +37,9 @@ type AttachLink struct { mu sync.Mutex pending map[uint64]chan Message - err error + // lost is the failure of the requests pending when the link ended; it is + // nil while the link is up. + lost error } // ErrLinkClosed is returned for requests on a link that has ended. @@ -52,18 +53,15 @@ func DialAttach(ctx context.Context, cfg AttachConfig) (*AttachLink, error) { } l := &AttachLink{onClosed: cfg.OnAttachmentClosed, done: make(chan struct{}), write: make(chan struct{}, 1), pending: map[uint64]chan Message{}} - sess, ctl, accepted, err := connect(ctx, dialerFor(cfg.Dial, cfg.URL, cfg.TLS), hello, &l.seq) + sess, ctl, err := connect(ctx, dialerFor(cfg.Dial, cfg.URL, cfg.TLS), hello, &l.seq) if err != nil { return nil, err } - l.sess, l.ctl, l.accepted = sess, ctl, accepted + l.sess, l.ctl = sess, ctl go l.read() return l, nil } -// Accepted returns the relay's HelloAccepted. -func (l *AttachLink) Accepted() HelloAccepted { return l.accepted } - // Done is closed when the link ends. func (l *AttachLink) Done() <-chan struct{} { return l.done } @@ -89,22 +87,18 @@ func (l *AttachLink) OpenService(ctx context.Context, o Open) (Stream, Opened, e stop := context.AfterFunc(ctx, func() { st.Reset() }) opened, err := func() (Opened, error) { var seq sandboxwire.RequestSequence - if err := WriteMessage(st, seq.Next(), o); err != nil { + id := seq.Next() + if err := WriteMessage(st, id, o); err != nil { return Opened{}, Uncertain(err) } - _, m, err := ReadMessage(st, MaxMessageBytes) + m, err := ReadReply(st, OpOpen, id) if err != nil { - return Opened{}, Uncertain(err) + return Opened{}, err } - switch r := m.(type) { - case Opened: - if r.AttachmentID == o.AttachmentID { - return r, nil - } - case Failure: - return Opened{}, r.Err() + if r := m.(Opened); r.AttachmentID == o.AttachmentID { + return r, nil } - return Opened{}, errPossibleViolation + return Opened{}, violation(nil) }() if !stop() { err = Uncertain(ctx.Err()) @@ -122,29 +116,19 @@ func (l *AttachLink) Renew(ctx context.Context, r RenewAttachment) (AttachmentRe if err != nil { return AttachmentRenewed{}, err } - renewed, ok := m.(AttachmentRenewed) - if !ok || renewed.AttachmentID != r.AttachmentID { - return AttachmentRenewed{}, errPossibleViolation + if renewed := m.(AttachmentRenewed); renewed.AttachmentID == r.AttachmentID { + return renewed, nil } - return renewed, nil + return AttachmentRenewed{}, violation(nil) } // CloseAttachment ends an attachment and all its streams. Closing an unknown // attachment succeeds. func (l *AttachLink) CloseAttachment(ctx context.Context, id sandboxwire.ID) error { - m, err := l.call(ctx, CloseAttachment{AttachmentID: id}) - if err != nil { - return err - } - if _, ok := m.(CloseAccepted); !ok { - return errPossibleViolation - } - return nil + _, err := l.call(ctx, CloseAttachment{AttachmentID: id}) + return err } -// errPossibleViolation answers a request whose response did not fit it. -var errPossibleViolation = &Error{Code: ProtocolViolation, Effect: sandboxwire.EffectPossible} - // Write states of a control request. Whichever of the writer and a // cancellation leaves writing first decides the request's fate. const ( @@ -162,8 +146,9 @@ func notSent(err error) *Error { // its *Error. A ctx that ends before the request is sent returns an *Error // with EffectNone and leaves the link up; the link's end returns // ErrLinkClosed. Once its frame began to be sent, a failure returns an *Error -// with EffectPossible; a write that ctx interrupts or that fails ends the -// link, because the control stream may hold a partial frame. +// with EffectPossible, a ProtocolViolation when the relay broke the protocol; +// a write that ctx interrupts or that fails ends the link, because the control +// stream may hold a partial frame. func (l *AttachLink) call(ctx context.Context, req Message) (Message, error) { if _, err := Encode(1, req); err != nil { return nil, err @@ -181,10 +166,10 @@ func (l *AttachLink) call(ctx context.Context, req Message) (Message, error) { return nil, notSent(err) } l.mu.Lock() - if l.err != nil { + if l.lost != nil { l.mu.Unlock() <-l.write - return nil, l.err + return nil, ErrLinkClosed } id := l.seq.Next() ch := make(chan Message, 1) @@ -220,15 +205,9 @@ func (l *AttachLink) call(ctx context.Context, req Message) (Message, error) { select { case m, ok := <-ch: if !ok { - return nil, Uncertain(ErrLinkClosed) - } - if f, failed := m.(Failure); failed { - return nil, f.Err() - } - if m.frameType() != sandboxwire.ResponseType(req.frameType()) { - return nil, errPossibleViolation + return nil, l.lost // set before ch was closed } - return m, nil + return reply(Op(req.frameType()), m) case <-ctx.Done(): return nil, Uncertain(ctx.Err()) } @@ -237,10 +216,11 @@ func (l *AttachLink) call(ctx context.Context, req Message) (Message, error) { // read dispatches control messages until the link ends. A request from the // relay or an unreadable frame ends the link. func (l *AttachLink) read() { + var lost error defer func() { l.sess.Close() l.mu.Lock() - l.err = ErrLinkClosed + l.lost = lost for id, ch := range l.pending { close(ch) delete(l.pending, id) @@ -249,8 +229,9 @@ func (l *AttachLink) read() { close(l.done) }() for { - id, m, err := ReadMessage(l.ctl, MaxMessageBytes) + id, m, err := ReadMessage(l.ctl) if err != nil { + lost = readError(err) return } switch r := m.(type) { @@ -260,6 +241,7 @@ func (l *AttachLink) read() { } default: if !sandboxwire.IsResponse(m.frameType()) { + lost = violation(nil) return } l.mu.Lock() diff --git a/internal/sandboxlink/protocol.go b/internal/sandboxlink/protocol.go index a9f4b57db..36dbfa32e 100644 --- a/internal/sandboxlink/protocol.go +++ b/internal/sandboxlink/protocol.go @@ -29,8 +29,7 @@ import ( const Version uint16 = 1 const ( - // MaxMessageBytes bounds every Link message payload. A relay never - // advertises a MaxFrameBytes below it. + // MaxMessageBytes bounds every Link message payload. MaxMessageBytes = 16 << 10 // MaxCredentialBytes bounds a serve or Runtime credential. MaxCredentialBytes = 4 << 10 @@ -316,13 +315,8 @@ type AttachHello struct { Credential []byte } -// HelloAccepted answers either Hello. MaxStreams bounds the link's concurrent -// service streams and MaxFrameBytes the payload of any frame on the link. -type HelloAccepted struct { - LinkID sandboxwire.ID - MaxStreams uint32 - MaxFrameBytes uint32 -} +// HelloAccepted answers either Hello. +type HelloAccepted struct{} // Open is the first message on a service stream the attach peer opens. A zero // ExpectedServerInstanceID means no expectation. @@ -357,7 +351,6 @@ type Opened struct { AttachmentID sandboxwire.ID ServerInstanceID sandboxwire.ID LeaseExpiresAt time.Time - MaxFrameBytes uint32 } // Bind is the first message on a stream the relay opens to the serve peer. It @@ -374,7 +367,6 @@ type Bind struct { AssignmentEpoch uint64 LeaseExpiresAt time.Time ExpectedServerInstanceID sandboxwire.ID - MaxFrameBytes uint32 Exports []ExportGrant Egress []EgressRule } @@ -596,11 +588,11 @@ func WriteMessage(w io.Writer, requestID uint64, m Message) error { return sandboxwire.WriteFrame(w, f) } -// ReadMessage reads and decodes one frame no larger than maxPayload. The +// ReadMessage reads and decodes one frame no larger than MaxMessageBytes. The // request ID is returned whenever a frame was read, even if it failed to // decode, so a reader can answer it. -func ReadMessage(r io.Reader, maxPayload uint32) (uint64, Message, error) { - f, err := sandboxwire.ReadFrame(r, maxPayload) +func ReadMessage(r io.Reader) (uint64, Message, error) { + f, err := sandboxwire.ReadFrame(r, MaxMessageBytes) if err != nil { return 0, nil, err } @@ -608,6 +600,51 @@ func ReadMessage(r io.Reader, maxPayload uint32) (uint64, Message, error) { return f.RequestID, m, err } +// violation is the failure of a request whose answer broke the protocol: +// ProtocolViolation with EffectPossible, since the request may have taken +// effect. +func violation(cause error) *Error { + return &Error{Code: ProtocolViolation, Effect: sandboxwire.EffectPossible, Cause: cause} +} + +// readError is the failure of a request when reading its answer fails. A +// failed stream leaves the request Uncertain; a frame that is malformed or does +// not decode is a violation. +func readError(err error) *Error { + if errors.Is(err, sandboxwire.ErrMalformed) || errors.Is(err, VersionMismatch) { + return violation(err) + } + return Uncertain(err) +} + +// ReadReply reads the response to the request op sent with requestID. A +// Failure returns its *Error, and a failed stream returns Uncertain. Any other +// frame, including one with another request ID or for another operation, is a +// violation. +func ReadReply(r io.Reader, op Op, requestID uint64) (Message, error) { + id, m, err := ReadMessage(r) + if err != nil { + return nil, readError(err) + } + if id != requestID { + return nil, violation(nil) + } + return reply(op, m) +} + +// reply returns m when it answers op and the *Error of a Failure of op. A +// response's frame type names its operation, so any other message is a +// violation. +func reply(op Op, m Message) (Message, error) { + if m.frameType() != sandboxwire.ResponseType(uint16(op)) { + return nil, violation(nil) + } + if f, failed := m.(Failure); failed { + return nil, f.Err() + } + return m, nil +} + func decodeRequest(r *reader, op Op) Message { switch op { case OpHello: @@ -635,7 +672,7 @@ func decodeRequest(r *reader, op Op) Message { return o case OpBind: b := Bind{AttachmentID: r.id(), Service: r.service(), Version: r.u16(), SessionID: r.id(), AssignmentID: r.id(), - AssignmentEpoch: r.u64(), LeaseExpiresAt: r.time(), ExpectedServerInstanceID: r.id(), MaxFrameBytes: r.u32()} + AssignmentEpoch: r.u64(), LeaseExpiresAt: r.time(), ExpectedServerInstanceID: r.id()} if r.present() != (b.Service == ServiceFile) && r.err == nil { r.err = invalid("exports presence does not match service %s", b.Service) } @@ -663,9 +700,9 @@ func decodeRequest(r *reader, op Op) Message { func decodeSuccess(r *reader, op Op) Message { switch op { case OpHello: - return HelloAccepted{LinkID: r.id(), MaxStreams: r.u32(), MaxFrameBytes: r.u32()} + return HelloAccepted{} case OpOpen: - return Opened{AttachmentID: r.id(), ServerInstanceID: r.id(), LeaseExpiresAt: r.time(), MaxFrameBytes: r.u32()} + return Opened{AttachmentID: r.id(), ServerInstanceID: r.id(), LeaseExpiresAt: r.time()} case OpBind: return Bound{} case OpRenewAttachment: @@ -695,11 +732,7 @@ func (h AttachHello) encode(e *sandboxwire.Encoder) { e.Bytes(h.Credential) } -func (a HelloAccepted) encode(e *sandboxwire.Encoder) { - e.ID(a.LinkID) - e.U32(a.MaxStreams) - e.U32(a.MaxFrameBytes) -} +func (HelloAccepted) encode(*sandboxwire.Encoder) {} func (o Open) encode(e *sandboxwire.Encoder) { e.Enum(uint16(o.Service)) @@ -720,7 +753,6 @@ func (o Opened) encode(e *sandboxwire.Encoder) { e.ID(o.AttachmentID) e.ID(o.ServerInstanceID) e.I64(o.LeaseExpiresAt.UnixMilli()) - e.U32(o.MaxFrameBytes) } func (b Bind) encode(e *sandboxwire.Encoder) { @@ -732,7 +764,6 @@ func (b Bind) encode(e *sandboxwire.Encoder) { e.U64(b.AssignmentEpoch) e.I64(b.LeaseExpiresAt.UnixMilli()) e.ID(b.ExpectedServerInstanceID) - e.U32(b.MaxFrameBytes) e.Present(b.Service == ServiceFile) if b.Service == ServiceFile { e.Count(len(b.Exports)) @@ -818,13 +849,6 @@ func checkLease(t time.Time) error { return nil } -func checkFrameBytes(n uint32) error { - if n < MaxMessageBytes || n > sandboxwire.MaxPayload { - return invalid("max frame bytes %d outside %d..%d", n, MaxMessageBytes, sandboxwire.MaxPayload) - } - return nil -} - func checkService(s Service, version uint16) error { if !s.Valid() || version == 0 { return invalid("service %d version %d", s, version) @@ -929,15 +953,7 @@ func (h AttachHello) validate() error { return checkIDs(h.RuntimeID) } -func (a HelloAccepted) validate() error { - if a.MaxStreams == 0 { - return invalid("zero max streams") - } - if err := checkFrameBytes(a.MaxFrameBytes); err != nil { - return err - } - return checkIDs(a.LinkID) -} +func (HelloAccepted) validate() error { return nil } func (o Open) validate() error { if err := checkService(o.Service, o.Version); err != nil { @@ -959,9 +975,6 @@ func (o Opened) validate() error { if err := checkLease(o.LeaseExpiresAt); err != nil { return err } - if err := checkFrameBytes(o.MaxFrameBytes); err != nil { - return err - } return checkIDs(o.AttachmentID, o.ServerInstanceID) } @@ -975,9 +988,6 @@ func (b Bind) validate() error { if err := checkLease(b.LeaseExpiresAt); err != nil { return err } - if err := checkFrameBytes(b.MaxFrameBytes); err != nil { - return err - } if err := checkExports(b.Service, b.Exports); err != nil { return err } diff --git a/internal/sandboxlink/protocol_test.go b/internal/sandboxlink/protocol_test.go index f80b415d9..02bf5e411 100644 --- a/internal/sandboxlink/protocol_test.go +++ b/internal/sandboxlink/protocol_test.go @@ -43,15 +43,15 @@ var goldenFrames = []golden{ {1, ServeHello{Version: 1, Credential: []byte("serve"), Resource: testResource, ServerInstanceID: testID(0x05), Services: []ServiceVersion{{ServiceFile, 1}, {ServiceNetwork, 1}}}}, {1, AttachHello{Version: 1, RuntimeID: testID(0x06), Credential: []byte("runtime")}}, - {1, HelloAccepted{LinkID: testID(0x0a), MaxStreams: 256, MaxFrameBytes: 1 << 20}}, + {1, HelloAccepted{}}, {1, Open{Service: ServiceFile, Version: 1, Resource: testResource, ExpectedServerInstanceID: testID(0x05), AttachmentID: testID(0x07), SessionID: testID(0x08), AssignmentID: testID(0x09), AssignmentEpoch: 3, AttachGrant: []byte("grant")}}, - {1, Opened{AttachmentID: testID(0x07), ServerInstanceID: testID(0x05), LeaseExpiresAt: testLease, MaxFrameBytes: 1 << 20}}, + {1, Opened{AttachmentID: testID(0x07), ServerInstanceID: testID(0x05), LeaseExpiresAt: testLease}}, {1, Failure{Op: OpOpen, Code: StaleGeneration, Effect: sandboxwire.EffectNone}}, {1, Bind{AttachmentID: testID(0x07), Service: ServiceFile, Version: 1, SessionID: testID(0x08), AssignmentID: testID(0x09), - AssignmentEpoch: 3, LeaseExpiresAt: testLease, ExpectedServerInstanceID: testID(0x05), MaxFrameBytes: 1 << 20, Exports: testExports}}, + AssignmentEpoch: 3, LeaseExpiresAt: testLease, ExpectedServerInstanceID: testID(0x05), Exports: testExports}}, {1, Bind{AttachmentID: testID(0x07), Service: ServiceNetwork, Version: 1, SessionID: testID(0x08), AssignmentID: testID(0x09), - AssignmentEpoch: 3, LeaseExpiresAt: testLease, ExpectedServerInstanceID: testID(0x05), MaxFrameBytes: 1 << 20, Egress: testEgress}}, + AssignmentEpoch: 3, LeaseExpiresAt: testLease, ExpectedServerInstanceID: testID(0x05), Egress: testEgress}}, {1, Bound{}}, {0, AttachmentClosed{AttachmentID: testID(0x07), Reason: CloseLeaseExpired}}, } @@ -87,7 +87,7 @@ func TestGolden(t *testing.T) { } r := bytes.NewReader(want) for _, g := range goldenFrames { - id, m, err := ReadMessage(r, sandboxwire.MaxPayload) + 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) } @@ -109,11 +109,11 @@ func frameOf(t *testing.T, i int, edit func(p []byte) []byte) sandboxwire.Frame func TestDecodeRejects(t *testing.T) { // Offsets into the Bind payloads: Service at 16 and exports presence at - // 88. In the file Bind (golden frame 6) the export count ends at 92, the - // first export runs from 93 to 102 with its ID at 97, and egress presence - // is at 112. In the network Bind (golden frame 7) egress presence is at 89, - // the rule count ends at 93, and the first rule's family is at 94, address - // at 96, prefix length at 100 and ports at 101 and 103. + // 84. In the file Bind (golden frame 6) the export count ends at 88, the + // first export runs from 89 to 98 with its ID at 93, and egress presence + // is at 108. In the network Bind (golden frame 7) egress presence is at 85, + // the rule count ends at 89, and the first rule's family is at 90, address + // at 92, prefix length at 96 and ports at 97 and 99. set := func(off int, v ...byte) func([]byte) []byte { return func(p []byte) []byte { copy(p[off:], v); return p } } @@ -128,23 +128,23 @@ func TestDecodeRejects(t *testing.T) { {"file without exports", 7, set(16, 0, 1)}, {"exports on network", 6, set(16, 0, 3)}, {"empty exports", 6, func(p []byte) []byte { - p[92] = 0 - return append(p[:93], 0) + p[88] = 0 + return append(p[:89], 0) }}, - {"invalid export ID", 6, set(97, 'W')}, + {"invalid export ID", 6, set(93, 'W')}, {"duplicate export", 6, func(p []byte) []byte { - q := append(bytes.Clone(p[:112]), p[93:103]...) - q[92] = 3 + q := append(bytes.Clone(p[:108]), p[89:99]...) + q[88] = 3 return append(q, 0) }}, - {"host bits set", 7, set(99, 1)}, - {"prefix too long", 7, set(100, 33)}, - {"unknown family", 7, set(94, 0, 3)}, - {"zero first port", 7, set(101, 0, 0)}, - {"first port above last", 7, set(101, 0x01, 0xbc)}, + {"host bits set", 7, set(95, 1)}, + {"prefix too long", 7, set(96, 33)}, + {"unknown family", 7, set(90, 0, 3)}, + {"zero first port", 7, set(97, 0, 0)}, + {"first port above last", 7, set(97, 0x01, 0xbc)}, {"duplicate rule", 7, func(p []byte) []byte { - p[93] = 3 - return append(p, p[94:105]...) + p[89] = 3 + return append(p, p[90:101]...) }}, } for _, c := range cases { diff --git a/internal/sandboxlink/relay/relay.go b/internal/sandboxlink/relay/relay.go index ac8ff4aab..51ce1fb6c 100644 --- a/internal/sandboxlink/relay/relay.go +++ b/internal/sandboxlink/relay/relay.go @@ -20,7 +20,8 @@ import ( ) const ( - defaultMaxStreams = 256 + // maxStreams bounds the concurrent service streams of each link. + maxStreams = 256 // spliceBuffer bounds the bytes a splice holds per direction, on top of // one yamux window per stream. spliceBuffer = 32 << 10 @@ -31,22 +32,11 @@ const ( answerQueue = 64 ) -// Config configures a Relay. -type Config struct { - Authority sandboxlink.Authority - // MaxStreams bounds the concurrent service streams of each link; zero - // selects 256. - MaxStreams uint32 - // MaxFrameBytes is the frame payload limit the relay advertises; zero - // selects sandboxwire.MaxPayload. - MaxFrameBytes uint32 -} - // Relay accepts Link peers on ServeHTTP. It keeps the current serve peer of // each resource, the attachments it has opened and their leases in memory; the // Authority stays the durable judge of every grant. type Relay struct { - cfg Config + auth sandboxlink.Authority ctx context.Context cancel context.CancelFunc @@ -63,27 +53,16 @@ type Relay struct { // closures is a set of AttachmentClosed events to write, by attachment. type closures map[sandboxwire.ID]sandboxlink.CloseReason -// New returns a relay for cfg. -func New(cfg Config) (*Relay, error) { - if cfg.Authority == nil { - return nil, errors.New("sandbox link relay: no authority") - } - if cfg.MaxStreams == 0 { - cfg.MaxStreams = defaultMaxStreams - } - if cfg.MaxFrameBytes == 0 { - cfg.MaxFrameBytes = sandboxwire.MaxPayload - } - if cfg.MaxFrameBytes < sandboxlink.MaxMessageBytes || cfg.MaxFrameBytes > sandboxwire.MaxPayload { - return nil, errors.New("sandbox link relay: max frame bytes out of range") - } +// New returns a relay that asks auth to authenticate peers and authorize their +// requests. +func New(auth sandboxlink.Authority) *Relay { ctx, cancel := context.WithCancel(context.Background()) - return &Relay{cfg: cfg, ctx: ctx, cancel: cancel, + return &Relay{auth: auth, ctx: ctx, cancel: cancel, generations: map[resourceKey]uint64{}, serves: map[resourceKey]*serveLink{}, closures: map[resourceKey]closures{}, attachments: map[sandboxwire.ID]*attachment{}, - }, nil + } } // Close ends every link and stops lease timers. @@ -325,7 +304,7 @@ func (rl *Relay) ServeHTTP(w http.ResponseWriter, r *http.Request) { if err != nil { return } - id, m, err := sandboxlink.ReadMessage(ctl, sandboxlink.MaxMessageBytes) + id, m, err := sandboxlink.ReadMessage(ctl) handshake.Stop() l := rl.newLink(sess, ctl) switch hello := m.(type) { @@ -347,10 +326,6 @@ func (rl *Relay) authorityContext() (context.Context, context.CancelFunc) { return context.WithTimeout(rl.ctx, sandboxlink.HandshakeTimeout) } -func (rl *Relay) accepted() sandboxlink.HelloAccepted { - return sandboxlink.HelloAccepted{LinkID: sandboxwire.NewID(), MaxStreams: rl.cfg.MaxStreams, MaxFrameBytes: rl.cfg.MaxFrameBytes} -} - // refusal returns the failure code for an Authority error. func refusal(err error) sandboxlink.Code { return sandboxlink.FailureFor(sandboxlink.OpHello, err).Code @@ -373,7 +348,7 @@ func (rl *Relay) serve(l *link, id uint64, hello sandboxlink.ServeHello) { l.sess.Close() }() go func() { - sandboxlink.ReadMessage(l.ctl, sandboxlink.MaxMessageBytes) + sandboxlink.ReadMessage(l.ctl) l.sess.Close() }() <-l.sess.CloseChan() @@ -394,7 +369,7 @@ func (rl *Relay) admitServe(l *link, id uint64, hello sandboxlink.ServeHello) (* epoch := rl.epoch rl.mu.Unlock() ctx, cancel := rl.authorityContext() - peer, err := rl.cfg.Authority.AuthenticateServe(ctx, hello) + peer, err := rl.auth.AuthenticateServe(ctx, hello) cancel() if err == nil && peer.Resource != hello.Resource { err = sandboxlink.Fail(sandboxlink.PermissionDenied) @@ -445,7 +420,7 @@ func (rl *Relay) installServeLocked(l *link, id uint64, hello sandboxlink.ServeH old.closures = nil } rl.serves[key] = sl - sl.send(id, rl.accepted()) + sl.send(id, sandboxlink.HelloAccepted{}) return sl, old, nil } @@ -453,7 +428,7 @@ func (rl *Relay) installServeLocked(l *link, id uint64, hello sandboxlink.ServeH // the link ends. Its attachments stay open until their leases expire. func (rl *Relay) attach(l *link, id uint64, hello sandboxlink.AttachHello) { ctx, cancel := rl.authorityContext() - peer, err := rl.cfg.Authority.AuthenticateAttach(ctx, hello) + peer, err := rl.auth.AuthenticateAttach(ctx, hello) cancel() if err == nil && peer.RuntimeID != hello.RuntimeID { err = sandboxlink.Fail(sandboxlink.PermissionDenied) @@ -465,7 +440,7 @@ func (rl *Relay) attach(l *link, id uint64, hello sandboxlink.AttachHello) { al := &attachLink{link: l, peer: peer} var seq sandboxwire.RequestSequence seq.Admit(id) // the Hello takes the first ID - al.send(id, rl.accepted()) + al.send(id, sandboxlink.HelloAccepted{}) go func() { for { st, err := l.sess.AcceptStream() @@ -476,7 +451,7 @@ func (rl *Relay) attach(l *link, id uint64, hello sandboxlink.AttachHello) { } }() for { - id, m, err := sandboxlink.ReadMessage(l.ctl, sandboxlink.MaxMessageBytes) + id, m, err := sandboxlink.ReadMessage(l.ctl) if err != nil { l.sess.Close() return @@ -539,7 +514,7 @@ func (rl *Relay) renewLease(al *attachLink, r sandboxlink.RenewAttachment) (sand return sandboxlink.AttachmentRenewed{}, sandboxlink.Fail(sandboxlink.PermissionDenied) } ctx, cancel := rl.authorityContext() - auth, err := rl.cfg.Authority.Renew(ctx, al.peer, r) + auth, err := rl.auth.Renew(ctx, al.peer, r) cancel() if err != nil { return sandboxlink.AttachmentRenewed{}, err @@ -644,7 +619,7 @@ func (sp *splice) abortLocked(code sandboxlink.Code) { // authorizes it, binds the serve peer and splices the two streams. func (rl *Relay) open(al *attachLink, st *yamux.Stream) { st.SetDeadline(time.Now().Add(sandboxlink.HandshakeTimeout)) - id, m, err := sandboxlink.ReadMessage(st, sandboxlink.MaxMessageBytes) + id, m, err := sandboxlink.ReadMessage(st) st.SetDeadline(time.Time{}) o, ok := m.(sandboxlink.Open) if err != nil || !ok { @@ -659,11 +634,11 @@ func (rl *Relay) open(al *attachLink, st *yamux.Stream) { defer rl.finish(sp) sl := sp.serve opened := sandboxlink.Opened{AttachmentID: o.AttachmentID, ServerInstanceID: sl.hello.ServerInstanceID, - LeaseExpiresAt: auth.LeaseExpiresAt, MaxFrameBytes: rl.cfg.MaxFrameBytes} + LeaseExpiresAt: auth.LeaseExpiresAt} bind := sandboxlink.Bind{AttachmentID: o.AttachmentID, Service: o.Service, Version: o.Version, SessionID: o.SessionID, AssignmentID: o.AssignmentID, AssignmentEpoch: o.AssignmentEpoch, LeaseExpiresAt: auth.LeaseExpiresAt, ExpectedServerInstanceID: sl.hello.ServerInstanceID, - MaxFrameBytes: rl.cfg.MaxFrameBytes, Exports: auth.Exports, Egress: auth.Egress} + Exports: auth.Exports, Egress: auth.Egress} err = rl.bind(sp, bind) rl.mu.Lock() if sp.aborted != 0 { @@ -694,7 +669,7 @@ func (rl *Relay) admit(al *attachLink, st *yamux.Stream, o sandboxlink.Open) (*s epoch := rl.epoch rl.mu.Unlock() ctx, cancel := rl.authorityContext() - auth, err := rl.cfg.Authority.AuthorizeOpen(ctx, al.peer, o) + auth, err := rl.auth.AuthorizeOpen(ctx, al.peer, o) cancel() if err != nil { return nil, auth, err @@ -728,7 +703,7 @@ func (rl *Relay) admitLocked(al *attachLink, st *yamux.Stream, o sandboxlink.Ope a := rl.attachments[o.AttachmentID] var code sandboxlink.Code switch { - case al.streams >= rl.cfg.MaxStreams || (sl != nil && sl.streams >= rl.cfg.MaxStreams): + case al.streams >= maxStreams || (sl != nil && sl.streams >= maxStreams): code = sandboxlink.LimitExceeded case rl.generations[key] > generation: code = sandboxlink.StaleGeneration @@ -786,18 +761,11 @@ func (rl *Relay) bind(sp *splice, b sandboxlink.Bind) error { if err := sandboxwire.WriteFrame(ss, f); err != nil { return sandboxlink.Uncertain(err) } - _, m, err := sandboxlink.ReadMessage(ss, sandboxlink.MaxMessageBytes) - if err != nil { - return sandboxlink.Uncertain(err) - } - switch r := m.(type) { - case sandboxlink.Bound: - ss.SetDeadline(time.Time{}) - return nil - case sandboxlink.Failure: - return r.Err() + if _, err := sandboxlink.ReadReply(ss, sandboxlink.OpBind, f.RequestID); err != nil { + return err } - return &sandboxlink.Error{Code: sandboxlink.ProtocolViolation, Effect: sandboxwire.EffectPossible} + ss.SetDeadline(time.Time{}) + return nil } // abortError answers an Open that a close interrupted during bind, whose own diff --git a/internal/sandboxlink/relay/relay_test.go b/internal/sandboxlink/relay/relay_test.go index 58bfa82fa..f7bff701b 100644 --- a/internal/sandboxlink/relay/relay_test.go +++ b/internal/sandboxlink/relay/relay_test.go @@ -14,7 +14,6 @@ import ( "time" "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" ) @@ -98,7 +97,7 @@ type fixture struct { func newFixture(t *testing.T) *fixture { auth := &hooked{Authority: sandboxlinktest.NewAuthority()} - f := &fixture{t: t, auth: auth, srv: sandboxlinktest.StartRelay(t, relay.Config{Authority: auth}), + f := &fixture{t: t, auth: auth, srv: sandboxlinktest.StartRelay(t, auth), runtime: sandboxwire.NewID(), closed: make(chan sandboxlink.AttachmentClosed, 16), lease: time.Minute} auth.AddRuntime([]byte("runtime credential"), f.runtime) link, err := sandboxlink.DialAttach(context.Background(), sandboxlink.AttachConfig{URL: f.srv.URL, TLS: f.srv.TLS, @@ -117,7 +116,7 @@ func newFixture(t *testing.T) *fixture { // stream. type servePeer struct { instance sandboxwire.ID - connected chan sandboxlink.HelloAccepted + connected chan struct{} conns chan net.Conn lost chan sandboxwire.ID restored chan sandboxwire.ID @@ -146,7 +145,7 @@ func (f *fixture) serve(generation uint64) *servePeer { func (f *fixture) startServe(generation uint64) *servePeer { credential := []byte(fmt.Sprintf("serve credential %d", generation)) f.auth.AddServe(credential, sandboxlink.ServePeer{PeerID: sandboxwire.NewID(), Resource: resource(generation)}) - p := &servePeer{instance: sandboxwire.NewID(), connected: make(chan sandboxlink.HelloAccepted, 16), conns: make(chan net.Conn, 16), + p := &servePeer{instance: sandboxwire.NewID(), connected: make(chan struct{}, 16), conns: make(chan net.Conn, 16), lost: make(chan sandboxwire.ID, 16), restored: make(chan sandboxwire.ID, 16), closed: make(chan sandboxlink.CloseReason, 128), binds: make(chan sandboxlink.Bind, 16), echoed: make(chan error, 16), seqs: make(chan uint64, 16), done: make(chan struct{})} echo := func(_ context.Context, b sandboxlink.Bind, _ uint64, s sandboxlink.Stream) { @@ -175,7 +174,7 @@ func (f *fixture) startServe(generation uint64) *servePeer { {Service: sandboxlink.ServiceFile, Version: 1, Serve: echo}, {Service: sandboxlink.ServiceProcess, Version: 1, Serve: resetAfterOne}, }, - OnConnected: func(a sandboxlink.HelloAccepted) { put(p.connected, a) }, + OnConnected: func() { put(p.connected, struct{}{}) }, OnAttachmentLost: func(id sandboxwire.ID) { put(p.lost, id) }, OnAttachmentRestored: func(id sandboxwire.ID) { put(p.restored, id) }, OnAttachmentClosed: func(_ sandboxwire.ID, r sandboxlink.CloseReason) { put(p.closed, r) }, @@ -518,3 +517,112 @@ func TestBindSequence(t *testing.T) { t.Fatalf("bind sequences %d, %d, then %d after a reconnect; want them to increase from above zero", earlier, later, reconnected) } } + +// A request fails with a final ProtocolViolation and EffectPossible when its +// answer carries another request ID, request ID zero or another operation. The +// relay passes a Bind's failure on to the Open. A CloseAttachment answer under +// another request ID is left out: the link drops it as a late answer. +func TestAnswersMatchTheirRequests(t *testing.T) { + for _, op := range []sandboxlink.Op{sandboxlink.OpHello, sandboxlink.OpOpen, sandboxlink.OpBind, sandboxlink.OpCloseAttachment} { + for _, kind := range []string{"wrong ID", "zero ID", "wrong op"} { + if op == sandboxlink.OpCloseAttachment && kind == "wrong ID" { + continue + } + t.Run(fmt.Sprintf("op %d %s", op, kind), func(t *testing.T) { + // answer writes m as the answer to request id, spoiled as kind + // says. A fake that fails leaves its caller without a violation. + answer := func(w io.Writer, id uint64, m sandboxlink.Message) { + switch kind { + case "wrong ID": + id++ + case "zero ID": + id = 0 + default: + m = sandboxlink.FailureFor(sandboxlink.OpRenewAttachment, sandboxlink.Fail(sandboxlink.ServiceUnavailable)) + } + fr, _ := sandboxlink.Encode(1, m) + fr.RequestID = id + sandboxwire.WriteFrame(w, fr) + } + ctx, cancel := context.WithTimeout(context.Background(), wait) + defer cancel() + f, grant := &fixture{t: t}, []byte("grant") + var err error + if op == sandboxlink.OpBind { + f = newFixture(t) + grant = f.grant(1) + credential := []byte("serve credential") + f.auth.AddServe(credential, sandboxlink.ServePeer{PeerID: sandboxwire.NewID(), Resource: resource(1)}) + conn, err := sandboxlink.DialWebSocket(ctx, f.srv.URL, f.srv.TLS) + if err != nil { + t.Fatal(err) + } + sess, _ := sandboxlink.ClientSession(conn) + t.Cleanup(func() { sess.Close() }) + ctl, err := sess.OpenStream(ctx) + if err == nil { + err = sandboxlink.WriteMessage(ctl, 1, sandboxlink.ServeHello{Version: sandboxlink.Version, Credential: credential, Resource: resource(1), + ServerInstanceID: sandboxwire.NewID(), Services: []sandboxlink.ServiceVersion{{Service: sandboxlink.ServiceFile, Version: 1}}}) + } + if err == nil { + _, err = sandboxlink.ReadReply(ctl, sandboxlink.OpHello, 1) + } + if err != nil { + t.Fatal(err) + } + go func() { + st, err := sess.AcceptStream() + if err != nil { + return + } + id, _, _ := sandboxlink.ReadMessage(st) + answer(st, id, sandboxlink.Bound{}) + }() + } else { + client, server := net.Pipe() + t.Cleanup(func() { server.Close() }) + go func() { + sess, _ := sandboxlink.ServerSession(server) + ctl, err := sess.AcceptStream() + if err != nil { + return + } + id, _, _ := sandboxlink.ReadMessage(ctl) + if op == sandboxlink.OpHello { + answer(ctl, id, sandboxlink.HelloAccepted{}) + return + } + sandboxlink.WriteMessage(ctl, id, sandboxlink.HelloAccepted{}) + if op == sandboxlink.OpCloseAttachment { + id, _, _ = sandboxlink.ReadMessage(ctl) + answer(ctl, id, sandboxlink.CloseAccepted{}) + return + } + st, err := sess.AcceptStream() + if err != nil { + return + } + id, m, _ := sandboxlink.ReadMessage(st) + o, _ := m.(sandboxlink.Open) + answer(st, id, sandboxlink.Opened{AttachmentID: o.AttachmentID, ServerInstanceID: sandboxwire.NewID(), LeaseExpiresAt: time.Now().Add(time.Minute)}) + }() + f.link, err = sandboxlink.DialAttach(ctx, sandboxlink.AttachConfig{RuntimeID: sandboxwire.NewID(), Credential: []byte("runtime credential"), + Dial: func(context.Context) (net.Conn, error) { return client, nil }}) + } + switch { + case op == sandboxlink.OpHello: + case err != nil: + t.Fatal(err) + case op == sandboxlink.OpCloseAttachment: + err = f.link.CloseAttachment(ctx, sandboxwire.NewID()) + default: + _, _, err = f.open(sandboxlink.ServiceFile, sandboxwire.NewID(), 1, grant) + } + var e *sandboxlink.Error + if !errors.As(err, &e) || e.Code != sandboxlink.ProtocolViolation || e.Effect != sandboxwire.EffectPossible || e.Code.Retryable() { + t.Fatalf("%v, want a ProtocolViolation with EffectPossible", err) + } + }) + } + } +} diff --git a/internal/sandboxlink/sandboxlinktest/authority.go b/internal/sandboxlink/sandboxlinktest/authority.go index 8e7360881..8e3081259 100644 --- a/internal/sandboxlink/sandboxlinktest/authority.go +++ b/internal/sandboxlink/sandboxlinktest/authority.go @@ -82,13 +82,6 @@ func (a *Authority) AddGrant(grant []byte, g Grant) { a.grants[string(grant)] = g } -// RemoveGrant withdraws an attachment grant. -func (a *Authority) RemoveGrant(grant []byte) { - a.mu.Lock() - defer a.mu.Unlock() - delete(a.grants, string(grant)) -} - func (a *Authority) AuthenticateServe(_ context.Context, hello sandboxlink.ServeHello) (sandboxlink.ServePeer, error) { a.mu.Lock() defer a.mu.Unlock() diff --git a/internal/sandboxlink/sandboxlinktest/server.go b/internal/sandboxlink/sandboxlinktest/server.go index 6c739ce50..90a800e1f 100644 --- a/internal/sandboxlink/sandboxlinktest/server.go +++ b/internal/sandboxlink/sandboxlinktest/server.go @@ -7,6 +7,7 @@ import ( "strings" "testing" + "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink" "github.com/MiniMax-AI/OpenAgentCore/internal/sandboxlink/relay" ) @@ -19,13 +20,9 @@ type Server struct { Relay *relay.Relay } -// StartRelay starts a relay for cfg and stops it when the test ends. -func StartRelay(t testing.TB, cfg relay.Config) *Server { - t.Helper() - rl, err := relay.New(cfg) - if err != nil { - t.Fatal(err) - } +// StartRelay starts a relay for auth and stops it when the test ends. +func StartRelay(t testing.TB, auth sandboxlink.Authority) *Server { + rl := relay.New(auth) srv := httptest.NewTLSServer(rl) t.Cleanup(srv.Close) t.Cleanup(func() { rl.Close() }) diff --git a/internal/sandboxlink/serve.go b/internal/sandboxlink/serve.go index 726a008c5..3bf89f073 100644 --- a/internal/sandboxlink/serve.go +++ b/internal/sandboxlink/serve.go @@ -42,7 +42,7 @@ type ServeConfig struct { // OnConnected reports each accepted Hello; OnDisconnected reports why a // link ended before the next attempt. - OnConnected func(HelloAccepted) + OnConnected func() OnDisconnected func(error) // An attachment is lost when its last open stream ends while it is still // attached, restored when a stream binds it again, and closed when the @@ -145,7 +145,7 @@ type served struct { // link runs one connection. connected reports whether the Hello was accepted. func (s *server) link(ctx context.Context, hello ServeHello) (connected bool, err error) { var seq sandboxwire.RequestSequence - sess, ctl, accepted, err := connect(ctx, s.dial, hello, &seq) + sess, ctl, err := connect(ctx, s.dial, hello, &seq) if err != nil { return false, err } @@ -153,7 +153,7 @@ func (s *server) link(ctx context.Context, hello ServeHello) (connected bool, er stop := context.AfterFunc(ctx, func() { sess.Close() }) defer stop() if s.cfg.OnConnected != nil { - s.cfg.OnConnected(accepted) + s.cfg.OnConnected() } done := make(chan error, 1) go func() { done <- s.control(ctl) }() @@ -180,7 +180,7 @@ func (s *server) link(ctx context.Context, hello ServeHello) (connected bool, er func (s *server) control(ctl *yamux.Stream) error { defer ctl.Session().Close() for { - _, m, err := ReadMessage(ctl, MaxMessageBytes) + _, m, err := ReadMessage(ctl) if err != nil { return err } @@ -195,7 +195,7 @@ func (s *server) control(ctl *yamux.Stream) error { // bind accepts one relay-opened stream and runs its handler. func (s *server) bind(ctx context.Context, st *yamux.Stream) { st.SetDeadline(time.Now().Add(HandshakeTimeout)) - id, m, err := ReadMessage(st, MaxMessageBytes) + id, m, err := ReadMessage(st) b, ok := m.(Bind) var h *ServiceHandler if ok { diff --git a/internal/sandboxlink/testdata/link_v1.hex b/internal/sandboxlink/testdata/link_v1.hex index 537cbc79c..7c1e707e6 100644 --- a/internal/sandboxlink/testdata/link_v1.hex +++ b/internal/sandboxlink/testdata/link_v1.hex @@ -1,5 +1,5 @@ # Link version 1 frames from goldenFrames in protocol_test.go, in order. Hex bytes; text after # is a comment. -# IDs repeat one byte: 01 tenant, 02 environment, 03 resource, 05 server instance, 06 Runtime, 07 attachment, 08 Session, 09 assignment, 0a link. +# IDs repeat one byte: 01 tenant, 02 environment, 03 resource, 05 server instance, 06 Runtime, 07 attachment, 08 Session, 09 assignment. # ServeHello 00000063 0001 0000 0000000000000001 # header: length 99, OpHello, flags, request 1 @@ -24,11 +24,8 @@ 00000007 72756e74696d65 # Credential "runtime" # HelloAccepted -0000001a 8001 0000 0000000000000001 # header: length 26, response to OpHello, flags, request 1 +00000002 8001 0000 0000000000000001 # header: length 2, response to OpHello, flags, request 1 0001 # result success -0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a0a # LinkID -00000100 # MaxStreams 256 -00100000 # MaxFrameBytes 1 MiB # Open 00000090 0002 0000 0000000000000001 # header: length 144, OpOpen, flags, request 1 @@ -48,12 +45,11 @@ 00000005 6772616e74 # AttachGrant "grant" # Opened -0000002e 8002 0000 0000000000000001 # header: length 46, response to OpOpen, flags, request 1 +0000002a 8002 0000 0000000000000001 # header: length 42, response to OpOpen, flags, request 1 0001 # result success 07070707070707070707070707070707 # AttachmentID 05050505050505050505050505050505 # ServerInstanceID 000001a0c4506c00 # LeaseExpiresAt 1790000000000 ms -00100000 # MaxFrameBytes 1 MiB # Failure answering Open 00000006 8002 0000 0000000000000001 # header: length 6, response to OpOpen, flags, request 1 @@ -62,7 +58,7 @@ 0001 # Effect EffectNone # Bind for a File stream -00000071 0003 0000 0000000000000001 # header: length 113, OpBind, flags, request 1 +0000006d 0003 0000 0000000000000001 # header: length 109, OpBind, flags, request 1 07070707070707070707070707070707 # AttachmentID 0001 # Service ServiceFile 0001 # Version 1 @@ -71,7 +67,6 @@ 0000000000000003 # AssignmentEpoch 3 000001a0c4506c00 # LeaseExpiresAt 1790000000000 ms 05050505050505050505050505050505 # ExpectedServerInstanceID -00100000 # MaxFrameBytes 1 MiB 01 # Exports present 00000002 # export count 2 00000005 776f726c64 00 # export "world", read-write @@ -79,7 +74,7 @@ 00 # Egress absent # Bind for a Network stream -00000080 0003 0000 0000000000000001 # header: length 128, OpBind, flags, request 1 +0000007c 0003 0000 0000000000000001 # header: length 124, OpBind, flags, request 1 07070707070707070707070707070707 # AttachmentID 0003 # Service ServiceNetwork 0001 # Version 1 @@ -88,7 +83,6 @@ 0000000000000003 # AssignmentEpoch 3 000001a0c4506c00 # LeaseExpiresAt 1790000000000 ms 05050505050505050505050505050505 # ExpectedServerInstanceID -00100000 # MaxFrameBytes 1 MiB 00 # Exports absent 01 # Egress present 00000002 # rule count 2 diff --git a/internal/sandboxlink/transport.go b/internal/sandboxlink/transport.go index 0cb71072a..f5d6c2cda 100644 --- a/internal/sandboxlink/transport.go +++ b/internal/sandboxlink/transport.go @@ -169,43 +169,35 @@ func ServerSession(conn net.Conn) (*yamux.Session, error) { // connect dials, starts yamux, opens the control stream and exchanges the // Hello, which takes the first ID of seq. A refused Hello returns the relay's // *Error. -func connect(ctx context.Context, dial Dialer, hello Message, seq *sandboxwire.RequestSequence) (*yamux.Session, *yamux.Stream, HelloAccepted, error) { +func connect(ctx context.Context, dial Dialer, hello Message, seq *sandboxwire.RequestSequence) (*yamux.Session, *yamux.Stream, error) { conn, err := dial(ctx) if err != nil { - return nil, nil, HelloAccepted{}, err + return nil, nil, err } sess, err := ClientSession(conn) if err != nil { conn.Close() - return nil, nil, HelloAccepted{}, err + return nil, nil, err } stop := context.AfterFunc(ctx, func() { sess.Close() }) defer stop() ctl, err := sess.OpenStream(ctx) if err == nil { ctl.SetDeadline(time.Now().Add(HandshakeTimeout)) - err = WriteMessage(ctl, seq.Next(), hello) - } - var m Message - if err == nil { - _, m, err = ReadMessage(ctl, MaxMessageBytes) + id := seq.Next() + if err = WriteMessage(ctl, id, hello); err == nil { + _, err = ReadReply(ctl, OpHello, id) + } } if err == nil { - switch r := m.(type) { - case HelloAccepted: - ctl.SetDeadline(time.Time{}) - return sess, ctl, r, nil - case Failure: - err = r.Err() - default: - err = Fail(ProtocolViolation) - } + ctl.SetDeadline(time.Time{}) + return sess, ctl, nil } sess.Close() if ctx.Err() != nil { err = ctx.Err() } - return nil, nil, HelloAccepted{}, err + return nil, nil, err } func dialerFor(dial Dialer, rawURL string, tlsConfig *tls.Config) Dialer { diff --git a/internal/sandboxwire/frame.go b/internal/sandboxwire/frame.go index 6a7653c23..376006182 100644 --- a/internal/sandboxwire/frame.go +++ b/internal/sandboxwire/frame.go @@ -20,8 +20,8 @@ const ( // MessageType uint16, Flags uint16 and RequestID uint64, in network byte // order. HeaderSize = 16 - // MaxPayload is the hard maximum frame payload. Advertised limits may be - // smaller. + // MaxPayload is the hard maximum frame payload. A protocol may bound its + // messages lower. MaxPayload = 1 << 20 // MaxChunk is the maximum file or process data chunk. MaxChunk = 64 << 10 diff --git a/internal/sandboxwire/sandboxwire_test.go b/internal/sandboxwire/sandboxwire_test.go index 54727f868..9868d1bcc 100644 --- a/internal/sandboxwire/sandboxwire_test.go +++ b/internal/sandboxwire/sandboxwire_test.go @@ -166,7 +166,7 @@ func TestRejectsMalformed(t *testing.T) { }{ // Each reader holds only a header, so these fail before any payload read. {"length above MaxPayload", errOf(ReadFrame(bytes.NewReader(header(MaxPayload+1, 0)), ^uint32(0)))}, - {"length above advertised limit", errOf(ReadFrame(bytes.NewReader(header(65, 0)), 64))}, + {"length above the caller's limit", errOf(ReadFrame(bytes.NewReader(header(65, 0)), 64))}, {"nonzero flags", errOf(ReadFrame(bytes.NewReader(header(0, 1)), MaxPayload))}, {"oversize write", WriteFrame(io.Discard, Frame{Type: 1, RequestID: 1, Payload: make([]byte, MaxPayload+1)})}, {"trailing bytes", NewDecoder([]byte{0}).Finish()},