Skip to content

Commit 898a3fc

Browse files
Copilotintel352
Copilot
and
intel352
authored
feat: dynamic table name support in db_query, db_exec, db_query_cached steps (#234)
* Initial plan * feat: add allow_dynamic_sql support to db_query, db_exec, and db_query_cached steps Co-authored-by: intel352 <77607+intel352@users.noreply.github.com> * fix: address review comments on pipeline_step_db_dynamic.go Co-authored-by: intel352 <77607+intel352@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>
1 parent 75f259d commit 898a3fc

8 files changed

Lines changed: 428 additions & 61 deletions

‎module/pipeline_step_db_dynamic.go‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
package module
2+
3+
import (
4+
"fmt"
5+
"strings"
6+
)
7+
8+
// validateSQLIdentifier checks that s is safe to interpolate directly into SQL as an
9+
// identifier (e.g. a table name). Only ASCII letters (A-Z, a-z), ASCII digits (0-9),
10+
// underscores (_) and hyphens (-) are permitted. This strict allowlist prevents SQL
11+
// injection when dynamic values are embedded in queries via allow_dynamic_sql.
12+
func validateSQLIdentifier(s string) error {
13+
if s == "" {
14+
return fmt.Errorf("dynamic SQL identifier must not be empty")
15+
}
16+
for _, c := range s {
17+
if (c < 'a' || c > 'z') &&
18+
(c < 'A' || c > 'Z') &&
19+
(c < '0' || c > '9') &&
20+
c != '_' && c != '-' {
21+
return fmt.Errorf("dynamic SQL identifier %q contains unsafe character %q (only ASCII letters, digits, underscores and hyphens are allowed)", s, string(c))
22+
}
23+
}
24+
return nil
25+
}
26+
27+
// resolveDynamicSQL resolves every {{ }} template expression found in query against
28+
// pc and validates that each resolved value is a safe SQL identifier. The validated
29+
// values are substituted back into the query in left-to-right order and the final
30+
// SQL string is returned.
31+
//
32+
// Each occurrence of a template expression is resolved independently, so
33+
// non-deterministic functions like {{uuid}} or {{now}} produce a distinct value
34+
// per occurrence.
35+
//
36+
// This is only called when allow_dynamic_sql is true (explicit opt-in). Callers
37+
// are responsible for ensuring that the query has already passed template parsing.
38+
func resolveDynamicSQL(tmpl *TemplateEngine, query string, pc *PipelineContext) (string, error) {
39+
if !strings.Contains(query, "{{") {
40+
return query, nil
41+
}
42+
43+
// Process template expressions in left-to-right order. Each occurrence is
44+
// resolved and validated independently to preserve correct semantics for
45+
// non-deterministic template functions (e.g. {{uuid}}, {{now}}).
46+
var result strings.Builder
47+
rest := query
48+
for {
49+
openIdx := strings.Index(rest, "{{")
50+
if openIdx < 0 {
51+
result.WriteString(rest)
52+
break
53+
}
54+
closeIdx := strings.Index(rest[openIdx:], "}}")
55+
if closeIdx < 0 {
56+
return "", fmt.Errorf("dynamic SQL: unclosed template action in query (missing closing '}}')")
57+
}
58+
closeIdx += openIdx
59+
60+
// Write the literal SQL text before this expression.
61+
result.WriteString(rest[:openIdx])
62+
63+
expr := rest[openIdx : closeIdx+2]
64+
65+
resolved, err := tmpl.Resolve(expr, pc)
66+
if err != nil {
67+
return "", fmt.Errorf("dynamic SQL: failed to resolve %q: %w", expr, err)
68+
}
69+
if err := validateSQLIdentifier(resolved); err != nil {
70+
return "", fmt.Errorf("dynamic SQL: %w", err)
71+
}
72+
result.WriteString(resolved)
73+
rest = rest[closeIdx+2:]
74+
}
75+
return result.String(), nil
76+
}

‎module/pipeline_step_db_exec.go‎

Lines changed: 33 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -10,13 +10,14 @@ import (
1010

1111
// DBExecStep executes parameterized SQL INSERT/UPDATE/DELETE against a named database service.
1212
type DBExecStep struct {
13-
name string
14-
database string
15-
query string
16-
params []string
17-
ignoreError bool
18-
app modular.Application
19-
tmpl *TemplateEngine
13+
name string
14+
database string
15+
query string
16+
params []string
17+
ignoreError bool
18+
allowDynamicSQL bool
19+
app modular.Application
20+
tmpl *TemplateEngine
2021
}
2122

2223
// NewDBExecStepFactory returns a StepFactory that creates DBExecStep instances.
@@ -32,8 +33,10 @@ func NewDBExecStepFactory() StepFactory {
3233
return nil, fmt.Errorf("db_exec step %q: 'query' is required", name)
3334
}
3435

35-
// Safety: reject template expressions in SQL to prevent injection
36-
if strings.Contains(query, "{{") {
36+
// Safety: reject template expressions in SQL to prevent injection,
37+
// unless allow_dynamic_sql is explicitly enabled.
38+
allowDynamicSQL, _ := config["allow_dynamic_sql"].(bool)
39+
if !allowDynamicSQL && strings.Contains(query, "{{") {
3740
return nil, fmt.Errorf("db_exec step %q: query must not contain template expressions (use params instead)", name)
3841
}
3942

@@ -51,20 +54,33 @@ func NewDBExecStepFactory() StepFactory {
5154
ignoreError, _ := config["ignore_error"].(bool)
5255

5356
return &DBExecStep{
54-
name: name,
55-
database: database,
56-
query: query,
57-
params: params,
58-
ignoreError: ignoreError,
59-
app: app,
60-
tmpl: NewTemplateEngine(),
57+
name: name,
58+
database: database,
59+
query: query,
60+
params: params,
61+
ignoreError: ignoreError,
62+
allowDynamicSQL: allowDynamicSQL,
63+
app: app,
64+
tmpl: NewTemplateEngine(),
6165
}, nil
6266
}
6367
}
6468

6569
func (s *DBExecStep) Name() string { return s.name }
6670

6771
func (s *DBExecStep) Execute(_ context.Context, pc *PipelineContext) (*StepResult, error) {
72+
// Resolve template expressions in the query early (before any DB access) when
73+
// dynamic SQL is enabled. This validates resolved identifiers against an
74+
// allowlist before any database interaction.
75+
query := s.query
76+
if s.allowDynamicSQL {
77+
var err error
78+
query, err = resolveDynamicSQL(s.tmpl, query, pc)
79+
if err != nil {
80+
return nil, fmt.Errorf("db_exec step %q: %w", s.name, err)
81+
}
82+
}
83+
6884
if s.app == nil {
6985
return nil, fmt.Errorf("db_exec step %q: no application context", s.name)
7086
}
@@ -102,7 +118,7 @@ func (s *DBExecStep) Execute(_ context.Context, pc *PipelineContext) (*StepResul
102118

103119
// Normalize SQL placeholders: users write $1,$2,$3 (PostgreSQL style),
104120
// engine converts to ? for SQLite automatically.
105-
query := normalizePlaceholders(s.query, driver)
121+
query = normalizePlaceholders(query, driver)
106122

107123
// Execute statement
108124
result, err := db.Exec(query, resolvedParams...)

‎module/pipeline_step_db_exec_test.go‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package module
33
import (
44
"context"
55
"database/sql"
6+
"strings"
67
"testing"
78

89
_ "modernc.org/sqlite"
@@ -195,6 +196,68 @@ func TestDBExecStep_RejectsTemplateInQuery(t *testing.T) {
195196
}
196197
}
197198

199+
func TestDBExecStep_DynamicTableName(t *testing.T) {
200+
db, err := sql.Open("sqlite", ":memory:")
201+
if err != nil {
202+
t.Fatalf("open db: %v", err)
203+
}
204+
defer db.Close()
205+
206+
_, err = db.Exec(`CREATE TABLE items_alpha (id TEXT PRIMARY KEY, name TEXT NOT NULL)`)
207+
if err != nil {
208+
t.Fatalf("create table: %v", err)
209+
}
210+
211+
app := mockAppWithDB("test-db", db)
212+
factory := NewDBExecStepFactory()
213+
step, err := factory("dynamic-insert", map[string]any{
214+
"database": "test-db",
215+
"query": `INSERT INTO items_{{.steps.auth.tenant}} (id, name) VALUES (?, ?)`,
216+
"params": []any{"i1", "Widget"},
217+
"allow_dynamic_sql": true,
218+
}, app)
219+
if err != nil {
220+
t.Fatalf("factory error: %v", err)
221+
}
222+
223+
pc := NewPipelineContext(nil, nil)
224+
pc.MergeStepOutput("auth", map[string]any{"tenant": "alpha"})
225+
226+
result, err := step.Execute(context.Background(), pc)
227+
if err != nil {
228+
t.Fatalf("execute error: %v", err)
229+
}
230+
231+
affected, _ := result.Output["affected_rows"].(int64)
232+
if affected != 1 {
233+
t.Errorf("expected affected_rows=1, got %v", affected)
234+
}
235+
}
236+
237+
func TestDBExecStep_DynamicSQL_RejectsInjection(t *testing.T) {
238+
factory := NewDBExecStepFactory()
239+
step, err := factory("injection-exec", map[string]any{
240+
"database": "test-db",
241+
"query": `DELETE FROM items_{{.steps.auth.tenant}} WHERE id = ?`,
242+
"params": []any{"i1"},
243+
"allow_dynamic_sql": true,
244+
}, nil)
245+
if err != nil {
246+
t.Fatalf("factory error: %v", err)
247+
}
248+
249+
pc := NewPipelineContext(nil, nil)
250+
pc.MergeStepOutput("auth", map[string]any{"tenant": "alpha'; DROP TABLE items;--"})
251+
252+
_, err = step.Execute(context.Background(), pc)
253+
if err == nil {
254+
t.Fatal("expected error for unsafe SQL identifier")
255+
}
256+
if !strings.Contains(err.Error(), "unsafe character") {
257+
t.Errorf("expected 'unsafe character' in error, got: %v", err)
258+
}
259+
}
260+
198261
func TestDBExecStep_MissingDatabase(t *testing.T) {
199262
factory := NewDBExecStepFactory()
200263
_, err := factory("no-db", map[string]any{

‎module/pipeline_step_db_query.go‎

Lines changed: 33 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,14 @@ type DBDriverProvider interface {
2525

2626
// DBQueryStep executes a parameterized SQL SELECT against a named database service.
2727
type DBQueryStep struct {
28-
name string
29-
database string
30-
query string
31-
params []string
32-
mode string // "list" or "single"
33-
app modular.Application
34-
tmpl *TemplateEngine
28+
name string
29+
database string
30+
query string
31+
params []string
32+
mode string // "list" or "single"
33+
allowDynamicSQL bool
34+
app modular.Application
35+
tmpl *TemplateEngine
3536
}
3637

3738
// NewDBQueryStepFactory returns a StepFactory that creates DBQueryStep instances.
@@ -47,8 +48,10 @@ func NewDBQueryStepFactory() StepFactory {
4748
return nil, fmt.Errorf("db_query step %q: 'query' is required", name)
4849
}
4950

50-
// Safety: reject template expressions in SQL to prevent injection
51-
if strings.Contains(query, "{{") {
51+
// Safety: reject template expressions in SQL to prevent injection,
52+
// unless allow_dynamic_sql is explicitly enabled.
53+
allowDynamicSQL, _ := config["allow_dynamic_sql"].(bool)
54+
if !allowDynamicSQL && strings.Contains(query, "{{") {
5255
return nil, fmt.Errorf("db_query step %q: query must not contain template expressions (use params instead)", name)
5356
}
5457

@@ -72,20 +75,33 @@ func NewDBQueryStepFactory() StepFactory {
7275
}
7376

7477
return &DBQueryStep{
75-
name: name,
76-
database: database,
77-
query: query,
78-
params: params,
79-
mode: mode,
80-
app: app,
81-
tmpl: NewTemplateEngine(),
78+
name: name,
79+
database: database,
80+
query: query,
81+
params: params,
82+
mode: mode,
83+
allowDynamicSQL: allowDynamicSQL,
84+
app: app,
85+
tmpl: NewTemplateEngine(),
8286
}, nil
8387
}
8488
}
8589

8690
func (s *DBQueryStep) Name() string { return s.name }
8791

8892
func (s *DBQueryStep) Execute(_ context.Context, pc *PipelineContext) (*StepResult, error) {
93+
// Resolve template expressions in the query early (before any DB access) when
94+
// dynamic SQL is enabled. This validates resolved identifiers against an
95+
// allowlist before any database interaction.
96+
query := s.query
97+
if s.allowDynamicSQL {
98+
var err error
99+
query, err = resolveDynamicSQL(s.tmpl, query, pc)
100+
if err != nil {
101+
return nil, fmt.Errorf("db_query step %q: %w", s.name, err)
102+
}
103+
}
104+
89105
// Resolve database service
90106
if s.app == nil {
91107
return nil, fmt.Errorf("db_query step %q: no application context", s.name)
@@ -124,7 +140,7 @@ func (s *DBQueryStep) Execute(_ context.Context, pc *PipelineContext) (*StepResu
124140

125141
// Normalize SQL placeholders: users write $1,$2,$3 (PostgreSQL style),
126142
// engine converts to ? for SQLite automatically.
127-
query := normalizePlaceholders(s.query, driver)
143+
query = normalizePlaceholders(query, driver)
128144

129145
// Execute query
130146
rows, err := db.Query(query, resolvedParams...)

0 commit comments

Comments
 (0)