diff --git a/addons/slack/cmd/slack/main.go b/addons/slack/cmd/slack/main.go index bc08822461..ca16e75348 100644 --- a/addons/slack/cmd/slack/main.go +++ b/addons/slack/cmd/slack/main.go @@ -10,6 +10,7 @@ package main import ( "os" + "github.com/mudler/LocalAI/addons/internal/hostclient" "github.com/mudler/LocalAI/addons/internal/protocol" "github.com/mudler/LocalAI/addons/slack/transport" "github.com/mudler/LocalAI/pkg/addons/contract" @@ -19,6 +20,9 @@ func main() { os.Exit(protocol.Run(os.Args, func(s protocol.Startup) contract.Runtime { rt := transport.New(&transport.API{}) rt.Socket.StateDir = s.DataDir + if h := hostclient.New(s); h != nil { + rt.Host = h + } return rt })) } diff --git a/addons/slack/manifest.json b/addons/slack/manifest.json index 44e653d9ad..82b1911891 100644 --- a/addons/slack/manifest.json +++ b/addons/slack/manifest.json @@ -18,6 +18,11 @@ "description": "Talk to one of your saved agents from Slack, in direct messages and in channel threads.", "senderHint": "A user ID (U...), a channel ID (C...) or an @name", "delivery": "pull", + "files": { + "receive": { "maxBytes": 1073741824 }, + "send": { "maxBytes": 1073741824 }, + "transfer": true + }, "page": "apps", "history": { "thread": true, "maxMessages": 30 }, "options": [ diff --git a/addons/slack/transport/api.go b/addons/slack/transport/api.go index 864ba1c77c..7fde77ea0d 100644 --- a/addons/slack/transport/api.go +++ b/addons/slack/transport/api.go @@ -7,6 +7,7 @@ package transport import ( + "bufio" "bytes" "context" "encoding/json" @@ -20,6 +21,7 @@ import ( "time" "github.com/mudler/LocalAI/addons/internal/connector" + "github.com/mudler/LocalAI/pkg/addons/contract" "github.com/mudler/LocalAI/pkg/httpclient" ) @@ -288,11 +290,18 @@ func (a *API) onFilesHost(address string) (*url.URL, error) { // upload: an upload address, the bytes, and the completion that posts the // file. It needs the scope files:write. func (a *API) Upload(ctx context.Context, bot, channel, thread, name string, data []byte) (string, error) { + return a.UploadStream(ctx, bot, channel, thread, name, int64(len(data)), bytes.NewReader(data)) +} + +// UploadStream is Upload for a file that is read as it is sent: body holds +// size bytes, which Slack is told first (the length of the upload). The body +// is read once; a failed upload starts again with a new body. +func (a *API) UploadStream(ctx context.Context, bot, channel, thread, name string, size int64, body io.Reader) (string, error) { var r struct { URL string `json:"upload_url"` ID string `json:"file_id"` } - if err := a.call(ctx, bot, "files.getUploadURLExternal", url.Values{"filename": {name}, "length": {strconv.Itoa(len(data))}}, &r); err != nil { + if err := a.call(ctx, bot, "files.getUploadURLExternal", url.Values{"filename": {name}, "length": {strconv.FormatInt(size, 10)}}, &r); err != nil { return "", err } // The address is signed for this one upload, so the bot token does not @@ -301,12 +310,13 @@ func (a *API) Upload(ctx context.Context, bot, channel, thread, name string, dat if err != nil || r.ID == "" { return "", errors.New("slack: the upload address is not on the files host of Slack") } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), bytes.NewReader(data)) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), io.NopCloser(body)) if err != nil { return "", errors.New("slack: could not build an upload request") } + req.ContentLength = size req.Header.Set("Content-Type", "application/octet-stream") - resp, err := a.client().Do(req) + resp, err := a.streamClient().Do(req) if err != nil { if ctx.Err() != nil { return "", ctx.Err() @@ -332,6 +342,56 @@ func (a *API) Upload(ctx context.Context, bot, channel, thread, name string, dat return r.ID, nil } +// streamClient is the client of a file that is streamed. It has no overall +// timeout, because a large file takes minutes; the context carries the bound. +func (a *API) streamClient() *http.Client { + if a.Client != nil { + return a.Client + } + return httpclient.NewWithTimeout(contract.ConnectorTransferTimeout) +} + +// DownloadStream opens a file of Slack for reading. The caller closes it and +// holds no more than it reads. The bot token goes with the request, so the +// address must be on the files host (see onFilesHost). +func (a *API) DownloadStream(ctx context.Context, bot, address string) (io.ReadCloser, error) { + u, err := a.onFilesHost(address) + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil) + if err != nil { + return nil, errors.New("slack: could not build a download request") + } + req.Header.Set("Authorization", "Bearer "+bot) + resp, err := a.streamClient().Do(req) + if err != nil { + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, errors.New("slack: the file could not be downloaded") + } + switch { + case resp.StatusCode == http.StatusTooManyRequests: + _ = resp.Body.Close() + return nil, &connector.RetryAfter{Seconds: retryAfter(resp.Header.Get("Retry-After"))} + case resp.StatusCode != http.StatusOK: + _ = resp.Body.Close() + return nil, fmt.Errorf("slack: the download answered %d", resp.StatusCode) + } + br := bufio.NewReader(resp.Body) + if head, _ := br.Peek(64); bytes.HasPrefix(bytes.TrimSpace(head), []byte(" max { + return connector.MediaStream{}, ErrTooLarge + } + body, err := p.r.API.DownloadStream(ctx, b, f.Download) + if err != nil { + return connector.MediaStream{}, err + } + name := f.Name + if name == "" { + name = f.ID + } + return connector.MediaStream{Name: name, MIME: f.Mimetype, Size: f.Size, Body: body}, nil +} + +// SendStream uploads a file that the host hands over by handle. +func (p platform) SendStream(ctx context.Context, req contract.ConnectorRequest, body io.Reader) (string, error) { + b, err := bot(req.Token) + if err != nil { + return "", err + } + return p.r.API.UploadStream(ctx, b, req.Chat, req.Thread, req.Transfer.Name, req.Transfer.Size, body) +} + // mark is the key of the "working" message of a conversation. It holds a hash // of the token, never the token. func (r *Runtime) mark(req contract.ConnectorRequest) string { diff --git a/addons/telegram/cmd/telegram/main.go b/addons/telegram/cmd/telegram/main.go index d84eb7e36f..5b820ce417 100644 --- a/addons/telegram/cmd/telegram/main.go +++ b/addons/telegram/cmd/telegram/main.go @@ -8,11 +8,18 @@ package main import ( "os" + "github.com/mudler/LocalAI/addons/internal/hostclient" "github.com/mudler/LocalAI/addons/internal/protocol" "github.com/mudler/LocalAI/addons/telegram/transport" "github.com/mudler/LocalAI/pkg/addons/contract" ) func main() { - os.Exit(protocol.Run(os.Args, func(protocol.Startup) contract.Runtime { return transport.Runtime{API: &transport.API{}} })) + os.Exit(protocol.Run(os.Args, func(s protocol.Startup) contract.Runtime { + rt := transport.Runtime{API: &transport.API{}} + if h := hostclient.New(s); h != nil { + rt.Host = h + } + return rt + })) } diff --git a/addons/telegram/manifest.json b/addons/telegram/manifest.json index 7a9ac532af..adc9667643 100644 --- a/addons/telegram/manifest.json +++ b/addons/telegram/manifest.json @@ -6,7 +6,7 @@ "os": "linux", "arch": "amd64", "host": { "min": 1, "max": 1 }, - "api": { "min": 1, "max": 1 }, + "api": { "min": 2, "max": 2 }, "protocol": { "min": 1, "max": 1 }, "dataFormat": { "min": 1, "max": 1 }, "capabilities": { @@ -18,6 +18,11 @@ "description": "Talk to one of your saved agents from a Telegram bot.", "senderHint": "A chat ID, a user ID or an @username", "delivery": "pull", + "files": { + "receive": { "maxBytes": 20971520 }, + "send": { "maxBytes": 52428800 }, + "transfer": true + }, "page": "bots", "drafts": true, "previewMs": 400 diff --git a/addons/telegram/transport/botapi.go b/addons/telegram/transport/botapi.go index f17b877f21..39f0f72d08 100644 --- a/addons/telegram/transport/botapi.go +++ b/addons/telegram/transport/botapi.go @@ -128,14 +128,20 @@ func (t *API) post(ctx context.Context, token, method, contentType string, body } func (t *API) do(ctx context.Context, token, method, contentType string, body []byte, wait time.Duration, out any) error { + return t.doReader(ctx, token, method, contentType, bytes.NewReader(body), int64(len(body)), wait, out) +} + +// doReader is do for a body that is read as it is sent: size bytes long. +func (t *API) doReader(ctx context.Context, token, method, contentType string, body io.Reader, size int64, wait time.Duration, out any) error { base := t.BaseURL if base == "" { base = DefaultBase } - req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimSuffix(base, "/")+"/bot"+url.PathEscape(token)+"/"+method, bytes.NewReader(body)) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimSuffix(base, "/")+"/bot"+url.PathEscape(token)+"/"+method, io.NopCloser(body)) if err != nil { return errors.New("telegram: could not build a Telegram request") } + req.ContentLength = size req.Header.Set("Content-Type", contentType) resp, err := t.client(wait).Do(req) if err != nil { @@ -494,6 +500,52 @@ func (t *API) SendFile(ctx context.Context, token string, chatID, thread int64, return messageID(raw), nil } +// photoMaxBytes is the largest photo that sendPhoto takes. A larger picture +// goes as a document, which Telegram does not recompress. +const photoMaxBytes = 10 << 20 + +// SendFileStream is SendFile for a file that is read as it is sent: body holds +// size bytes. The multipart head and tail are small and built here, so the +// request has a length and the file is never held in memory. Telegram answers +// 413 for a file over the limit of the Bot API server, which is ErrTooLarge. +func (t *API) SendFileStream(ctx context.Context, token string, chatID, thread int64, name, mimeType string, size int64, body io.Reader) (int64, error) { + method, field := "sendDocument", "document" + if photoTypes[mimeType] && size <= photoMaxBytes { + method, field = "sendPhoto", "photo" + } + var buf bytes.Buffer + w := multipart.NewWriter(&buf) + _ = w.WriteField("chat_id", strconv.FormatInt(chatID, 10)) + if thread != 0 { + _ = w.WriteField("message_thread_id", strconv.FormatInt(thread, 10)) + } + h := textproto.MIMEHeader{} + h.Set("Content-Disposition", mime.FormatMediaType("form-data", map[string]string{"name": field, "filename": name})) + if mimeType == "" { + mimeType = "application/octet-stream" + } + h.Set("Content-Type", mimeType) + if _, err := w.CreatePart(h); err != nil { + return 0, errors.New("telegram: could not build the upload") + } + head := buf.Len() + if err := w.Close(); err != nil { + return 0, errors.New("telegram: could not build the upload") + } + raw := buf.Bytes() + var out json.RawMessage + r := io.MultiReader(bytes.NewReader(raw[:head]), io.LimitReader(body, size), bytes.NewReader(raw[head:])) + err := redact(t.doReader(ctx, token, method, w.FormDataContentType(), r, int64(len(raw))+size, contract.ConnectorTransferTimeout, &out), token) + var ce *callError + if errors.As(err, &ce) && ce.Code == http.StatusRequestEntityTooLarge { + return 0, fmt.Errorf("telegram: the file has %d bytes: %w", size, ErrTooLarge) + } + if err != nil { + return 0, err + } + return messageID(out), nil +} + // Edit replaces the text of a message the bot sent with plain text. func (t *API) Edit(ctx context.Context, token string, chatID, messageID int64, text string) error { return t.edit(ctx, token, chatID, messageID, text, "") diff --git a/addons/telegram/transport/media.go b/addons/telegram/transport/media.go index 52a47d3b12..f88a168564 100644 --- a/addons/telegram/transport/media.go +++ b/addons/telegram/transport/media.go @@ -17,8 +17,9 @@ import ( ) // MaxDownloadBytes is the largest file that Telegram lets a bot download -// with getFile. The add-on hands over less: contract.MaxConnectorFileBytes, -// so that the file fits the envelope of the connector route. +// with getFile on the public Bot API (20 MB), as the Bots FAQ states: +// https://core.telegram.org/bots/faq. Without a handle the add-on hands over +// less: contract.MaxConnectorFileBytes, so that the file fits the envelope. const MaxDownloadBytes = 20 << 20 // ErrTooLarge means a file of a message is larger than the add-on hands @@ -90,13 +91,14 @@ func mediaRefs(m tgMessage) []string { return nil } -// bestPhoto is the largest size of a photo that the add-on can hand over. +// bestPhoto is the largest size of a photo that the add-on can hand over +// (at most MaxDownloadBytes; the host's limit is checked when it is read). // Telegram sends each photo in several sizes; when none says that it fits, // the smallest one is tried. func bestPhoto(sizes []tgPhoto) tgPhoto { best, found := sizes[0], false for _, p := range sizes { - if p.Size <= 0 || p.Size > contract.MaxConnectorFileBytes { + if p.Size <= 0 || p.Size > MaxDownloadBytes { continue } if !found || p.Width*p.Height > best.Width*best.Height { @@ -188,6 +190,82 @@ func (t *API) Media(ctx context.Context, token, ref string) (contract.ConnectorF return contract.ConnectorFile{Name: contract.SafeFileName(strings.ReplaceAll(name, "_", "-") + ext), MIME: mimeType, Data: data}, nil } +// MediaStream opens the file of a reference for streaming to the host. The +// size that getFile states is checked against max before a byte is read, and +// the type is decided from the first bytes, which stay in the stream. The name +// is made here, as in Media. +func (t *API) MediaStream(ctx context.Context, token, ref string, max int64) (connector.MediaStream, error) { + kind, id, ok := strings.Cut(ref, ":") + if !ok || !refKinds[kind] || !fileID.MatchString(id) { + return connector.MediaStream{}, errors.New("telegram: not a file reference") + } + var f struct { + UniqueID string `json:"file_unique_id"` + Size int64 `json:"file_size"` + Path string `json:"file_path"` + } + err := t.call(ctx, token, "getFile", map[string]any{"file_id": id}, 0, &f) + switch { + case refusedBecause(err, "file is too big"): + return connector.MediaStream{}, fmt.Errorf("telegram: Telegram lets a bot download %d MB at most: %w", MaxDownloadBytes>>20, ErrTooLarge) + case err != nil: + return connector.MediaStream{}, err + case f.Size > max: + return connector.MediaStream{}, fmt.Errorf("telegram: the file has %d bytes: %w", f.Size, ErrTooLarge) + case !filePath.MatchString(f.Path): + return connector.MediaStream{}, errors.New("telegram: Telegram named no file to download") + } + body, err := t.getStream(ctx, token, f.Path) + if err != nil { + return connector.MediaStream{}, redact(err, token) + } + rd, head := connector.Sniff(body, 512) + mimeType, ext := fileType(kind, head) + if mimeType == "" { + _ = body.Close() + return connector.MediaStream{}, fmt.Errorf("telegram: the %s is not of a type the add-on reads", strings.ReplaceAll(kind, "_", " ")) + } + name := kind + if fileID.MatchString(f.UniqueID) { + name += "-" + f.UniqueID + } + return connector.MediaStream{Name: contract.SafeFileName(strings.ReplaceAll(name, "_", "-") + ext), MIME: mimeType, Size: f.Size, + Body: struct { + io.Reader + io.Closer + }{rd, body}}, nil +} + +// getStream opens the download of a file of the bot. The context carries the +// bound of the transfer; the client has no total timeout. +func (t *API) getStream(ctx context.Context, token, path string) (io.ReadCloser, error) { + base := t.BaseURL + if base == "" { + base = DefaultBase + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimSuffix(base, "/")+"/file/bot"+url.PathEscape(token)+"/"+path, nil) + if err != nil { + return nil, errors.New("telegram: could not build a download request") + } + resp, err := t.client(contract.ConnectorTransferTimeout).Do(req) + if err != nil { + var ue *url.Error + if errors.As(err, &ue) { + err = ue.Err + } + if ctx.Err() != nil { + return nil, ctx.Err() + } + return nil, fmt.Errorf("telegram: the download failed: %w", err) + } + if resp.StatusCode != http.StatusOK { + _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, 4<<10)) + _ = resp.Body.Close() + return nil, fmt.Errorf("telegram: the download answered %d", resp.StatusCode) + } + return resp.Body, nil +} + // download reads a file of the bot, at most contract.MaxConnectorFileBytes. // The token is in the path, as for the methods, and is taken out of every // error. diff --git a/addons/telegram/transport/media_test.go b/addons/telegram/transport/media_test.go index 91d905e73a..70d5d1533d 100644 --- a/addons/telegram/transport/media_test.go +++ b/addons/telegram/transport/media_test.go @@ -55,7 +55,7 @@ var _ = Describe("the files of messages", func() { ups, err := f.client().Updates(ctx, secretToken, 1, 0) Expect(err).NotTo(HaveOccurred()) Expect(ups).To(HaveLen(7)) - Expect(ups[0].Media).To(Equal([]string{"photo:big"}), "the largest size that fits") + Expect(ups[0].Media).To(Equal([]string{"photo:huge"}), "the largest size that a bot may download") Expect(ups[0].Text).To(Equal("what is this?")) Expect(ups[0].Unsupported).To(BeFalse()) Expect(ups[1].Media).To(Equal([]string{"voice:v1"})) @@ -71,7 +71,9 @@ var _ = Describe("the files of messages", func() { It("finds a mention of the bot in the caption of a photo in a group", func() { f := newFakeBot() - f.reply["getMe"] = func(map[string]any) (int, string) { return 200, `{"ok":true,"result":{"id":123456,"username":"helper_bot"}}` } + f.reply["getMe"] = func(map[string]any) (int, string) { + return 200, `{"ok":true,"result":{"id":123456,"username":"helper_bot"}}` + } f.reply["getUpdates"] = func(map[string]any) (int, string) { return 200, updates(`{"update_id":1,"message":{"chat":{"id":-300,"type":"supergroup"},"from":{"id":7},"caption":"@helper_bot read this","caption_entities":[{"type":"mention","offset":0,"length":11}],"photo":[{"file_id":"p1","file_size":10,"width":1,"height":1}]}}`) } diff --git a/addons/telegram/transport/runtime.go b/addons/telegram/transport/runtime.go index e8545e2751..360972cd21 100644 --- a/addons/telegram/transport/runtime.go +++ b/addons/telegram/transport/runtime.go @@ -2,6 +2,7 @@ package transport import ( "context" + "io" "strconv" "github.com/mudler/LocalAI/addons/internal/connector" @@ -21,6 +22,9 @@ const Kind = "telegram" type Runtime struct { idle.Runtime API *API + // Host is the host operation channel, with which a file moves by handle. + // Nil: the add-on keeps the inline path. + Host connector.Host } var _ contract.ConnectorRuntime = Runtime{} @@ -32,7 +36,7 @@ func (r Runtime) Connector(ctx context.Context, req contract.ConnectorRequest) ( if api == nil { api = &API{} } - return connector.Serve(ctx, platform{api}, req) + return connector.ServeWith(ctx, platform{api}, r.Host, req) } // platform is the Bot API as a connector.Platform. Telegram names chats and @@ -100,6 +104,21 @@ func (p platform) SendFile(ctx context.Context, req contract.ConnectorRequest) ( return messageRef(id), err } +// MediaStream streams a file of a message to the host by handle. +func (p platform) MediaStream(ctx context.Context, token, ref string, max int64) (connector.MediaStream, error) { + return p.api.MediaStream(ctx, token, ref, max) +} + +// SendStream sends a file that the host hands over by handle. +func (p platform) SendStream(ctx context.Context, req contract.ConnectorRequest, body io.Reader) (string, error) { + chat, thread, err := where(req) + if err != nil { + return "", err + } + id, err := p.api.SendFileStream(ctx, req.Token, chat, thread, req.Transfer.Name, req.Transfer.MIME, req.Transfer.Size, body) + return messageRef(id), err +} + // SendRich sends one part of an answer as MarkdownV2, or as plain text when // Telegram refuses the markup. func (p platform) SendRich(ctx context.Context, req contract.ConnectorRequest) (string, error) { diff --git a/deploy/addons/host-policy.json b/deploy/addons/host-policy.json index 152459252f..354075796a 100644 --- a/deploy/addons/host-policy.json +++ b/deploy/addons/host-policy.json @@ -6,6 +6,7 @@ "sandbox": ["workspace.", "egress."], "office": ["workspace."], "web": ["workspace."], - "telegram": ["connector."] + "telegram": ["connector."], + "slack": ["connector."] } } diff --git a/pkg/addons/packaging/host_policy_test.go b/pkg/addons/packaging/host_policy_test.go index 13994fcd18..d4b8fb179a 100644 --- a/pkg/addons/packaging/host_policy_test.go +++ b/pkg/addons/packaging/host_policy_test.go @@ -18,7 +18,7 @@ var _ = Describe("official host policy deployment", func() { grants, err := config.LoadAddonHostPolicy(filepath.Join(root, "deploy", "addons", "host-policy.json")) Expect(err).NotTo(HaveOccurred()) // Deployment-owned exception: historical identity grants, not core defaults. - legacy := "fine-tuning=finetune.,quantization=quantize.,sandbox=workspace.,sandbox=egress.,office=workspace.,web=workspace.,telegram=connector." + legacy := "fine-tuning=finetune.,quantization=quantize.,sandbox=workspace.,sandbox=egress.,office=workspace.,web=workspace.,telegram=connector.,slack=connector." want, err := config.ParseAddonHostOps(legacy) Expect(err).NotTo(HaveOccurred()) Expect(grants).To(Equal(want))