diff --git a/.env.example b/.env.example index ea13f0d..b16533e 100644 --- a/.env.example +++ b/.env.example @@ -1,4 +1,15 @@ LISTEN_ADDR=:8090 REDIS_URL=redis://localhost:6379 + +# Required. Shared with the CRM, which sends it as X-Bot-Runtime-Secret on /events. +# An empty value would authenticate any caller that omits the header. BOT_RUNTIME_SECRET= + AI_CALL_TIMEOUT_SECONDS=30 + +# Optional. Extra hosts allowed to serve incoming media, comma-separated, no scheme +# or port (e.g. "minio.internal,cdn.example.com"). By default only the host of the +# event's postback_url is allowed. Set this when blobs are served elsewhere +# (ActiveStorage redirect mode, CDN); media on an unlisted host is skipped and +# logged as pipeline.ai.attachment.blocked_url, and the text reply still goes out. +MEDIA_HOST_ALLOWLIST= diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..8c74f6b --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,41 @@ +name: CI + +# Nothing ran the Go suite on a PR before this: test/e2e sat non-compiling from +# EVO-558 to EVO-2180 without a single red check. Redis is a service container +# because the repository tests need a real one. + +on: + pull_request: + push: + branches: [develop, main] + +jobs: + test: + runs-on: ubuntu-latest + services: + redis: + image: redis:7-alpine + ports: ['6379:6379'] + options: >- + --health-cmd "redis-cli ping" + --health-interval 5s + --health-timeout 3s + --health-retries 10 + env: + REDIS_TEST_URL: redis://localhost:6379 + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version-file: go.mod + cache: true + + - name: Build + run: go build ./... + + - name: Vet + run: go vet ./... + + - name: Test + run: go test ./... diff --git a/internal/config/config.go b/internal/config/config.go index 76592f4..9781220 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -27,7 +27,12 @@ func Load() (*Config, error) { if err != nil { return nil, err } - botRuntimeSecret := os.Getenv("BOT_RUNTIME_SECRET") + // Required: SecretMiddleware compares the header against this, so an empty value + // authenticates every caller that omits it. + botRuntimeSecret, err := mustGetEnv("BOT_RUNTIME_SECRET") + if err != nil { + return nil, err + } aiCallTimeout, err := getEnvIntOrDefault("AI_CALL_TIMEOUT_SECONDS", 30) if err != nil { return nil, err diff --git a/pkg/ai/model/a2a.go b/pkg/ai/model/a2a.go index 5e6de87..64d0a62 100644 --- a/pkg/ai/model/a2a.go +++ b/pkg/ai/model/a2a.go @@ -10,6 +10,9 @@ type A2ARequest struct { Message string // aggregated buffer content (FR-15) Metadata map[string]any // CRM metadata passed through to processor (tools context) Attachments []Attachment // EVO-2180: incoming media to forward as A2A file parts + // PostbackURL anchors the media host allowlist (see allowedMediaHosts); it is + // not used for the A2A call itself. + PostbackURL string } // Attachment is an incoming media item (image/audio/…) the adapter downloads and diff --git a/pkg/ai/service/ai_adapter.go b/pkg/ai/service/ai_adapter.go index 65b5b6d..21deacd 100644 --- a/pkg/ai/service/ai_adapter.go +++ b/pkg/ai/service/ai_adapter.go @@ -13,6 +13,7 @@ import ( "mime" "net/http" neturl "net/url" + "os" "path" "strings" "time" @@ -49,6 +50,11 @@ const ( attachmentsTotalTimeFactor = 3 ) +// mediaHostAllowlistEnv names extra hosts authorized to serve incoming media, +// comma-separated, for deployments serving blobs off the CRM host (ActiveStorage +// redirect mode, CDN). The default anchor is the event's own postback host. +const mediaHostAllowlistEnv = "MEDIA_HOST_ALLOWLIST" + // maxBackoff caps the exponential backoff between retries so a large // AI_CALL_RETRY_BASE_MS or retry count cannot balloon the wait. const maxBackoff = 5 * time.Second @@ -333,12 +339,26 @@ func (a *aiAdapter) buildFileParts(ctx context.Context, req *model.A2ARequest) [ budgetCtx, cancelBudget := context.WithTimeout(ctx, attachmentsTotalTimeFactor*perDownload) defer cancelBudget() + // Built once per turn: the client closes over the authorized hosts. + hosts := allowedMediaHosts(req.PostbackURL) + client := a.mediaClient(hosts) + parts := make([]model.JSONRPCPart, 0, len(req.Attachments)) remaining := maxAttachmentsTotalBytes for i, att := range req.Attachments { if att.URL == "" { continue } + if err := checkMediaURL(att.URL, hosts); err != nil { + slog.Warn("pipeline.ai.attachment.blocked_url", + "contact_id", req.ContactID, + "conversation_id", req.ConversationID, + "file_type", att.FileType, + "error", err, + "hint", "set "+mediaHostAllowlistEnv+" when blobs are served off the CRM host", + ) + continue + } if budgetCtx.Err() != nil { slog.Warn("pipeline.ai.attachment.budget_exhausted", "contact_id", req.ContactID, @@ -360,12 +380,15 @@ func (a *aiAdapter) buildFileParts(ctx context.Context, req *model.A2ARequest) [ ) break } - data, respContentType, err := a.downloadAttachment(budgetCtx, att.URL, perDownload, limit) + data, respContentType, err := downloadAttachment(budgetCtx, client, att.URL, perDownload, limit) if err != nil { + // Status separates the common 404-from-an-expired-signed-link case from + // an unreachable host; they logged identically before. slog.Warn("pipeline.ai.attachment.download_failed", "contact_id", req.ContactID, "conversation_id", req.ConversationID, "file_type", att.FileType, + "status", statusOf(err), "error", err, ) continue @@ -412,10 +435,74 @@ func (a *aiAdapter) attachmentTimeout() time.Duration { return d } -// downloadAttachment GETs the URL with the adapter's client and the given timeout, -// reading at most limit bytes. It returns the body and the response Content-Type so -// the caller can decide what the bytes actually are. -func (a *aiAdapter) downloadAttachment(ctx context.Context, url string, timeout time.Duration, limit int) ([]byte, string, error) { +// allowedMediaHosts returns the hostnames authorized to serve this event's media. +// An empty set authorizes nothing: failing closed beats fetching a URL the caller +// chose. +func allowedMediaHosts(postbackURL string) map[string]struct{} { + hosts := make(map[string]struct{}, 2) + if u, err := neturl.Parse(postbackURL); err == nil { + if h := strings.ToLower(u.Hostname()); h != "" { + hosts[h] = struct{}{} + } + } + for _, h := range strings.Split(os.Getenv(mediaHostAllowlistEnv), ",") { + if h = strings.ToLower(strings.TrimSpace(h)); h != "" { + hosts[h] = struct{}{} + } + } + return hosts +} + +// checkMediaURL reports why a media URL must not be fetched, or nil when it may be. +func checkMediaURL(rawURL string, hosts map[string]struct{}) error { + u, err := neturl.Parse(rawURL) + if err != nil { + return fmt.Errorf("unparseable media url: %w", err) + } + if scheme := strings.ToLower(u.Scheme); scheme != "http" && scheme != "https" { + return fmt.Errorf("scheme %q is not allowed for media", u.Scheme) + } + host := strings.ToLower(u.Hostname()) + if host == "" { + return errors.New("media url has no host") + } + if _, ok := hosts[host]; !ok { + return fmt.Errorf("host %q is not authorized to serve media for this event", host) + } + return nil +} + +// mediaClient shares the adapter's transport but re-runs checkMediaURL on every +// redirect hop, so an authorized host cannot 302 the download onto an internal one. +func (a *aiAdapter) mediaClient(hosts map[string]struct{}) *http.Client { + return &http.Client{ + Transport: a.client.Transport, + CheckRedirect: func(req *http.Request, via []*http.Request) error { + if len(via) >= 10 { + return errors.New("stopped after 10 redirects") + } + return checkMediaURL(req.URL.String(), hosts) + }, + } +} + +// httpStatusError carries the status of a non-200 media response. +type httpStatusError struct{ status int } + +func (e *httpStatusError) Error() string { return fmt.Sprintf("unexpected status %d", e.status) } + +// statusOf returns the HTTP status of a download error, or 0 if there was no response. +func statusOf(err error) int { + var se *httpStatusError + if errors.As(err, &se) { + return se.status + } + return 0 +} + +// downloadAttachment GETs the URL with the given client and timeout, reading at most +// limit bytes. It returns the body and the response Content-Type. +func downloadAttachment(ctx context.Context, client *http.Client, url string, timeout time.Duration, limit int) ([]byte, string, error) { dlCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() @@ -423,13 +510,13 @@ func (a *aiAdapter) downloadAttachment(ctx context.Context, url string, timeout if err != nil { return nil, "", fmt.Errorf("new_request: %w", err) } - resp, err := a.client.Do(httpReq) + resp, err := client.Do(httpReq) if err != nil { return nil, "", fmt.Errorf("do: %w", err) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return nil, "", fmt.Errorf("unexpected status %d", resp.StatusCode) + return nil, "", &httpStatusError{status: resp.StatusCode} } // +1 so an exactly-at-cap read is distinguishable from an oversize one. data, err := io.ReadAll(io.LimitReader(resp.Body, int64(limit)+1)) diff --git a/pkg/ai/service/ai_adapter_media_test.go b/pkg/ai/service/ai_adapter_media_test.go index def62d5..6957953 100644 --- a/pkg/ai/service/ai_adapter_media_test.go +++ b/pkg/ai/service/ai_adapter_media_test.go @@ -54,6 +54,7 @@ func TestCall_ForwardsAttachmentAsFilePart(t *testing.T) { adapter := aiService.NewAIAdapter(30, 0, 1) _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ OutgoingURL: proc.URL + "/api/v1/a2a/agent-1", + PostbackURL: proc.URL, ContactID: 1, ConversationID: 2, ApiKey: "k", @@ -98,6 +99,7 @@ func TestCall_AttachmentDownloadFailure_SendsTextOnly(t *testing.T) { adapter := aiService.NewAIAdapter(30, 0, 1) _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ OutgoingURL: proc.URL + "/api/v1/a2a/agent-1", + PostbackURL: proc.URL, ContactID: 1, ConversationID: 2, ApiKey: "k", @@ -159,7 +161,7 @@ func TestCall_TotalAttachmentBudget_DropsExcessAndStillSends(t *testing.T) { adapter := aiService.NewAIAdapter(30, 0, 1) if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ - OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "album", Attachments: atts, + OutgoingURL: proc.URL, PostbackURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "album", Attachments: atts, }); err != nil { t.Fatalf("a media budget overflow must not error the call: %v", err) } @@ -195,7 +197,7 @@ func TestCall_AttachmentTimeBudget_IsBounded(t *testing.T) { adapter := aiService.NewAIAdapter(1, 0, 1) start := time.Now() if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ - OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", Attachments: atts, + OutgoingURL: proc.URL, PostbackURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", Attachments: atts, }); err != nil { t.Fatalf("unreachable media must not error the call: %v", err) } @@ -221,7 +223,7 @@ func TestCall_HTMLResponse_IsNotForwardedAsMedia(t *testing.T) { adapter := aiService.NewAIAdapter(30, 0, 1) if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ - OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + OutgoingURL: proc.URL, PostbackURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", Attachments: []aiModel.Attachment{{URL: htmlSrv.URL + "/photo.jpg", ContentType: "image/jpeg", FileType: "image"}}, }); err != nil { t.Fatalf("Call: %v", err) @@ -260,7 +262,7 @@ func TestCall_MimeTypeResolution(t *testing.T) { adapter := aiService.NewAIAdapter(30, 0, 1) if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ - OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + OutgoingURL: proc.URL, PostbackURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", Attachments: []aiModel.Attachment{{URL: srv.URL + tc.urlPath, ContentType: tc.declared, FileType: "image"}}, }); err != nil { t.Fatalf("Call: %v", err) @@ -294,7 +296,7 @@ func TestCall_OversizeAttachment_SendsTextOnly(t *testing.T) { adapter := aiService.NewAIAdapter(30, 0, 1) if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ - OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + OutgoingURL: proc.URL, PostbackURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", Attachments: []aiModel.Attachment{{URL: srv.URL + "/big.png", ContentType: "image/png", FileType: "image"}}, }); err != nil { t.Fatalf("an oversize attachment must not error the call: %v", err) diff --git a/pkg/ai/service/ai_adapter_ssrf_test.go b/pkg/ai/service/ai_adapter_ssrf_test.go new file mode 100644 index 0000000..6f8f79f --- /dev/null +++ b/pkg/ai/service/ai_adapter_ssrf_test.go @@ -0,0 +1,193 @@ +package service_test + +// The attachment URL and the outgoing_url arrive in the same /events payload, so an +// unvalidated download is a read primitive aimed by its caller. These pin the guard. + +import ( + "context" + "encoding/base64" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + aiModel "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/model" + aiService "github.com/EvolutionAPI/evo-bot-runtime/pkg/ai/service" +) + +// capturingProc records the parts the adapter posts. +func capturingProc(parts *[]aiModel.JSONRPCPart) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req aiModel.JSONRPCRequest + _ = json.NewDecoder(r.Body).Decode(&req) + *parts = req.Params.Message.Parts + _ = json.NewEncoder(w).Encode(aiModel.A2AResponse{ + Result: &aiModel.A2AResult{ + Artifacts: []aiModel.A2AArtifact{{Parts: []aiModel.A2APart{{Type: "text", Text: "ok"}}}}, + }, + }) + })) +} + +func fileParts(parts []aiModel.JSONRPCPart) []aiModel.JSONRPCPart { + out := make([]aiModel.JSONRPCPart, 0, len(parts)) + for _, p := range parts { + if p.Type == "file" { + out = append(out, p) + } + } + return out +} + +// The exfiltration shape: internal attachment URL, attacker-chosen outgoing_url. +func TestCall_ForeignHostAttachment_IsNotFetched(t *testing.T) { + var reached bool + internal := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reached = true + w.Header().Set("Content-Type", "text/plain") + _, _ = w.Write([]byte("internal-credentials")) + })) + defer internal.Close() + + var parts []aiModel.JSONRPCPart + proc := capturingProc(&parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + // A CRM on a host that does not serve the attachment below. + PostbackURL: "http://evo-crm.internal:3000/webhooks/bot_runtime/postback/7", + Attachments: []aiModel.Attachment{{URL: internal.URL + "/latest/meta-data/", ContentType: "image/png"}}, + }); err != nil { + t.Fatalf("a blocked attachment must not error the call: %v", err) + } + + if reached { + t.Error("the unauthorized host was fetched — the URL guard did not run") + } + if got := fileParts(parts); len(got) != 0 { + t.Fatalf("file parts = %d, want the attachment dropped", len(got)) + } + if len(parts) != 1 || parts[0].Type != "text" { + t.Fatalf("parts = %+v, want the text reply to survive the block", parts) + } +} + +// Blob storage does not always live on the CRM host, so the allowlist must work. +func TestCall_AllowlistedHost_IsFetched(t *testing.T) { + blob := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "image/png") + _, _ = w.Write([]byte("png-bytes")) + })) + defer blob.Close() + + var parts []aiModel.JSONRPCPart + proc := capturingProc(&parts) + defer proc.Close() + + t.Setenv("MEDIA_HOST_ALLOWLIST", "127.0.0.1") + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + PostbackURL: "http://evo-crm.internal:3000/webhooks/bot_runtime/postback/7", + Attachments: []aiModel.Attachment{{URL: blob.URL + "/photo.png", ContentType: "image/png"}}, + }); err != nil { + t.Fatalf("Call: %v", err) + } + + got := fileParts(parts) + if len(got) != 1 { + t.Fatalf("file parts = %d, want the allowlisted host forwarded", len(got)) + } + decoded, _ := base64.StdEncoding.DecodeString(got[0].File.Bytes) + if string(decoded) != "png-bytes" { + t.Errorf("decoded = %q, want the blob bytes", decoded) + } +} + +// Checking the declared URL alone is not enough: a 302 could walk it internal. +func TestCall_RedirectOffTheAuthorizedHost_IsRefused(t *testing.T) { + var reached bool + internal := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reached = true + _, _ = w.Write([]byte("internal-after-redirect")) + })) + defer internal.Close() + + // httptest always binds 127.0.0.1, so the target uses a host string the + // allowlist does not carry. Without the redirect check this would succeed. + target := strings.Replace(internal.URL, "127.0.0.1", "localhost", 1) + redirector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Redirect(w, r, target+"/secret", http.StatusFound) + })) + defer redirector.Close() + + var parts []aiModel.JSONRPCPart + proc := capturingProc(&parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + // Authorizes 127.0.0.1 (the redirector) but not "localhost" (the target). + PostbackURL: redirector.URL, + Attachments: []aiModel.Attachment{{URL: redirector.URL + "/photo.png", ContentType: "image/png"}}, + }); err != nil { + t.Fatalf("a refused redirect must not error the call: %v", err) + } + + if reached { + t.Error("the redirect target was fetched — the redirect is not re-checked") + } + if got := fileParts(parts); len(got) != 0 { + t.Fatalf("file parts = %d, want the redirected download dropped", len(got)) + } +} + +// Only http/https may be addressed. +func TestCall_NonHTTPScheme_IsRejected(t *testing.T) { + var parts []aiModel.JSONRPCPart + proc := capturingProc(&parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + PostbackURL: proc.URL, + Attachments: []aiModel.Attachment{{URL: "file:///etc/passwd", ContentType: "image/png"}}, + }); err != nil { + t.Fatalf("Call: %v", err) + } + if got := fileParts(parts); len(got) != 0 { + t.Fatalf("file parts = %d, want the file:// URL dropped", len(got)) + } +} + +// Nothing authorized means nothing fetched — failing closed is the point. +func TestCall_NoAuthorizedHost_ForwardsNothing(t *testing.T) { + var reached bool + blob := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + reached = true + })) + defer blob.Close() + + var parts []aiModel.JSONRPCPart + proc := capturingProc(&parts) + defer proc.Close() + + adapter := aiService.NewAIAdapter(30, 0, 1) + if _, err := adapter.Call(context.Background(), &aiModel.A2ARequest{ + OutgoingURL: proc.URL, ContactID: 1, ConversationID: 2, Message: "hi", + Attachments: []aiModel.Attachment{{URL: blob.URL + "/photo.png", ContentType: "image/png"}}, + }); err != nil { + t.Fatalf("Call: %v", err) + } + if reached { + t.Error("an attachment was fetched with no authorized host configured") + } + if len(parts) != 1 || parts[0].Type != "text" { + t.Fatalf("parts = %+v, want text only", parts) + } +} diff --git a/pkg/pipeline/repository/redis_pipeline_repository.go b/pkg/pipeline/repository/redis_pipeline_repository.go index 95838f0..48ca9d5 100644 --- a/pkg/pipeline/repository/redis_pipeline_repository.go +++ b/pkg/pipeline/repository/redis_pipeline_repository.go @@ -82,7 +82,12 @@ func (r *redisPipelineRepository) AppendAttachments(ctx context.Context, contact } values = append(values, b) } - return r.rdb.RPush(ctx, attachBufferKey(contactID, conversationID), values...).Err() + key := attachBufferKey(contactID, conversationID) + if err := r.rdb.RPush(ctx, key, values...).Err(); err != nil { + return err + } + // Leak backstop for turns that die before ClearState; not the debounce window. + return r.rdb.Expire(ctx, key, attachBufferTTL).Err() } // GetAttachments returns the media aggregated during the debounce window. EVO-2180. @@ -175,6 +180,10 @@ func bufferKey(contactID, conversationID int64) string { return fmt.Sprintf("bot_runtime:buffer:%d:%d", contactID, conversationID) } +// attachBufferTTL outlives any turn but stays under the 15-minute TTL the CRM signs +// the media URLs with, so an orphaned entry never outlives its links. +const attachBufferTTL = 10 * time.Minute + func attachBufferKey(contactID, conversationID int64) string { return fmt.Sprintf("bot_runtime:attach:%d:%d", contactID, conversationID) } diff --git a/pkg/pipeline/service/pipeline_service.go b/pkg/pipeline/service/pipeline_service.go index 63c610a..38b1474 100644 --- a/pkg/pipeline/service/pipeline_service.go +++ b/pkg/pipeline/service/pipeline_service.go @@ -398,6 +398,7 @@ func (s *pipelineService) runAIStage(ctx context.Context, contactID, conversatio Message: buffer, Metadata: metadata, Attachments: aiAttachments, + PostbackURL: postbackURL, // anchors the media host allowlist }) if err != nil { switch {