diff --git a/addons/internal/connector/route.go b/addons/internal/connector/route.go index 84db92d7fe..af84a13a30 100644 --- a/addons/internal/connector/route.go +++ b/addons/internal/connector/route.go @@ -11,6 +11,7 @@ import ( "context" "errors" "fmt" + "io" "time" "github.com/mudler/LocalAI/pkg/addons/contract" @@ -111,6 +112,13 @@ type Splitter interface { // means. A request for another kind, without a token or with an unknown op is // refused. func Serve(ctx context.Context, p Platform, req contract.ConnectorRequest) (contract.ConnectorResponse, error) { + return ServeWith(ctx, p, nil, req) +} + +// ServeWith is Serve for an add-on that has the host operation channel, so it +// can move file bytes by handle (see ConnectorTransfer). A request with a +// handle that the Platform cannot serve is refused. +func ServeWith(ctx context.Context, p Platform, host Host, req contract.ConnectorRequest) (contract.ConnectorResponse, error) { if req.Kind != p.Kind() || req.Token == "" { return contract.ConnectorResponse{}, fmt.Errorf("%w: unknown kind or no token", contract.ErrRefused) } @@ -149,6 +157,18 @@ func Serve(ctx context.Context, p Platform, req contract.ConnectorRequest) (cont out.BotID, out.BotUsername = bot.ID, bot.Username } case contract.ConnectorOpSendFile: + if req.Transfer != nil { + ss, ok := p.(StreamSender) + if !ok || req.File != nil || req.Transfer.Handle == "" || req.Transfer.Size <= 0 || req.Transfer.Name != contract.SafeFileName(req.Transfer.Name) { + return contract.ConnectorResponse{}, fmt.Errorf("%w: no file, or one this platform cannot send by handle", contract.ErrRefused) + } + err = Retry(ctx, func() error { + id, err := ss.SendStream(ctx, req, Get(ctx, host, req.Transfer)) + out.MessageID = id + return err + }) + break + } fs, ok := p.(FileSender) if !ok || !validFile(req.File) { return contract.ConnectorResponse{}, fmt.Errorf("%w: no file, or one this platform cannot send", contract.ErrRefused) @@ -159,6 +179,10 @@ func Serve(ctx context.Context, p Platform, req contract.ConnectorRequest) (cont return err }) case contract.ConnectorOpMedia: + if req.Transfer != nil { + out.Received, err = receive(ctx, p, host, req) + break + } m, ok := p.(MediaReader) if !ok || req.Media == "" { return contract.ConnectorResponse{}, fmt.Errorf("%w: no media", contract.ErrRefused) @@ -188,12 +212,36 @@ func Serve(ctx context.Context, p Platform, req contract.ConnectorRequest) (cont default: return contract.ConnectorResponse{}, fmt.Errorf("%w: unknown op", contract.ErrRefused) } + if errors.Is(err, contract.ErrRefused) { + return contract.ConnectorResponse{}, err + } if err != nil { return contract.ConnectorResponse{Error: Typed(err)}, nil } return out, nil } +// receive streams the file of a reference to the host by the handle. +func receive(ctx context.Context, p Platform, host Host, req contract.ConnectorRequest) (*contract.ConnectorReceived, error) { + sr, ok := p.(StreamReader) + if !ok || req.Media == "" || req.Transfer.Handle == "" || req.Transfer.MaxBytes <= 0 { + return nil, fmt.Errorf("%w: no media, or a platform that cannot stream it", contract.ErrRefused) + } + m, err := sr.MediaStream(ctx, req.Token, req.Media, req.Transfer.MaxBytes) + if err != nil { + return nil, err + } + defer func() { _ = m.Body.Close() }() + if m.Size > req.Transfer.MaxBytes { + return nil, ErrTooLarge + } + // One byte past the bound shows a file that is larger than it said. + if _, err := Put(ctx, host, req.Transfer, io.LimitReader(m.Body, req.Transfer.MaxBytes+1)); err != nil { + return nil, err + } + return &contract.ConnectorReceived{Name: contract.SafeFileName(m.Name), MIME: m.MIME}, nil +} + type optionsKey struct{} // OptionsFrom returns the options in effect that the host sent with the op diff --git a/addons/internal/connector/transfer.go b/addons/internal/connector/transfer.go new file mode 100644 index 0000000000..77c6c3becc --- /dev/null +++ b/addons/internal/connector/transfer.go @@ -0,0 +1,154 @@ +// SPDX-License-Identifier: MIT +package connector + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + + "github.com/mudler/LocalAI/pkg/addons/contract" +) + +// Host is the host operation channel of the add-on. *hostclient.Client +// implements it. +type Host interface { + Do(ctx context.Context, operationID, op string, args, out any) error +} + +// transferTries is how often one chunk is sent before the transfer fails. A +// repeat is safe: the host answers where it is, and a chunk it already has is +// not written twice. +const transferTries = 3 + +// ErrNoHost means the add-on got a handle but has no host operation channel. +var ErrNoHost = errors.New("the host gave this add-on no channel for file transfer") + +func hostError(err error) error { + var he *contract.HostError + if errors.As(err, &he) && he.Code == contract.HostTooLarge { + return fmt.Errorf("%w: %v", ErrTooLarge, err) + } + return err +} + +// Put streams r to the host through the handle of an inbound file and ends +// the file. It holds one chunk in memory, whatever the size of the file, and +// returns the bytes sent. The host refuses the file when it passes the bound +// of the handle, and Put then returns ErrTooLarge. +func Put(ctx context.Context, h Host, t *contract.ConnectorTransfer, r io.Reader) (int64, error) { + if h == nil { + return 0, ErrNoHost + } + buf := make([]byte, contract.TransferChunkBytes) + var off int64 + for { + n, rerr := io.ReadFull(r, buf) + last := rerr != nil // a short read or the end + if rerr != nil && !errors.Is(rerr, io.EOF) && !errors.Is(rerr, io.ErrUnexpectedEOF) { + return off, rerr + } + if n == 0 && !last { + continue + } + var res contract.ConnectorFilePutResult + var err error + for try := 0; try < transferTries; try++ { + err = h.Do(ctx, t.Handle, contract.ConnectorFilePutOp, contract.ConnectorFilePut{Handle: t.Handle, Offset: off, Data: buf[:n], Final: last}, &res) + if err == nil || ctx.Err() != nil || errors.Is(hostError(err), ErrTooLarge) { + break + } + } + if err != nil { + return off, hostError(err) + } + if res.Received != off+int64(n) { + // The host is not where the add-on is, and a stream cannot be + // rewound; the platform download starts again. + return off, fmt.Errorf("the host has %d bytes and the add-on sent %d", res.Received, off+int64(n)) + } + off += int64(n) + if last { + return off, nil + } + } +} + +// Get returns a reader of the outbound file of the handle. Each Read pulls at +// most one chunk, so a platform upload that reads from it streams the file +// and holds one chunk. A repeated read of a chunk is safe. +func Get(ctx context.Context, h Host, t *contract.ConnectorTransfer) io.Reader { + return &pull{ctx: ctx, h: h, t: t} +} + +type pull struct { + ctx context.Context + h Host + t *contract.ConnectorTransfer + off int64 + buf []byte + eof bool +} + +func (p *pull) Read(b []byte) (int, error) { + if p.h == nil { + return 0, ErrNoHost + } + for len(p.buf) == 0 { + if p.eof { + return 0, io.EOF + } + var res contract.ConnectorFileGetResult + var err error + for try := 0; try < transferTries; try++ { + err = p.h.Do(p.ctx, p.t.Handle, contract.ConnectorFileGetOp, contract.ConnectorFileGet{Handle: p.t.Handle, Offset: p.off, Length: contract.TransferChunkBytes}, &res) + if err == nil || p.ctx.Err() != nil { + break + } + } + if err != nil { + return 0, err + } + p.buf, p.eof = res.Data, res.EOF + p.off += int64(len(res.Data)) + if len(res.Data) == 0 && !res.EOF { + return 0, errors.New("the host sent an empty chunk before the end of the file") + } + } + n := copy(b, p.buf) + p.buf = p.buf[n:] + return n, nil +} + +// Sniff returns a reader that gives the same bytes as r and the first bytes +// of r, at most n, for a platform that decides the type of a file from them. +func Sniff(r io.Reader, n int) (io.Reader, []byte) { + br := bufio.NewReaderSize(r, max(n, 16)) + head, _ := br.Peek(n) + return br, head +} + +// MediaStream is a file that a platform streams to the add-on. +type MediaStream struct { + Name, MIME string + // Size is the size the platform states, or zero when it states none. + Size int64 + // Body is the stream of bytes; Serve closes it. + Body io.ReadCloser +} + +// StreamReader is a Platform that streams the files of messages for the op +// media when the host gives a handle. max is the bound of the handle: the +// platform should refuse a larger file as soon as it knows the size. A +// Platform that does not implement it keeps the inline path. +type StreamReader interface { + MediaStream(ctx context.Context, token, ref string, max int64) (MediaStream, error) +} + +// StreamSender is a Platform that sends a file for the op send_file from a +// stream. body is the file of req.Transfer: t.Size bytes. Serve builds a new +// body for every attempt. +type StreamSender interface { + SendStream(ctx context.Context, req contract.ConnectorRequest, body io.Reader) (string, error) +} diff --git a/addons/internal/connector/transfer_test.go b/addons/internal/connector/transfer_test.go new file mode 100644 index 0000000000..dc857b1e04 --- /dev/null +++ b/addons/internal/connector/transfer_test.go @@ -0,0 +1,163 @@ +// SPDX-License-Identifier: MIT +package connector_test + +import ( + "bytes" + "context" + "errors" + "io" + "strings" + + "github.com/mudler/LocalAI/addons/internal/connector" + "github.com/mudler/LocalAI/pkg/addons/contract" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// memHost is the host side of a handle in memory: it keeps what was put, and +// serves what is to be got. fail makes the next calls fail. +type memHost struct { + got bytes.Buffer + final bool + src []byte + fail int + max int64 + tooLarge bool + calls int +} + +func (h *memHost) Do(_ context.Context, _, op string, args, out any) error { + h.calls++ + if h.fail > 0 { + h.fail-- + return errors.New("network") + } + switch a := args.(type) { + case contract.ConnectorFilePut: + if h.tooLarge || (h.max > 0 && int64(h.got.Len()+len(a.Data)) > h.max) { + return &contract.HostError{Code: contract.HostTooLarge, Message: "too large"} + } + if a.Offset == int64(h.got.Len()) { + h.got.Write(a.Data) + h.final = a.Final + } + *(out.(*contract.ConnectorFilePutResult)) = contract.ConnectorFilePutResult{Received: int64(h.got.Len())} + case contract.ConnectorFileGet: + end := min(a.Offset+a.Length, int64(len(h.src))) + *(out.(*contract.ConnectorFileGetResult)) = contract.ConnectorFileGetResult{Data: h.src[a.Offset:end], EOF: end == int64(len(h.src))} + } + return nil +} + +var _ = Describe("file transfer by handle", func() { + tr := &contract.ConnectorTransfer{Handle: "h", MaxBytes: 1 << 30} + + It("puts a stream in chunks and ends it, also when its size is a multiple of the chunk", func() { + for _, n := range []int{1, contract.TransferChunkBytes - 1, contract.TransferChunkBytes, 3*contract.TransferChunkBytes + 7} { + h := &memHost{} + data := bytes.Repeat([]byte{'x'}, n) + sent, err := connector.Put(context.Background(), h, tr, bytes.NewReader(data)) + Expect(err).NotTo(HaveOccurred()) + Expect(sent).To(BeEquivalentTo(n)) + Expect(h.got.Bytes()).To(Equal(data)) + Expect(h.final).To(BeTrue()) + } + }) + + It("sends a chunk again after a failed call, once the host says where it is", func() { + h := &memHost{fail: 2} + _, err := connector.Put(context.Background(), h, tr, strings.NewReader("hello")) + Expect(err).NotTo(HaveOccurred()) + Expect(h.got.String()).To(Equal("hello")) + h = &memHost{fail: 3} + _, err = connector.Put(context.Background(), h, tr, strings.NewReader("hello")) + Expect(err).To(HaveOccurred(), "three tries, then the transfer fails") + }) + + It("reports a file over the bound of the handle as too large", func() { + _, err := connector.Put(context.Background(), &memHost{tooLarge: true}, tr, strings.NewReader("hello")) + Expect(err).To(MatchError(connector.ErrTooLarge)) + }) + + It("needs a host", func() { + _, err := connector.Put(context.Background(), nil, tr, strings.NewReader("x")) + Expect(err).To(MatchError(connector.ErrNoHost)) + _, err = io.ReadAll(connector.Get(context.Background(), nil, tr)) + Expect(err).To(MatchError(connector.ErrNoHost)) + }) + + It("gets a file chunk by chunk as a reader", func() { + src := bytes.Repeat([]byte("abc"), contract.TransferChunkBytes) + h := &memHost{src: src} + got, err := io.ReadAll(connector.Get(context.Background(), h, tr)) + Expect(err).NotTo(HaveOccurred()) + Expect(got).To(Equal(src)) + Expect(h.calls).To(Equal(3), "one call per chunk") + }) +}) + +// streaming is a Platform that streams. +type streaming struct { + script + body string + size int64 + upload []byte + gotName string +} + +func (s *streaming) MediaStream(_ context.Context, _, _ string, _ int64) (connector.MediaStream, error) { + return connector.MediaStream{Name: "../a.txt", MIME: "text/plain", Size: s.size, Body: io.NopCloser(strings.NewReader(s.body))}, nil +} + +func (s *streaming) SendStream(_ context.Context, req contract.ConnectorRequest, body io.Reader) (string, error) { + s.upload, _ = io.ReadAll(body) + s.gotName = req.Transfer.Name + return "f1", nil +} + +var _ = Describe("the connector route with a handle", func() { + serve := func(p connector.Platform, h connector.Host, req contract.ConnectorRequest) (contract.ConnectorResponse, error) { + req.Kind, req.Token = "chat", "t" + return connector.ServeWith(context.Background(), p, h, req) + } + + It("streams the file of an op media to the host and answers with the name only", func() { + h := &memHost{} + out, err := serve(&streaming{body: "hello", size: 5}, h, contract.ConnectorRequest{Op: contract.ConnectorOpMedia, Media: "m", Transfer: &contract.ConnectorTransfer{Handle: "h", MaxBytes: 100}}) + Expect(err).NotTo(HaveOccurred()) + Expect(out.Error).To(BeNil()) + Expect(out.File).To(BeNil()) + Expect(out.Received).To(Equal(&contract.ConnectorReceived{Name: "a.txt", MIME: "text/plain"})) + Expect(h.got.String()).To(Equal("hello")) + }) + + It("refuses a file the platform says is over the bound, before any byte moves", func() { + h := &memHost{} + out, _ := serve(&streaming{body: "hello", size: 500}, h, contract.ConnectorRequest{Op: contract.ConnectorOpMedia, Media: "m", Transfer: &contract.ConnectorTransfer{Handle: "h", MaxBytes: 100}}) + Expect(out.Error.Code).To(Equal(contract.ConnectorTooLarge)) + Expect(h.calls).To(BeZero()) + }) + + It("stops a platform that states a small size and sends more", func() { + h := &memHost{max: 100} + out, _ := serve(&streaming{body: strings.Repeat("x", 500), size: 5}, h, contract.ConnectorRequest{Op: contract.ConnectorOpMedia, Media: "m", Transfer: &contract.ConnectorTransfer{Handle: "h", MaxBytes: 100}}) + Expect(out.Error.Code).To(Equal(contract.ConnectorTooLarge)) + }) + + It("sends the file of an op send_file from the host's chunks", func() { + p := &streaming{} + h := &memHost{src: []byte("report body")} + out, err := serve(p, h, contract.ConnectorRequest{Op: contract.ConnectorOpSendFile, Chat: "c", Transfer: &contract.ConnectorTransfer{Handle: "h", Name: "r.txt", MIME: "text/plain", Size: 11}}) + Expect(err).NotTo(HaveOccurred()) + Expect(out.MessageID).To(Equal("f1")) + Expect(string(p.upload)).To(Equal("report body")) + Expect(p.gotName).To(Equal("r.txt")) + }) + + It("refuses a handle on a platform that cannot stream, and a name that is not safe", func() { + _, err := serve(&script{}, &memHost{}, contract.ConnectorRequest{Op: contract.ConnectorOpMedia, Media: "m", Transfer: &contract.ConnectorTransfer{Handle: "h", MaxBytes: 1}}) + Expect(err).To(MatchError(contract.ErrRefused)) + _, err = serve(&streaming{}, &memHost{}, contract.ConnectorRequest{Op: contract.ConnectorOpSendFile, Transfer: &contract.ConnectorTransfer{Handle: "h", Name: "../x", Size: 1}}) + Expect(err).To(MatchError(contract.ErrRefused)) + }) +}) diff --git a/core/application/addon_gateway.go b/core/application/addon_gateway.go index 373449f28d..cf751a6b63 100644 --- a/core/application/addon_gateway.go +++ b/core/application/addon_gateway.go @@ -302,7 +302,7 @@ func (a *Application) startAddonGateway() error { if mode == auth.AddonModeLocalOwner { xlog.Warn(localOwnerWarning) } - g := &gateway.Gateway{Limits: c.Limits, Manager: a.addonManager, ConnectorAuthority: connectorAuthority{a}, Revalidate: auth.AddonRevalidate(a.authDB, mode)} + g := &gateway.Gateway{Transfers: a.transfers(), Limits: c.Limits, Manager: a.addonManager, ConnectorAuthority: connectorAuthority{a}, Revalidate: auth.AddonRevalidate(a.authDB, mode)} if cfg.InstanceID != cfg.OwnerID { g.Forward = carrier a.addonGateway = g diff --git a/core/application/addon_policy_reader.go b/core/application/addon_policy_reader.go index cae940d7fb..614f2c6ef4 100644 --- a/core/application/addon_policy_reader.go +++ b/core/application/addon_policy_reader.go @@ -68,6 +68,16 @@ func (a *Application) studioToolPins(ctx context.Context) (map[string]string, er // the owner policy. Without add-ons there is no owner policy and no connector // to move a file, so the default stands. func (a *Application) studioConnectorFilesCeiling(ctx context.Context) (int64, error) { + n, err := a.policyFilesCeiling(ctx) + if err != nil { + return 0, err + } + // A file that a sender shares becomes a file of the chat, so the ceiling + // never passes what one file of a workspace may be. + return min(n, a.studioUploadLimit()), nil +} + +func (a *Application) policyFilesCeiling(ctx context.Context) (int64, error) { if err := ctx.Err(); err != nil { return 0, err } diff --git a/core/application/application.go b/core/application/application.go index 3f5d35d6bb..e6caa36081 100644 --- a/core/application/application.go +++ b/core/application/application.go @@ -26,6 +26,7 @@ import ( "github.com/mudler/LocalAI/core/services/addons/backendlib" "github.com/mudler/LocalAI/core/services/addons/catalog" "github.com/mudler/LocalAI/core/services/addons/gateway" + "github.com/mudler/LocalAI/core/services/addons/hostops" "github.com/mudler/LocalAI/core/services/addons/hostsettings" "github.com/mudler/LocalAI/core/services/addons/packages" "github.com/mudler/LocalAI/core/services/addons/pins" @@ -113,9 +114,12 @@ type Application struct { backendLibrary *backendlib.Installer addonManager *addons.Manager addonHostOps *addonruntime.HostOps - addonDataRoot string // parent of the private data directory of each add-on; empty when none runs here - addonsProbe addonsProbeState - modelsProbe modelsProbeState + // addonTransfers holds the handles of connector file transfers; see transfers. + addonTransfers *hostops.Transfers + addonTransfersOnce sync.Once + addonDataRoot string // parent of the private data directory of each add-on; empty when none runs here + addonsProbe addonsProbeState + modelsProbe modelsProbeState // libraryStore is the library sources file when no add-on owner runs here. libraryStore *catalog.Sources diff --git a/core/application/studio_connectors.go b/core/application/studio_connectors.go index 91a2b79e0f..9b17872ca7 100644 --- a/core/application/studio_connectors.go +++ b/core/application/studio_connectors.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "time" "github.com/mudler/xlog" @@ -63,7 +64,16 @@ func (t connectorTransport) Call(ctx context.Context, scope studio.ConnectorCall // function as a message the manager pulls itself. A dedicated add-on owner // refuses it instead. func (a *Application) RegisterAddonConnector() error { - if a.addonHostOps == nil || a.studioConnectors == nil { + if a.addonHostOps == nil { + return nil + } + // The two file operations serve every connector add-on that declares + // files.transfer; the handles come with the connector calls. They are + // registered wherever add-ons run, also on a dedicated owner. + if err := hostops.RegisterConnectorTransfer(a.addonHostOps.Broker, a.transfers()); err != nil { + return err + } + if a.studioConnectors == nil { return nil } if a.ownsAddonsOnly() { @@ -130,3 +140,17 @@ func (a *Application) startStudioConnectors(svc *studio.Service) { a.studioConnectors, a.studioConnectorsStop = c, cancel go c.Run(ctx) } + +// transfers is the table of live connector file handles of this process. The +// gateway checks and ends them; the broker serves the bytes. Its blob store is +// the one of the workspace, set when the files start. +func (a *Application) transfers() *hostops.Transfers { + a.addonTransfersOnce.Do(func() { + t := hostops.NewTransfers(nil) + t.Audit = func(o hostops.TransferOutcome) { + xlog.Info("Connector file transfer", "direction", o.Direction, "binding", o.BindingID, "bytes", o.Bytes, "result", o.Result, "duration", o.Duration.Round(time.Millisecond)) + } + a.addonTransfers = t + }) + return a.addonTransfers +} diff --git a/core/application/studio_files.go b/core/application/studio_files.go index 82d2641acd..40241708ce 100644 --- a/core/application/studio_files.go +++ b/core/application/studio_files.go @@ -108,6 +108,7 @@ func (a *Application) initStudioFilesWithS3(newS3 func(context.Context, storage. if err != nil { return stop("workspace blob store", err, "Files are off: their storage could not be opened.") } + a.transfers().UseBlobs(blobs) ws, closer, err := a.openWorkspaceIndex(root, blobs) if err != nil { return stop("workspace index", err, "Files are off: the workspace could not be started.") diff --git a/core/services/addons/gateway/connectors.go b/core/services/addons/gateway/connectors.go index 5f10306953..aa5ec1f32f 100644 --- a/core/services/addons/gateway/connectors.go +++ b/core/services/addons/gateway/connectors.go @@ -13,6 +13,7 @@ import ( "time" "github.com/mudler/LocalAI/core/services/addons" + "github.com/mudler/LocalAI/core/services/addons/hostops" "github.com/mudler/LocalAI/core/services/studio/connectors" "github.com/mudler/LocalAI/pkg/addons/contract" ) @@ -51,7 +52,7 @@ func (g *Gateway) connectorSource() ConnectorSource { func (g *Gateway) ConnectorKinds(ctx context.Context) ([]connectors.Kind, error) { if g.Forward != nil { var out []connectors.Kind - err := g.Forward.connectorRequest(ctx, "GET", "/addons/connectors/kinds", nil, &out) + err := g.Forward.connectorRequest(ctx, "GET", "/addons/connectors/kinds", nil, &out, connectorTimeout) return out, err } return g.localConnectorKinds(ctx) @@ -77,13 +78,17 @@ func (g *Gateway) ConnectorCall(ctx context.Context, s connectors.ConnectorCallS req.Token = "" } var out contract.ConnectorResponse - err := g.Forward.connectorRequest(ctx, "POST", "/addons/connectors/call", connectorCall{s, req}, &out) + err := g.Forward.connectorRequest(ctx, "POST", "/addons/connectors/call", connectorCall{s, req}, &out, req.CallTimeout()) return out, err } return g.localConnectorCall(ctx, s, req) } func (g *Gateway) localConnectorCall(ctx context.Context, s connectors.ConnectorCallScope, req contract.ConnectorRequest) (contract.ConnectorResponse, error) { - ctx, cancel := context.WithTimeout(ctx, connectorTimeout) + timeout := connectorTimeout + if req.Transfer != nil { + timeout = req.CallTimeout() + } + ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() source := g.connectorSource() if source == nil || g.ConnectorAuthority == nil { @@ -109,7 +114,18 @@ func (g *Gateway) localConnectorCall(ctx context.Context, s connectors.Connector } return contract.ConnectorResponse{}, ErrForbidden } + if req.Transfer != nil && (kinds[i].Files == nil || !kinds[i].Files.Transfer) { + // A handle only goes to an add-on that declared that it takes one. + return contract.ConnectorResponse{}, ErrForbidden + } + end, err := g.beginTransfer(s, req) + if err != nil { + return contract.ConnectorResponse{}, err + } out, err := source.Connector(ctx, s.AddonID, req) + if end != nil { + out = finishTransfer(end, out) + } if err != nil { // A definite refusal of the add-on keeps its type; anything else could // not tell, so the caller treats it as an outage, with the cause kept. @@ -120,8 +136,8 @@ func (g *Gateway) localConnectorCall(ctx context.Context, s connectors.Connector } return out, nil } -func (c *Carrier) connectorRequest(ctx context.Context, method, path string, body any, out any) error { - ctx, cancel := context.WithTimeout(ctx, connectorTimeout) +func (c *Carrier) connectorRequest(ctx context.Context, method, path string, body any, out any, timeout time.Duration) error { + ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() var raw []byte if body != nil { @@ -136,7 +152,13 @@ func (c *Carrier) connectorRequest(ctx context.Context, method, path string, bod return ErrUnavailable } req.Header.Set("Content-Type", "application/json") - resp, err := c.client.Do(req) + // A call that moves a file runs longer than the client's own timeout; the + // context carries the bound. + client := c.client + if timeout > connectorTimeout { + client = c.long + } + resp, err := client.Do(req) if err != nil { return ErrUnavailable } @@ -172,6 +194,10 @@ func (g *Gateway) serveConnectors(w http.ResponseWriter, r *http.Request) bool { typedError(w, ErrForbidden) return true } + if call.Request.Transfer != nil { + // The server's write timeout is sized for short calls. + _ = http.NewResponseController(w).SetWriteDeadline(time.Now().Add(contract.ConnectorTransferTimeout + time.Minute)) + } out, err := g.localConnectorCall(r.Context(), call.Scope, call.Request) if err != nil { typedError(w, err) @@ -193,3 +219,51 @@ func decodeConnector(r io.Reader, out any) error { } return decode(bytes.NewReader(raw), out) } + +// beginTransfer registers the handle of a call that carries one, on the +// process that runs the add-on. The frontend made the handle and checked the +// chat's access; this process binds it to the add-on of the call, so no other +// add-on can use it, and to the content or the bound the scope names. +func (g *Gateway) beginTransfer(s connectors.ConnectorCallScope, req contract.ConnectorRequest) (func() hostops.TransferOutcome, error) { + if req.Transfer == nil { + return nil, nil + } + grant := hostops.TransferGrant{Handle: req.Transfer.Handle, AddonID: s.AddonID, BindingID: s.BindingID} + switch req.Op { + case contract.ConnectorOpMedia: + grant.Direction, grant.MaxBytes = hostops.TransferIn, req.Transfer.MaxBytes + case contract.ConnectorOpSendFile: + if s.Content == nil || req.File != nil { + return nil, ErrForbidden + } + grant.Direction, grant.SHA, grant.Size = hostops.TransferOut, s.Content.SHA, s.Content.Size + default: + return nil, ErrForbidden + } + if g.Transfers == nil { + return nil, ErrUnavailable + } + end, err := g.Transfers.Begin(grant) + if err != nil { + if errors.Is(err, hostops.ErrBadRequest) || errors.Is(err, hostops.ErrConflict) { + return nil, ErrForbidden + } + return nil, ErrUnavailable + } + return end, nil +} + +// finishTransfer ends the handle after the add-on answered. For an inbound +// file it writes the size and the hash from the host's own record into the +// answer; an add-on that did not finish the file gets no Received. +func finishTransfer(end func() hostops.TransferOutcome, out contract.ConnectorResponse) contract.ConnectorResponse { + o := end() + if out.Received != nil { + if o.Direction != hostops.TransferIn || o.Result != hostops.TransferDone || o.SHA == "" { + out.Received = nil + } else { + out.Received.Size, out.Received.SHA256 = o.Bytes, o.SHA + } + } + return out +} diff --git a/core/services/addons/gateway/connectors_test.go b/core/services/addons/gateway/connectors_test.go index 7978246865..d9029c1b09 100644 --- a/core/services/addons/gateway/connectors_test.go +++ b/core/services/addons/gateway/connectors_test.go @@ -69,7 +69,7 @@ var _ = Describe("Forwarded connector answers", func() { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(status) })) defer server.Close() c := &Carrier{config: CarrierConfig{OwnerAddress: server.URL}, client: server.Client()} - return c.connectorRequest(context.Background(), "GET", "/addons/connectors/kinds", nil, &[]connectors.Kind{}) + return c.connectorRequest(context.Background(), "GET", "/addons/connectors/kinds", nil, &[]connectors.Kind{}, connectorTimeout) } It("keeps a refusal apart from an outage", func() { diff --git a/core/services/addons/gateway/connectors_transfer_test.go b/core/services/addons/gateway/connectors_transfer_test.go new file mode 100644 index 0000000000..c9683de21f --- /dev/null +++ b/core/services/addons/gateway/connectors_transfer_test.go @@ -0,0 +1,111 @@ +// SPDX-License-Identifier: MIT +package gateway + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "net/http" + "net/http/httptest" + + "github.com/mudler/LocalAI/core/services/addons" + "github.com/mudler/LocalAI/core/services/addons/hostops" + "github.com/mudler/LocalAI/core/services/studio/connectors" + "github.com/mudler/LocalAI/core/services/studio/workspace/blob" + "github.com/mudler/LocalAI/pkg/addons/contract" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// handleSource is an add-on that takes handles: it runs fn during the call. +type handleSource struct { + transfer bool + fn func(contract.ConnectorRequest) contract.ConnectorResponse +} + +func (s handleSource) ConnectorKinds(context.Context) ([]addons.ConnectorKind, error) { + k := contract.ConnectorKind{ID: "slack", Delivery: contract.DeliveryPull} + if s.transfer { + k.Files = &contract.ConnectorFiles{Receive: &contract.FileAbility{MaxBytes: 1 << 30}, Transfer: true} + } + return []addons.ConnectorKind{{AddonID: "slack", ConnectorKind: k}}, nil +} + +func (s handleSource) Connector(_ context.Context, _ string, r contract.ConnectorRequest) (contract.ConnectorResponse, error) { + return s.fn(r), nil +} + +var _ = Describe("Connector calls that carry a file handle", func() { + scope := connectors.ConnectorCallScope{AddonID: "slack", Kind: "slack", BindingID: "b1"} + var ( + t *hostops.Transfers + host http.Handler + ) + BeforeEach(func() { + blobs, err := blob.NewFS(GinkgoT().TempDir()) + Expect(err).NotTo(HaveOccurred()) + t = hostops.NewTransfers(blobs) + b := hostops.New(hostops.NewInFlight(), func(string, string) bool { return true }, contract.DefaultLimits()) + Expect(hostops.RegisterConnectorTransfer(b, t)).To(Succeed()) + host = b.Handler("slack", "k", []string{"connector."}) + }) + put := func(h string, off int64, data []byte, final bool) int { + raw, _ := json.Marshal(contract.ConnectorFilePut{Handle: h, Offset: off, Data: data, Final: final}) + body, _ := json.Marshal(contract.HostRequest{OperationID: "o", Op: contract.ConnectorFilePutOp, Arguments: raw}) + req := httptest.NewRequest(http.MethodPost, "/host", bytes.NewReader(body)) + req.Header.Set(contract.HostCredentialHeader, "k") + rec := httptest.NewRecorder() + host.ServeHTTP(rec, req) + return rec.Code + } + media := func(s handleSource, tr *hostops.Transfers) (contract.ConnectorResponse, error) { + g := &Gateway{Connectors: s, ConnectorAuthority: allowAllConnectors{}, Transfers: tr} + return g.ConnectorCall(context.Background(), scope, contract.ConnectorRequest{Op: contract.ConnectorOpMedia, Kind: "slack", Token: "t", + Transfer: &contract.ConnectorTransfer{Handle: "hh", MaxBytes: 100}}) + } + + It("registers the handle for the call and gives the host's size and hash, not the add-on's", func() { + src := handleSource{transfer: true, fn: func(r contract.ConnectorRequest) contract.ConnectorResponse { + Expect(t.Live()).To(Equal(1)) + Expect(put(r.Transfer.Handle, 0, []byte("hello"), true)).To(Equal(http.StatusOK)) + return contract.ConnectorResponse{Received: &contract.ConnectorReceived{Name: "a.txt", Size: 999, SHA256: "forged"}} + }} + out, err := media(src, t) + Expect(err).NotTo(HaveOccurred()) + sum := sha256.Sum256([]byte("hello")) + Expect(out.Received.SHA256).To(Equal(hex.EncodeToString(sum[:]))) + Expect(out.Received.Size).To(BeEquivalentTo(5)) + Expect(t.Live()).To(BeZero(), "the handle ends with the call") + }) + + It("drops an answer for a file the add-on did not finish", func() { + src := handleSource{transfer: true, fn: func(r contract.ConnectorRequest) contract.ConnectorResponse { + Expect(put(r.Transfer.Handle, 0, []byte("half"), false)).To(Equal(http.StatusOK)) + return contract.ConnectorResponse{Received: &contract.ConnectorReceived{Name: "a.txt", Size: 4, SHA256: "forged"}} + }} + out, err := media(src, t) + Expect(err).NotTo(HaveOccurred()) + Expect(out.Received).To(BeNil()) + Expect(t.Live()).To(BeZero()) + }) + + It("refuses a handle for a kind that did not declare transfer, and without a table", func() { + src := handleSource{fn: func(contract.ConnectorRequest) contract.ConnectorResponse { return contract.ConnectorResponse{} }} + _, err := media(src, t) + Expect(err).To(MatchError(ErrForbidden)) + src.transfer = true + _, err = media(src, nil) + Expect(err).To(MatchError(ErrUnavailable)) + }) + + It("refuses a handle on an op that moves no file, and an outbound handle with no content", func() { + src := handleSource{transfer: true, fn: func(contract.ConnectorRequest) contract.ConnectorResponse { return contract.ConnectorResponse{} }} + g := &Gateway{Connectors: src, ConnectorAuthority: allowAllConnectors{}, Transfers: t} + _, err := g.ConnectorCall(context.Background(), scope, contract.ConnectorRequest{Op: contract.ConnectorOpSend, Kind: "slack", Transfer: &contract.ConnectorTransfer{Handle: "x"}}) + Expect(err).To(MatchError(ErrForbidden)) + _, err = g.ConnectorCall(context.Background(), scope, contract.ConnectorRequest{Op: contract.ConnectorOpSendFile, Kind: "slack", Transfer: &contract.ConnectorTransfer{Handle: "x", Size: 1}}) + Expect(err).To(MatchError(ErrForbidden), "the frontend names the content in the scope") + }) +}) diff --git a/core/services/addons/gateway/gateway.go b/core/services/addons/gateway/gateway.go index d0e1ccf6cc..e5c2000201 100644 --- a/core/services/addons/gateway/gateway.go +++ b/core/services/addons/gateway/gateway.go @@ -11,6 +11,7 @@ import ( "sync" "github.com/mudler/LocalAI/core/services/addons" + "github.com/mudler/LocalAI/core/services/addons/hostops" "github.com/mudler/LocalAI/pkg/addons/contract" ) @@ -25,6 +26,10 @@ var ErrNotSent = fmt.Errorf("the add-on is not running now and nothing was sent: // Gateway deliberately admits lifecycle commands only until resource admission // and proposal confirmation are installed. Installation admin is not data access. type Gateway struct { + // Transfers holds the handles of connector file transfers on the process + // that runs the add-ons. Without it a call that carries a handle is + // unavailable. + Transfers *hostops.Transfers Connectors ConnectorSource ConnectorAuthority ConnectorAuthority // HostSettings is owner-only durable policy; frontends forward, never fall back. diff --git a/core/services/addons/hostops/admission.go b/core/services/addons/hostops/admission.go index e9c42beebe..e0805be710 100644 --- a/core/services/addons/hostops/admission.go +++ b/core/services/addons/hostops/admission.go @@ -54,5 +54,5 @@ func Requests(req *contract.HostOperations, registered, grants []string) Request // must not mistake that startup ordering for unsupported operations. The broker // still requires a registered handler and revalidates every request's authority. func SupportedOperations() []string { - return strings.Fields(`finetune.start finetune.get finetune.list finetune.checkpoints finetune.dataset finetune.export finetune.progress finetune.stop quantize.start quantize.get quantize.list quantize.import quantize.progress quantize.stop workspace.materialize workspace.commit workspace.export workspace.import workspace.changes workspace.admin egress.check egress.audit connector.inbound`) + return strings.Fields(`finetune.start finetune.get finetune.list finetune.checkpoints finetune.dataset finetune.export finetune.progress finetune.stop quantize.start quantize.get quantize.list quantize.import quantize.progress quantize.stop workspace.materialize workspace.commit workspace.export workspace.import workspace.changes workspace.admin egress.check egress.audit connector.inbound connector.file.put connector.file.get`) } diff --git a/core/services/addons/hostops/admission_test.go b/core/services/addons/hostops/admission_test.go index a03360ca3e..ebc71d108d 100644 --- a/core/services/addons/hostops/admission_test.go +++ b/core/services/addons/hostops/admission_test.go @@ -37,6 +37,7 @@ func registeredBuiltinOperations() []string { })).To(Succeed()) Expect(RegisterEgress(b, &fakeEgress{})).To(Succeed()) Expect(RegisterConnector(b, admissionInbox{})).To(Succeed()) + Expect(RegisterConnectorTransfer(b, NewTransfers(nil))).To(Succeed()) // Each registration error is checked above, so a duplicate cannot be hidden // by the broker's map. Ops includes dynamic progress/stop and inbound specs. return b.Ops() diff --git a/core/services/addons/hostops/broker.go b/core/services/addons/hostops/broker.go index 39d9c1f866..ec6433160a 100644 --- a/core/services/addons/hostops/broker.go +++ b/core/services/addons/hostops/broker.go @@ -327,6 +327,10 @@ func (b *Broker) serveInbound(w http.ResponseWriter, r *http.Request, spec Spec, status, code, msg = http.StatusForbidden, contract.HostScope, "not allowed" case errors.Is(err, ErrNotFound): status, code, msg = http.StatusNotFound, contract.HostNotFound, "not found" + case errors.Is(err, ErrTransferTooLarge): + status, code, msg = http.StatusRequestEntityTooLarge, contract.HostTooLarge, "the file is larger than the limit" + case errors.Is(err, ErrConflict): + status, code, msg = http.StatusConflict, contract.HostConflict, "conflict" case errors.Is(err, ErrBadRequest): status, code, msg = http.StatusBadRequest, contract.HostBadRequest, err.Error() } diff --git a/core/services/addons/hostops/transfer.go b/core/services/addons/hostops/transfer.go new file mode 100644 index 0000000000..d7abdf586b --- /dev/null +++ b/core/services/addons/hostops/transfer.go @@ -0,0 +1,450 @@ +// SPDX-License-Identifier: MIT +package hostops + +import ( + "bytes" + "context" + "crypto/subtle" + "encoding/json" + "errors" + "fmt" + "io" + "sync" + "time" + + "github.com/mudler/LocalAI/core/services/studio/workspace" + "github.com/mudler/LocalAI/core/services/studio/workspace/blob" + "github.com/mudler/LocalAI/pkg/addons/contract" +) + +// Direction of a transfer, from the host. +const ( + // TransferIn moves a file from an add-on to the host: the add-on puts. + TransferIn = "in" + // TransferOut moves a file from the host to an add-on: the add-on gets. + TransferOut = "out" +) + +// Results of a transfer, as the audit shows them. +const ( + TransferDone = "done" + TransferTooLarge = "too_large" + TransferExpired = "expired" + TransferRefused = "refused" + TransferIncomplete = "incomplete" +) + +// Default times of a handle. +const ( + // DefaultTransferIdle ends a handle that has seen no call for this long. + DefaultTransferIdle = 2 * time.Minute + // DefaultTransferLife ends a handle this long after it was made, whatever + // it does. A connector call is shorter than this. + DefaultTransferLife = time.Hour +) + +// ErrTransferTooLarge means a put went past the bound of its handle. +var ErrTransferTooLarge = errors.New("the file is larger than the limit of the handle") + +// TransferGrant is what the host that issued a handle records about it. The +// owner of the add-ons registers it when a connector call that carries a +// ConnectorTransfer arrives. +type TransferGrant struct { + Handle string + Direction string + // AddonID is the only add-on that may use the handle. + AddonID string + // BindingID is for the audit. + BindingID string + // MaxBytes bounds an inbound file. + MaxBytes int64 + // SHA and Size name the outbound content, a blob of the workspace store. + SHA string + Size int64 +} + +// TransferOutcome is what a handle ended with. It never holds a name or any +// content. +type TransferOutcome struct { + Direction string + BindingID string + Bytes int64 + Result string + Duration time.Duration + // SHA is the hash of a finished inbound file. + SHA string +} + +// FileScan is the hook after the last byte of an inbound file. An error +// rejects the file. It is a no-op until an owner sets one. +type FileScan func(ctx context.Context, sha string, size int64) error + +// Transfers is the table of live handles of one process, and the bytes behind +// them. Inbound bytes stream into the blob store of the workspace, which in a +// cluster is the shared object store, so no node needs a shared disk. Outbound +// bytes stream out of the same store. A file is never held in memory: a put +// or a get moves one chunk. +type Transfers struct { + // Blobs is the blob store of the workspace. UseBlobs sets it when the + // files start, which can be after the table is made. + Blobs blob.Store + Scan FileScan + // Audit gets the outcome of every handle when it ends. Nil: none. + Audit func(TransferOutcome) + // Idle and Life default to DefaultTransferIdle and DefaultTransferLife. + Idle, Life time.Duration + + mu sync.Mutex + m map[string]*xfer +} + +type xfer struct { + t *Transfers + g TransferGrant + start time.Time + + mu sync.Mutex // one call at a time per handle + received int64 + pw *io.PipeWriter + done chan struct{} // closed when the blob Put of an inbound handle returned + sha string + size int64 + putErr error + final bool + rd io.ReadSeekCloser + cancel context.CancelFunc + idle *time.Timer + life *time.Timer + ended bool + result string +} + +// NewTransfers builds a table over the blob store of the workspace. +func NewTransfers(blobs blob.Store) *Transfers { + return &Transfers{Blobs: blobs, m: map[string]*xfer{}} +} + +// UseBlobs sets the blob store. Handles that begin before it is set are +// refused as unavailable. +func (t *Transfers) UseBlobs(b blob.Store) { + t.mu.Lock() + t.Blobs = b + t.mu.Unlock() +} + +func (t *Transfers) store() blob.Store { + t.mu.Lock() + defer t.mu.Unlock() + return t.Blobs +} + +func (t *Transfers) idle() time.Duration { + if t.Idle > 0 { + return t.Idle + } + return DefaultTransferIdle +} + +func (t *Transfers) life() time.Duration { + if t.Life > 0 { + return t.Life + } + return DefaultTransferLife +} + +// Begin registers a handle. The returned function ends it: it drops a file +// that is not finished and reports the outcome. For a finished inbound file the +// outcome holds the hash and the size. Calling it again returns the same +// outcome. A handle that is already registered is refused, so a handle is +// never used for two calls. +func (t *Transfers) Begin(g TransferGrant) (end func() TransferOutcome, err error) { + if t == nil || t.store() == nil { + return nil, errors.New("file transfer is not available here") + } + if g.Handle == "" || len(g.Handle) > contract.MaxTransferHandle || g.AddonID == "" { + return nil, fmt.Errorf("%w: bad handle", ErrBadRequest) + } + switch g.Direction { + case TransferIn: + if g.MaxBytes <= 0 { + return nil, fmt.Errorf("%w: no bound", ErrBadRequest) + } + case TransferOut: + if g.SHA == "" || g.Size <= 0 { + return nil, fmt.Errorf("%w: no content", ErrBadRequest) + } + default: + return nil, fmt.Errorf("%w: bad direction", ErrBadRequest) + } + x := &xfer{t: t, g: g, start: time.Now()} + t.mu.Lock() + if t.m == nil { + t.m = map[string]*xfer{} + } + if _, dup := t.m[g.Handle]; dup { + t.mu.Unlock() + return nil, fmt.Errorf("%w: handle in use", ErrConflict) + } + t.m[g.Handle] = x + t.mu.Unlock() + x.idle = time.AfterFunc(t.idle(), func() { x.finish(TransferExpired) }) + x.life = time.AfterFunc(t.life(), func() { x.finish(TransferExpired) }) + return func() TransferOutcome { return x.finish("") }, nil +} + +// Live reports how many handles exist; a test and the sweep use it. +func (t *Transfers) Live() int { + t.mu.Lock() + defer t.mu.Unlock() + return len(t.m) +} + +func (t *Transfers) lookup(addonID, handle string) (*xfer, error) { + t.mu.Lock() + x := t.m[handle] + t.mu.Unlock() + // The same answer for a handle that does not exist and one of another + // add-on: a handle cannot be probed. + if x == nil || subtle.ConstantTimeCompare([]byte(x.g.AddonID), []byte(addonID)) != 1 { + return nil, ErrNotFound + } + return x, nil +} + +// finish ends the handle once. why is the result when the handle did not end +// by its own end function. +func (x *xfer) finish(why string) TransferOutcome { + x.idle.Stop() + x.life.Stop() + // A put or a get in progress holds mu; an expiry must not wait for a + // stuck one, so it cancels first. + if x.cancel != nil { + x.cancel() + } + if x.pw != nil { + _ = x.pw.CloseWithError(context.Canceled) + } + x.mu.Lock() + defer x.mu.Unlock() + if x.ended { + return x.outcome() + } + x.ended = true + x.t.mu.Lock() + delete(x.t.m, x.g.Handle) + x.t.mu.Unlock() + if x.rd != nil { + _ = x.rd.Close() + } + if x.done != nil { + <-x.done // the blob Put drops its temp file when its reader fails + } + switch { + case x.result != "": + case why != "": + x.result = why + case x.g.Direction == TransferIn && !x.final: + x.result = TransferIncomplete + default: + x.result = TransferDone + } + o := x.outcome() + if x.t.Audit != nil { + x.t.Audit(o) + } + return o +} + +func (x *xfer) outcome() TransferOutcome { + o := TransferOutcome{Direction: x.g.Direction, BindingID: x.g.BindingID, Bytes: x.received, Result: x.result, Duration: time.Since(x.start)} + if x.g.Direction == TransferOut { + o.Bytes = x.size + } + if x.final && x.result != TransferTooLarge && x.result != TransferRefused { + o.SHA = x.sha + } + return o +} + +func (x *xfer) touch() { x.idle.Reset(x.t.idle()) } + +// put writes the next chunk of an inbound file. +func (x *xfer) put(ctx context.Context, p contract.ConnectorFilePut) (int64, error) { + if x.g.Direction != TransferIn { + return 0, ErrNotFound + } + x.mu.Lock() + defer x.mu.Unlock() + if x.ended || x.final { + return 0, ErrConflict + } + x.touch() + if len(p.Data) > contract.TransferChunkBytes { + return 0, fmt.Errorf("%w: chunk too large", ErrBadRequest) + } + if p.Offset != x.received { + // The add-on resumes where the host is. + return x.received, nil + } + if x.received+int64(len(p.Data)) > x.g.MaxBytes { + x.result = TransferTooLarge + x.abortLocked() + return x.received, ErrTransferTooLarge + } + if x.pw == nil { + pr, pw := io.Pipe() + bctx, cancel := context.WithCancel(context.Background()) + x.pw, x.cancel, x.done = pw, cancel, make(chan struct{}) + go func() { + defer close(x.done) + sha, size, err := x.t.store().Put(bctx, pr, x.g.MaxBytes) + x.sha, x.size, x.putErr = sha, size, err + _ = pr.CloseWithError(errors.New("closed")) + }() + } + if len(p.Data) > 0 { + if _, err := io.Copy(x.pw, bytes.NewReader(p.Data)); err != nil { + return x.received, x.failedLocked(err) + } + x.received += int64(len(p.Data)) + } + if !p.Final { + return x.received, nil + } + if x.received == 0 { + x.result = TransferRefused + x.abortLocked() + return 0, fmt.Errorf("%w: an empty file", ErrBadRequest) + } + _ = x.pw.Close() + <-x.done + switch { + case errors.Is(x.putErr, workspace.ErrTooLarge): + x.result = TransferTooLarge + return x.received, ErrTransferTooLarge + case x.putErr != nil: + return x.received, x.putErr + case x.size != x.received: + return x.received, errors.New("the stored size differs from the bytes received") + } + if x.t.Scan != nil { + if err := x.t.Scan(ctx, x.sha, x.size); err != nil { + x.result = TransferRefused + return x.received, fmt.Errorf("%w: the file was refused", ErrBadRequest) + } + } + x.final = true + return x.received, nil +} + +// failedLocked turns a failed write into the error of the blob store, which +// ended the pipe, so that a bound reached inside the store is too large. +func (x *xfer) failedLocked(err error) error { + <-x.done + if errors.Is(x.putErr, workspace.ErrTooLarge) { + x.result = TransferTooLarge + return ErrTransferTooLarge + } + if x.putErr != nil { + return x.putErr + } + return err +} + +func (x *xfer) abortLocked() { + if x.pw != nil { + _ = x.pw.CloseWithError(context.Canceled) + if x.cancel != nil { + x.cancel() + } + <-x.done + } +} + +// get reads the next chunk of an outbound file. +func (x *xfer) get(ctx context.Context, g contract.ConnectorFileGet) (contract.ConnectorFileGetResult, error) { + if x.g.Direction != TransferOut { + return contract.ConnectorFileGetResult{}, ErrNotFound + } + x.mu.Lock() + defer x.mu.Unlock() + if x.ended { + return contract.ConnectorFileGetResult{}, ErrConflict + } + x.touch() + if g.Offset < 0 || g.Offset > x.g.Size || g.Length < 1 || g.Length > contract.TransferChunkBytes { + return contract.ConnectorFileGetResult{}, fmt.Errorf("%w: offset or length", ErrBadRequest) + } + if x.rd == nil { + r, err := x.t.store().Open(ctx, x.g.SHA) + if err != nil { + return contract.ConnectorFileGetResult{}, err + } + n, err := r.Seek(0, io.SeekEnd) + if err != nil || n != x.g.Size { + _ = r.Close() + return contract.ConnectorFileGetResult{}, errors.New("the file changed since the handle was made") + } + x.rd, x.size = r, n + } + if _, err := x.rd.Seek(g.Offset, io.SeekStart); err != nil { + return contract.ConnectorFileGetResult{}, err + } + buf := make([]byte, min(g.Length, x.g.Size-g.Offset)) + n, err := io.ReadFull(x.rd, buf) + if err != nil && !errors.Is(err, io.ErrUnexpectedEOF) && !errors.Is(err, io.EOF) { + return contract.ConnectorFileGetResult{}, err + } + buf = buf[:n] + return contract.ConnectorFileGetResult{Data: buf, EOF: g.Offset+int64(n) >= x.g.Size}, nil +} + +// RegisterConnectorTransfer registers the two operations with which a +// connector add-on moves file bytes by handle. Both are inbound class: the +// credential of the live incarnation authorizes them, and the handle binds +// them to one connector call. The arguments name no user. +func RegisterConnectorTransfer(b *Broker, t *Transfers) error { + if t == nil { + return errors.New("connector file transfer needs a table") + } + decode := func(raw json.RawMessage, v any) error { + d := json.NewDecoder(bytes.NewReader(raw)) + d.DisallowUnknownFields() + if err := d.Decode(v); err != nil { + return ErrBadRequest + } + return nil + } + if err := b.Register(Spec{Op: contract.ConnectorFilePutOp, Scope: ScopeInbound, Timeout: 2 * time.Minute, Handler: func(ctx context.Context, c Call) (json.RawMessage, error) { + var p contract.ConnectorFilePut + if err := decode(c.Arguments, &p); err != nil { + return nil, err + } + x, err := t.lookup(c.AddonID, p.Handle) + if err != nil { + return nil, err + } + got, err := x.put(ctx, p) + if err != nil { + return nil, err + } + return json.Marshal(contract.ConnectorFilePutResult{Received: got}) + }}); err != nil { + return err + } + return b.Register(Spec{Op: contract.ConnectorFileGetOp, Scope: ScopeInbound, Timeout: 2 * time.Minute, Handler: func(ctx context.Context, c Call) (json.RawMessage, error) { + var g contract.ConnectorFileGet + if err := decode(c.Arguments, &g); err != nil { + return nil, err + } + x, err := t.lookup(c.AddonID, g.Handle) + if err != nil { + return nil, err + } + r, err := x.get(ctx, g) + if err != nil { + return nil, err + } + return json.Marshal(r) + }}) +} diff --git a/core/services/addons/hostops/transfer_test.go b/core/services/addons/hostops/transfer_test.go new file mode 100644 index 0000000000..ed88485956 --- /dev/null +++ b/core/services/addons/hostops/transfer_test.go @@ -0,0 +1,284 @@ +// SPDX-License-Identifier: MIT +package hostops + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "math/rand" + "net/http" + "net/http/httptest" + "path/filepath" + "runtime" + "time" + + "github.com/mudler/LocalAI/core/services/storage" + "github.com/mudler/LocalAI/core/services/studio/workspace/blob" + "github.com/mudler/LocalAI/pkg/addons/contract" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +type xferFixture struct { + t *Transfers + h http.Handler + outs []TransferOutcome + blobs blob.Store + dir string + shared *storage.FilesystemStore +} + +func newXferFixture() *xferFixture { + dir := GinkgoT().TempDir() + shared, err := storage.NewFilesystemStore(filepath.Join(dir, "objects")) + Expect(err).NotTo(HaveOccurred()) + // The owner node has its own cache disk; the object store is the only + // thing the nodes share. + blobs := blob.NewObjectStore(shared, filepath.Join(dir, "owner-cache")) + f := &xferFixture{blobs: blobs, dir: dir, shared: shared} + f.t = NewTransfers(blobs) + f.t.Audit = func(o TransferOutcome) { f.outs = append(f.outs, o) } + b := New(NewInFlight(), always, contract.DefaultLimits()) + Expect(RegisterConnectorTransfer(b, f.t)).To(Succeed()) + f.h = b.Handler("slack", "secret", []string{"connector."}) + return f +} + +// call posts one host operation as the add-on addon. +func (f *xferFixture) call(addon, op string, args any) (int, contract.HostResponse) { + raw, _ := json.Marshal(args) + body, _ := json.Marshal(contract.HostRequest{OperationID: "x", Op: op, Arguments: raw}) + req := httptest.NewRequest(http.MethodPost, "/host", bytes.NewReader(body)) + req.Header.Set(contract.HostCredentialHeader, "secret") + rec := httptest.NewRecorder() + h := f.h + if addon != "slack" { + b := New(NewInFlight(), always, contract.DefaultLimits()) + Expect(RegisterConnectorTransfer(b, f.t)).To(Succeed()) + h = b.Handler(addon, "secret", []string{"connector."}) + } + h.ServeHTTP(rec, req) + var out contract.HostResponse + _ = json.Unmarshal(rec.Body.Bytes(), &out) + return rec.Code, out +} + +func (f *xferFixture) put(handle string, off int64, data []byte, final bool) (int, int64) { + code, r := f.call("slack", contract.ConnectorFilePutOp, contract.ConnectorFilePut{Handle: handle, Offset: off, Data: data, Final: final}) + var res contract.ConnectorFilePutResult + if r.Result != nil { + _ = json.Unmarshal(r.Result, &res) + } + return code, res.Received +} + +func (f *xferFixture) get(handle string, off, n int64) (int, contract.ConnectorFileGetResult) { + code, r := f.call("slack", contract.ConnectorFileGetOp, contract.ConnectorFileGet{Handle: handle, Offset: off, Length: n}) + var res contract.ConnectorFileGetResult + if r.Result != nil { + _ = json.Unmarshal(r.Result, &res) + } + return code, res +} + +// chunks streams n bytes of a seeded sequence, chunk bytes at a time, and +// never holds more than one chunk. +func chunks(n, chunk int64, each func(off int64, b []byte, last bool)) string { + rng := rand.New(rand.NewSource(7)) + h := sha256.New() + buf := make([]byte, chunk) + for off := int64(0); off < n; off += chunk { + b := buf[:min(chunk, n-off)] + _, _ = rng.Read(b) + h.Write(b) + each(off, b, off+int64(len(b)) >= n) + } + return hex.EncodeToString(h.Sum(nil)) +} + +func heap() uint64 { + var m runtime.MemStats + runtime.GC() + runtime.ReadMemStats(&m) + return m.HeapAlloc +} + +var _ = Describe("connector file transfer by handle", func() { + const mb = 1 << 20 + + It("moves 20 MB in and 20 MB out with bounded memory", func() { + f := newXferFixture() + end, err := f.t.Begin(TransferGrant{Handle: "h1", Direction: TransferIn, AddonID: "slack", BindingID: "b1", MaxBytes: 50 * mb}) + Expect(err).NotTo(HaveOccurred()) + base, peak := heap(), uint64(0) + var count int + want := chunks(20*mb, contract.TransferChunkBytes, func(off int64, b []byte, last bool) { + code, got := f.put("h1", off, append([]byte(nil), b...), last) + Expect(code).To(Equal(http.StatusOK)) + Expect(got).To(Equal(off + int64(len(b)))) + if count++; count%16 == 0 { + peak = max(peak, heap()) + } + }) + out := end() + Expect(out.Result).To(Equal(TransferDone)) + Expect(out.Bytes).To(Equal(int64(20 * mb))) + Expect(out.SHA).To(Equal(want)) + Expect(peak).To(BeNumerically("<", base+16*mb), "the file is never held in memory") + Expect(f.t.Live()).To(BeZero()) + + // Another node: same object store, its own cache disk. + node2 := blob.NewObjectStore(f.shared, filepath.Join(f.dir, "frontend-cache")) + r, err := node2.Open(GinkgoT().Context(), out.SHA) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(func() { _ = r.Close() }) + h := sha256.New() + n, err := io.Copy(h, r) + Expect(err).NotTo(HaveOccurred()) + Expect(n).To(Equal(int64(20 * mb))) + Expect(hex.EncodeToString(h.Sum(nil))).To(Equal(want)) + + // Out: the add-on pulls the same file by a new handle. + end2, err := f.t.Begin(TransferGrant{Handle: "h2", Direction: TransferOut, AddonID: "slack", SHA: out.SHA, Size: n}) + Expect(err).NotTo(HaveOccurred()) + hh := sha256.New() + var off int64 + base, peak = heap(), 0 + for i := 0; ; i++ { + code, res := f.get("h2", off, contract.TransferChunkBytes) + Expect(code).To(Equal(http.StatusOK)) + hh.Write(res.Data) + off += int64(len(res.Data)) + if i%16 == 0 { + peak = max(peak, heap()) + } + if res.EOF { + break + } + } + Expect(off).To(Equal(n)) + Expect(hex.EncodeToString(hh.Sum(nil))).To(Equal(want)) + Expect(peak).To(BeNumerically("<", base+16*mb)) + Expect(end2().Result).To(Equal(TransferDone)) + Expect(f.outs).To(HaveLen(2)) + }) + + It("resumes at the offset the host has after a gap", func() { + f := newXferFixture() + end, _ := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: mb}) + code, got := f.put("h", 0, []byte("abcd"), false) + Expect(code).To(Equal(http.StatusOK)) + Expect(got).To(BeEquivalentTo(4)) + _, got = f.put("h", 10, []byte("zz"), false) + Expect(got).To(BeEquivalentTo(4), "a gap answers where the host is and writes nothing") + _, got = f.put("h", 4, []byte("efgh"), true) + Expect(got).To(BeEquivalentTo(8)) + o := end() + sum := sha256.Sum256([]byte("abcdefgh")) + Expect(o.SHA).To(Equal(hex.EncodeToString(sum[:]))) + }) + + It("refuses an oversize file and drops the partial one", func() { + f := newXferFixture() + end, _ := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 10}) + code, _ := f.put("h", 0, []byte("12345678"), false) + Expect(code).To(Equal(http.StatusOK)) + code, _ = f.put("h", 8, []byte("9012"), true) + Expect(code).To(Equal(http.StatusRequestEntityTooLarge)) + o := end() + Expect(o.Result).To(Equal(TransferTooLarge)) + Expect(o.SHA).To(BeEmpty()) + code, _ = f.put("h", 0, []byte("x"), false) + Expect(code).To(Equal(http.StatusNotFound), "the handle ended") + }) + + It("refuses a file whose last chunk says it is short of the bytes sent, an empty file, and a write after final", func() { + f := newXferFixture() + end, _ := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + code, _ := f.put("h", 0, nil, true) + Expect(code).To(Equal(http.StatusBadRequest), "empty") + Expect(end().Result).To(Equal(TransferRefused)) + + end, _ = f.t.Begin(TransferGrant{Handle: "g", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + code, _ = f.put("g", 0, []byte("ab"), true) + Expect(code).To(Equal(http.StatusOK)) + code, _ = f.put("g", 2, []byte("cd"), false) + Expect(code).To(Equal(http.StatusConflict), "write after final") + Expect(end().Result).To(Equal(TransferDone)) + }) + + It("refuses another add-on, an unknown handle and the wrong direction with one answer", func() { + f := newXferFixture() + end, _ := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + defer end() + other, _ := f.call("telegram", contract.ConnectorFilePutOp, contract.ConnectorFilePut{Handle: "h", Data: []byte("x")}) + unknown, _ := f.call("slack", contract.ConnectorFilePutOp, contract.ConnectorFilePut{Handle: "nope", Data: []byte("x")}) + Expect(other).To(Equal(http.StatusNotFound)) + Expect(unknown).To(Equal(other)) + code, _ := f.get("h", 0, 10) + Expect(code).To(Equal(http.StatusNotFound), "a write handle cannot read") + }) + + It("does not let a handle be used for two calls", func() { + f := newXferFixture() + end, err := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + Expect(err).NotTo(HaveOccurred()) + _, err = f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + Expect(err).To(HaveOccurred()) + end() + end() // idempotent + Expect(f.outs).To(HaveLen(1)) + _, err = f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + Expect(err).NotTo(HaveOccurred(), "a handle of an ended call is free; a new call makes its own random one") + }) + + It("expires a handle that sees no call and removes the partial file", func() { + f := newXferFixture() + f.t.Idle = 80 * time.Millisecond + _, err := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + Expect(err).NotTo(HaveOccurred()) + code, _ := f.put("h", 0, []byte("partial"), false) + Expect(code).To(Equal(http.StatusOK)) + Eventually(f.t.Live, "2s", "20ms").Should(BeZero()) + code, _ = f.put("h", 7, []byte("x"), false) + Expect(code).To(Equal(http.StatusNotFound)) + Expect(f.outs).To(HaveLen(1)) + Expect(f.outs[0].Result).To(Equal(TransferExpired)) + tmp, _ := filepath.Glob(filepath.Join(f.dir, "owner-cache", "tmp", "*")) + Expect(tmp).To(BeEmpty(), "no partial file is left") + }) + + It("refuses a file the scan hook rejects", func() { + f := newXferFixture() + f.t.Scan = func(_ context.Context, _ string, _ int64) error { return io.ErrClosedPipe } + end, _ := f.t.Begin(TransferGrant{Handle: "h", Direction: TransferIn, AddonID: "slack", MaxBytes: 100}) + code, _ := f.put("h", 0, []byte("virus"), true) + Expect(code).To(Equal(http.StatusBadRequest)) + o := end() + Expect(o.Result).To(Equal(TransferRefused)) + Expect(o.SHA).To(BeEmpty()) + }) + + It("refuses an outbound handle whose content changed size and an offset past the end", func() { + f := newXferFixture() + sha, size, err := f.blobs.Put(GinkgoT().Context(), bytes.NewReader([]byte("hello world")), 0) + Expect(err).NotTo(HaveOccurred()) + bad, _ := f.t.Begin(TransferGrant{Handle: "bad", Direction: TransferOut, AddonID: "slack", SHA: sha, Size: size + 1}) + code, _ := f.get("bad", 0, 5) + Expect(code).To(Equal(http.StatusServiceUnavailable)) + bad() + end, _ := f.t.Begin(TransferGrant{Handle: "ok", Direction: TransferOut, AddonID: "slack", SHA: sha, Size: size}) + defer end() + code, res := f.get("ok", 6, contract.TransferChunkBytes+1) + Expect(code).To(Equal(http.StatusBadRequest), "a length above the chunk is refused") + code, res = f.get("ok", 6, 5) + Expect(code).To(Equal(http.StatusOK)) + Expect(string(res.Data)).To(Equal("world")) + Expect(res.EOF).To(BeTrue()) + code, _ = f.get("ok", size+1, 1) + Expect(code).To(Equal(http.StatusBadRequest)) + }) +}) diff --git a/core/services/addons/runtime/client.go b/core/services/addons/runtime/client.go index fedc6643dc..4c84a49e9c 100644 --- a/core/services/addons/runtime/client.go +++ b/core/services/addons/runtime/client.go @@ -266,7 +266,7 @@ func (c *Client) Connector(ctx context.Context, v contract.ConnectorRequest) (co if c.registration.Capabilities.Connector == nil { return r, ErrNotSent } - e := c.call(ctx, "/connector", nil, v, &r, contract.ConnectorTimeout) + e := c.call(ctx, "/connector", nil, v, &r, v.CallTimeout()) return r, e } func (c *Client) Preview(ctx context.Context, v contract.Envelope) (contract.Preview, error) { diff --git a/core/services/studio/connectors/scope.go b/core/services/studio/connectors/scope.go index 4fc656d0f4..a804b5ef35 100644 --- a/core/services/studio/connectors/scope.go +++ b/core/services/studio/connectors/scope.go @@ -12,6 +12,16 @@ type ConnectorCallScope struct { AgentID string `json:"agentId,omitempty"` AddonID string `json:"addonId"` Kind string `json:"kind"` + // Content names the file an outbound transfer reads: a blob of the + // workspace store. The frontend that checked the chat's access sets it; + // it never reaches the add-on, which gets only the handle. + Content *TransferContent `json:"content,omitempty"` +} + +// TransferContent is the stored content behind an outbound handle. +type TransferContent struct { + SHA string `json:"sha"` + Size int64 `json:"size"` } const ( diff --git a/pkg/addons/contract/connector.go b/pkg/addons/contract/connector.go index d559132fdd..fba032a3da 100644 --- a/pkg/addons/contract/connector.go +++ b/pkg/addons/contract/connector.go @@ -435,6 +435,11 @@ type ConnectorRequest struct { // send_file: the file. Its name is SafeFileName of itself and its data // is 1 to MaxConnectorFileBytes long. File *ConnectorFile `json:"file,omitempty"` + // Transfer is the handle for the bytes of a file, set only for a kind + // that declares files.transfer. media: the add-on writes the file with + // ConnectorFilePutOp. send_file: File is nil and the add-on reads the + // file with ConnectorFileGetOp. + Transfer *ConnectorTransfer `json:"transfer,omitempty"` } // ConnectorFile is a file that the op media returns or the op send_file @@ -468,6 +473,9 @@ type ConnectorResponse struct { BotUsername string `json:"botUsername,omitempty"` // media File *ConnectorFile `json:"file,omitempty"` + // Received answers the op media of a request with a Transfer: the file + // was written by handle. The host adds Size and SHA256 from its record. + Received *ConnectorReceived `json:"received,omitempty"` // history: the messages, oldest first, at most the Limit asked for. // Truncated says that older messages exist that are not returned. Messages []ConnectorHistoryMessage `json:"messages,omitempty"` @@ -503,6 +511,18 @@ type ConnectorRuntime interface { // MaxPollSeconds for the platform. const ConnectorTimeout = (MaxPollSeconds + 20) * time.Second +// ConnectorTransferTimeout is how long a call that carries a Transfer may +// run: it moves a whole file chunk by chunk, and a large file takes minutes. +const ConnectorTransferTimeout = 15 * time.Minute + +// CallTimeout is the time one connector call may run. +func (r ConnectorRequest) CallTimeout() time.Duration { + if r.Transfer != nil { + return ConnectorTransferTimeout + } + return ConnectorTimeout +} + // SafeFileName returns a file name that every platform takes: the last // element of a path, without control characters, quotes or path separators, // cut to MaxConnectorFileName bytes on a character boundary and keeping the diff --git a/pkg/addons/contract/connectorfiles.go b/pkg/addons/contract/connectorfiles.go index bcdc7859f8..dede757f01 100644 --- a/pkg/addons/contract/connectorfiles.go +++ b/pkg/addons/contract/connectorfiles.go @@ -40,6 +40,11 @@ type ConnectorFiles struct { Receive *FileAbility `json:"receive,omitempty"` // Send is the ability to upload files from the bot. Send *FileAbility `json:"send,omitempty"` + // Transfer says that the add-on moves file bytes by a short-lived handle + // (ConnectorTransfer, ConnectorFilePutOp, ConnectorFileGetOp) and not in + // the connector envelope. Only then may a limit pass MaxConnectorFileBytes. + // A kind that declares files without it keeps the inline path. + Transfer bool `json:"transfer,omitempty"` } var mediaTypePattern = regexp.MustCompile(`^[a-z0-9][a-z0-9!#$&^_.+-]{0,63}/(\*|[a-z0-9][a-z0-9!#$&^_.+-]{0,63})$`) @@ -79,3 +84,74 @@ func (f *ConnectorFiles) validate(api VersionRange) error { } return f.Send.validate("send") } + +// Host operations with which an add-on moves the bytes of a file by handle. A +// handle is issued by the host with one connector call (ConnectorRequest.Transfer), +// is single use, bound to that call, that add-on and that binding, and ends +// when the call ends. The add-on needs the operation prefix "connector." in +// its host policy. +const ( + // ConnectorFilePutOp writes the next chunk of an inbound file. + ConnectorFilePutOp = "connector.file.put" + // ConnectorFileGetOp reads the next chunk of an outbound file. + ConnectorFileGetOp = "connector.file.get" + // TransferChunkBytes is the largest chunk of data of one put or get. It + // is far below the payload limit of a host operation after base64. + TransferChunkBytes = 256 << 10 + // MaxTransferHandle bounds the length of a handle. + MaxTransferHandle = 128 +) + +// ConnectorTransfer is the handle that the host gives to the add-on with the +// ops media (inbound, write only) and send_file (outbound, read only). +type ConnectorTransfer struct { + // Handle is random and means nothing to the add-on. + Handle string `json:"handle"` + // MaxBytes is the largest inbound file the host takes, the effective + // limit. The add-on should refuse a larger file before it downloads it. + MaxBytes int64 `json:"maxBytes,omitempty"` + // Name, MIME and Size describe the outbound file. Name is already safe. + Name string `json:"name,omitempty"` + MIME string `json:"mime,omitempty"` + Size int64 `json:"size,omitempty"` +} + +// ConnectorFilePut is the argument object of ConnectorFilePutOp. Offset must +// equal the bytes the host already has; when it does not, the host answers +// with Received and the add-on resumes there. Final ends the file. +type ConnectorFilePut struct { + Handle string `json:"handle"` + Offset int64 `json:"offset"` + Data []byte `json:"data,omitempty"` + Final bool `json:"final,omitempty"` +} + +// ConnectorFilePutResult is the answer of ConnectorFilePutOp. +type ConnectorFilePutResult struct { + Received int64 `json:"received"` +} + +// ConnectorFileGet is the argument object of ConnectorFileGetOp. Length is at +// most TransferChunkBytes. +type ConnectorFileGet struct { + Handle string `json:"handle"` + Offset int64 `json:"offset"` + Length int64 `json:"length"` +} + +// ConnectorFileGetResult is the answer of ConnectorFileGetOp. EOF is true +// when Data ends the file. +type ConnectorFileGetResult struct { + Data []byte `json:"data,omitempty"` + EOF bool `json:"eof,omitempty"` +} + +// ConnectorReceived is what the add-on answers to the op media when it wrote +// the file by handle: Name and MIME. The host that holds the handle overwrites +// Size and SHA256 with its own record, so the add-on cannot choose them. +type ConnectorReceived struct { + Name string `json:"name"` + MIME string `json:"mime,omitempty"` + Size int64 `json:"size,omitempty"` + SHA256 string `json:"sha256,omitempty"` +}