Skip to content

Commit e093b82

Browse files
Michael McNeesclaude
authored and
Michael McNees
committed
chore: merge main + address Codex review feedback
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>
2 parents 9242ff6 + 52d3860 commit e093b82

9 files changed

Lines changed: 139 additions & 42 deletions

File tree

‎.github/codeql/codeql-config.yml‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
name: "Mantle CodeQL config"
2+
3+
query-filters:
4+
- exclude:
5+
id: go/weak-sensitive-data-hashing
6+
paths:
7+
- packages/engine/internal/connector/**

‎packages/engine/internal/connector/kubernetes.go‎

Lines changed: 17 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -119,7 +119,9 @@ func k8sDoRequest(ctx context.Context, client *http.Client, method, url, token s
119119
return result, resp.StatusCode, nil
120120
}
121121

122-
// K8sApplyConnector applies a Kubernetes resource manifest (PUT, fallback to POST on 404).
122+
// K8sApplyConnector applies a Kubernetes resource manifest using server-side apply
123+
// (PATCH with application/apply-patch+yaml and fieldManager=mantle).
124+
// Creates or updates without requiring resourceVersion in the manifest.
123125
// Params: manifest (required — map[string]any Kubernetes resource manifest).
124126
// Output: {"ok": true, "name": "...", "kind": "..."}
125127
type K8sApplyConnector struct {
@@ -171,26 +173,25 @@ func (c *K8sApplyConnector) Execute(ctx context.Context, params map[string]any)
171173

172174
client := k8sInsecureClient(c.Client)
173175
path := k8sResourcePath(apiVersion, kind, namespace, name)
174-
url := baseURL + path
176+
applyURL := baseURL + path + "?fieldManager=mantle&force=true"
175177

176-
_, statusCode, err := k8sDoRequest(ctx, client, http.MethodPut, url, token, body)
178+
req, err := http.NewRequestWithContext(ctx, http.MethodPatch, applyURL, bytes.NewReader(body))
177179
if err != nil {
178180
return nil, fmt.Errorf("k8s/apply: %w", err)
179181
}
182+
req.Header.Set("Authorization", "Bearer "+token)
183+
// Server-side apply content type: creates or updates without requiring resourceVersion.
184+
req.Header.Set("Content-Type", "application/apply-patch+yaml")
185+
req.Header.Set("Accept", "application/json")
180186

181-
if statusCode == http.StatusNotFound {
182-
// Resource does not exist — POST to collection endpoint.
183-
collectionPath := k8sResourcePath(apiVersion, kind, namespace, "")
184-
collectionURL := baseURL + collectionPath
185-
_, statusCode, err = k8sDoRequest(ctx, client, http.MethodPost, collectionURL, token, body)
186-
if err != nil {
187-
return nil, fmt.Errorf("k8s/apply: %w", err)
188-
}
189-
if statusCode < 200 || statusCode >= 300 {
190-
return nil, fmt.Errorf("k8s/apply: POST returned status %d", statusCode)
191-
}
192-
} else if statusCode < 200 || statusCode >= 300 {
193-
return nil, fmt.Errorf("k8s/apply: PUT returned status %d", statusCode)
187+
resp, err := client.Do(req)
188+
if err != nil {
189+
return nil, fmt.Errorf("k8s/apply: %w", err)
190+
}
191+
defer resp.Body.Close()
192+
respBody, _ := io.ReadAll(io.LimitReader(resp.Body, DefaultMaxResponseBytes))
193+
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
194+
return nil, fmt.Errorf("k8s/apply: server-side apply returned %d: %s", resp.StatusCode, truncate(string(respBody), 200))
194195
}
195196

196197
return map[string]any{

‎packages/engine/internal/connector/kubernetes_test.go‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,69 @@ func testK8sCredential() map[string]string {
1515
}
1616
}
1717

18+
// --- K8sApplyConnector ---
19+
20+
func TestK8sApplyConnector_ServerSideApply(t *testing.T) {
21+
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
22+
if r.Method != http.MethodPatch {
23+
t.Errorf("expected PATCH, got %s", r.Method)
24+
}
25+
if r.Header.Get("Content-Type") != "application/apply-patch+yaml" {
26+
t.Errorf("unexpected Content-Type: %s", r.Header.Get("Content-Type"))
27+
}
28+
if r.URL.Query().Get("fieldManager") != "mantle" {
29+
t.Errorf("expected fieldManager=mantle, got %q", r.URL.Query().Get("fieldManager"))
30+
}
31+
if r.URL.Query().Get("force") != "true" {
32+
t.Errorf("expected force=true, got %q", r.URL.Query().Get("force"))
33+
}
34+
w.Header().Set("Content-Type", "application/json")
35+
w.WriteHeader(http.StatusOK)
36+
}))
37+
defer ts.Close()
38+
39+
c := &K8sApplyConnector{Client: ts.Client(), baseURL: ts.URL}
40+
out, err := c.Execute(t.Context(), map[string]any{
41+
"_credential": testK8sCredential(),
42+
"manifest": map[string]any{
43+
"apiVersion": "v1",
44+
"kind": "ConfigMap",
45+
"metadata": map[string]any{"name": "my-config", "namespace": "default"},
46+
},
47+
})
48+
if err != nil {
49+
t.Fatalf("unexpected error: %v", err)
50+
}
51+
if out["ok"] != true {
52+
t.Errorf("expected ok=true, got %v", out["ok"])
53+
}
54+
}
55+
56+
func TestK8sApplyConnector_MissingManifest(t *testing.T) {
57+
c := &K8sApplyConnector{}
58+
_, err := c.Execute(t.Context(), map[string]any{
59+
"_credential": testK8sCredential(),
60+
})
61+
if err == nil {
62+
t.Fatal("expected error for missing manifest")
63+
}
64+
}
65+
66+
func TestK8sApplyConnector_MissingMetadataName(t *testing.T) {
67+
c := &K8sApplyConnector{baseURL: "http://unused"}
68+
_, err := c.Execute(t.Context(), map[string]any{
69+
"_credential": testK8sCredential(),
70+
"manifest": map[string]any{
71+
"apiVersion": "v1",
72+
"kind": "ConfigMap",
73+
"metadata": map[string]any{},
74+
},
75+
})
76+
if err == nil {
77+
t.Fatal("expected error for missing metadata.name")
78+
}
79+
}
80+
1881
// --- K8sCreateJobConnector ---
1982

2083
func TestK8sCreateJobConnector_CreatesJob(t *testing.T) {

‎packages/engine/internal/connector/mailchimp.go‎

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,8 @@ package connector
33
import (
44
"bytes"
55
"context"
6+
"crypto/md5" // #nosec G501 -- Mailchimp API requires MD5(lowercase(email)) as subscriber hash; not a security primitive
7+
"encoding/hex"
68
"encoding/json"
79
"fmt"
810
"io"
@@ -122,8 +124,12 @@ func (c *MailchimpAddMemberConnector) Execute(ctx context.Context, params map[st
122124
return nil, fmt.Errorf("mailchimp/add_member: marshaling request: %w", err)
123125
}
124126

125-
req, err := http.NewRequestWithContext(ctx, "POST",
126-
c.apiURL(dc, fmt.Sprintf("/lists/%s/members", listID)),
127+
// CodeQL[go/weak-sensitive-data-hashing]
128+
hash := md5.Sum([]byte(strings.ToLower(email))) // #nosec G401 G501 -- Mailchimp API mandates MD5(lowercase(email)) as subscriber hash
129+
subscriberHash := hex.EncodeToString(hash[:])
130+
131+
req, err := http.NewRequestWithContext(ctx, "PUT",
132+
c.apiURL(dc, fmt.Sprintf("/lists/%s/members/%s", listID, subscriberHash)),
127133
bytes.NewReader(reqJSON),
128134
)
129135
if err != nil {

‎packages/engine/internal/connector/mailchimp_test.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,8 +73,10 @@ func TestMailchimpListMembersConnector_InvalidAPIKeyFormat(t *testing.T) {
7373
}
7474

7575
func TestMailchimpAddMemberConnector_AddsMember(t *testing.T) {
76+
// md5("bob@example.com") = 4b9bb80620f03eb3719e0a061c14283d
77+
const wantPath = "/3.0/lists/list1/members/4b9bb80620f03eb3719e0a061c14283d"
7678
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
77-
if r.Method != "POST" || r.URL.Path != "/3.0/lists/list1/members" {
79+
if r.Method != "PUT" || r.URL.Path != wantPath {
7880
t.Errorf("unexpected %s %s", r.Method, r.URL.Path)
7981
}
8082
var body map[string]any

‎packages/engine/internal/connector/mysql.go‎

Lines changed: 6 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -60,15 +60,15 @@ func extractMySQLCredential(params map[string]any) (dsn string, err error) {
6060
return dsn, nil
6161
}
6262

63-
// scanSQLRows scans database/sql rows into a slice of maps.
64-
func scanSQLRows(rows *sql.Rows) ([]map[string]any, error) {
63+
// scanSQLRows scans database/sql rows into a slice of maps, stopping at maxRows.
64+
func scanSQLRows(rows *sql.Rows, maxRows int) ([]map[string]any, error) {
6565
cols, err := rows.Columns()
6666
if err != nil {
6767
return nil, err
6868
}
6969

7070
var result []map[string]any
71-
for rows.Next() {
71+
for rows.Next() && len(result) < maxRows {
7272
vals := make([]any, len(cols))
7373
ptrs := make([]any, len(cols))
7474
for i := range vals {
@@ -123,14 +123,11 @@ func (c *MySQLQueryConnector) Execute(ctx context.Context, params map[string]any
123123
}
124124
defer rows.Close()
125125

126-
result, err := scanSQLRows(rows)
126+
result, err := scanSQLRows(rows, maxRows)
127127
if err != nil {
128128
return nil, fmt.Errorf("mysql/query: scanning rows: %w", err)
129129
}
130130

131-
if len(result) > maxRows {
132-
result = result[:maxRows]
133-
}
134131
if result == nil {
135132
result = []map[string]any{}
136133
}
@@ -273,14 +270,11 @@ func (c *MSSQLQueryConnector) Execute(ctx context.Context, params map[string]any
273270
}
274271
defer rows.Close()
275272

276-
result, err := scanSQLRows(rows)
273+
result, err := scanSQLRows(rows, maxRows)
277274
if err != nil {
278275
return nil, fmt.Errorf("mssql/query: scanning rows: %w", err)
279276
}
280277

281-
if len(result) > maxRows {
282-
result = result[:maxRows]
283-
}
284278
if result == nil {
285279
result = []map[string]any{}
286280
}
@@ -417,14 +411,11 @@ func (c *RedshiftQueryConnector) Execute(ctx context.Context, params map[string]
417411
}
418412
defer rows.Close()
419413

420-
result, err := scanSQLRows(rows)
414+
result, err := scanSQLRows(rows, maxRows)
421415
if err != nil {
422416
return nil, fmt.Errorf("redshift/query: scanning rows: %w", err)
423417
}
424418

425-
if len(result) > maxRows {
426-
result = result[:maxRows]
427-
}
428419
if result == nil {
429420
result = []map[string]any{}
430421
}

‎packages/engine/internal/connector/okta.go‎

Lines changed: 1 addition & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -121,10 +121,7 @@ func (c *OktaCreateUserConnector) Execute(ctx context.Context, params map[string
121121
return nil, fmt.Errorf("okta/create_user: marshaling request: %w", err)
122122
}
123123

124-
endpoint := c.apiURL(domain, "/api/v1/users")
125-
if activate {
126-
endpoint += "?activate=true"
127-
}
124+
endpoint := fmt.Sprintf("%s?activate=%v", c.apiURL(domain, "/api/v1/users"), activate)
128125

129126
req, err := http.NewRequestWithContext(ctx, "POST", endpoint, bytes.NewReader(reqJSON))
130127
if err != nil {

‎packages/engine/internal/connector/okta_test.go‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,9 @@ func TestOktaCreateUserConnector_CreatesUser(t *testing.T) {
7878
if r.Method != "POST" {
7979
t.Errorf("expected POST, got %s", r.Method)
8080
}
81+
if r.URL.Query().Get("activate") != "true" {
82+
t.Errorf("expected activate=true, got %q", r.URL.Query().Get("activate"))
83+
}
8184
var body map[string]any
8285
json.NewDecoder(r.Body).Decode(&body)
8386
profile, _ := body["profile"].(map[string]any)
@@ -108,6 +111,31 @@ func TestOktaCreateUserConnector_CreatesUser(t *testing.T) {
108111
}
109112
}
110113

114+
func TestOktaCreateUserConnector_ActivateFalse(t *testing.T) {
115+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
116+
if r.URL.Query().Get("activate") != "false" {
117+
t.Errorf("expected activate=false, got %q", r.URL.Query().Get("activate"))
118+
}
119+
w.Header().Set("Content-Type", "application/json")
120+
w.WriteHeader(http.StatusOK)
121+
json.NewEncoder(w).Encode(map[string]any{"id": "00u2", "status": "STAGED"})
122+
}))
123+
defer srv.Close()
124+
125+
c := &OktaCreateUserConnector{baseURL: srv.URL}
126+
out, err := c.Execute(t.Context(), map[string]any{
127+
"_credential": map[string]string{"domain": "dev.okta.com", "token": "tok"},
128+
"activate": false,
129+
"profile": map[string]any{"login": "staged@example.com", "email": "staged@example.com"},
130+
})
131+
if err != nil {
132+
t.Fatal(err)
133+
}
134+
if out["id"] != "00u2" {
135+
t.Errorf("expected id=00u2, got %v", out["id"])
136+
}
137+
}
138+
111139
func TestOktaCreateUserConnector_MissingProfile(t *testing.T) {
112140
c := &OktaCreateUserConnector{baseURL: "http://unused"}
113141
_, err := c.Execute(t.Context(), map[string]any{

‎packages/engine/internal/server/server.go‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -250,11 +250,13 @@ func (s *Server) Start(ctx context.Context) error {
250250
rl := NewRateLimiter(100, 200)
251251
apiHandler = rl.Middleware(apiHandler)
252252

253-
// Top-level mux: health and metrics endpoints bypass auth so that Kubernetes
254-
// probes and Prometheus scrapers work without an API key. All other requests
255-
// fall through to the auth-wrapped apiHandler.
253+
// Metrics behind auth — path labels include workflow names which are sensitive.
254+
// Kubernetes probes bypass auth; Prometheus must be configured with an API key.
255+
apiMux.Handle("/metrics", promhttp.Handler())
256+
257+
// Top-level mux: health probes bypass auth so Kubernetes can reach them
258+
// without credentials. All other requests fall through to apiHandler.
256259
mux := http.NewServeMux()
257-
mux.Handle("/metrics", promhttp.Handler())
258260
mux.Handle("/healthz", health.HealthzHandler())
259261
// /readyz checks DB connectivity only; worker/reaper liveness is intentionally
260262
// excluded to prevent flapping between poll cycles.

0 commit comments

Comments
 (0)