feat(connector): enterprise and cloud connector batch (batch 3)#149
Merged
Conversation
…ks, 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 <noreply@anthropic.com>
- 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 <noreply@anthropic.com>
- 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 <noreply@anthropic.com>
…kerfile - 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 <noreply@anthropic.com>
Adds 39 new connector actions across 17 integrations: Microsoft: entra/list_users, entra/create_user, entra/add_group_member, teams/send_message, teams/send_adaptive_card CRM/Ticketing: hubspot/create_contact, hubspot/search_contacts, jira/create_issue, jira/search_issues, salesforce/query, salesforce/create_record Google: drive/upload, drive/list_files, sheets/read_range, sheets/append_rows, gcp/publish, gcp/invoke_cloud_run GitHub: github/dispatch_workflow (added to existing file) Cloud: azure/blob_upload, azure/blob_download, azure/invoke_function, databricks/execute_sql, databricks/submit_job, bigquery/query, bigquery/insert_rows AWS: aws/invoke_lambda, aws/send_sqs, aws/publish_sns Messaging: kafka/produce, kafka/consume Database: snowflake/query, mysql/query, mysql/execute, mssql/query, mssql/execute, redshift/query Kubernetes: k8s/apply, k8s/create_job, k8s/get_pod_status All connectors follow the established pattern: baseURL injection for testing, dual-shape credential extraction, LimitReader on HTTP responses, 170 unit tests via httptest.NewServer with no external dependencies. Closes #113 Closes #97 Closes #107 Closes #106 Closes #93 Closes #87 Closes #86 Closes #85 Closes #88 Closes #92 Closes #91 Closes #89 Closes #90 Closes #78 Closes #79 Closes #82 Closes #105 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Contributor
|
Warning Review limit reached
More reviews will be available in 19 minutes and 48 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (14)
📝 WalkthroughWalkthroughThis PR implements 17+ new connectors across AWS, Azure, GCP, Google Workspace, enterprise SaaS platforms, databases, Kafka, Kubernetes, and Microsoft Entra, updating Go dependencies and extending the connector registry with complete credential extraction, HTTP handling, error wrapping, and test coverage for each implementation. ChangesMulti-cloud and enterprise connector implementations
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
✨ Finishing Touches🧪 Generate unit tests (beta)
|
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 9242ff64e2
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| path := k8sResourcePath(apiVersion, kind, namespace, name) | ||
| url := baseURL + path | ||
|
|
||
| _, statusCode, err := k8sDoRequest(ctx, client, http.MethodPut, url, token, body) |
There was a problem hiding this comment.
When k8s/apply targets an object that already exists, this sends the provided manifest with PUT; normal declarative manifests usually do not include the live metadata.resourceVersion, so the Kubernetes update endpoint rejects them instead of applying the desired changes. This means workflows can create a resource on first run but fail on the common second run/update path; use server-side apply/patch semantics or fetch and preserve resourceVersion before updating.
Useful? React with 👍 / 👎.
Collaborator
Author
There was a problem hiding this comment.
Fixed: replaced PUT+POST-fallback with a PATCH using Content-Type: application/apply-patch+yaml and ?fieldManager=mantle&force=true — server-side apply creates or updates without requiring resourceVersion in the manifest.
| } | ||
| defer rows.Close() | ||
|
|
||
| result, err := scanSQLRows(rows) |
There was a problem hiding this comment.
For MySQL queries that return many rows, max_rows does not actually limit what is read from the database because scanSQLRows consumes the entire result set before the caller truncates it. An unbounded SELECT can therefore allocate all rows in memory despite the default 1000-row cap; the scan loop should stop at maxRows or the query should be constrained before materializing results.
Useful? React with 👍 / 👎.
Collaborator
Author
There was a problem hiding this comment.
Fixed: scanSQLRows now accepts a maxRows parameter and stops iterating after that many rows (matching how the Snowflake connector already works), so large result sets never materialize in full. Post-scan truncation removed from all three callers (mysql/mssql/redshift).
Merge conflicts resolved: - connector.go: keep batch-3 registrations (ours) - server.go: take /metrics-behind-auth fix from main - mailchimp.go, mailchimp_test.go, okta.go, okta_test.go: take fixed versions from main (PUT upsert, activate param, MD5 subscriber hash) - go.mod: keep batch-3 deps (snowflake, franz-go, zeebo/xxh3) - go.sum: keep batch-3 superset Codex feedback addressed: - k8s/apply: replace PUT+POST-fallback with server-side apply (PATCH with Content-Type: application/apply-patch+yaml, fieldManager=mantle, force=true) so declarative manifests create-or-update correctly without requiring resourceVersion - mysql/mssql/redshift query: scanSQLRows now stops at maxRows during iteration instead of materializing all rows then truncating, preventing unbounded memory allocation on large result sets Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
There was a problem hiding this comment.
Pull request overview
This PR adds a large batch of new “shared connector pattern” actions to the engine (Microsoft, Google, CRM/ticketing, cloud/data, messaging, databases, and Kubernetes), registering them in the connector registry and adding unit tests plus dependency updates to support the new integrations.
Changes:
- Added new connectors for Teams, Entra ID, HubSpot, Jira, Salesforce, Google Drive/Sheets, GCP Pub/Sub + Cloud Run, Azure Blob + Functions, Databricks, BigQuery, AWS (Lambda/SQS/SNS), Kafka, Kubernetes, Snowflake, MySQL/MSSQL/Redshift.
- Registered the new actions in the connector Registry and extended GitHub with
github/dispatch_workflow. - Added extensive httptest-based unit tests and updated Go module dependencies / workspace versioning.
Reviewed changes
Copilot reviewed 38 out of 39 changed files in this pull request and generated 23 comments.
Show a summary per file
| File | Description |
|---|---|
| packages/engine/internal/connector/teams.go | Teams webhook connectors (send text + adaptive cards). |
| packages/engine/internal/connector/teams_test.go | Unit tests for Teams connectors + registry checks. |
| packages/engine/internal/connector/snowflake.go | Snowflake SQL query connector (database/sql + gosnowflake). |
| packages/engine/internal/connector/snowflake_test.go | Basic validation tests + registry check for Snowflake connector. |
| packages/engine/internal/connector/salesforce.go | Salesforce SOQL query + create record connectors. |
| packages/engine/internal/connector/salesforce_test.go | Salesforce connector tests + registry check. |
| packages/engine/internal/connector/mysql.go | MySQL query/execute + MSSQL query/execute + Redshift query connectors. |
| packages/engine/internal/connector/mysql_test.go | Validation tests + registry checks for MySQL/MSSQL/Redshift connectors. |
| packages/engine/internal/connector/kubernetes.go | REST-based Kubernetes apply/create job/get pod status connectors. |
| packages/engine/internal/connector/kubernetes_test.go | Kubernetes connector tests + registry check. |
| packages/engine/internal/connector/kafka.go | Kafka produce/consume connectors (franz-go). |
| packages/engine/internal/connector/kafka_test.go | Validation tests + registry check for Kafka connectors. |
| packages/engine/internal/connector/jira.go | Jira create issue + search issues connectors. |
| packages/engine/internal/connector/jira_test.go | Jira connector tests + registry check. |
| packages/engine/internal/connector/hubspot.go | HubSpot create contact + search contacts connectors. |
| packages/engine/internal/connector/hubspot_test.go | HubSpot connector tests + registry check. |
| packages/engine/internal/connector/googlesheets.go | Google Sheets read range + append rows connectors. |
| packages/engine/internal/connector/googlesheets_test.go | Google Sheets connector tests + registry check. |
| packages/engine/internal/connector/googledrive.go | Google Drive upload + list files connectors. |
| packages/engine/internal/connector/googledrive_test.go | Google Drive connector tests + registry check. |
| packages/engine/internal/connector/github.go | Adds github/dispatch_workflow connector. |
| packages/engine/internal/connector/github_test.go | Tests for github/dispatch_workflow + registry update. |
| packages/engine/internal/connector/gcpconnector.go | GCP Pub/Sub publish + Cloud Run invoke connectors. |
| packages/engine/internal/connector/gcpconnector_test.go | GCP connector tests + registry check. |
| packages/engine/internal/connector/entra.go | Entra list/create users + add group member connectors. |
| packages/engine/internal/connector/entra_test.go | Entra connector tests + registry check. |
| packages/engine/internal/connector/databricks.go | Databricks SQL statements + submit job connectors. |
| packages/engine/internal/connector/databricks_test.go | Databricks connector tests + registry check. |
| packages/engine/internal/connector/bigquery.go | BigQuery query + insertAll connectors. |
| packages/engine/internal/connector/bigquery_test.go | BigQuery connector tests + registry check. |
| packages/engine/internal/connector/azureblob.go | Azure Blob upload/download + Azure Functions invoke connectors. |
| packages/engine/internal/connector/azureblob_test.go | Azure connector tests + registry check. |
| packages/engine/internal/connector/aws.go | AWS Lambda invoke + SQS send + SNS publish connectors. |
| packages/engine/internal/connector/aws_test.go | AWS connector validation tests + registry check. |
| packages/engine/internal/connector/connector.go | Registers all newly added actions in the Registry. |
| packages/engine/go.mod | Adds/upgrades dependencies for new connectors (AWS services, DB drivers, franz-go, snowflake). |
| go.work | Updates workspace Go version and removes toolchain pin. |
| go.work.sum | Updates workspace dependency checksums. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| return nil, fmt.Errorf("teams/send_message: %w", err) | ||
| } | ||
| defer resp.Body.Close() | ||
| io.Copy(io.Discard, resp.Body) //nolint:errcheck |
Collaborator
Author
There was a problem hiding this comment.
Fixed: drain body via io.LimitReader(resp.Body, DefaultMaxResponseBytes) with blank assignment to avoid both unbounded reads and unhandled error.
| return nil, fmt.Errorf("teams/send_adaptive_card: %w", err) | ||
| } | ||
| defer resp.Body.Close() | ||
| io.Copy(io.Discard, resp.Body) //nolint:errcheck |
Collaborator
Author
There was a problem hiding this comment.
Fixed: drain body via io.LimitReader(resp.Body, DefaultMaxResponseBytes) with blank assignment to avoid both unbounded reads and unhandled error.
| return nil, fmt.Errorf("entra/add_group_member: %w", err) | ||
| } | ||
| defer resp.Body.Close() | ||
| io.Copy(io.Discard, resp.Body) //nolint:errcheck |
Collaborator
Author
There was a problem hiding this comment.
Fixed: replaced io.Copy(io.Discard, resp.Body) with _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, DefaultMaxResponseBytes)).
Collaborator
Author
There was a problem hiding this comment.
Fixed in bd4d387 — replaced io.Copy(io.Discard, ...) with io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) and now includes the response body in the error message when status != 204.
| } | ||
| return &http.Client{ | ||
| Transport: &http.Transport{ | ||
| TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec |
Collaborator
Author
There was a problem hiding this comment.
Changed //nolint:gosec → // #nosec G402 so gosec native suppression is recognized. TLS verification is intentionally skipped here because K8s clusters commonly use self-signed certs; callers that need verification should supply their own *http.Client.
Collaborator
Author
There was a problem hiding this comment.
Fixed in bd4d387 — added MinVersion: tls.VersionTLS12 to the TLS config in k8sInsecureClient alongside InsecureSkipVerify.
| // Core API (apiVersion="v1"): /api/v1/namespaces/{ns}/{resourceType}/{name} | ||
| // Other APIs: /apis/{apiVersion}/namespaces/{ns}/{resourceType}/{name} | ||
| func k8sResourcePath(apiVersion, kind, namespace, name string) string { | ||
| resourceType := strings.ToLower(kind) + "s" |
Collaborator
Author
There was a problem hiding this comment.
Fixed: added k8sPlural(kind) which correctly handles kinds ending in 's' (+es, e.g. ingress→ingresses) and 'y' (→ies, e.g. networkpolicy→networkpolicies), with the default case (+s).
| } | ||
|
|
||
| // KafkaProduceConnector produces a message to a Kafka topic. | ||
| // Params: topic (required), message (required), key (optional), partition (optional int). |
Collaborator
Author
There was a problem hiding this comment.
Fixed: removed 'partition (optional int)' from the docstring — the param is not implemented and franz-go's Produce selects partition via the record's key hash by default.
| return nil | ||
| } | ||
| var n int | ||
| fmt.Sscanf(s, "%d", &n) |
Collaborator
Author
There was a problem hiding this comment.
Fixed: parseIntString now checks the error from fmt.Sscanf and returns nil for non-numeric input instead of silently returning 0.
| return "", fmt.Errorf("credential must contain a 'database' field") | ||
| } | ||
|
|
||
| dsn = fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?parseTime=true", user, password, host, port, database) |
Collaborator
Author
There was a problem hiding this comment.
Fixed: switched from fmt.Sprintf DSN string concatenation to mysqldrv.Config{...}.FormatDSN() which properly escapes passwords containing '@', ':', '/', or '?' characters.
Comment on lines
+318
to
+324
| rowsAffected, _ := res.RowsAffected() | ||
| lastInsertID, _ := res.LastInsertId() | ||
|
|
||
| return map[string]any{ | ||
| "rows_affected": rowsAffected, | ||
| "last_insert_id": lastInsertID, | ||
| }, nil |
Collaborator
Author
There was a problem hiding this comment.
Fixed: last_insert_id is only included in the result map when res.LastInsertId() succeeds — SQL Server returns an error for this call, so MSSQL execute results will omit the key rather than return a misleading 0.
Comment on lines
+8
to
+9
| sf "github.com/snowflakedb/gosnowflake" | ||
| _ "github.com/snowflakedb/gosnowflake" // register "snowflake" driver |
Collaborator
Author
There was a problem hiding this comment.
Fixed: removed the blank import — the named import sf "github.com/snowflakedb/gosnowflake" already executes init() which registers the 'snowflake' database/sql driver.
Collaborator
Author
There was a problem hiding this comment.
Fixed in bd4d387 — replaced the duplicated row-scanning loop in SnowflakeQueryConnector.Execute with a call to the shared scanSQLRows helper from mysql.go.
Gosec (unblocks CI): - kubernetes.go: change //nolint:gosec → // #nosec G402 for TLS InsecureSkipVerify so gosec native suppression is recognized Copilot feedback applied: - teams.go: drain response body via io.LimitReader (bounded discard) - entra.go: same bounded discard for add_group_member response - kubernetes.go: add k8sPlural() to handle irregular pluralization (Ingress→ingresses, NetworkPolicy→networkpolicies); handle io.ReadAll error when reading server-side apply response - kubernetes_test.go: replace recursive containsStr with strings.Contains - kafka.go: remove 'partition' from KafkaProduceConnector docstring (param is not implemented) - bigquery.go: handle fmt.Sscanf error in parseIntString - mysql.go: use mysqldrv.Config.FormatDSN() for safe DSN encoding (handles passwords with '@', ':', '/' characters); omit last_insert_id from MSSQL execute result when driver returns error (SQL Server does not support LastInsertId) - snowflake.go: remove duplicate blank import (named import already registers the driver via init()) - aws.go, provider_bedrock.go: add #nosec G115 for safe int32 narrowing Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 17
🧹 Nitpick comments (3)
packages/engine/internal/connector/kubernetes.go (1)
60-65: ⚡ Quick winAdd
MinVersionto TLS config to prevent downgrade attacks.Even with
InsecureSkipVerify: true, setting a minimum TLS version prevents protocol downgrade attacks. The static analysis tool correctly flags this.🔒 Proposed fix
return &http.Client{ Transport: &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, // `#nosec` G402 -- K8s clusters commonly use self-signed certs; callers that need cert verification should supply their own *http.Client + TLSClientConfig: &tls.Config{ + InsecureSkipVerify: true, // `#nosec` G402 -- K8s clusters commonly use self-signed certs; callers that need cert verification should supply their own *http.Client + MinVersion: tls.VersionTLS12, + }, }, }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/engine/internal/connector/kubernetes.go` around lines 60 - 65, The TLS config used when constructing the http.Client sets InsecureSkipVerify but lacks a minimum TLS version; update the tls.Config created for Transport.TLSClientConfig to include a MinVersion (e.g., tls.VersionTLS12 or higher) to prevent protocol downgrade attacks—modify the tls.Config in the http.Transport construction (the TLSClientConfig value) to set MinVersion appropriately while keeping InsecureSkipVerify as-is.packages/engine/internal/connector/entra.go (1)
212-216: 💤 Low valueError response body discarded before status check.
Unlike
list_usersandcreate_userwhich include truncated response body in error messages, this connector discards the body before checking the status. On non-204 responses (e.g., 400/403/404), the error message loses potentially useful debugging info from the API.🔧 Optional: Read body before status check to include in error
resp, err := httpClient(c.Client).Do(req) if err != nil { return nil, fmt.Errorf("entra/add_group_member: %w", err) } defer resp.Body.Close() - _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, DefaultMaxResponseBytes)) if resp.StatusCode != http.StatusNoContent { + body, _ := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes)) - return nil, fmt.Errorf("entra/add_group_member: API returned %d", resp.StatusCode) + return nil, fmt.Errorf("entra/add_group_member: API returned %d: %s", resp.StatusCode, truncate(string(body), 500)) } + _, _ = io.Copy(io.Discard, resp.Body) return map[string]any{"ok": true}, nil🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/engine/internal/connector/entra.go` around lines 212 - 216, The code currently discards resp.Body before checking status in the add_group_member flow; change it to read a truncated body (using io.ReadAll with io.LimitReader and DefaultMaxResponseBytes) into a variable, then check resp.StatusCode and include the truncated body content in the returned error (similar to list_users/create_user). Locate the block around the add_group_member HTTP response handling (resp, resp.Body, DefaultMaxResponseBytes) and replace the io.Copy(io.Discard, ...) call with reading into a []byte, convert to string for the fmt.Errorf message when resp.StatusCode != http.StatusNoContent, preserving the truncation limit.packages/engine/internal/connector/snowflake.go (1)
106-129: ⚡ Quick winReuse
scanSQLRowshere instead of duplicating it.This loop is the same behavior already implemented in
packages/engine/internal/connector/mysql.go, so future fixes to row normalization or max-row handling can easily drift between connectors. Calling the shared helper keeps the SQL connectors aligned.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/engine/internal/connector/snowflake.go` around lines 106 - 129, The duplicated row-scanning loop in snowflake.go should be replaced by a call to the shared helper scanSQLRows to avoid drift; retrieve cols as you already do, then call scanSQLRows(rows, cols, maxRows) (or the existing scanSQLRows signature in packages/engine/internal/connector/mysql.go) and propagate/wrap its error with the same "snowflake/query" context, returning the resulting []map[string]any; remove the manual vals/ptrs/rows.Scan loop and the rows.Err() check since scanSQLRows handles normalization and max-row logic.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@packages/engine/internal/connector/azureblob.go`:
- Around line 225-247: The current response handling after calling
httpClient(c.Client).Do(req) in azureblob.go treats any HTTP response as
success; update the logic in the invoke path (the block that reads resp via
io.ReadAll with DefaultMaxResponseBytes and unmarshals JSON) to check
resp.StatusCode and return an error for non-2xx responses (e.g., if
resp.StatusCode < 200 || resp.StatusCode >= 300) instead of returning a map
payload; include the status code and a concise snippet of the response body (or
the unmarshalled JSON if available) in the error message so callers of this
connector see failed invocations rather than treating 401/404/500 responses as
successful outputs.
- Around line 200-206: The code builds endpoint by string-concatenating
functionKey which breaks on reserved chars and also never treats non-2xx
responses as errors; fix by parsing functionURL with net/url (use url.Parse on
functionURL), modify the URL's Query to set q.Set("code", functionKey) (or
remove when functionKey empty), then use url.String() for endpoint; in
AzureInvokeFunctionConnector.Execute check resp.StatusCode and return a non-nil
error for non-2xx responses instead of always returning
map[string]any{"body":..., "status": resp.StatusCode}, nil so failures surface
to callers.
In `@packages/engine/internal/connector/bigquery.go`:
- Around line 116-137: BigQueryQueryConnector.Execute currently unmarshals only
one POST /{projectId}/queries response and returns rows/total_rows, which can be
incomplete; update Execute to check the response's jobComplete flag and
pageToken and, if jobComplete is false or a pageToken exists, poll GET
/queries/{jobId} (or follow pagination) until jobComplete is true and there is
no next page, accumulating and flattening rows with flattenBigQueryRows for each
response and recomputing totalRows using extractInt(parseIntString(...)) from
the latest totalRows value; ensure errors from subsequent requests are handled
and merged rows/schema are returned together using the existing
schema/flattening helpers (refer to BigQueryQueryConnector.Execute,
flattenBigQueryRows, parseIntString, extractInt).
- Around line 141-149: flattenBigQueryRows currently assumes fm["v"] is a string
which loses NULLs and nested RECORD/REPEATED values; update flattenBigQueryRows
to inspect fm["v"]'s dynamic type (in the loop over fields) and handle: nil ->
append "" (or "NULL" per desired semantics); string -> append as now;
numeric/bool -> fmt.Sprint(v) and append; []any (REPEATED) -> iterate and
stringify each element (joining with a separator) or recursively stringify
elements; map[string]any (RECORD) -> detect a nested BigQuery field map
(contains "f") and either recursively flatten that sub-fields or marshal to
JSON/stringify in a stable way; ensure you reference the existing locals
(fields, fm, v, rowVals) and avoid silent failures by not discarding
type-assertion errors.
In `@packages/engine/internal/connector/gcpconnector.go`:
- Around line 132-154: The code currently returns a map for all HTTP responses,
so non-2xx Cloud Run responses are treated as successes; update the response
handling in the gcp/invoke_cloud_run flow to treat non-2xx statuses as errors:
after reading body and detecting contentType (the block using resp.Header.Get,
parseJSONBody and parseErr), if resp.StatusCode is not in the 200-299 range
construct and return an error that includes the status code and response body
(or parsed JSON result when parseJSONBody succeeded) rather than returning the
map; keep the existing returns for successful 2xx responses and reuse symbols
httpClient, DefaultMaxResponseBytes and parseJSONBody to locate the change.
In `@packages/engine/internal/connector/github.go`:
- Around line 224-225: The URL path currently built with
fmt.Sprintf("/repos/%s/%s/actions/workflows/%s/dispatches", owner, repo,
workflowID) can break for workflow file names with spaces or special chars;
change it to escape workflowID (and optionally owner/repo if untrusted) using
url.PathEscape before formatting, update the path variable accordingly, and
ensure net/url is added to imports; the subsequent
http.NewRequestWithContext(ctx, "POST", c.apiURL(path), ...) will then use the
safe, escaped path.
In `@packages/engine/internal/connector/googledrive.go`:
- Around line 59-79: The current multipart body is assembled with a fixed
boundary ("mantle_multipart_boundary") which can be corrupted if file content
contains the delimiter; replace the manual boundary and string writes with
mime/multipart.Writer to generate a safe boundary and properly create parts.
Specifically, in the block that builds buf and uses metaBody, content, mimeType
and boundary, create a multipart.NewWriter(&buf), use writer.CreatePart with
appropriate textproto.MIMEHeader for the metadata part ("Content-Type:
application/json") and for the file/content part ("Content-Type: "+mimeType),
write metaBody and content into those parts, call writer.Close() to finalize,
then build the request with http.NewRequestWithContext and set the Content-Type
header to "multipart/related; boundary="+writer.Boundary() (or get the boundary
from the writer) instead of the fixed boundary. Ensure imports mime/multipart
and net/textproto are added.
In `@packages/engine/internal/connector/jira.go`:
- Line 184: The Jira connector is calling the wrong REST path; in the function
that builds the request (the http.NewRequestWithContext line in the Jira
connector, e.g., where c.apiURL is used to POST issue search), change the
endpoint from "/rest/api/3/issue/search" to the correct "/rest/api/3/search" (or
"/rest/api/3/search/jql" if JQL-specific), update any helper that formats the
URL (c.apiURL) if it hardcodes the path, and update corresponding tests/mocks
that expect the old path so they assert the new path instead.
In `@packages/engine/internal/connector/kubernetes.go`:
- Around line 357-359: The URL is built directly from unescaped namespace and
podName (seen where client := k8sInsecureClient(c.Client) and url :=
fmt.Sprintf("%s/api/v1/namespaces/%s/pods/%s", baseURL, namespace, podName)),
which can break for special characters; fix by applying url.PathEscape to
namespace and podName (e.g., escapedNamespace := url.PathEscape(namespace),
escapedPodName := url.PathEscape(podName)) and use those in the fmt.Sprintf
call, and add the net/url import if missing.
- Around line 81-98: The k8sResourcePath function is vulnerable because it
interpolates namespace and name without escaping and always inserts the
/namespaces/{ns} segment so cluster-scoped kinds fail; update k8sResourcePath to
call url.PathEscape on namespace and name before building the path (use
url.PathEscape(namespace) and url.PathEscape(name)), and change path
construction to omit the "namespaces/{ns}" segment for cluster-scoped resources
by treating an empty/absent namespace as cluster-scoped (i.e., if namespace ==
"" then return "%s/%s" or "%s/%s/%s" without the namespaces segment), or augment
K8sApplyConnector to detect cluster-scoped kinds (e.g., Namespace, ClusterRole,
PersistentVolume) and call k8sResourcePath with an empty namespace; ensure the
resourceType logic (k8sPlural(kind)) and apiVersion branching remain unchanged.
In `@packages/engine/internal/connector/mysql.go`:
- Around line 19-24: extractMySQLCredential is mutating the caller-owned params
map by deleting "_credential", making retries non-idempotent; change the
function to copy the credential value out into a local variable and return it
without calling delete(params, "_credential"), and remove the same mutation
(delete(params, "_credential")) in the other occurrences in this file (search
for delete(params, "_credential") around the other blocks referenced) so all
credential extractions return the DSN but leave the original params map
unchanged.
- Around line 101-145: The Execute method currently only runs a raw query string
from params["query"] and must support bound parameters: add a new params key
(e.g., "args" or "params") that accepts a slice of values ([]any) or
[]interface{}, validate/extract it in extractMySQLCredential/Execute flow, and
pass that slice into db.QueryContext(ctx, query, args...) and
db.ExecContext(ctx, query, args...) where applicable (also update the other
methods/blocks at the indicated ranges). Ensure the extracted args are nil-safe
(use no-arg call when absent) and that scanSQLRows/return payloads remain
unchanged; reference the Execute function, the params map access for "query" and
the calls to QueryContext/ExecContext to locate where to thread the args
through.
- Around line 379-380: The current extractRedshiftCredential builds connStr by
interpolating host,user,password,database directly which breaks on
spaces/quotes/backslashes; instead build a proper pgx DSN: use pgx.ParseConfig
or pgx.ParseConfig + config.ConnString()/pgx.ConnConfig to produce a
safely-escaped DSN (or construct a libpq URL via net/url.URL and set Query for
sslmode=require) and then pass that to sql.Open("pgx", connStr); if you must
keep the keyword/value format, single-quote each value and escape any '\'' and
backslashes per libpq rules before assembling connStr to avoid
injection/formatting issues.
In `@packages/engine/internal/connector/provider_bedrock.go`:
- Around line 130-133: The narrowing from int to int32 for req.MaxTokens in the
block that assigns input.InferenceConfig (brtypes.InferenceConfiguration{
MaxTokens: aws.Int32(int32(req.MaxTokens)) }) must explicitly guard against
values > math.MaxInt32 (or a model-specific cap) instead of relying on the
`#nosec` suppression; update the code to check req.MaxTokens before the cast and
either clip it to int32(max) or return an error when it exceeds the safe upper
bound, then pass the safe int32 into aws.Int32(...) and remove the `#nosec`
comment to ensure no silent truncation occurs.
In `@packages/engine/internal/connector/salesforce.go`:
- Around line 144-145: The path is built from unescaped user input (objectType)
which can alter the intended URL; update the code that constructs the path (the
fmt.Sprintf("/sobjects/%s", objectType) call used before
http.NewRequestWithContext and c.dataURL) to either validate objectType against
the allowed Salesforce sObject name pattern or call url.PathEscape(objectType)
and use the escaped value in the fmt.Sprintf, ensuring the final request path is
safe from segments like "/", "?", or "..".
In `@packages/engine/internal/connector/snowflake.go`:
- Around line 62-138: The Execute method in SnowflakeQueryConnector only runs
raw SQL from params["query"] and lacks support for bound parameters; update
Execute to accept a parameter slice (e.g., params["args"] or params["binds"])
typed as []any/[]interface{} (validate type and convert if necessary), and pass
that slice into db.QueryContext(ctx, query, args...) instead of calling
db.QueryContext with only the query; keep existing error handling and maxRows
logic, ensure you handle nil/empty args (call QueryContext with no extra args)
and include the param name (args/binds) and variable when locating code inside
SnowflakeQueryConnector.Execute and where db.QueryContext is invoked.
- Around line 14-20: The function extractSnowflakeCredential is mutating the
caller's params by calling delete(params, "_credential"), causing retries to
lose credentials; remove that deletion and stop mutating params — simply read
params["_credential"] (or, if you must modify, work on a shallow copy of params
inside extractSnowflakeCredential) so
extractSnowflakeCredential(account,user,password,database,schema,warehouse)
returns credentials without deleting the _credential key from the original map.
---
Nitpick comments:
In `@packages/engine/internal/connector/entra.go`:
- Around line 212-216: The code currently discards resp.Body before checking
status in the add_group_member flow; change it to read a truncated body (using
io.ReadAll with io.LimitReader and DefaultMaxResponseBytes) into a variable,
then check resp.StatusCode and include the truncated body content in the
returned error (similar to list_users/create_user). Locate the block around the
add_group_member HTTP response handling (resp, resp.Body,
DefaultMaxResponseBytes) and replace the io.Copy(io.Discard, ...) call with
reading into a []byte, convert to string for the fmt.Errorf message when
resp.StatusCode != http.StatusNoContent, preserving the truncation limit.
In `@packages/engine/internal/connector/kubernetes.go`:
- Around line 60-65: The TLS config used when constructing the http.Client sets
InsecureSkipVerify but lacks a minimum TLS version; update the tls.Config
created for Transport.TLSClientConfig to include a MinVersion (e.g.,
tls.VersionTLS12 or higher) to prevent protocol downgrade attacks—modify the
tls.Config in the http.Transport construction (the TLSClientConfig value) to set
MinVersion appropriately while keeping InsecureSkipVerify as-is.
In `@packages/engine/internal/connector/snowflake.go`:
- Around line 106-129: The duplicated row-scanning loop in snowflake.go should
be replaced by a call to the shared helper scanSQLRows to avoid drift; retrieve
cols as you already do, then call scanSQLRows(rows, cols, maxRows) (or the
existing scanSQLRows signature in packages/engine/internal/connector/mysql.go)
and propagate/wrap its error with the same "snowflake/query" context, returning
the resulting []map[string]any; remove the manual vals/ptrs/rows.Scan loop and
the rows.Err() check since scanSQLRows handles normalization and max-row logic.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 3b7c88e0-661a-4fd1-8aa1-95a9ccbbea53
⛔ Files ignored due to path filters (3)
go.workis excluded by!**/*.workgo.work.sumis excluded by!**/*.sumpackages/engine/go.sumis excluded by!**/*.sum
📒 Files selected for processing (37)
packages/engine/go.modpackages/engine/internal/connector/aws.gopackages/engine/internal/connector/aws_test.gopackages/engine/internal/connector/azureblob.gopackages/engine/internal/connector/azureblob_test.gopackages/engine/internal/connector/bigquery.gopackages/engine/internal/connector/bigquery_test.gopackages/engine/internal/connector/connector.gopackages/engine/internal/connector/databricks.gopackages/engine/internal/connector/databricks_test.gopackages/engine/internal/connector/entra.gopackages/engine/internal/connector/entra_test.gopackages/engine/internal/connector/gcpconnector.gopackages/engine/internal/connector/gcpconnector_test.gopackages/engine/internal/connector/github.gopackages/engine/internal/connector/github_test.gopackages/engine/internal/connector/googledrive.gopackages/engine/internal/connector/googledrive_test.gopackages/engine/internal/connector/googlesheets.gopackages/engine/internal/connector/googlesheets_test.gopackages/engine/internal/connector/hubspot.gopackages/engine/internal/connector/hubspot_test.gopackages/engine/internal/connector/jira.gopackages/engine/internal/connector/jira_test.gopackages/engine/internal/connector/kafka.gopackages/engine/internal/connector/kafka_test.gopackages/engine/internal/connector/kubernetes.gopackages/engine/internal/connector/kubernetes_test.gopackages/engine/internal/connector/mysql.gopackages/engine/internal/connector/mysql_test.gopackages/engine/internal/connector/provider_bedrock.gopackages/engine/internal/connector/salesforce.gopackages/engine/internal/connector/salesforce_test.gopackages/engine/internal/connector/snowflake.gopackages/engine/internal/connector/snowflake_test.gopackages/engine/internal/connector/teams.gopackages/engine/internal/connector/teams_test.go
Comment on lines
+14
to
+20
| func extractSnowflakeCredential(params map[string]any) (account, user, password, database, schema, warehouse string, err error) { | ||
| raw, ok := params["_credential"] | ||
| if !ok || raw == nil { | ||
| return "", "", "", "", "", "", fmt.Errorf("credential is required") | ||
| } | ||
| delete(params, "_credential") | ||
|
|
Contributor
There was a problem hiding this comment.
Don't delete _credential from the input map.
This mutates caller state and creates the same retry hazard as the other SQL connectors: once a first attempt fails, subsequent retries can fail early because the credential was removed from params.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/engine/internal/connector/snowflake.go` around lines 14 - 20, The
function extractSnowflakeCredential is mutating the caller's params by calling
delete(params, "_credential"), causing retries to lose credentials; remove that
deletion and stop mutating params — simply read params["_credential"] (or, if
you must modify, work on a shallow copy of params inside
extractSnowflakeCredential) so
extractSnowflakeCredential(account,user,password,database,schema,warehouse)
returns credentials without deleting the _credential key from the original map.
Collaborator
Author
There was a problem hiding this comment.
Same as the MySQL comment — credential deletion is intentional as a security boundary to prevent secret data from flowing into query parameters or payloads.
Contributor
There was a problem hiding this comment.
@michaelmcnees, understood — the deletion is an intentional security measure to scrub credential data from the params map after extraction, preventing secrets from inadvertently propagating into query arguments or downstream payloads. I'll keep this in mind for future reviews across the connector package.
🐇 ✦
✏️ Learnings added
Learnt from: michaelmcnees
Repo: dvflw/mantle PR: 149
File: packages/engine/internal/connector/snowflake.go:14-20
Timestamp: 2026-05-31T20:46:39.120Z
Learning: In the dvflw/mantle codebase (packages/engine/internal/connector/), the pattern of calling `delete(params, "_credential")` inside credential extraction helpers (e.g., `extractSnowflakeCredential`, `extractMySQLCredential`, and similar functions across connector files) is intentional. It acts as a security boundary to prevent secret/credential data from flowing into query parameters, HTTP payloads, or other downstream uses of the `params` map. Do not flag this as a mutation hazard or retry hazard in future reviews — the deletion is by design.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
🧠 Learnings used
Learnt from: michaelmcnees
Repo: dvflw/mantle PR: 45
File: internal/connector/email_receive.go:126-135
Timestamp: 2026-03-25T19:44:56.827Z
Learning: When reviewing Go code that uses go-imap v2 `github.com/emersion/go-imap/v2/imapclient.Client.Store(numSet, store, options)`, do not require a separate `UIDStore` method call. The method decides between `UID STORE` and plain `STORE` at runtime based on `numSet`’s concrete type: passing an `imap.UIDSet` should be treated as correct and will send `UID STORE`, while `imap.SeqSet` will send plain `STORE`. Only flag `client.Store` usage if `numSet` is of an unexpected type or the call signature is otherwise incorrect.
- azure/invoke_function: error on non-2xx responses; url.QueryEscape function key - gcp/invoke_cloud_run: error on non-2xx responses - github/dispatch_workflow: url.PathEscape workflowID in path - drive/upload: use multipart.NewWriter for random boundary (prevents boundary collision) - k8s: url.PathEscape namespace/name in resource paths and get_pod_status - salesforce: url.PathEscape objectType in create_record path Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
BigQuery: - Poll until jobComplete with context-bounded retry (500ms interval, configurable for tests) - Paginate through pageToken until max_rows satisfied - Return []map[string]any rows keyed by schema field names (was [][]string positional) - NULL values → nil, RECORD types → nested map, REPEATED fields → slice - bqParseValue recursively handles nested RECORD and REPEATED+RECORD combinations MySQL, MSSQL, Redshift, Snowflake: - Accept optional args: []any for parameterized query/execute - MySQL/MSSQL use ? placeholders; Redshift uses $1/$2; Snowflake uses ? - Passes args... to QueryContext/ExecContext — driver handles escaping Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
- kubernetes: support cluster-scoped resources (empty namespace omits /namespaces/{ns}),
add MinVersion TLS 1.2, propagate "default" namespace only in inherently-namespaced
connectors (create_job, get_pod_status)
- mysql: fix Redshift DSN to use pgx URL format so special chars in credentials are
properly percent-encoded
- snowflake: replace duplicated row-scanning loop with shared scanSQLRows helper
- entra: read response body before status check in add_group_member so errors include
the API's error message
- bedrock: add explicit math.MaxInt32 bounds check before int32 narrowing cast and
remove #nosec suppression
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
github/dispatch_workflowto existing file)All 39 actions use the shared connector pattern:
baseURLinjection for testing, dual-shape_credentialextraction,io.LimitReaderon HTTP responses. 170 unit tests viahttptest.NewServer— no external services required.Test plan
go build ./...passes (verified locally)go test ./internal/connector/... -run "TestEntra|TestTeams|TestHubSpot|TestJira|TestSalesforce|TestGitHub|TestGoogleDrive|TestGoogleSheets|TestGCP|TestAzure|TestDatabricks|TestBigQuery|TestAWS|TestKafka|TestK8s|TestSnowflake|TestMySQL|TestMSSQL|TestRedshift|TestRegistry"— 170 tests pass (verified locally)Closes #113
Closes #97
Closes #107
Closes #106
Closes #93
Closes #87
Closes #86
Closes #85
Closes #88
Closes #92
Closes #91
Closes #89
Closes #90
Closes #78
Closes #79
Closes #82
Closes #105
🤖 Generated with Claude Code
Summary by CodeRabbit
New Features