From 7403cbd9a7bd428c7bae6f79530056c2384669a8 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 20:22:20 -0400 Subject: [PATCH 01/10] feat(connector): Stripe, Okta, Shopify, Mailchimp, OneDrive, QuickBooks, and RabbitMQ connectors Adds 17 new connector actions across 7 services: - stripe/create_charge, stripe/create_customer, stripe/create_refund - okta/list_users, okta/create_user - shopify/list_orders, shopify/list_products, shopify/create_order - mailchimp/list_members, mailchimp/add_member - onedrive/upload, sharepoint/list_items - quickbooks/create_invoice, quickbooks/list_invoices - rabbitmq/publish, rabbitmq/consume Also adds parseJSONBody shared helper to tools.go and github.com/rabbitmq/amqp091-go v1.11.0 as the AMQP protocol client. Closes #94, #108, #109, #111, #112, #114, #115 Co-Authored-By: Claude Sonnet 4.6 --- packages/engine/go.mod | 1 + packages/engine/go.sum | 2 + .../engine/internal/connector/connector.go | 16 ++ .../engine/internal/connector/mailchimp.go | 186 +++++++++++++++ .../internal/connector/mailchimp_test.go | 164 +++++++++++++ packages/engine/internal/connector/okta.go | 185 +++++++++++++++ .../engine/internal/connector/okta_test.go | 173 ++++++++++++++ .../engine/internal/connector/onedrive.go | 153 ++++++++++++ .../internal/connector/onedrive_test.go | 176 ++++++++++++++ .../engine/internal/connector/quickbooks.go | 199 ++++++++++++++++ .../internal/connector/quickbooks_test.go | 161 +++++++++++++ .../engine/internal/connector/rabbitmq.go | 134 +++++++++++ .../internal/connector/rabbitmq_test.go | 93 ++++++++ packages/engine/internal/connector/shopify.go | 224 ++++++++++++++++++ .../engine/internal/connector/shopify_test.go | 183 ++++++++++++++ packages/engine/internal/connector/stripe.go | 180 ++++++++++++++ .../engine/internal/connector/stripe_test.go | 190 +++++++++++++++ packages/engine/internal/connector/tools.go | 9 + 18 files changed, 2429 insertions(+) create mode 100644 packages/engine/internal/connector/mailchimp.go create mode 100644 packages/engine/internal/connector/mailchimp_test.go create mode 100644 packages/engine/internal/connector/okta.go create mode 100644 packages/engine/internal/connector/okta_test.go create mode 100644 packages/engine/internal/connector/onedrive.go create mode 100644 packages/engine/internal/connector/onedrive_test.go create mode 100644 packages/engine/internal/connector/quickbooks.go create mode 100644 packages/engine/internal/connector/quickbooks_test.go create mode 100644 packages/engine/internal/connector/rabbitmq.go create mode 100644 packages/engine/internal/connector/rabbitmq_test.go create mode 100644 packages/engine/internal/connector/shopify.go create mode 100644 packages/engine/internal/connector/shopify_test.go create mode 100644 packages/engine/internal/connector/stripe.go create mode 100644 packages/engine/internal/connector/stripe_test.go diff --git a/packages/engine/go.mod b/packages/engine/go.mod index 2daed27..0e0cbd1 100644 --- a/packages/engine/go.mod +++ b/packages/engine/go.mod @@ -126,6 +126,7 @@ require ( github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.66.1 // indirect github.com/prometheus/procfs v0.19.2 // indirect + github.com/rabbitmq/amqp091-go v1.11.0 // indirect github.com/redis/go-redis/v9 v9.20.0 // indirect github.com/sagikazarmark/locafero v0.11.0 // indirect github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 // indirect diff --git a/packages/engine/go.sum b/packages/engine/go.sum index ce669e9..ebc4f7b 100644 --- a/packages/engine/go.sum +++ b/packages/engine/go.sum @@ -285,6 +285,8 @@ github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9Z github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= github.com/prometheus/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4Vws= github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw= +github.com/rabbitmq/amqp091-go v1.11.0 h1:HxIctVm9Gid/Vtn706necmZ7Wj6pgGI2eqplRbEY8O8= +github.com/rabbitmq/amqp091-go v1.11.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= github.com/redis/go-redis/v9 v9.20.0 h1:WnQYxLkgO2xiXTCJY0ldIiI8dNqCDlQAG+AtaH7a2a0= github.com/redis/go-redis/v9 v9.20.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= diff --git a/packages/engine/internal/connector/connector.go b/packages/engine/internal/connector/connector.go index c7199f0..0a98ca7 100644 --- a/packages/engine/internal/connector/connector.go +++ b/packages/engine/internal/connector/connector.go @@ -65,6 +65,22 @@ func NewRegistry() *Registry { r.Register("redis/publish", &RedisPublishConnector{}) r.Register("mongodb/find", &MongoFindConnector{}) r.Register("mongodb/aggregate", &MongoAggregateConnector{}) + r.Register("stripe/create_charge", &StripeCreateChargeConnector{}) + r.Register("stripe/create_customer", &StripeCreateCustomerConnector{}) + r.Register("stripe/create_refund", &StripeCreateRefundConnector{}) + r.Register("okta/list_users", &OktaListUsersConnector{}) + r.Register("okta/create_user", &OktaCreateUserConnector{}) + r.Register("quickbooks/create_invoice", &QuickBooksCreateInvoiceConnector{}) + r.Register("quickbooks/list_invoices", &QuickBooksListInvoicesConnector{}) + r.Register("onedrive/upload", &OneDriveUploadConnector{}) + r.Register("sharepoint/list_items", &SharePointListItemsConnector{}) + r.Register("rabbitmq/publish", &RabbitMQPublishConnector{}) + r.Register("rabbitmq/consume", &RabbitMQConsumeConnector{}) + r.Register("shopify/list_orders", &ShopifyListOrdersConnector{}) + r.Register("shopify/list_products", &ShopifyListProductsConnector{}) + r.Register("shopify/create_order", &ShopifyCreateOrderConnector{}) + r.Register("mailchimp/list_members", &MailchimpListMembersConnector{}) + r.Register("mailchimp/add_member", &MailchimpAddMemberConnector{}) return r } diff --git a/packages/engine/internal/connector/mailchimp.go b/packages/engine/internal/connector/mailchimp.go new file mode 100644 index 0000000..8a12dc5 --- /dev/null +++ b/packages/engine/internal/connector/mailchimp.go @@ -0,0 +1,186 @@ +package connector + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "strings" +) + +// MailchimpListMembersConnector lists members of a Mailchimp audience list. +type MailchimpListMembersConnector struct { + Client *http.Client + baseURL string +} + +func (c *MailchimpListMembersConnector) apiURL(dc, path string) string { + if c.baseURL != "" { + return c.baseURL + path + } + return fmt.Sprintf("https://%s.api.mailchimp.com/3.0%s", dc, path) +} + +func (c *MailchimpListMembersConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + dc, apiKey, err := extractMailchimpCredential(params) + if err != nil { + return nil, fmt.Errorf("mailchimp/list_members: %w", err) + } + + listID, _ := params["list_id"].(string) + if listID == "" { + return nil, fmt.Errorf("mailchimp/list_members: list_id is required") + } + + endpoint := c.apiURL(dc, fmt.Sprintf("/lists/%s/members", listID)) + if count, ok := extractInt(params["count"]); ok && count > 0 { + endpoint += fmt.Sprintf("?count=%d", count) + } + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("mailchimp/list_members: creating request: %w", err) + } + req.SetBasicAuth("anystring", apiKey) + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("mailchimp/list_members: %w", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("mailchimp/list_members: reading response: %w", err) + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("mailchimp/list_members: Mailchimp API returned %d: %s", resp.StatusCode, truncate(string(body), 500)) + } + + var result struct { + Members []any `json:"members"` + TotalItems int `json:"total_items"` + } + if err := json.Unmarshal(body, &result); err != nil { + return nil, fmt.Errorf("mailchimp/list_members: parsing response: %w", err) + } + return map[string]any{ + "members": result.Members, + "count": len(result.Members), + "total_items": result.TotalItems, + }, nil +} + +// MailchimpAddMemberConnector adds or updates a subscriber in a Mailchimp audience list. +type MailchimpAddMemberConnector struct { + Client *http.Client + baseURL string +} + +func (c *MailchimpAddMemberConnector) apiURL(dc, path string) string { + if c.baseURL != "" { + return c.baseURL + path + } + return fmt.Sprintf("https://%s.api.mailchimp.com/3.0%s", dc, path) +} + +func (c *MailchimpAddMemberConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + dc, apiKey, err := extractMailchimpCredential(params) + if err != nil { + return nil, fmt.Errorf("mailchimp/add_member: %w", err) + } + + listID, _ := params["list_id"].(string) + if listID == "" { + return nil, fmt.Errorf("mailchimp/add_member: list_id is required") + } + email, _ := params["email"].(string) + if email == "" { + return nil, fmt.Errorf("mailchimp/add_member: email is required") + } + + status := "subscribed" + if s, ok := params["status"].(string); ok && s != "" { + status = s + } + + member := map[string]any{ + "email_address": email, + "status": status, + } + if mergeFields, ok := params["merge_fields"].(map[string]any); ok { + member["merge_fields"] = mergeFields + } + if tags, ok := params["tags"].([]any); ok && len(tags) > 0 { + member["tags"] = tags + } + + reqJSON, err := json.Marshal(member) + if err != nil { + return nil, fmt.Errorf("mailchimp/add_member: marshaling request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", + c.apiURL(dc, fmt.Sprintf("/lists/%s/members", listID)), + bytes.NewReader(reqJSON), + ) + if err != nil { + return nil, fmt.Errorf("mailchimp/add_member: creating request: %w", err) + } + req.SetBasicAuth("anystring", apiKey) + req.Header.Set("Content-Type", "application/json") + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("mailchimp/add_member: %w", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("mailchimp/add_member: reading response: %w", err) + } + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + return nil, fmt.Errorf("mailchimp/add_member: Mailchimp API returned %d: %s", resp.StatusCode, truncate(string(body), 500)) + } + + return parseJSONBody(body, "mailchimp/add_member") +} + +func extractMailchimpCredential(params map[string]any) (dc, apiKey string, err error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return "", "", fmt.Errorf("credential is required") + } + delete(params, "_credential") + + var cred map[string]string + switch v := raw.(type) { + case map[string]string: + cred = v + case map[string]any: + cred = make(map[string]string, len(v)) + for k, val := range v { + if s, ok := val.(string); ok { + cred[k] = s + } + } + default: + return "", "", fmt.Errorf("credential is required") + } + + apiKey = cred["api_key"] + if apiKey == "" { + return "", "", fmt.Errorf("credential must contain an 'api_key' field") + } + + // Data center is the suffix after the last '-' in the API key (e.g. "abc123-us1" → "us1"). + parts := strings.Split(apiKey, "-") + if len(parts) < 2 { + return "", "", fmt.Errorf("credential api_key format invalid: expected '-' (e.g. abc123-us1)") + } + dc = parts[len(parts)-1] + return dc, apiKey, nil +} diff --git a/packages/engine/internal/connector/mailchimp_test.go b/packages/engine/internal/connector/mailchimp_test.go new file mode 100644 index 0000000..4c04108 --- /dev/null +++ b/packages/engine/internal/connector/mailchimp_test.go @@ -0,0 +1,164 @@ +package connector + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestMailchimpListMembersConnector_ListsMembers(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" || r.URL.Path != "/3.0/lists/abc123/members" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + _, pass, ok := r.BasicAuth() + if !ok || pass != "testkey-us1" { + t.Errorf("unexpected basic auth password: %s", pass) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "members": []any{map[string]any{"id": "m1", "email_address": "alice@example.com"}}, + "total_items": 1, + }) + })) + defer srv.Close() + + // Override baseURL to use the test server; DC is "us1" from the key. + c := &MailchimpListMembersConnector{baseURL: srv.URL + "/3.0"} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "testkey-us1"}, + "list_id": "abc123", + }) + if err != nil { + t.Fatal(err) + } + if out["count"].(int) != 1 { + t.Errorf("expected count=1, got %v", out["count"]) + } + if out["total_items"].(int) != 1 { + t.Errorf("expected total_items=1, got %v", out["total_items"]) + } +} + +func TestMailchimpListMembersConnector_MissingListID(t *testing.T) { + c := &MailchimpListMembersConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "key-us1"}, + }) + if err == nil { + t.Fatal("expected error for missing list_id") + } +} + +func TestMailchimpListMembersConnector_MissingCredential(t *testing.T) { + c := &MailchimpListMembersConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "list_id": "abc", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestMailchimpListMembersConnector_InvalidAPIKeyFormat(t *testing.T) { + c := &MailchimpListMembersConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "nokeydc"}, + "list_id": "abc", + }) + if err == nil { + t.Fatal("expected error for invalid api_key format") + } +} + +func TestMailchimpAddMemberConnector_AddsMember(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" || r.URL.Path != "/3.0/lists/list1/members" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + if body["email_address"] != "bob@example.com" { + t.Errorf("unexpected email: %v", body["email_address"]) + } + if body["status"] != "subscribed" { + t.Errorf("unexpected status: %v", body["status"]) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]any{"id": "m2", "email_address": "bob@example.com", "status": "subscribed"}) + })) + defer srv.Close() + + c := &MailchimpAddMemberConnector{baseURL: srv.URL + "/3.0"} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "key-us1"}, + "list_id": "list1", + "email": "bob@example.com", + }) + if err != nil { + t.Fatal(err) + } + if out["id"] != "m2" { + t.Errorf("expected id=m2, got %v", out["id"]) + } +} + +func TestMailchimpAddMemberConnector_MissingEmail(t *testing.T) { + c := &MailchimpAddMemberConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "key-us1"}, + "list_id": "abc", + }) + if err == nil { + t.Fatal("expected error for missing email") + } +} + +func TestMailchimpAddMemberConnector_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"title":"Invalid Resource","status":400}`, http.StatusBadRequest) + })) + defer srv.Close() + + c := &MailchimpAddMemberConnector{baseURL: srv.URL + "/3.0"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "key-us1"}, + "list_id": "bad", + "email": "x@y.com", + }) + if err == nil { + t.Fatal("expected error for 400") + } +} + +func TestMailchimpListMembersConnector_MapAnyCredential(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, pass, _ := r.BasicAuth() + if pass != "mapkey-us2" { + t.Errorf("unexpected auth: %s", pass) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"members": []any{}, "total_items": 0}) + })) + defer srv.Close() + + c := &MailchimpListMembersConnector{baseURL: srv.URL + "/3.0"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"api_key": "mapkey-us2"}, + "list_id": "abc", + }) + if err != nil { + t.Fatal(err) + } +} + +func TestRegistry_MailchimpConnectors(t *testing.T) { + r := NewRegistry() + for _, action := range []string{"mailchimp/list_members", "mailchimp/add_member"} { + if _, err := r.Get(action); err != nil { + t.Errorf("%s not registered: %v", action, err) + } + } +} diff --git a/packages/engine/internal/connector/okta.go b/packages/engine/internal/connector/okta.go new file mode 100644 index 0000000..0f2b6ed --- /dev/null +++ b/packages/engine/internal/connector/okta.go @@ -0,0 +1,185 @@ +package connector + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" +) + +// OktaListUsersConnector lists users from an Okta organization. +type OktaListUsersConnector struct { + Client *http.Client + baseURL string +} + +func (c *OktaListUsersConnector) apiURL(domain, path string) string { + if c.baseURL != "" { + return c.baseURL + path + } + return "https://" + domain + path +} + +func (c *OktaListUsersConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + domain, token, err := extractOktaCredential(params) + if err != nil { + return nil, fmt.Errorf("okta/list_users: %w", err) + } + + query := url.Values{} + if q, ok := params["q"].(string); ok && q != "" { + query.Set("q", q) + } + if filter, ok := params["filter"].(string); ok && filter != "" { + query.Set("filter", filter) + } + if limit, ok := extractInt(params["limit"]); ok && limit > 0 { + query.Set("limit", fmt.Sprintf("%d", limit)) + } + + endpoint := c.apiURL(domain, "/api/v1/users") + if len(query) > 0 { + endpoint += "?" + query.Encode() + } + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("okta/list_users: creating request: %w", err) + } + req.Header.Set("Authorization", "SSWS "+token) + req.Header.Set("Accept", "application/json") + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("okta/list_users: %w", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("okta/list_users: reading response: %w", err) + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("okta/list_users: Okta API returned %d: %s", resp.StatusCode, truncate(string(body), 500)) + } + + var users []any + if err := json.Unmarshal(body, &users); err != nil { + return nil, fmt.Errorf("okta/list_users: parsing response: %w", err) + } + return map[string]any{"users": users, "count": len(users)}, nil +} + +// OktaCreateUserConnector creates a new user in an Okta organization. +type OktaCreateUserConnector struct { + Client *http.Client + baseURL string +} + +func (c *OktaCreateUserConnector) apiURL(domain, path string) string { + if c.baseURL != "" { + return c.baseURL + path + } + return "https://" + domain + path +} + +func (c *OktaCreateUserConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + domain, token, err := extractOktaCredential(params) + if err != nil { + return nil, fmt.Errorf("okta/create_user: %w", err) + } + + profile, _ := params["profile"].(map[string]any) + if profile == nil { + return nil, fmt.Errorf("okta/create_user: profile is required") + } + if _, ok := profile["login"]; !ok { + return nil, fmt.Errorf("okta/create_user: profile.login is required") + } + if _, ok := profile["email"]; !ok { + return nil, fmt.Errorf("okta/create_user: profile.email is required") + } + + body := map[string]any{"profile": profile} + if credentials, ok := params["credentials"].(map[string]any); ok { + body["credentials"] = credentials + } + if groupIDs, ok := params["group_ids"].([]any); ok && len(groupIDs) > 0 { + body["groupIds"] = groupIDs + } + + activate := true + if a, ok := params["activate"].(bool); ok { + activate = a + } + + reqJSON, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("okta/create_user: marshaling request: %w", err) + } + + endpoint := c.apiURL(domain, "/api/v1/users") + if activate { + endpoint += "?activate=true" + } + + req, err := http.NewRequestWithContext(ctx, "POST", endpoint, bytes.NewReader(reqJSON)) + if err != nil { + return nil, fmt.Errorf("okta/create_user: creating request: %w", err) + } + req.Header.Set("Authorization", "SSWS "+token) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("okta/create_user: %w", err) + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("okta/create_user: reading response: %w", err) + } + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + return nil, fmt.Errorf("okta/create_user: Okta API returned %d: %s", resp.StatusCode, truncate(string(respBody), 500)) + } + + return parseJSONBody(respBody, "okta/create_user") +} + +func extractOktaCredential(params map[string]any) (domain, token string, err error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return "", "", fmt.Errorf("credential is required") + } + delete(params, "_credential") + + var cred map[string]string + switch v := raw.(type) { + case map[string]string: + cred = v + case map[string]any: + cred = make(map[string]string, len(v)) + for k, val := range v { + if s, ok := val.(string); ok { + cred[k] = s + } + } + default: + return "", "", fmt.Errorf("credential is required") + } + + domain = cred["domain"] + if domain == "" { + return "", "", fmt.Errorf("credential must contain a 'domain' field (e.g. dev-xxx.okta.com)") + } + token = cred["token"] + if token == "" { + return "", "", fmt.Errorf("credential must contain a 'token' field") + } + return domain, token, nil +} diff --git a/packages/engine/internal/connector/okta_test.go b/packages/engine/internal/connector/okta_test.go new file mode 100644 index 0000000..4b49d2a --- /dev/null +++ b/packages/engine/internal/connector/okta_test.go @@ -0,0 +1,173 @@ +package connector + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestOktaListUsersConnector_ListsUsers(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" || r.URL.Path != "/api/v1/users" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + if r.Header.Get("Authorization") != "SSWS ssws-token" { + t.Errorf("unexpected Authorization: %s", r.Header.Get("Authorization")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode([]any{ + map[string]any{"id": "00u1", "profile": map[string]any{"login": "alice@example.com"}}, + map[string]any{"id": "00u2", "profile": map[string]any{"login": "bob@example.com"}}, + }) + })) + defer srv.Close() + + c := &OktaListUsersConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"domain": "dev-xxx.okta.com", "token": "ssws-token"}, + }) + if err != nil { + t.Fatal(err) + } + if out["count"].(int) != 2 { + t.Errorf("expected count=2, got %v", out["count"]) + } +} + +func TestOktaListUsersConnector_WithQuery(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("q") != "alice" { + t.Errorf("expected q=alice, got %s", r.URL.Query().Get("q")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode([]any{}) + })) + defer srv.Close() + + c := &OktaListUsersConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"domain": "dev.okta.com", "token": "tok"}, + "q": "alice", + }) + if err != nil { + t.Fatal(err) + } +} + +func TestOktaListUsersConnector_MissingCredential(t *testing.T) { + c := &OktaListUsersConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{}) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestOktaListUsersConnector_MissingDomain(t *testing.T) { + c := &OktaListUsersConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "tok"}, + }) + if err == nil { + t.Fatal("expected error for missing domain") + } +} + +func TestOktaCreateUserConnector_CreatesUser(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" { + t.Errorf("expected POST, got %s", r.Method) + } + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + profile, _ := body["profile"].(map[string]any) + if profile["login"] != "alice@example.com" { + t.Errorf("unexpected login: %v", profile["login"]) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]any{"id": "00u1", "status": "ACTIVE"}) + })) + defer srv.Close() + + c := &OktaCreateUserConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"domain": "dev.okta.com", "token": "tok"}, + "profile": map[string]any{ + "login": "alice@example.com", + "email": "alice@example.com", + "firstName": "Alice", + "lastName": "Smith", + }, + }) + if err != nil { + t.Fatal(err) + } + if out["id"] != "00u1" { + t.Errorf("expected id=00u1, got %v", out["id"]) + } +} + +func TestOktaCreateUserConnector_MissingProfile(t *testing.T) { + c := &OktaCreateUserConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"domain": "dev.okta.com", "token": "tok"}, + }) + if err == nil { + t.Fatal("expected error for missing profile") + } +} + +func TestOktaCreateUserConnector_MissingLogin(t *testing.T) { + c := &OktaCreateUserConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"domain": "dev.okta.com", "token": "tok"}, + "profile": map[string]any{"email": "alice@example.com"}, + }) + if err == nil { + t.Fatal("expected error for missing profile.login") + } +} + +func TestOktaListUsersConnector_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"errorCode":"E0000011","errorSummary":"Invalid token"}`, http.StatusUnauthorized) + })) + defer srv.Close() + + c := &OktaListUsersConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"domain": "dev.okta.com", "token": "bad"}, + }) + if err == nil { + t.Fatal("expected error for 401") + } +} + +func TestOktaListUsersConnector_MapAnyCredential(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "SSWS maptoken" { + t.Errorf("unexpected auth: %s", r.Header.Get("Authorization")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode([]any{}) + })) + defer srv.Close() + + c := &OktaListUsersConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"domain": "dev.okta.com", "token": "maptoken"}, + }) + if err != nil { + t.Fatal(err) + } +} + +func TestRegistry_OktaConnectors(t *testing.T) { + r := NewRegistry() + for _, action := range []string{"okta/list_users", "okta/create_user"} { + if _, err := r.Get(action); err != nil { + t.Errorf("%s not registered: %v", action, err) + } + } +} diff --git a/packages/engine/internal/connector/onedrive.go b/packages/engine/internal/connector/onedrive.go new file mode 100644 index 0000000..6bbe1fa --- /dev/null +++ b/packages/engine/internal/connector/onedrive.go @@ -0,0 +1,153 @@ +package connector + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" +) + +const graphBaseURL = "https://graph.microsoft.com/v1.0" + +// OneDriveUploadConnector uploads a file to OneDrive via the Microsoft Graph API. +type OneDriveUploadConnector struct { + Client *http.Client + baseURL string +} + +func (c *OneDriveUploadConnector) apiURL(path string) string { + base := c.baseURL + if base == "" { + base = graphBaseURL + } + return base + path +} + +func (c *OneDriveUploadConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + token, err := extractBearerToken(params) + if err != nil { + return nil, fmt.Errorf("onedrive/upload: %w", err) + } + + filePath, _ := params["path"].(string) + if filePath == "" { + return nil, fmt.Errorf("onedrive/upload: path is required") + } + content, _ := params["content"].(string) + if content == "" { + return nil, fmt.Errorf("onedrive/upload: content is required") + } + + var endpoint string + if driveID, ok := params["drive_id"].(string); ok && driveID != "" { + endpoint = c.apiURL(fmt.Sprintf("/drives/%s/root:/%s:/content", + url.PathEscape(driveID), url.PathEscape(filePath))) + } else { + endpoint = c.apiURL(fmt.Sprintf("/me/drive/root:/%s:/content", url.PathEscape(filePath))) + } + + contentType := "application/octet-stream" + if ct, ok := params["content_type"].(string); ok && ct != "" { + contentType = ct + } + + req, err := http.NewRequestWithContext(ctx, "PUT", endpoint, bytes.NewReader([]byte(content))) + if err != nil { + return nil, fmt.Errorf("onedrive/upload: creating request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", contentType) + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("onedrive/upload: %w", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("onedrive/upload: reading response: %w", err) + } + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + return nil, fmt.Errorf("onedrive/upload: Graph API returned %d: %s", resp.StatusCode, truncate(string(body), 500)) + } + + return parseJSONBody(body, "onedrive/upload") +} + +// SharePointListItemsConnector lists items from a SharePoint list via the Microsoft Graph API. +type SharePointListItemsConnector struct { + Client *http.Client + baseURL string +} + +func (c *SharePointListItemsConnector) apiURL(path string) string { + base := c.baseURL + if base == "" { + base = graphBaseURL + } + return base + path +} + +func (c *SharePointListItemsConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + token, err := extractBearerToken(params) + if err != nil { + return nil, fmt.Errorf("sharepoint/list_items: %w", err) + } + + siteID, _ := params["site_id"].(string) + if siteID == "" { + return nil, fmt.Errorf("sharepoint/list_items: site_id is required") + } + listID, _ := params["list_id"].(string) + if listID == "" { + return nil, fmt.Errorf("sharepoint/list_items: list_id is required") + } + + endpoint := c.apiURL(fmt.Sprintf("/sites/%s/lists/%s/items", + url.PathEscape(siteID), url.PathEscape(listID))) + + query := url.Values{} + query.Set("expand", "fields") + if top, ok := extractInt(params["top"]); ok && top > 0 { + query.Set("$top", fmt.Sprintf("%d", top)) + } + if filter, ok := params["filter"].(string); ok && filter != "" { + query.Set("$filter", filter) + } + endpoint += "?" + query.Encode() + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("sharepoint/list_items: creating request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+token) + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("sharepoint/list_items: %w", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("sharepoint/list_items: reading response: %w", err) + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("sharepoint/list_items: Graph API returned %d: %s", resp.StatusCode, truncate(string(body), 500)) + } + + var result struct { + Value []any `json:"value"` + } + if err := json.Unmarshal(body, &result); err != nil { + return nil, fmt.Errorf("sharepoint/list_items: parsing response: %w", err) + } + return map[string]any{ + "items": result.Value, + "count": len(result.Value), + }, nil +} diff --git a/packages/engine/internal/connector/onedrive_test.go b/packages/engine/internal/connector/onedrive_test.go new file mode 100644 index 0000000..c1a0d26 --- /dev/null +++ b/packages/engine/internal/connector/onedrive_test.go @@ -0,0 +1,176 @@ +package connector + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestOneDriveUploadConnector_UploadsFile(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "PUT" { + t.Errorf("expected PUT, got %s", r.Method) + } + if r.Header.Get("Authorization") != "Bearer graph-token" { + t.Errorf("unexpected Authorization: %s", r.Header.Get("Authorization")) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + json.NewEncoder(w).Encode(map[string]any{ + "id": "file-id-1", + "name": "hello.txt", + "size": 13, + }) + })) + defer srv.Close() + + c := &OneDriveUploadConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "graph-token"}, + "path": "documents/hello.txt", + "content": "Hello, Mantle!", + }) + if err != nil { + t.Fatal(err) + } + if out["id"] != "file-id-1" { + t.Errorf("expected id=file-id-1, got %v", out["id"]) + } +} + +func TestOneDriveUploadConnector_MissingPath(t *testing.T) { + c := &OneDriveUploadConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "tok"}, + "content": "data", + }) + if err == nil { + t.Fatal("expected error for missing path") + } +} + +func TestOneDriveUploadConnector_MissingContent(t *testing.T) { + c := &OneDriveUploadConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "tok"}, + "path": "file.txt", + }) + if err == nil { + t.Fatal("expected error for missing content") + } +} + +func TestOneDriveUploadConnector_MissingCredential(t *testing.T) { + c := &OneDriveUploadConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "path": "file.txt", + "content": "data", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestOneDriveUploadConnector_WithDriveID(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/drives/drive123/root:/notes.txt:/content" { + t.Errorf("unexpected path: %s", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + json.NewEncoder(w).Encode(map[string]any{"id": "f2", "name": "notes.txt"}) + })) + defer srv.Close() + + c := &OneDriveUploadConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "tok"}, + "path": "notes.txt", + "content": "note content", + "drive_id": "drive123", + }) + if err != nil { + t.Fatal(err) + } +} + +func TestSharePointListItemsConnector_ListsItems(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("expected GET, got %s", r.Method) + } + if r.Header.Get("Authorization") != "Bearer sp-token" { + t.Errorf("unexpected Authorization: %s", r.Header.Get("Authorization")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "value": []any{ + map[string]any{"id": "1", "fields": map[string]any{"Title": "Row 1"}}, + map[string]any{"id": "2", "fields": map[string]any{"Title": "Row 2"}}, + }, + }) + })) + defer srv.Close() + + c := &SharePointListItemsConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "sp-token"}, + "site_id": "site1", + "list_id": "list1", + }) + if err != nil { + t.Fatal(err) + } + if out["count"].(int) != 2 { + t.Errorf("expected count=2, got %v", out["count"]) + } +} + +func TestSharePointListItemsConnector_MissingSiteID(t *testing.T) { + c := &SharePointListItemsConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "tok"}, + "list_id": "list1", + }) + if err == nil { + t.Fatal("expected error for missing site_id") + } +} + +func TestSharePointListItemsConnector_MissingListID(t *testing.T) { + c := &SharePointListItemsConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "tok"}, + "site_id": "site1", + }) + if err == nil { + t.Fatal("expected error for missing list_id") + } +} + +func TestOneDriveUploadConnector_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"error":{"code":"AccessDenied"}}`, http.StatusForbidden) + })) + defer srv.Close() + + c := &OneDriveUploadConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "bad"}, + "path": "file.txt", + "content": "data", + }) + if err == nil { + t.Fatal("expected error for 403") + } +} + +func TestRegistry_OneDriveSharePointConnectors(t *testing.T) { + r := NewRegistry() + for _, action := range []string{"onedrive/upload", "sharepoint/list_items"} { + if _, err := r.Get(action); err != nil { + t.Errorf("%s not registered: %v", action, err) + } + } +} diff --git a/packages/engine/internal/connector/quickbooks.go b/packages/engine/internal/connector/quickbooks.go new file mode 100644 index 0000000..dae6c95 --- /dev/null +++ b/packages/engine/internal/connector/quickbooks.go @@ -0,0 +1,199 @@ +package connector + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" +) + +const quickbooksBaseURL = "https://quickbooks.api.intuit.com/v3/company" + +// QuickBooksCreateInvoiceConnector creates an invoice in QuickBooks Online. +type QuickBooksCreateInvoiceConnector struct { + Client *http.Client + baseURL string +} + +func (c *QuickBooksCreateInvoiceConnector) apiURL(realmID, path string) string { + if c.baseURL != "" { + return c.baseURL + path + } + return fmt.Sprintf("%s/%s%s", quickbooksBaseURL, realmID, path) +} + +func (c *QuickBooksCreateInvoiceConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + realmID, token, err := extractQuickBooksCredential(params) + if err != nil { + return nil, fmt.Errorf("quickbooks/create_invoice: %w", err) + } + + lineItems, _ := params["line"].([]any) + if len(lineItems) == 0 { + return nil, fmt.Errorf("quickbooks/create_invoice: line is required") + } + + invoice := map[string]any{"Line": lineItems} + if customer, ok := params["customer_ref"].(map[string]any); ok { + invoice["CustomerRef"] = customer + } + if dueDate, ok := params["due_date"].(string); ok && dueDate != "" { + invoice["DueDate"] = dueDate + } + if txnDate, ok := params["txn_date"].(string); ok && txnDate != "" { + invoice["TxnDate"] = txnDate + } + if memo, ok := params["customer_memo"].(string); ok && memo != "" { + invoice["CustomerMemo"] = map[string]any{"value": memo} + } + + return qboPost(ctx, c.Client, c.apiURL(realmID, "/invoice"), token, invoice, "quickbooks/create_invoice") +} + +// QuickBooksListInvoicesConnector queries invoices from QuickBooks Online. +type QuickBooksListInvoicesConnector struct { + Client *http.Client + baseURL string +} + +func (c *QuickBooksListInvoicesConnector) apiURL(realmID, path string) string { + if c.baseURL != "" { + return c.baseURL + path + } + return fmt.Sprintf("%s/%s%s", quickbooksBaseURL, realmID, path) +} + +func (c *QuickBooksListInvoicesConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + realmID, token, err := extractQuickBooksCredential(params) + if err != nil { + return nil, fmt.Errorf("quickbooks/list_invoices: %w", err) + } + + query := "SELECT * FROM Invoice" + if where, ok := params["where"].(string); ok && where != "" { + query += " WHERE " + where + } + if orderBy, ok := params["order_by"].(string); ok && orderBy != "" { + query += " ORDERBY " + orderBy + } + maxResults := 20 + if m, ok := extractInt(params["max_results"]); ok && m > 0 { + maxResults = m + } + query += fmt.Sprintf(" MAXRESULTS %d", maxResults) + if startPos, ok := extractInt(params["start_position"]); ok && startPos > 0 { + query += fmt.Sprintf(" STARTPOSITION %d", startPos) + } + + qv := url.Values{} + qv.Set("query", query) + qv.Set("minorversion", "65") + endpoint := c.apiURL(realmID, "/query?"+qv.Encode()) + + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("quickbooks/list_invoices: creating request: %w", err) + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Accept", "application/json") + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("quickbooks/list_invoices: %w", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("quickbooks/list_invoices: reading response: %w", err) + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("quickbooks/list_invoices: QuickBooks API returned %d: %s", resp.StatusCode, truncate(string(body), 500)) + } + + var result struct { + QueryResponse struct { + Invoice []any `json:"Invoice"` + TotalCount int `json:"totalCount"` + } `json:"QueryResponse"` + } + if err := json.Unmarshal(body, &result); err != nil { + return nil, fmt.Errorf("quickbooks/list_invoices: parsing response: %w", err) + } + invoices := result.QueryResponse.Invoice + if invoices == nil { + invoices = []any{} + } + return map[string]any{ + "invoices": invoices, + "count": len(invoices), + "total_count": result.QueryResponse.TotalCount, + }, nil +} + +func extractQuickBooksCredential(params map[string]any) (realmID, token string, err error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return "", "", fmt.Errorf("credential is required") + } + delete(params, "_credential") + + var cred map[string]string + switch v := raw.(type) { + case map[string]string: + cred = v + case map[string]any: + cred = make(map[string]string, len(v)) + for k, val := range v { + if s, ok := val.(string); ok { + cred[k] = s + } + } + default: + return "", "", fmt.Errorf("credential is required") + } + + realmID = cred["realm_id"] + if realmID == "" { + return "", "", fmt.Errorf("credential must contain a 'realm_id' field") + } + token = cred["access_token"] + if token == "" { + return "", "", fmt.Errorf("credential must contain an 'access_token' field") + } + return realmID, token, nil +} + +func qboPost(ctx context.Context, client *http.Client, endpoint, token string, body map[string]any, action string) (map[string]any, error) { + reqJSON, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("%s: marshaling request: %w", action, err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", endpoint+"?minorversion=65", bytes.NewReader(reqJSON)) + if err != nil { + return nil, fmt.Errorf("%s: creating request: %w", action, err) + } + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "application/json") + + resp, err := httpClient(client).Do(req) + if err != nil { + return nil, fmt.Errorf("%s: %w", action, err) + } + defer resp.Body.Close() + + respBody, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("%s: reading response: %w", action, err) + } + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + return nil, fmt.Errorf("%s: QuickBooks API returned %d: %s", action, resp.StatusCode, truncate(string(respBody), 500)) + } + + return parseJSONBody(respBody, action) +} diff --git a/packages/engine/internal/connector/quickbooks_test.go b/packages/engine/internal/connector/quickbooks_test.go new file mode 100644 index 0000000..c52878f --- /dev/null +++ b/packages/engine/internal/connector/quickbooks_test.go @@ -0,0 +1,161 @@ +package connector + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestQuickBooksCreateInvoiceConnector_CreatesInvoice(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" { + t.Errorf("expected POST, got %s", r.Method) + } + if r.Header.Get("Authorization") != "Bearer qb-token" { + t.Errorf("unexpected Authorization: %s", r.Header.Get("Authorization")) + } + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + lineItems, _ := body["Line"].([]any) + if len(lineItems) != 1 { + t.Errorf("expected 1 line item, got %d", len(lineItems)) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "Invoice": map[string]any{"Id": "1", "DocNumber": "INV-001"}, + }) + })) + defer srv.Close() + + c := &QuickBooksCreateInvoiceConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"realm_id": "123456", "access_token": "qb-token"}, + "line": []any{ + map[string]any{ + "Amount": 100.0, + "DetailType": "SalesItemLineDetail", + "SalesItemLineDetail": map[string]any{"ItemRef": map[string]any{"value": "1", "name": "Services"}}, + }, + }, + }) + if err != nil { + t.Fatal(err) + } + if out["Invoice"] == nil { + t.Errorf("expected Invoice in response, got %v", out) + } +} + +func TestQuickBooksCreateInvoiceConnector_MissingLine(t *testing.T) { + c := &QuickBooksCreateInvoiceConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"realm_id": "123", "access_token": "tok"}, + }) + if err == nil { + t.Fatal("expected error for missing line") + } +} + +func TestQuickBooksCreateInvoiceConnector_MissingCredential(t *testing.T) { + c := &QuickBooksCreateInvoiceConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "line": []any{map[string]any{"Amount": 100}}, + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestQuickBooksCreateInvoiceConnector_MissingRealmID(t *testing.T) { + c := &QuickBooksCreateInvoiceConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"access_token": "tok"}, + "line": []any{map[string]any{"Amount": 100}}, + }) + if err == nil { + t.Fatal("expected error for missing realm_id") + } +} + +func TestQuickBooksListInvoicesConnector_ListsInvoices(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" { + t.Errorf("expected GET, got %s", r.Method) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "QueryResponse": map[string]any{ + "Invoice": []any{map[string]any{"Id": "1"}, map[string]any{"Id": "2"}}, + "totalCount": 2, + }, + }) + })) + defer srv.Close() + + c := &QuickBooksListInvoicesConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"realm_id": "123", "access_token": "tok"}, + }) + if err != nil { + t.Fatal(err) + } + if out["count"].(int) != 2 { + t.Errorf("expected count=2, got %v", out["count"]) + } +} + +func TestQuickBooksListInvoicesConnector_MissingAccessToken(t *testing.T) { + c := &QuickBooksListInvoicesConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"realm_id": "123"}, + }) + if err == nil { + t.Fatal("expected error for missing access_token") + } +} + +func TestQuickBooksCreateInvoiceConnector_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"Fault":{"Error":[{"Message":"Token expired"}]}}`, http.StatusUnauthorized) + })) + defer srv.Close() + + c := &QuickBooksCreateInvoiceConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"realm_id": "123", "access_token": "bad"}, + "line": []any{map[string]any{"Amount": 100}}, + }) + if err == nil { + t.Fatal("expected error for 401") + } +} + +func TestQuickBooksCreateInvoiceConnector_MapAnyCredential(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer maptoken" { + t.Errorf("unexpected Authorization: %s", r.Header.Get("Authorization")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"Invoice": map[string]any{"Id": "2"}}) + })) + defer srv.Close() + + c := &QuickBooksCreateInvoiceConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"realm_id": "123", "access_token": "maptoken"}, + "line": []any{map[string]any{"Amount": 100}}, + }) + if err != nil { + t.Fatal(err) + } +} + +func TestRegistry_QuickBooksConnectors(t *testing.T) { + r := NewRegistry() + for _, action := range []string{"quickbooks/create_invoice", "quickbooks/list_invoices"} { + if _, err := r.Get(action); err != nil { + t.Errorf("%s not registered: %v", action, err) + } + } +} diff --git a/packages/engine/internal/connector/rabbitmq.go b/packages/engine/internal/connector/rabbitmq.go new file mode 100644 index 0000000..4eaf3ce --- /dev/null +++ b/packages/engine/internal/connector/rabbitmq.go @@ -0,0 +1,134 @@ +package connector + +import ( + "context" + "fmt" + "time" + + amqp "github.com/rabbitmq/amqp091-go" +) + +// RabbitMQPublishConnector publishes a message to a RabbitMQ exchange or queue. +type RabbitMQPublishConnector struct{} + +func (c *RabbitMQPublishConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + conn, ch, err := newRabbitMQChannel(params) + if err != nil { + return nil, fmt.Errorf("rabbitmq/publish: %w", err) + } + defer conn.Close() + defer ch.Close() + + exchange, _ := params["exchange"].(string) + routingKey, _ := params["routing_key"].(string) + if exchange == "" && routingKey == "" { + return nil, fmt.Errorf("rabbitmq/publish: exchange or routing_key is required") + } + + body, _ := params["body"].(string) + if body == "" { + return nil, fmt.Errorf("rabbitmq/publish: body is required") + } + + contentType := "text/plain" + if ct, ok := params["content_type"].(string); ok && ct != "" { + contentType = ct + } + + msg := amqp.Publishing{ + ContentType: contentType, + Body: []byte(body), + DeliveryMode: amqp.Persistent, + Timestamp: time.Now(), + } + + if err := ch.PublishWithContext(ctx, exchange, routingKey, false, false, msg); err != nil { + return nil, fmt.Errorf("rabbitmq/publish: %w", err) + } + return map[string]any{"ok": true}, nil +} + +// RabbitMQConsumeConnector consumes up to N messages from a RabbitMQ queue (non-blocking poll). +type RabbitMQConsumeConnector struct{} + +func (c *RabbitMQConsumeConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + conn, ch, err := newRabbitMQChannel(params) + if err != nil { + return nil, fmt.Errorf("rabbitmq/consume: %w", err) + } + defer conn.Close() + defer ch.Close() + + queue, _ := params["queue"].(string) + if queue == "" { + return nil, fmt.Errorf("rabbitmq/consume: queue is required") + } + + maxMessages := 10 + if m, ok := extractInt(params["max_messages"]); ok && m > 0 { + maxMessages = m + } + + autoAck := true + if a, ok := params["auto_ack"].(bool); ok { + autoAck = a + } + + var messages []any + for i := 0; i < maxMessages; i++ { + msg, ok, err := ch.Get(queue, autoAck) + if err != nil { + return nil, fmt.Errorf("rabbitmq/consume: %w", err) + } + if !ok { + break + } + messages = append(messages, map[string]any{ + "body": string(msg.Body), + "content_type": msg.ContentType, + "delivery_tag": msg.DeliveryTag, + "routing_key": msg.RoutingKey, + "exchange": msg.Exchange, + }) + } + + if messages == nil { + messages = []any{} + } + return map[string]any{"messages": messages, "count": len(messages)}, nil +} + +// newRabbitMQChannel dials a RabbitMQ connection and opens a channel. +// Credential: {url: "amqp://user:pass@host:5672/"} +func newRabbitMQChannel(params map[string]any) (*amqp.Connection, *amqp.Channel, error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return nil, nil, fmt.Errorf("credential is required") + } + delete(params, "_credential") + + var amqpURL string + switch cred := raw.(type) { + case map[string]string: + amqpURL = cred["url"] + case map[string]any: + amqpURL, _ = cred["url"].(string) + default: + return nil, nil, fmt.Errorf("credential is required") + } + if amqpURL == "" { + return nil, nil, fmt.Errorf("credential must contain a 'url' field (amqp://...)") + } + + conn, err := amqp.Dial(amqpURL) + if err != nil { + return nil, nil, fmt.Errorf("connecting to RabbitMQ: %w", err) + } + + ch, err := conn.Channel() + if err != nil { + conn.Close() + return nil, nil, fmt.Errorf("opening channel: %w", err) + } + return conn, ch, nil +} diff --git a/packages/engine/internal/connector/rabbitmq_test.go b/packages/engine/internal/connector/rabbitmq_test.go new file mode 100644 index 0000000..caec439 --- /dev/null +++ b/packages/engine/internal/connector/rabbitmq_test.go @@ -0,0 +1,93 @@ +package connector + +import ( + "testing" +) + +func TestRabbitMQPublishConnector_MissingCredential(t *testing.T) { + c := &RabbitMQPublishConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "exchange": "events", + "routing_key": "order.created", + "body": "hello", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestRabbitMQPublishConnector_MissingURL(t *testing.T) { + c := &RabbitMQPublishConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"host": "localhost"}, + "exchange": "events", + "body": "hello", + }) + if err == nil { + t.Fatal("expected error for missing url in credential") + } +} + +func TestRabbitMQPublishConnector_MissingExchangeAndRoutingKey(t *testing.T) { + c := &RabbitMQPublishConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "amqp://localhost:5672/"}, + "body": "hello", + }) + if err == nil { + t.Fatal("expected error for missing exchange and routing_key") + } +} + +func TestRabbitMQPublishConnector_MissingBody(t *testing.T) { + c := &RabbitMQPublishConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "amqp://localhost:5672/"}, + "routing_key": "myqueue", + }) + if err == nil { + t.Fatal("expected error for missing body") + } +} + +func TestRabbitMQConsumeConnector_MissingCredential(t *testing.T) { + c := &RabbitMQConsumeConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "queue": "myqueue", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestRabbitMQConsumeConnector_MissingQueue(t *testing.T) { + c := &RabbitMQConsumeConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"url": "amqp://localhost:5672/"}, + }) + if err == nil { + t.Fatal("expected error for missing queue") + } +} + +func TestRabbitMQPublishConnector_MapAnyCredential(t *testing.T) { + c := &RabbitMQPublishConnector{} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"url": "amqp://localhost:5672/"}, + "routing_key": "myqueue", + "body": "hello", + }) + // Expect a connection error (no real RabbitMQ running), not a parse error. + if err == nil { + t.Fatal("expected connection error") + } +} + +func TestRegistry_RabbitMQConnectors(t *testing.T) { + r := NewRegistry() + for _, action := range []string{"rabbitmq/publish", "rabbitmq/consume"} { + if _, err := r.Get(action); err != nil { + t.Errorf("%s not registered: %v", action, err) + } + } +} diff --git a/packages/engine/internal/connector/shopify.go b/packages/engine/internal/connector/shopify.go new file mode 100644 index 0000000..2304fed --- /dev/null +++ b/packages/engine/internal/connector/shopify.go @@ -0,0 +1,224 @@ +package connector + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" +) + +const shopifyAPIVersion = "2024-01" + +// ShopifyListOrdersConnector lists orders from a Shopify store. +type ShopifyListOrdersConnector struct { + Client *http.Client + baseURL string +} + +func (c *ShopifyListOrdersConnector) apiURL(shop, path string) string { + if c.baseURL != "" { + return c.baseURL + path + } + return fmt.Sprintf("https://%s.myshopify.com/admin/api/%s%s", shop, shopifyAPIVersion, path) +} + +func (c *ShopifyListOrdersConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + shop, token, err := extractShopifyCredential(params) + if err != nil { + return nil, fmt.Errorf("shopify/list_orders: %w", err) + } + + query := url.Values{} + if status, ok := params["status"].(string); ok && status != "" { + query.Set("status", status) + } + if limit, ok := extractInt(params["limit"]); ok && limit > 0 { + query.Set("limit", fmt.Sprintf("%d", limit)) + } + if sinceID, ok := params["since_id"].(string); ok && sinceID != "" { + query.Set("since_id", sinceID) + } + + endpoint := c.apiURL(shop, "/orders.json") + if len(query) > 0 { + endpoint += "?" + query.Encode() + } + + out, err := shopifyGet(ctx, c.Client, endpoint, token, "shopify/list_orders") + if err != nil { + return nil, err + } + + orders, _ := out["orders"].([]any) + out["count"] = len(orders) + return out, nil +} + +// ShopifyListProductsConnector lists products from a Shopify store. +type ShopifyListProductsConnector struct { + Client *http.Client + baseURL string +} + +func (c *ShopifyListProductsConnector) apiURL(shop, path string) string { + if c.baseURL != "" { + return c.baseURL + path + } + return fmt.Sprintf("https://%s.myshopify.com/admin/api/%s%s", shop, shopifyAPIVersion, path) +} + +func (c *ShopifyListProductsConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + shop, token, err := extractShopifyCredential(params) + if err != nil { + return nil, fmt.Errorf("shopify/list_products: %w", err) + } + + query := url.Values{} + if limit, ok := extractInt(params["limit"]); ok && limit > 0 { + query.Set("limit", fmt.Sprintf("%d", limit)) + } + if productType, ok := params["product_type"].(string); ok && productType != "" { + query.Set("product_type", productType) + } + if vendor, ok := params["vendor"].(string); ok && vendor != "" { + query.Set("vendor", vendor) + } + + endpoint := c.apiURL(shop, "/products.json") + if len(query) > 0 { + endpoint += "?" + query.Encode() + } + + out, err := shopifyGet(ctx, c.Client, endpoint, token, "shopify/list_products") + if err != nil { + return nil, err + } + + products, _ := out["products"].([]any) + out["count"] = len(products) + return out, nil +} + +// ShopifyCreateOrderConnector creates an order in a Shopify store. +type ShopifyCreateOrderConnector struct { + Client *http.Client + baseURL string +} + +func (c *ShopifyCreateOrderConnector) apiURL(shop, path string) string { + if c.baseURL != "" { + return c.baseURL + path + } + return fmt.Sprintf("https://%s.myshopify.com/admin/api/%s%s", shop, shopifyAPIVersion, path) +} + +func (c *ShopifyCreateOrderConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + shop, token, err := extractShopifyCredential(params) + if err != nil { + return nil, fmt.Errorf("shopify/create_order: %w", err) + } + + lineItems, _ := params["line_items"].([]any) + if len(lineItems) == 0 { + return nil, fmt.Errorf("shopify/create_order: line_items is required") + } + + order := map[string]any{"line_items": lineItems} + if email, ok := params["email"].(string); ok && email != "" { + order["email"] = email + } + if customer, ok := params["customer"].(map[string]any); ok { + order["customer"] = customer + } + if note, ok := params["note"].(string); ok && note != "" { + order["note"] = note + } + + reqJSON, err := json.Marshal(map[string]any{"order": order}) + if err != nil { + return nil, fmt.Errorf("shopify/create_order: marshaling request: %w", err) + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.apiURL(shop, "/orders.json"), bytes.NewReader(reqJSON)) + if err != nil { + return nil, fmt.Errorf("shopify/create_order: creating request: %w", err) + } + req.Header.Set("X-Shopify-Access-Token", token) + req.Header.Set("Content-Type", "application/json") + + resp, err := httpClient(c.Client).Do(req) + if err != nil { + return nil, fmt.Errorf("shopify/create_order: %w", err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("shopify/create_order: reading response: %w", err) + } + if resp.StatusCode != http.StatusCreated && resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("shopify/create_order: Shopify API returned %d: %s", resp.StatusCode, truncate(string(body), 500)) + } + + return parseJSONBody(body, "shopify/create_order") +} + +func extractShopifyCredential(params map[string]any) (shop, token string, err error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return "", "", fmt.Errorf("credential is required") + } + delete(params, "_credential") + + var cred map[string]string + switch v := raw.(type) { + case map[string]string: + cred = v + case map[string]any: + cred = make(map[string]string, len(v)) + for k, val := range v { + if s, ok := val.(string); ok { + cred[k] = s + } + } + default: + return "", "", fmt.Errorf("credential is required") + } + + shop = cred["shop"] + if shop == "" { + return "", "", fmt.Errorf("credential must contain a 'shop' field (subdomain only, e.g. mystore)") + } + token = cred["token"] + if token == "" { + return "", "", fmt.Errorf("credential must contain a 'token' field") + } + return shop, token, nil +} + +func shopifyGet(ctx context.Context, client *http.Client, endpoint, token, action string) (map[string]any, error) { + req, err := http.NewRequestWithContext(ctx, "GET", endpoint, nil) + if err != nil { + return nil, fmt.Errorf("%s: creating request: %w", action, err) + } + req.Header.Set("X-Shopify-Access-Token", token) + + resp, err := httpClient(client).Do(req) + if err != nil { + return nil, fmt.Errorf("%s: %w", action, err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("%s: reading response: %w", action, err) + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("%s: Shopify API returned %d: %s", action, resp.StatusCode, truncate(string(body), 500)) + } + + return parseJSONBody(body, action) +} diff --git a/packages/engine/internal/connector/shopify_test.go b/packages/engine/internal/connector/shopify_test.go new file mode 100644 index 0000000..b66641f --- /dev/null +++ b/packages/engine/internal/connector/shopify_test.go @@ -0,0 +1,183 @@ +package connector + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestShopifyListOrdersConnector_ListsOrders(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "GET" || r.URL.Path != "/orders.json" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + if r.Header.Get("X-Shopify-Access-Token") != "shpat_token" { + t.Errorf("unexpected token: %s", r.Header.Get("X-Shopify-Access-Token")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "orders": []any{ + map[string]any{"id": 1001, "name": "#1001"}, + map[string]any{"id": 1002, "name": "#1002"}, + }, + }) + })) + defer srv.Close() + + c := &ShopifyListOrdersConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"shop": "mystore", "token": "shpat_token"}, + }) + if err != nil { + t.Fatal(err) + } + if out["count"].(int) != 2 { + t.Errorf("expected count=2, got %v", out["count"]) + } +} + +func TestShopifyListOrdersConnector_WithStatus(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("status") != "open" { + t.Errorf("expected status=open, got %s", r.URL.Query().Get("status")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"orders": []any{}}) + })) + defer srv.Close() + + c := &ShopifyListOrdersConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"shop": "mystore", "token": "tok"}, + "status": "open", + }) + if err != nil { + t.Fatal(err) + } +} + +func TestShopifyListOrdersConnector_MissingCredential(t *testing.T) { + c := &ShopifyListOrdersConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{}) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestShopifyListOrdersConnector_MissingShop(t *testing.T) { + c := &ShopifyListOrdersConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"token": "tok"}, + }) + if err == nil { + t.Fatal("expected error for missing shop") + } +} + +func TestShopifyListProductsConnector_ListsProducts(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/products.json" { + t.Errorf("unexpected path: %s", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{ + "products": []any{ + map[string]any{"id": 2001, "title": "Widget"}, + }, + }) + })) + defer srv.Close() + + c := &ShopifyListProductsConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"shop": "mystore", "token": "tok"}, + }) + if err != nil { + t.Fatal(err) + } + if out["count"].(int) != 1 { + t.Errorf("expected count=1, got %v", out["count"]) + } +} + +func TestShopifyCreateOrderConnector_CreatesOrder(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" || r.URL.Path != "/orders.json" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + var body map[string]any + json.NewDecoder(r.Body).Decode(&body) + order, _ := body["order"].(map[string]any) + lineItems, _ := order["line_items"].([]any) + if len(lineItems) != 1 { + t.Errorf("expected 1 line item, got %d", len(lineItems)) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusCreated) + json.NewEncoder(w).Encode(map[string]any{"order": map[string]any{"id": 3001, "name": "#3001"}}) + })) + defer srv.Close() + + c := &ShopifyCreateOrderConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"shop": "mystore", "token": "tok"}, + "line_items": []any{map[string]any{"variant_id": 1, "quantity": 1}}, + }) + if err != nil { + t.Fatal(err) + } +} + +func TestShopifyCreateOrderConnector_MissingLineItems(t *testing.T) { + c := &ShopifyCreateOrderConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"shop": "mystore", "token": "tok"}, + }) + if err == nil { + t.Fatal("expected error for missing line_items") + } +} + +func TestShopifyListOrdersConnector_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"errors":"[API] Invalid API key or access token"}`, http.StatusUnauthorized) + })) + defer srv.Close() + + c := &ShopifyListOrdersConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"shop": "mystore", "token": "bad"}, + }) + if err == nil { + t.Fatal("expected error for 401") + } +} + +func TestShopifyListOrdersConnector_MapAnyCredential(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("X-Shopify-Access-Token") != "maptoken" { + t.Errorf("unexpected token: %s", r.Header.Get("X-Shopify-Access-Token")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"orders": []any{}}) + })) + defer srv.Close() + + c := &ShopifyListOrdersConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"shop": "mystore", "token": "maptoken"}, + }) + if err != nil { + t.Fatal(err) + } +} + +func TestRegistry_ShopifyConnectors(t *testing.T) { + r := NewRegistry() + for _, action := range []string{"shopify/list_orders", "shopify/list_products", "shopify/create_order"} { + if _, err := r.Get(action); err != nil { + t.Errorf("%s not registered: %v", action, err) + } + } +} diff --git a/packages/engine/internal/connector/stripe.go b/packages/engine/internal/connector/stripe.go new file mode 100644 index 0000000..ffa3bfe --- /dev/null +++ b/packages/engine/internal/connector/stripe.go @@ -0,0 +1,180 @@ +package connector + +import ( + "context" + "fmt" + "io" + "net/http" + "net/url" + "strings" +) + +const stripeBaseURL = "https://api.stripe.com/v1" + +// StripeCreateChargeConnector creates a one-time charge via the Stripe API. +type StripeCreateChargeConnector struct { + Client *http.Client + baseURL string +} + +func (c *StripeCreateChargeConnector) apiURL(path string) string { + base := c.baseURL + if base == "" { + base = stripeBaseURL + } + return base + path +} + +func (c *StripeCreateChargeConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + apiKey, err := extractStripeCredential(params) + if err != nil { + return nil, fmt.Errorf("stripe/create_charge: %w", err) + } + + amount, ok := extractInt(params["amount"]) + if !ok || amount <= 0 { + return nil, fmt.Errorf("stripe/create_charge: amount is required (positive integer in smallest currency unit)") + } + currency, _ := params["currency"].(string) + if currency == "" { + return nil, fmt.Errorf("stripe/create_charge: currency is required") + } + + form := url.Values{} + form.Set("amount", fmt.Sprintf("%d", amount)) + form.Set("currency", strings.ToLower(currency)) + if source, ok := params["source"].(string); ok && source != "" { + form.Set("source", source) + } + if customer, ok := params["customer"].(string); ok && customer != "" { + form.Set("customer", customer) + } + if desc, ok := params["description"].(string); ok && desc != "" { + form.Set("description", desc) + } + + return stripePost(ctx, c.Client, c.apiURL("/charges"), apiKey, form, "stripe/create_charge") +} + +// StripeCreateCustomerConnector creates a customer object in Stripe. +type StripeCreateCustomerConnector struct { + Client *http.Client + baseURL string +} + +func (c *StripeCreateCustomerConnector) apiURL(path string) string { + base := c.baseURL + if base == "" { + base = stripeBaseURL + } + return base + path +} + +func (c *StripeCreateCustomerConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + apiKey, err := extractStripeCredential(params) + if err != nil { + return nil, fmt.Errorf("stripe/create_customer: %w", err) + } + + form := url.Values{} + if email, ok := params["email"].(string); ok && email != "" { + form.Set("email", email) + } + if name, ok := params["name"].(string); ok && name != "" { + form.Set("name", name) + } + if desc, ok := params["description"].(string); ok && desc != "" { + form.Set("description", desc) + } + if phone, ok := params["phone"].(string); ok && phone != "" { + form.Set("phone", phone) + } + + return stripePost(ctx, c.Client, c.apiURL("/customers"), apiKey, form, "stripe/create_customer") +} + +// StripeCreateRefundConnector refunds a charge via the Stripe API. +type StripeCreateRefundConnector struct { + Client *http.Client + baseURL string +} + +func (c *StripeCreateRefundConnector) apiURL(path string) string { + base := c.baseURL + if base == "" { + base = stripeBaseURL + } + return base + path +} + +func (c *StripeCreateRefundConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { + apiKey, err := extractStripeCredential(params) + if err != nil { + return nil, fmt.Errorf("stripe/create_refund: %w", err) + } + + charge, _ := params["charge"].(string) + if charge == "" { + return nil, fmt.Errorf("stripe/create_refund: charge is required") + } + + form := url.Values{} + form.Set("charge", charge) + if amount, ok := extractInt(params["amount"]); ok && amount > 0 { + form.Set("amount", fmt.Sprintf("%d", amount)) + } + if reason, ok := params["reason"].(string); ok && reason != "" { + form.Set("reason", reason) + } + + return stripePost(ctx, c.Client, c.apiURL("/refunds"), apiKey, form, "stripe/create_refund") +} + +func extractStripeCredential(params map[string]any) (string, error) { + raw, ok := params["_credential"] + if !ok || raw == nil { + return "", fmt.Errorf("credential is required") + } + delete(params, "_credential") + + var apiKey string + switch cred := raw.(type) { + case map[string]string: + apiKey = cred["api_key"] + case map[string]any: + apiKey, _ = cred["api_key"].(string) + default: + return "", fmt.Errorf("credential is required") + } + if apiKey == "" { + return "", fmt.Errorf("credential must contain an 'api_key' field") + } + return apiKey, nil +} + +// stripePost sends a form-encoded POST to the Stripe API and returns the parsed JSON response. +func stripePost(ctx context.Context, client *http.Client, endpoint, apiKey string, form url.Values, action string) (map[string]any, error) { + req, err := http.NewRequestWithContext(ctx, "POST", endpoint, strings.NewReader(form.Encode())) + if err != nil { + return nil, fmt.Errorf("%s: creating request: %w", action, err) + } + req.SetBasicAuth(apiKey, "") + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + resp, err := httpClient(client).Do(req) + if err != nil { + return nil, fmt.Errorf("%s: %w", action, err) + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) + if err != nil { + return nil, fmt.Errorf("%s: reading response: %w", action, err) + } + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("%s: Stripe API returned %d: %s", action, resp.StatusCode, truncate(string(body), 500)) + } + + return parseJSONBody(body, action) +} diff --git a/packages/engine/internal/connector/stripe_test.go b/packages/engine/internal/connector/stripe_test.go new file mode 100644 index 0000000..584485a --- /dev/null +++ b/packages/engine/internal/connector/stripe_test.go @@ -0,0 +1,190 @@ +package connector + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestStripeCreateChargeConnector_CreatesCharge(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != "POST" || r.URL.Path != "/charges" { + t.Errorf("unexpected %s %s", r.Method, r.URL.Path) + } + user, _, ok := r.BasicAuth() + if !ok || user != "sk_test_key" { + t.Errorf("unexpected auth user: %s", user) + } + if err := r.ParseForm(); err != nil { + t.Fatal(err) + } + if r.FormValue("amount") != "2000" { + t.Errorf("unexpected amount: %s", r.FormValue("amount")) + } + if r.FormValue("currency") != "usd" { + t.Errorf("unexpected currency: %s", r.FormValue("currency")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"id": "ch_123", "status": "succeeded", "amount": 2000}) + })) + defer srv.Close() + + c := &StripeCreateChargeConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "sk_test_key"}, + "amount": 2000, + "currency": "usd", + "source": "tok_visa", + }) + if err != nil { + t.Fatal(err) + } + if out["id"] != "ch_123" { + t.Errorf("expected id=ch_123, got %v", out["id"]) + } +} + +func TestStripeCreateChargeConnector_MissingAmount(t *testing.T) { + c := &StripeCreateChargeConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "sk_test"}, + "currency": "usd", + }) + if err == nil { + t.Fatal("expected error for missing amount") + } +} + +func TestStripeCreateChargeConnector_MissingCurrency(t *testing.T) { + c := &StripeCreateChargeConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "sk_test"}, + "amount": 2000, + }) + if err == nil { + t.Fatal("expected error for missing currency") + } +} + +func TestStripeCreateChargeConnector_MissingCredential(t *testing.T) { + c := &StripeCreateChargeConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "amount": 2000, + "currency": "usd", + }) + if err == nil { + t.Fatal("expected error for missing credential") + } +} + +func TestStripeCreateChargeConnector_APIError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, `{"error":{"message":"No such token"}}`, http.StatusBadRequest) + })) + defer srv.Close() + + c := &StripeCreateChargeConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "sk_bad"}, + "amount": 500, + "currency": "usd", + "source": "tok_bad", + }) + if err == nil { + t.Fatal("expected error for 400") + } +} + +func TestStripeCreateCustomerConnector_CreatesCustomer(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/customers" { + t.Errorf("unexpected path: %s", r.URL.Path) + } + r.ParseForm() + if r.FormValue("email") != "alice@example.com" { + t.Errorf("unexpected email: %s", r.FormValue("email")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"id": "cus_abc", "email": "alice@example.com"}) + })) + defer srv.Close() + + c := &StripeCreateCustomerConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "sk_test"}, + "email": "alice@example.com", + "name": "Alice", + }) + if err != nil { + t.Fatal(err) + } + if out["id"] != "cus_abc" { + t.Errorf("expected id=cus_abc, got %v", out["id"]) + } +} + +func TestStripeCreateRefundConnector_CreatesRefund(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + r.ParseForm() + if r.FormValue("charge") != "ch_123" { + t.Errorf("unexpected charge: %s", r.FormValue("charge")) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"id": "re_abc", "status": "succeeded"}) + })) + defer srv.Close() + + c := &StripeCreateRefundConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "sk_test"}, + "charge": "ch_123", + }) + if err != nil { + t.Fatal(err) + } + if out["id"] != "re_abc" { + t.Errorf("expected id=re_abc, got %v", out["id"]) + } +} + +func TestStripeCreateRefundConnector_MissingCharge(t *testing.T) { + c := &StripeCreateRefundConnector{baseURL: "http://unused"} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"api_key": "sk_test"}, + }) + if err == nil { + t.Fatal("expected error for missing charge") + } +} + +func TestStripeCreateChargeConnector_MapAnyCredential(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + user, _, _ := r.BasicAuth() + if user != "sk_mapkey" { + t.Errorf("unexpected auth user: %s", user) + } + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(map[string]any{"id": "ch_1", "status": "succeeded"}) + })) + defer srv.Close() + + c := &StripeCreateChargeConnector{baseURL: srv.URL} + _, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]any{"api_key": "sk_mapkey"}, + "amount": 100, + "currency": "usd", + }) + if err != nil { + t.Fatal(err) + } +} + +func TestRegistry_StripeConnectors(t *testing.T) { + r := NewRegistry() + for _, action := range []string{"stripe/create_charge", "stripe/create_customer", "stripe/create_refund"} { + if _, err := r.Get(action); err != nil { + t.Errorf("%s not registered: %v", action, err) + } + } +} diff --git a/packages/engine/internal/connector/tools.go b/packages/engine/internal/connector/tools.go index 40d3836..1d38bd0 100644 --- a/packages/engine/internal/connector/tools.go +++ b/packages/engine/internal/connector/tools.go @@ -348,6 +348,15 @@ func truncateOlderToolResults(messages []map[string]any, targetBytes int) int { return truncated } +// parseJSONBody unmarshals a JSON response body into map[string]any. +func parseJSONBody(body []byte, action string) (map[string]any, error) { + var result map[string]any + if err := json.Unmarshal(body, &result); err != nil { + return nil, fmt.Errorf("%s: parsing response: %w", action, err) + } + return result, nil +} + // extractBearerToken pulls a "token" field from the _credential param. // Accepts both map[string]string (engine-injected) and map[string]any // (JSON/CEL-deserialised) credential shapes, and deletes _credential from params. From 8561593ca29cafddd4ae9ffd2d7d3bc879803fb1 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 20:29:55 -0400 Subject: [PATCH 02/10] fix(connector): address Codex review comments on OneDrive and RabbitMQ - OneDrive: remove url.PathEscape on filePath to preserve literal "/" separators in Graph API path templates (escaping produced %2F which resolved to a file literally named "documents%2Fhello.txt") - RabbitMQ: ack each message inside the consume loop when auto_ack=false so messages are not requeued when the channel closes on return Co-Authored-By: Claude Sonnet 4.6 --- packages/engine/internal/connector/onedrive.go | 8 ++++++-- packages/engine/internal/connector/rabbitmq.go | 5 +++++ 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/packages/engine/internal/connector/onedrive.go b/packages/engine/internal/connector/onedrive.go index 6bbe1fa..3ffcc76 100644 --- a/packages/engine/internal/connector/onedrive.go +++ b/packages/engine/internal/connector/onedrive.go @@ -41,12 +41,16 @@ func (c *OneDriveUploadConnector) Execute(ctx context.Context, params map[string return nil, fmt.Errorf("onedrive/upload: content is required") } + // filePath must NOT be PathEscaped here: Graph path addressing uses + // literal "/" separators inside the /root:/...:/content template, so + // escaping them produces %2F which resolves to a file literally named + // "documents%2Fhello.txt" instead of hello.txt inside the documents folder. var endpoint string if driveID, ok := params["drive_id"].(string); ok && driveID != "" { endpoint = c.apiURL(fmt.Sprintf("/drives/%s/root:/%s:/content", - url.PathEscape(driveID), url.PathEscape(filePath))) + url.PathEscape(driveID), filePath)) } else { - endpoint = c.apiURL(fmt.Sprintf("/me/drive/root:/%s:/content", url.PathEscape(filePath))) + endpoint = c.apiURL(fmt.Sprintf("/me/drive/root:/%s:/content", filePath)) } contentType := "application/octet-stream" diff --git a/packages/engine/internal/connector/rabbitmq.go b/packages/engine/internal/connector/rabbitmq.go index 4eaf3ce..10cb5c3 100644 --- a/packages/engine/internal/connector/rabbitmq.go +++ b/packages/engine/internal/connector/rabbitmq.go @@ -90,6 +90,11 @@ func (c *RabbitMQConsumeConnector) Execute(ctx context.Context, params map[strin "routing_key": msg.RoutingKey, "exchange": msg.Exchange, }) + if !autoAck { + if err := msg.Ack(false); err != nil { + return nil, fmt.Errorf("rabbitmq/consume: acking message: %w", err) + } + } } if messages == nil { From dd60b805fcdc6f3be295187e37fe3f4d334fcf89 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 20:34:18 -0400 Subject: [PATCH 03/10] fix(connector): address CodeRabbit review on saas-connector batch - rabbitmq: validate params before dialing (exchange/routing_key/body for publish, queue for consume) so invalid requests fail without a connection - rabbitmq: use amqp.DialConfig with net.Dialer.DialContext so the dial respects context cancellation - rabbitmq: default auto_ack=false (at-least-once); true opts into at-most-once - onedrive: add escapePath helper to encode individual path segments without escaping "/" separators, so nested folders work with Graph path addressing - onedrive_test: assert request URL path to catch future slash-escaping regressions - quickbooks: reject where/order_by values that contain structural keywords (maxresults, startposition, orderby) to prevent clause injection - shopify: extract shopifyAPIURL package-level helper, removing three identical apiURL methods - go.mod: go mod tidy promotes amqp091-go from indirect to direct Co-Authored-By: Claude Sonnet 4.6 --- packages/engine/go.mod | 10 +++-- packages/engine/go.sum | 17 ++++++++ .../engine/internal/connector/onedrive.go | 21 +++++++--- .../internal/connector/onedrive_test.go | 3 ++ .../engine/internal/connector/quickbooks.go | 13 ++++++ .../engine/internal/connector/rabbitmq.go | 42 ++++++++++++------- packages/engine/internal/connector/shopify.go | 34 +++++---------- 7 files changed, 91 insertions(+), 49 deletions(-) diff --git a/packages/engine/go.mod b/packages/engine/go.mod index 0e0cbd1..527f714 100644 --- a/packages/engine/go.mod +++ b/packages/engine/go.mod @@ -22,11 +22,14 @@ require ( github.com/jackc/pgx/v5 v5.8.0 github.com/pressly/goose/v3 v3.27.0 github.com/prometheus/client_golang v1.23.2 + github.com/rabbitmq/amqp091-go v1.11.0 + github.com/redis/go-redis/v9 v9.20.0 github.com/spf13/cobra v1.10.2 github.com/spf13/viper v1.21.0 github.com/stretchr/testify v1.11.1 github.com/testcontainers/testcontainers-go v0.41.0 github.com/testcontainers/testcontainers-go/modules/postgres v0.41.0 + go.mongodb.org/mongo-driver/v2 v2.6.0 golang.org/x/oauth2 v0.35.0 golang.org/x/term v0.41.0 golang.org/x/time v0.15.0 @@ -126,8 +129,6 @@ require ( github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.66.1 // indirect github.com/prometheus/procfs v0.19.2 // indirect - github.com/rabbitmq/amqp091-go v1.11.0 // indirect - github.com/redis/go-redis/v9 v9.20.0 // indirect github.com/sagikazarmark/locafero v0.11.0 // indirect github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 // indirect github.com/sethvargo/go-retry v0.3.0 // indirect @@ -142,8 +143,11 @@ require ( github.com/tklauser/go-sysconf v0.3.16 // indirect github.com/tklauser/numcpus v0.11.0 // indirect github.com/xanzy/ssh-agent v0.3.3 // indirect + github.com/xdg-go/pbkdf2 v1.0.0 // indirect + github.com/xdg-go/scram v1.2.0 // indirect + github.com/xdg-go/stringprep v1.0.4 // indirect + github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect github.com/yusufpapurcu/wmi v1.2.4 // indirect - go.mongodb.org/mongo-driver/v2 v2.6.0 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.61.0 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.65.0 // indirect diff --git a/packages/engine/go.sum b/packages/engine/go.sum index ebc4f7b..e49e3ae 100644 --- a/packages/engine/go.sum +++ b/packages/engine/go.sum @@ -89,6 +89,10 @@ github.com/aws/smithy-go v1.24.2 h1:FzA3bu/nt/vDvmnkg+R8Xl46gmzEDam6mZ1hzmwXFng= github.com/aws/smithy-go v1.24.2/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= @@ -212,6 +216,8 @@ github.com/keybase/go-keychain v0.0.1 h1:way+bWYa6lDppZoZcgMbYsvC7GxljxrskdNInRt github.com/keybase/go-keychain v0.0.1/go.mod h1:PdEILRW3i9D8JcdM+FmY6RwkHGnhHxXwkPPMeUgOK1k= github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c= github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -341,9 +347,19 @@ github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9R github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ= github.com/xanzy/ssh-agent v0.3.3 h1:+/15pJfg/RsTxqYcX6fHqOXZwwMP+2VyYWJeWM2qQFM= github.com/xanzy/ssh-agent v0.3.3/go.mod h1:6dzNDKs0J9rVPHPhaGCukekBHKqfl+L3KghI1Bc68Uw= +github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c= +github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= +github.com/xdg-go/scram v1.2.0 h1:bYKF2AEwG5rqd1BumT4gAnvwU/M9nBp2pTSxeZw7Wvs= +github.com/xdg-go/scram v1.2.0/go.mod h1:3dlrS0iBaWKYVt2ZfA4cj48umJZ+cAEbR6/SjLA88I8= +github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8= +github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM= +github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 h1:ilQV1hzziu+LLM3zUTJ0trRztfwgjqKnBWNtSRkbmwM= +github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78/go.mod h1:aL8wCCfTfSfmXjznFBSZNN13rSJjlIOI1fUNAtF7rmI= github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= go.mongodb.org/mongo-driver/v2 v2.6.0 h1:b9sJOYrkmt4l8bY43ZenFBcPlhYIjaOfYHLtbB/5qi8= go.mongodb.org/mongo-driver/v2 v2.6.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= @@ -426,6 +442,7 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ= golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8= golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= diff --git a/packages/engine/internal/connector/onedrive.go b/packages/engine/internal/connector/onedrive.go index 3ffcc76..fc7f8f5 100644 --- a/packages/engine/internal/connector/onedrive.go +++ b/packages/engine/internal/connector/onedrive.go @@ -8,10 +8,21 @@ import ( "io" "net/http" "net/url" + "strings" ) const graphBaseURL = "https://graph.microsoft.com/v1.0" +// escapePath percent-encodes each segment of a slash-delimited path individually, +// preserving "/" as the folder separator for Microsoft Graph path-based addressing. +func escapePath(p string) string { + segs := strings.Split(p, "/") + for i, s := range segs { + segs[i] = url.PathEscape(s) + } + return strings.Join(segs, "/") +} + // OneDriveUploadConnector uploads a file to OneDrive via the Microsoft Graph API. type OneDriveUploadConnector struct { Client *http.Client @@ -41,16 +52,14 @@ func (c *OneDriveUploadConnector) Execute(ctx context.Context, params map[string return nil, fmt.Errorf("onedrive/upload: content is required") } - // filePath must NOT be PathEscaped here: Graph path addressing uses - // literal "/" separators inside the /root:/...:/content template, so - // escaping them produces %2F which resolves to a file literally named - // "documents%2Fhello.txt" instead of hello.txt inside the documents folder. + // escapePath preserves "/" as folder separators while encoding special characters + // within each segment (spaces, #, etc.) for Microsoft Graph path-based addressing. var endpoint string if driveID, ok := params["drive_id"].(string); ok && driveID != "" { endpoint = c.apiURL(fmt.Sprintf("/drives/%s/root:/%s:/content", - url.PathEscape(driveID), filePath)) + url.PathEscape(driveID), escapePath(filePath))) } else { - endpoint = c.apiURL(fmt.Sprintf("/me/drive/root:/%s:/content", filePath)) + endpoint = c.apiURL(fmt.Sprintf("/me/drive/root:/%s:/content", escapePath(filePath))) } contentType := "application/octet-stream" diff --git a/packages/engine/internal/connector/onedrive_test.go b/packages/engine/internal/connector/onedrive_test.go index c1a0d26..07c815f 100644 --- a/packages/engine/internal/connector/onedrive_test.go +++ b/packages/engine/internal/connector/onedrive_test.go @@ -12,6 +12,9 @@ func TestOneDriveUploadConnector_UploadsFile(t *testing.T) { if r.Method != "PUT" { t.Errorf("expected PUT, got %s", r.Method) } + if r.URL.Path != "/me/drive/root:/documents/hello.txt:/content" { + t.Errorf("unexpected path: %s", r.URL.Path) + } if r.Header.Get("Authorization") != "Bearer graph-token" { t.Errorf("unexpected Authorization: %s", r.Header.Get("Authorization")) } diff --git a/packages/engine/internal/connector/quickbooks.go b/packages/engine/internal/connector/quickbooks.go index dae6c95..55b5e21 100644 --- a/packages/engine/internal/connector/quickbooks.go +++ b/packages/engine/internal/connector/quickbooks.go @@ -8,6 +8,7 @@ import ( "io" "net/http" "net/url" + "strings" ) const quickbooksBaseURL = "https://quickbooks.api.intuit.com/v3/company" @@ -74,9 +75,21 @@ func (c *QuickBooksListInvoicesConnector) Execute(ctx context.Context, params ma query := "SELECT * FROM Invoice" if where, ok := params["where"].(string); ok && where != "" { + wl := strings.ToLower(where) + for _, kw := range []string{"maxresults", "startposition", "orderby"} { + if strings.Contains(wl, kw) { + return nil, fmt.Errorf("quickbooks/list_invoices: invalid where clause") + } + } query += " WHERE " + where } if orderBy, ok := params["order_by"].(string); ok && orderBy != "" { + ol := strings.ToLower(orderBy) + for _, kw := range []string{"maxresults", "startposition", "where"} { + if strings.Contains(ol, kw) { + return nil, fmt.Errorf("quickbooks/list_invoices: invalid order_by clause") + } + } query += " ORDERBY " + orderBy } maxResults := 20 diff --git a/packages/engine/internal/connector/rabbitmq.go b/packages/engine/internal/connector/rabbitmq.go index 10cb5c3..076a0cd 100644 --- a/packages/engine/internal/connector/rabbitmq.go +++ b/packages/engine/internal/connector/rabbitmq.go @@ -3,6 +3,7 @@ package connector import ( "context" "fmt" + "net" "time" amqp "github.com/rabbitmq/amqp091-go" @@ -12,13 +13,6 @@ import ( type RabbitMQPublishConnector struct{} func (c *RabbitMQPublishConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { - conn, ch, err := newRabbitMQChannel(params) - if err != nil { - return nil, fmt.Errorf("rabbitmq/publish: %w", err) - } - defer conn.Close() - defer ch.Close() - exchange, _ := params["exchange"].(string) routingKey, _ := params["routing_key"].(string) if exchange == "" && routingKey == "" { @@ -30,6 +24,13 @@ func (c *RabbitMQPublishConnector) Execute(ctx context.Context, params map[strin return nil, fmt.Errorf("rabbitmq/publish: body is required") } + conn, ch, err := newRabbitMQChannel(ctx, params) + if err != nil { + return nil, fmt.Errorf("rabbitmq/publish: %w", err) + } + defer conn.Close() + defer ch.Close() + contentType := "text/plain" if ct, ok := params["content_type"].(string); ok && ct != "" { contentType = ct @@ -49,27 +50,30 @@ func (c *RabbitMQPublishConnector) Execute(ctx context.Context, params map[strin } // RabbitMQConsumeConnector consumes up to N messages from a RabbitMQ queue (non-blocking poll). +// auto_ack defaults to false (at-least-once): each message is acknowledged after it is +// collected. Set auto_ack=true to opt into at-most-once delivery. type RabbitMQConsumeConnector struct{} func (c *RabbitMQConsumeConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { - conn, ch, err := newRabbitMQChannel(params) + queue, _ := params["queue"].(string) + if queue == "" { + return nil, fmt.Errorf("rabbitmq/consume: queue is required") + } + + conn, ch, err := newRabbitMQChannel(ctx, params) if err != nil { return nil, fmt.Errorf("rabbitmq/consume: %w", err) } defer conn.Close() defer ch.Close() - queue, _ := params["queue"].(string) - if queue == "" { - return nil, fmt.Errorf("rabbitmq/consume: queue is required") - } - maxMessages := 10 if m, ok := extractInt(params["max_messages"]); ok && m > 0 { maxMessages = m } - autoAck := true + // Default false: broker requires explicit Ack; set true to opt into at-most-once. + autoAck := false if a, ok := params["auto_ack"].(bool); ok { autoAck = a } @@ -104,8 +108,9 @@ func (c *RabbitMQConsumeConnector) Execute(ctx context.Context, params map[strin } // newRabbitMQChannel dials a RabbitMQ connection and opens a channel. +// The dial respects ctx cancellation via a context-aware net.Dialer. // Credential: {url: "amqp://user:pass@host:5672/"} -func newRabbitMQChannel(params map[string]any) (*amqp.Connection, *amqp.Channel, error) { +func newRabbitMQChannel(ctx context.Context, params map[string]any) (*amqp.Connection, *amqp.Channel, error) { raw, ok := params["_credential"] if !ok || raw == nil { return nil, nil, fmt.Errorf("credential is required") @@ -125,7 +130,12 @@ func newRabbitMQChannel(params map[string]any) (*amqp.Connection, *amqp.Channel, return nil, nil, fmt.Errorf("credential must contain a 'url' field (amqp://...)") } - conn, err := amqp.Dial(amqpURL) + d := &net.Dialer{} + conn, err := amqp.DialConfig(amqpURL, amqp.Config{ + Dial: func(network, addr string) (net.Conn, error) { + return d.DialContext(ctx, network, addr) + }, + }) if err != nil { return nil, nil, fmt.Errorf("connecting to RabbitMQ: %w", err) } diff --git a/packages/engine/internal/connector/shopify.go b/packages/engine/internal/connector/shopify.go index 2304fed..d5e94f8 100644 --- a/packages/engine/internal/connector/shopify.go +++ b/packages/engine/internal/connector/shopify.go @@ -12,19 +12,19 @@ import ( const shopifyAPIVersion = "2024-01" +func shopifyAPIURL(baseURL, shop, path string) string { + if baseURL != "" { + return baseURL + path + } + return fmt.Sprintf("https://%s.myshopify.com/admin/api/%s%s", shop, shopifyAPIVersion, path) +} + // ShopifyListOrdersConnector lists orders from a Shopify store. type ShopifyListOrdersConnector struct { Client *http.Client baseURL string } -func (c *ShopifyListOrdersConnector) apiURL(shop, path string) string { - if c.baseURL != "" { - return c.baseURL + path - } - return fmt.Sprintf("https://%s.myshopify.com/admin/api/%s%s", shop, shopifyAPIVersion, path) -} - func (c *ShopifyListOrdersConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { shop, token, err := extractShopifyCredential(params) if err != nil { @@ -42,7 +42,7 @@ func (c *ShopifyListOrdersConnector) Execute(ctx context.Context, params map[str query.Set("since_id", sinceID) } - endpoint := c.apiURL(shop, "/orders.json") + endpoint := shopifyAPIURL(c.baseURL, shop, "/orders.json") if len(query) > 0 { endpoint += "?" + query.Encode() } @@ -63,13 +63,6 @@ type ShopifyListProductsConnector struct { baseURL string } -func (c *ShopifyListProductsConnector) apiURL(shop, path string) string { - if c.baseURL != "" { - return c.baseURL + path - } - return fmt.Sprintf("https://%s.myshopify.com/admin/api/%s%s", shop, shopifyAPIVersion, path) -} - func (c *ShopifyListProductsConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { shop, token, err := extractShopifyCredential(params) if err != nil { @@ -87,7 +80,7 @@ func (c *ShopifyListProductsConnector) Execute(ctx context.Context, params map[s query.Set("vendor", vendor) } - endpoint := c.apiURL(shop, "/products.json") + endpoint := shopifyAPIURL(c.baseURL, shop, "/products.json") if len(query) > 0 { endpoint += "?" + query.Encode() } @@ -108,13 +101,6 @@ type ShopifyCreateOrderConnector struct { baseURL string } -func (c *ShopifyCreateOrderConnector) apiURL(shop, path string) string { - if c.baseURL != "" { - return c.baseURL + path - } - return fmt.Sprintf("https://%s.myshopify.com/admin/api/%s%s", shop, shopifyAPIVersion, path) -} - func (c *ShopifyCreateOrderConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { shop, token, err := extractShopifyCredential(params) if err != nil { @@ -142,7 +128,7 @@ func (c *ShopifyCreateOrderConnector) Execute(ctx context.Context, params map[st return nil, fmt.Errorf("shopify/create_order: marshaling request: %w", err) } - req, err := http.NewRequestWithContext(ctx, "POST", c.apiURL(shop, "/orders.json"), bytes.NewReader(reqJSON)) + req, err := http.NewRequestWithContext(ctx, "POST", shopifyAPIURL(c.baseURL, shop, "/orders.json"), bytes.NewReader(reqJSON)) if err != nil { return nil, fmt.Errorf("shopify/create_order: creating request: %w", err) } From b37d9fefbd68597b59cfcc484832170e7a6c8762 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 20:45:56 -0400 Subject: [PATCH 04/10] fix: exempt /metrics, /healthz, /readyz from auth; numeric UID in Dockerfile - server: register /metrics, /healthz, and /readyz on a top-level mux so Kubernetes probes and Prometheus scrapers bypass API key auth entirely. All other routes fall through to the auth/rate-limit-wrapped apiMux. - Dockerfile: create the mantle user with explicit -u 10001 -g 10001 and switch to USER 10001:10001 so kubelet can verify runAsNonRoot: true without a numeric runAsUser override in the Helm chart. Closes #139 Closes #137 Co-Authored-By: Claude Sonnet 4.6 --- packages/engine/Dockerfile | 6 +- packages/engine/internal/server/server.go | 68 ++++++++++++----------- 2 files changed, 40 insertions(+), 34 deletions(-) diff --git a/packages/engine/Dockerfile b/packages/engine/Dockerfile index 362204d..5e0e004 100644 --- a/packages/engine/Dockerfile +++ b/packages/engine/Dockerfile @@ -26,8 +26,8 @@ RUN CGO_ENABLED=0 GOOS=linux go build \ FROM alpine:3.21 RUN apk add --no-cache ca-certificates tzdata \ - && addgroup -S mantle \ - && adduser -S -G mantle -h /home/mantle -s /bin/sh mantle + && addgroup -S -g 10001 mantle \ + && adduser -S -u 10001 -G mantle -h /home/mantle -s /bin/sh mantle COPY --from=builder /mantle /usr/local/bin/mantle @@ -39,7 +39,7 @@ EXPOSE 8080 HEALTHCHECK --interval=10s --timeout=3s --start-period=5s --retries=3 \ CMD wget -qO- http://localhost:8080/healthz || exit 1 -USER mantle +USER 10001:10001 WORKDIR /home/mantle ENTRYPOINT ["mantle"] diff --git a/packages/engine/internal/server/server.go b/packages/engine/internal/server/server.go index d553559..8b5e68c 100644 --- a/packages/engine/internal/server/server.go +++ b/packages/engine/internal/server/server.go @@ -108,43 +108,43 @@ func metricsMiddleware(next http.Handler) http.Handler { // Start starts the HTTP server, cron scheduler, and webhook handler. // It blocks until the context is cancelled. func (s *Server) Start(ctx context.Context) error { - mux := http.NewServeMux() - - // Prometheus metrics endpoint. - mux.Handle("/metrics", promhttp.Handler()) + // apiMux holds all routes that require authentication (and rate limiting). + // Health and metrics endpoints are registered on the top-level mux below + // so that Kubernetes probes and Prometheus scrapes bypass auth entirely. + apiMux := http.NewServeMux() // Git push webhook endpoint — registered before the generic /hooks/ prefix // so the more-specific path is explicit (Go's ServeMux uses longest-prefix // match regardless of registration order, but explicit ordering is clearer). - mux.Handle("/hooks/git/", &GitWebhookHandler{ + apiMux.Handle("/hooks/git/", &GitWebhookHandler{ DB: s.DB, Store: s.RepoStore, Driver: s.GitDriver, }) // Webhook endpoints. - mux.HandleFunc("/hooks/", s.webhooks.ServeHTTP) + apiMux.HandleFunc("/hooks/", s.webhooks.ServeHTTP) // API endpoints. - mux.HandleFunc("POST /api/v1/run/{workflow}", s.handleRun) - mux.HandleFunc("POST /api/v1/cancel/{execution}", s.handleCancel) - mux.HandleFunc("GET /api/v1/executions", s.handleListExecutions) - mux.HandleFunc("GET /api/v1/executions/{id}", s.handleGetExecution) + apiMux.HandleFunc("POST /api/v1/run/{workflow}", s.handleRun) + apiMux.HandleFunc("POST /api/v1/cancel/{execution}", s.handleCancel) + apiMux.HandleFunc("GET /api/v1/executions", s.handleListExecutions) + apiMux.HandleFunc("GET /api/v1/executions/{id}", s.handleGetExecution) // Workflow definition endpoints. - mux.HandleFunc("GET /api/v1/workflows", s.handleListWorkflows) - mux.HandleFunc("GET /api/v1/workflows/{name}", s.handleGetWorkflow) - mux.HandleFunc("GET /api/v1/workflows/{name}/versions", s.handleListWorkflowVersions) - mux.HandleFunc("GET /api/v1/workflows/{name}/versions/{version}", s.handleGetWorkflowVersion) + apiMux.HandleFunc("GET /api/v1/workflows", s.handleListWorkflows) + apiMux.HandleFunc("GET /api/v1/workflows/{name}", s.handleGetWorkflow) + apiMux.HandleFunc("GET /api/v1/workflows/{name}/versions", s.handleListWorkflowVersions) + apiMux.HandleFunc("GET /api/v1/workflows/{name}/versions/{version}", s.handleGetWorkflowVersion) // Budget endpoints. - mux.HandleFunc("GET /api/v1/budgets", s.handleListBudgets) - mux.HandleFunc("PUT /api/v1/budgets/{provider}", s.handleSetBudget) - mux.HandleFunc("DELETE /api/v1/budgets/{provider}", s.handleDeleteBudget) - mux.HandleFunc("GET /api/v1/budgets/usage", s.handleGetUsage) + apiMux.HandleFunc("GET /api/v1/budgets", s.handleListBudgets) + apiMux.HandleFunc("PUT /api/v1/budgets/{provider}", s.handleSetBudget) + apiMux.HandleFunc("DELETE /api/v1/budgets/{provider}", s.handleDeleteBudget) + apiMux.HandleFunc("GET /api/v1/budgets/usage", s.handleGetUsage) // OpenAPI spec endpoint. - mux.HandleFunc("GET /api/v1/openapi.json", handleOpenAPISpec) + apiMux.HandleFunc("GET /api/v1/openapi.json", handleOpenAPISpec) // Start distributed engine components (worker + reaper). if cfg := config.FromContext(ctx); cfg != nil { @@ -238,24 +238,30 @@ func (s *Server) Start(ctx context.Context) error { go s.pollQueueDepth(ctx, cfg.Engine.ReaperInterval) } - // Health endpoints. /readyz checks DB connectivity only; worker/reaper - // liveness is no longer gated here to avoid flapping between poll cycles. - mux.Handle("/healthz", health.HealthzHandler()) - mux.Handle("/readyz", health.ReadyzHandler(s.DB)) - - // Wrap with metrics middleware (innermost, runs for every request). - handler := metricsMiddleware(mux) + // Wrap the API mux with metrics middleware, auth, and rate limiting. + apiHandler := metricsMiddleware(apiMux) if s.AuthStore != nil { if s.Auditor != nil { - handler = auth.AuthMiddleware(s.AuthStore, s.OIDCValidator, handler, s.Auditor) + apiHandler = auth.AuthMiddleware(s.AuthStore, s.OIDCValidator, apiHandler, s.Auditor) } else { - handler = auth.AuthMiddleware(s.AuthStore, s.OIDCValidator, handler) + apiHandler = auth.AuthMiddleware(s.AuthStore, s.OIDCValidator, apiHandler) } } - - // Apply rate limiting (after auth so rate limit keys can use API key prefix). rl := NewRateLimiter(100, 200) - handler = rl.Middleware(handler) + apiHandler = rl.Middleware(apiHandler) + + // Top-level mux: health and metrics endpoints bypass auth so that Kubernetes + // probes and Prometheus scrapers work without an API key. All other requests + // fall through to the auth-wrapped apiHandler. + mux := http.NewServeMux() + mux.Handle("/metrics", promhttp.Handler()) + mux.Handle("/healthz", health.HealthzHandler()) + // /readyz checks DB connectivity only; worker/reaper liveness is intentionally + // excluded to prevent flapping between poll cycles. + mux.Handle("/readyz", health.ReadyzHandler(s.DB)) + mux.Handle("/", apiHandler) + + handler := mux s.httpServer = &http.Server{ Addr: s.Address, From 3180bce32632e3ab9848fb557a8bdbf6e07f60f7 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 21:04:00 -0400 Subject: [PATCH 05/10] fix(connector): Okta activate param, Mailchimp upsert endpoint MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Okta: always pass ?activate= — omitting it when false causes Okta to activate the user by default, triggering unintended provisioning side effects. Mailchimp: switch add_member from POST /members to PUT /members/{hash} (MD5 of lowercased email). The POST endpoint returns a duplicate-member error on reruns; the PUT upsert endpoint is idempotent. Co-Authored-By: Claude Sonnet 4.6 --- .../engine/internal/connector/mailchimp.go | 9 ++++-- .../internal/connector/mailchimp_test.go | 4 ++- packages/engine/internal/connector/okta.go | 5 +--- .../engine/internal/connector/okta_test.go | 28 +++++++++++++++++++ 4 files changed, 39 insertions(+), 7 deletions(-) diff --git a/packages/engine/internal/connector/mailchimp.go b/packages/engine/internal/connector/mailchimp.go index 8a12dc5..07adc14 100644 --- a/packages/engine/internal/connector/mailchimp.go +++ b/packages/engine/internal/connector/mailchimp.go @@ -3,6 +3,8 @@ package connector import ( "bytes" "context" + "crypto/md5" + "encoding/hex" "encoding/json" "fmt" "io" @@ -122,8 +124,11 @@ func (c *MailchimpAddMemberConnector) Execute(ctx context.Context, params map[st return nil, fmt.Errorf("mailchimp/add_member: marshaling request: %w", err) } - req, err := http.NewRequestWithContext(ctx, "POST", - c.apiURL(dc, fmt.Sprintf("/lists/%s/members", listID)), + hash := md5.Sum([]byte(strings.ToLower(email))) + subscriberHash := hex.EncodeToString(hash[:]) + + req, err := http.NewRequestWithContext(ctx, "PUT", + c.apiURL(dc, fmt.Sprintf("/lists/%s/members/%s", listID, subscriberHash)), bytes.NewReader(reqJSON), ) if err != nil { diff --git a/packages/engine/internal/connector/mailchimp_test.go b/packages/engine/internal/connector/mailchimp_test.go index 4c04108..bdc49c6 100644 --- a/packages/engine/internal/connector/mailchimp_test.go +++ b/packages/engine/internal/connector/mailchimp_test.go @@ -73,8 +73,10 @@ func TestMailchimpListMembersConnector_InvalidAPIKeyFormat(t *testing.T) { } func TestMailchimpAddMemberConnector_AddsMember(t *testing.T) { + // md5("bob@example.com") = 4b9bb80620f03eb3719e0a061c14283d + const wantPath = "/3.0/lists/list1/members/4b9bb80620f03eb3719e0a061c14283d" srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.Method != "POST" || r.URL.Path != "/3.0/lists/list1/members" { + if r.Method != "PUT" || r.URL.Path != wantPath { t.Errorf("unexpected %s %s", r.Method, r.URL.Path) } var body map[string]any diff --git a/packages/engine/internal/connector/okta.go b/packages/engine/internal/connector/okta.go index 0f2b6ed..bbc4ca8 100644 --- a/packages/engine/internal/connector/okta.go +++ b/packages/engine/internal/connector/okta.go @@ -121,10 +121,7 @@ func (c *OktaCreateUserConnector) Execute(ctx context.Context, params map[string return nil, fmt.Errorf("okta/create_user: marshaling request: %w", err) } - endpoint := c.apiURL(domain, "/api/v1/users") - if activate { - endpoint += "?activate=true" - } + endpoint := fmt.Sprintf("%s?activate=%v", c.apiURL(domain, "/api/v1/users"), activate) req, err := http.NewRequestWithContext(ctx, "POST", endpoint, bytes.NewReader(reqJSON)) if err != nil { diff --git a/packages/engine/internal/connector/okta_test.go b/packages/engine/internal/connector/okta_test.go index 4b49d2a..f0753c3 100644 --- a/packages/engine/internal/connector/okta_test.go +++ b/packages/engine/internal/connector/okta_test.go @@ -78,6 +78,9 @@ func TestOktaCreateUserConnector_CreatesUser(t *testing.T) { if r.Method != "POST" { t.Errorf("expected POST, got %s", r.Method) } + if r.URL.Query().Get("activate") != "true" { + t.Errorf("expected activate=true, got %q", r.URL.Query().Get("activate")) + } var body map[string]any json.NewDecoder(r.Body).Decode(&body) profile, _ := body["profile"].(map[string]any) @@ -108,6 +111,31 @@ func TestOktaCreateUserConnector_CreatesUser(t *testing.T) { } } +func TestOktaCreateUserConnector_ActivateFalse(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Query().Get("activate") != "false" { + t.Errorf("expected activate=false, got %q", r.URL.Query().Get("activate")) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]any{"id": "00u2", "status": "STAGED"}) + })) + defer srv.Close() + + c := &OktaCreateUserConnector{baseURL: srv.URL} + out, err := c.Execute(t.Context(), map[string]any{ + "_credential": map[string]string{"domain": "dev.okta.com", "token": "tok"}, + "activate": false, + "profile": map[string]any{"login": "staged@example.com", "email": "staged@example.com"}, + }) + if err != nil { + t.Fatal(err) + } + if out["id"] != "00u2" { + t.Errorf("expected id=00u2, got %v", out["id"]) + } +} + func TestOktaCreateUserConnector_MissingProfile(t *testing.T) { c := &OktaCreateUserConnector{baseURL: "http://unused"} _, err := c.Execute(t.Context(), map[string]any{ From 78d17b654f78266eab34eff1b020516c4b85c86a Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 21:12:23 -0400 Subject: [PATCH 06/10] fix(connector): suppress gosec/CodeQL false positives on Mailchimp MD5 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Mailchimp subscriber hash is MD5(lowercase(email)) per Mailchimp's own API specification — it is a lookup identifier, not a security hash. Add inline suppression annotations for both gosec G401 and CodeQL go/weak-sensitive-data-hashing to clarify the non-security intent. Co-Authored-By: Claude Sonnet 4.6 --- packages/engine/internal/connector/mailchimp.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/engine/internal/connector/mailchimp.go b/packages/engine/internal/connector/mailchimp.go index 07adc14..11bb70d 100644 --- a/packages/engine/internal/connector/mailchimp.go +++ b/packages/engine/internal/connector/mailchimp.go @@ -124,6 +124,9 @@ func (c *MailchimpAddMemberConnector) Execute(ctx context.Context, params map[st return nil, fmt.Errorf("mailchimp/add_member: marshaling request: %w", err) } + // Mailchimp requires MD5(lowercase(email)) as the member identifier — this is an API + // contract, not a security hash. MD5 is mandated by the Mailchimp API spec. + // #nosec G401 -- codeql[go/weak-sensitive-data-hashing] -- noncryptographic API identifier required by Mailchimp hash := md5.Sum([]byte(strings.ToLower(email))) subscriberHash := hex.EncodeToString(hash[:]) From 5afd9b29e614bbeee24ac49ccc66a06e7f7b0ff8 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 21:19:49 -0400 Subject: [PATCH 07/10] fix(connector): correct gosec nosec placement for Mailchimp MD5 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit G501 fires on the import line and G401 on the call site — both need their own same-line #nosec annotations. The combined //nolint+#nosec format wasn't parsed correctly by gosec. Co-Authored-By: Claude Sonnet 4.6 --- packages/engine/internal/connector/mailchimp.go | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/packages/engine/internal/connector/mailchimp.go b/packages/engine/internal/connector/mailchimp.go index 11bb70d..071ede8 100644 --- a/packages/engine/internal/connector/mailchimp.go +++ b/packages/engine/internal/connector/mailchimp.go @@ -3,7 +3,7 @@ package connector import ( "bytes" "context" - "crypto/md5" + "crypto/md5" // #nosec G501 -- Mailchimp API requires MD5(lowercase(email)) as subscriber hash; not a security primitive "encoding/hex" "encoding/json" "fmt" @@ -124,10 +124,8 @@ func (c *MailchimpAddMemberConnector) Execute(ctx context.Context, params map[st return nil, fmt.Errorf("mailchimp/add_member: marshaling request: %w", err) } - // Mailchimp requires MD5(lowercase(email)) as the member identifier — this is an API - // contract, not a security hash. MD5 is mandated by the Mailchimp API spec. - // #nosec G401 -- codeql[go/weak-sensitive-data-hashing] -- noncryptographic API identifier required by Mailchimp - hash := md5.Sum([]byte(strings.ToLower(email))) + // Mailchimp API requires MD5(lowercase(email)) as the subscriber hash — API identifier, not a security hash. + hash := md5.Sum([]byte(strings.ToLower(email))) // #nosec G401 G501 subscriberHash := hex.EncodeToString(hash[:]) req, err := http.NewRequestWithContext(ctx, "PUT", From 643a93b57de1302b2d56c5651dd105dfbce4baf3 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 21:24:23 -0400 Subject: [PATCH 08/10] fix(connector): add CodeQL inline suppression for Mailchimp MD5 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add CodeQL[go/weak-sensitive-data-hashing] inline suppression alongside the gosec #nosec annotation. The earlier attempt used lowercase 'codeql' which GitHub Code Scanning does not recognize — the exact token is 'CodeQL[rule-id]' (capital C, Q, L). Co-Authored-By: Claude Sonnet 4.6 --- packages/engine/internal/connector/mailchimp.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/engine/internal/connector/mailchimp.go b/packages/engine/internal/connector/mailchimp.go index 071ede8..1ff4043 100644 --- a/packages/engine/internal/connector/mailchimp.go +++ b/packages/engine/internal/connector/mailchimp.go @@ -125,7 +125,7 @@ func (c *MailchimpAddMemberConnector) Execute(ctx context.Context, params map[st } // Mailchimp API requires MD5(lowercase(email)) as the subscriber hash — API identifier, not a security hash. - hash := md5.Sum([]byte(strings.ToLower(email))) // #nosec G401 G501 + hash := md5.Sum([]byte(strings.ToLower(email))) // #nosec G401 G501 -- CodeQL[go/weak-sensitive-data-hashing] subscriberHash := hex.EncodeToString(hash[:]) req, err := http.NewRequestWithContext(ctx, "PUT", From 8e3c90ae38307cc6b1b335978edecd7bf110cf07 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 21:31:44 -0400 Subject: [PATCH 09/10] fix(connector): CodeQL suppression for Mailchimp MD5 (standalone comment) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Use standalone // CodeQL[go/weak-sensitive-data-hashing] on the preceding line as GitHub docs specify — no other text on that line. Also add .github/codeql/codeql-config.yml as a fallback that excludes the go/weak-sensitive-data-hashing query for the connector package. Co-Authored-By: Claude Sonnet 4.6 --- .github/codeql/codeql-config.yml | 7 +++++++ packages/engine/internal/connector/mailchimp.go | 4 ++-- 2 files changed, 9 insertions(+), 2 deletions(-) create mode 100644 .github/codeql/codeql-config.yml diff --git a/.github/codeql/codeql-config.yml b/.github/codeql/codeql-config.yml new file mode 100644 index 0000000..8639535 --- /dev/null +++ b/.github/codeql/codeql-config.yml @@ -0,0 +1,7 @@ +name: "Mantle CodeQL config" + +query-filters: + - exclude: + id: go/weak-sensitive-data-hashing + paths: + - packages/engine/internal/connector/** diff --git a/packages/engine/internal/connector/mailchimp.go b/packages/engine/internal/connector/mailchimp.go index 1ff4043..a46f604 100644 --- a/packages/engine/internal/connector/mailchimp.go +++ b/packages/engine/internal/connector/mailchimp.go @@ -124,8 +124,8 @@ func (c *MailchimpAddMemberConnector) Execute(ctx context.Context, params map[st return nil, fmt.Errorf("mailchimp/add_member: marshaling request: %w", err) } - // Mailchimp API requires MD5(lowercase(email)) as the subscriber hash — API identifier, not a security hash. - hash := md5.Sum([]byte(strings.ToLower(email))) // #nosec G401 G501 -- CodeQL[go/weak-sensitive-data-hashing] + // CodeQL[go/weak-sensitive-data-hashing] + hash := md5.Sum([]byte(strings.ToLower(email))) // #nosec G401 G501 -- Mailchimp API mandates MD5(lowercase(email)) as subscriber hash subscriberHash := hex.EncodeToString(hash[:]) req, err := http.NewRequestWithContext(ctx, "PUT", From 258e276d7cc188ef1517af3e8579a9bb1535b122 Mon Sep 17 00:00:00 2001 From: Michael McNees Date: Sat, 30 May 2026 21:35:54 -0400 Subject: [PATCH 10/10] fix(server): move /metrics behind auth to prevent workflow name leakage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Prometheus path labels include workflow names (via metricsMiddleware → normalizePath). Serving /metrics on the unauthenticated top-level mux let anyone enumerate workflow identifiers without credentials. Moved /metrics to apiMux so it goes through the same auth middleware as all other API routes. Health probes (/healthz, /readyz) remain on the unauthenticated mux for Kubernetes probe access. Prometheus scrapers must now supply an API key. Co-Authored-By: Claude Sonnet 4.6 --- packages/engine/internal/server/server.go | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/packages/engine/internal/server/server.go b/packages/engine/internal/server/server.go index 8b5e68c..3012edf 100644 --- a/packages/engine/internal/server/server.go +++ b/packages/engine/internal/server/server.go @@ -250,11 +250,13 @@ func (s *Server) Start(ctx context.Context) error { rl := NewRateLimiter(100, 200) apiHandler = rl.Middleware(apiHandler) - // Top-level mux: health and metrics endpoints bypass auth so that Kubernetes - // probes and Prometheus scrapers work without an API key. All other requests - // fall through to the auth-wrapped apiHandler. + // Metrics behind auth — path labels include workflow names which are sensitive. + // Kubernetes probes bypass auth; Prometheus must be configured with an API key. + apiMux.Handle("/metrics", promhttp.Handler()) + + // Top-level mux: health probes bypass auth so Kubernetes can reach them + // without credentials. All other requests fall through to apiHandler. mux := http.NewServeMux() - mux.Handle("/metrics", promhttp.Handler()) mux.Handle("/healthz", health.HealthzHandler()) // /readyz checks DB connectivity only; worker/reaper liveness is intentionally // excluded to prevent flapping between poll cycles.