fix(pipes): escape non-scalar param values

CHANGELOG.md

@@ -27,6 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
   Column authorization now lives in the one place that already enumerates *every* column reference — the SQL builder (`query.Build`, made policy-aware) — so the projection, aggregation arguments, `filters`, `group_by`, `order_by`, and `time_range` are all checked against the role's allowlist in a single pass, and no current or future clause can silently skip the check. A full-row read is now requested explicitly with `"select_all": true`, which for a column-restricted role expands to exactly that role's allowed columns, so denied columns never reach ClickHouse; unrestricted roles and admin keep `SELECT *`; a role allowed the table but no columns fails closed with `403`. **Omitting `columns` now returns nothing** (an empty result) rather than expanding to a projection — so a hidden column can't leak by being left out at all. Because the generated SQL is now role-specific, the read cache key partitions by column visibility for free — one role can no longer be served another's cached columns. The structured-query and live-stream paths now share the same per-column decision (`policy.IsColumnAllowed`, plus new `AllowedProjection`/`RestrictsColumns` helpers), so the two read surfaces enforce identical column visibility and can't drift apart. New unit, handler, and end-to-end (SDK → WaveHouse → ClickHouse) tests cover each bypass shape as explicit red-team cases. One intended behavior change falls out of authorizing `order_by`/`group_by`/`filter`: a column-restricted role can no longer order, group, or filter by a column it cannot read — those clauses must reference only readable columns (unrestricted roles and admin are unaffected). The SDK's bare `.fetch()` maps to `select_all`, so it still returns a full row set (the same-release #270 fix already dropped its implicit `ORDER BY received_timestamp` default).
 - **Structured queries accept any identifier ClickHouse accepts (bring-your-own schemas), with every identifier backtick-quoted to stay injection-safe** (`internal/chsql/` (new), `internal/query/builder.go`, `internal/query/ast.go`, `internal/policy/policy.go`, `internal/api/structured_query.go`, `clients/ts/src/query-builder.ts`, `clients/ts/src/types.ts`, `docs/src/content/docs/access-control.md`, `docs/src/content/docs/api.md`, plus tests in `internal/query/{builder,ast}_test.go`, `internal/api/structured_query_test.go`, `internal/policy/policy_test.go`, `internal/chsql/chsql_test.go`, `tests/integration/identifier_roundtrip_test.go` (new), `tests/e2e/sdk/{query,cache,ingest}.test.ts`): table names, column names, and aggregation aliases may now contain dots, spaces, unicode, SQL keywords, even embedded backticks/backslashes — anything ClickHouse permits in a quoted identifier. The builder and the policy row-filter path route every identifier through one shared `chsql.QuoteIdent` that escapes exactly as ClickHouse's own `backQuote()` does (`\` → `\\`, `` ` `` → `` \` ``, verified against `SHOW CREATE TABLE` on a live server). This replaces the prior safe-identifier regex, which rejected legitimate columns in existing customer schemas. It also **subsumes the earlier aggregation-alias injection hardening**: an alias such as `n FROM secrets --` is now neutralized by quoting (it becomes an inert column label) rather than rejected, so callers keep arbitrary alias names while the reparent/statement-break/subquery payloads still cannot break out. Column and table references stay bounded by schema membership (the injection boundary for those) plus the per-column allowlist (#223); aliases carry no membership but are fully contained by quoting. The lone refusal is an identifier literally containing `?` — clickhouse-go's positional value-binder counts every `?` in the query text, even inside a quoted identifier, so such a name is rejected fail-closed rather than risk shifting value parameters (tracked in [#279](https://github.com/Wave-RF/WaveHouse/issues/279)). Two related API changes fall out of treating names literally: in a query, **`*` is now a literal column name** (quoted, gated by schema membership), not a wildcard — the all-columns read moved to an explicit **`select_all`** boolean (`columns` and `select_all` are mutually exclusive; omitting both selects nothing), and the SDK gains `.selectAll()` with a bare `.fetch()` mapping to it. `count(*)` still means count-all, and `allow_columns: ["*"]` is still the policy wildcard. The `columns` field also now accepts a single string as shorthand for a one-element list. Values are unchanged (positional `?`, driver-escaped). New real-ClickHouse round-trip tests cover weird column/table/alias names (including embedded backticks and backslashes), injection containment, and the `*`/`select_all` contract; server-side `{name:Identifier}` params were evaluated and rejected because clickhouse-go cannot mix them with positional value binding.
+- **Pipe parameters can no longer inject SQL through a non-scalar value, and a declared parameter `type` is now enforced** (`internal/pipes/pipes.go`, plus tests in `internal/pipes/pipes_test.go`, `internal/api/pipes_test.go`, `tests/e2e/sdk/admin.test.ts`, and docs `docs/src/content/docs/{pipes,api,architecture}.md`): closes #317. Pipes bind parameters by inlining escaped literals into the SQL template (no positional `?` binds), and `formatParamValue`'s fallback branch emitted whatever Go's `fmt.Sprintf("%v")` produced — so a JSON **array** or **object** supplied as a parameter value reached ClickHouse **raw and unescaped**. In a string-context placeholder an array element's inner `'` terminated the literal, giving a clean `UNION`/boolean injection that bypassed the per-column allowlist the structured-query path enforces; `BindParams` also never checked the declared `ParamDef.Type`, so a `type: number` parameter accepted an array all the same. The execute route (`GET/POST /v1/pipes/{name}`) is authorized only by the pipe's `allowed_roles` (commonly `[public]`), so this was reachable unauthenticated. `formatParamValue` is now recursive and fails closed: every scalar leaf — including each element of an array — is escaped, a JSON array renders as a parenthesized `(v1, v2, …)` list (the same `IN (…)` shape the structured-query builder emits, so list parameters such as `WHERE id IN {{ids}}` now work safely), and a value with no scalar SQL form — a JSON object, or an empty array that would render as the invalid `IN ()` — is rejected with a `400` rather than emitted raw. Independently, a declared `type` (`string`/`number`/`boolean`/`array`) is validated before formatting (a wrong-kind value is a `400`), while query-string params (always strings) still accept the numeric/boolean spelling. New unit tests cover the array/object/empty/nested-list and type-matrix cases plus the original injection payloads as red-team regressions, new handler tests cover the array-binds, object-rejected, and type-mismatch paths, and an end-to-end SDK test executes an `IN`-list pipe against real ClickHouse with a quote-bearing element to exercise escaping.
 
 ### Fixed
 

docs/src/content/docs/api.md

@@ -487,6 +487,9 @@ JSON array of result rows, with `X-Cache: HIT` or `X-Cache: MISS` indicating whe
 | 404 | `{"error":"pipe not found"}` | Pipe name not registered |
 | 403 | `{"error":"forbidden"}` | Role not in pipe's `allowed_roles` (and not the admin role). Fails closed: a request with no role (no token, or a JWT missing `auth.role_claim`) is denied unless a `default_role` resolves it into the list; a pipe with no `allowed_roles` denies everyone but the admin role. |
 | 400 | `{"error":"missing required parameter: x"}` | Required parameter not supplied |
+| 400 | `{"error":"parameter \"x\": expected number, got array"}` | A value violates the parameter's declared `type` (`string`/`number`/`boolean`/`array`) |
+| 400 | `{"error":"parameter \"x\": unsupported parameter type object"}` | A non-scalar value with no SQL literal form — a JSON object. A JSON **array** is valid and renders as an `IN`-style `(…)` list. |
+| 400 | `{"error":"parameter \"x\": array parameter must not be empty"}` | An empty array — it would render as the invalid `IN ()`. |
 
 ---
 

docs/src/content/docs/architecture.md

@@ -135,7 +135,7 @@ The package's design invariants — stdout always 100%, WARN+ERROR always export
 
 ### `pipes/` — Named Query Pipes
 
-- **pipes.go** — `NamedQuery` type with SQL template and parameter definitions, `Store` backed by NATS KV bucket `WAVEHOUSE_PIPES`. Supports `.sql` file directory bootstrap. `BindParams()` resolves `{{param}}` / `{{param:default}}` placeholders by inlining escaped literal values into the SQL (strings are single-quote-escaped).
+- **pipes.go** — `NamedQuery` type with SQL template and parameter definitions, `Store` backed by NATS KV bucket `WAVEHOUSE_PIPES`. Supports `.sql` file directory bootstrap. `BindParams()` resolves `{{param}}` / `{{param:default}}` placeholders by inlining escaped literal values into the SQL (strings single-quote-escaped; arrays rendered as escaped `(…)` `IN`-lists). A declared `ParamDef.Type` is enforced, and a non-scalar value with no SQL form (a JSON object, or an empty array) is rejected rather than emitted raw.
 
 ### `query/` — Structured Query Engine
 

docs/src/content/docs/pipes.mdx

@@ -50,7 +50,7 @@ The `parameters` array formally declares a parameter with metadata:
 | Field | Type | Description |
 | ----- | ---- | ----------- |
 | `name` | string | Placeholder name; matches `{{name}}` in the SQL. |
-| `type` | string | Documents the expected type (`"string"`, `"number"`, `"boolean"`). Descriptive — it advertises intent to callers and the SDK; binding itself keys off the runtime value. |
+| `type` | string | Expected value type: `"string"`, `"number"`, `"boolean"`, or `"array"`. **Enforced** — a value of the wrong kind (e.g. an array where a `"number"` is declared) is rejected with a `400`. An omitted or unrecognized type is not constrained. |
 | `required` | bool | If `true`, the caller must supply a value, even if the placeholder doesn't appear in the SQL. A missing required parameter is a `400`. |
 | `default` | any | Value used when the caller supplies none. |
 
@@ -84,12 +84,30 @@ Bound values are **inlined directly into the SQL string** (not sent as positiona
 | Supplied value | Rendered as | Note |
 | -------------- | ----------- | ---- |
 | string | `'escaped'` | single-quoted; `'` and `\` are escaped |
-| numeric string (e.g. `"100"`) | `100` | emitted bare, so `?limit=100` from a query string works as a number |
+| numeric string (e.g. `"100"`) | `100` | emitted bare, so `?limit=100` from a query string works as a number (a parameter declared `type: string` is quoted as `'100'` instead) |
 | number | `42` / `42.5` | whole floats render as integers |
 | boolean | `1` / `0` | ClickHouse-style |
+| array | `('a', 'b')` | parenthesized list of escaped elements — for `IN` clauses (see below) |
 | null | `NULL` | |
 
-Because string values are single-quote-escaped, a parameter value can't break out of its literal and inject SQL. The SQL *structure* still comes only from the operator-authored template.
+Every leaf value is escaped the same way — including each element of an array — so a parameter value can't break out of its literal and inject SQL. The SQL *structure* still comes only from the operator-authored template. Values with no safe scalar form are **rejected** with a `400`: a JSON object, and an empty array (which would render as the invalid `IN ()`).
+
+#### Array parameters and `IN` lists
+
+A parameter whose value is a JSON array renders as a ClickHouse list literal — `(v1, v2, …)`, every element escaped — ready for an `IN` clause. Write the placeholder **without** surrounding parentheses; the value brings its own:
+
+```sql
+SELECT page, count() AS views FROM clicks WHERE page IN {{pages}} GROUP BY page
+```
+
+```bash
+curl -X POST http://localhost:8080/v1/pipes/by_pages \
+  -H "Content-Type: application/json" \
+  -d '{"pages": ["/home", "/pricing", "/docs"]}'
+# binds to: … WHERE page IN ('/home', '/pricing', '/docs')
+```
+
+Array values must come from a **JSON body** (`POST`) — query-string parameters are always scalar strings, so `GET …?pages=a&pages=b` binds only the first occurrence (`a`) as a string, ignoring the rest. Declare the parameter as `"type": "array"` to require a list (a scalar value is then a `400`); leave the type off to accept either a scalar or a list at the same placeholder.
 
 ## Authorizing a pipe
 
@@ -158,6 +176,8 @@ The response is a JSON array of rows. Results flow through the shared in-process
 | 404 | `{"error":"pipe not found"}` | No pipe registered under that name |
 | 403 | `{"error":"forbidden"}` | Caller's role isn't in `allowed_roles` (and isn't admin) |
 | 400 | `{"error":"missing required parameter: x"}` | A required parameter wasn't supplied |
+| 400 | `{"error":"parameter \"x\": expected number, got array"}` | A value violates the parameter's declared `type` |
+| 400 | `{"error":"parameter \"x\": unsupported parameter type object"}` | A non-scalar with no SQL form (a JSON object, or an empty array) |
 
 ## Bootstrapping from `.sql` files
 

internal/api/pipes_test.go

@@ -377,6 +377,77 @@ func TestPipesHandler_Execute_NoAllowedRoles_NonAdminDenied(t *testing.T) {
 	testutil.AssertJSONErrorResponse(t, w)
 }
 
+// TestPipesHandler_Execute_ArrayParamBinds: an array body param renders into an
+// IN list and passes binding (failing only later at the nil ClickHouse conn).
+func TestPipesHandler_Execute_ArrayParamBinds(t *testing.T) {
+	t.Parallel()
+	store := pipes.NewMemoryStore(
+		&pipes.NamedQuery{
+			Name:       "by_ids",
+			SQL:        "SELECT * FROM clicks WHERE id IN {{ids}}",
+			Parameters: []pipes.ParamDef{{Name: "ids", Type: "array", Required: true}},
+		},
+	)
+	h := NewPipesHandler(store, policy.NewMemoryStore(&policy.Policy{}), nil, nil, 0, testutil.NopLogger())
+
+	w := httptest.NewRecorder()
+	body := map[string]any{"ids": []any{"a", "b"}}
+	r := pipesRequest(t, http.MethodPost, "/v1/pipes/by_ids/execute", "by_ids", body)
+	r = r.WithContext(auth.WithRole(r.Context(), "admin"))
+
+	safeHandle(h.Execute, w, r)
+
+	// Binding succeeded — the only failure left is the nil conn, never a 400.
+	assert.NotEqual(t, http.StatusBadRequest, w.Code)
+	assert.NotEqual(t, http.StatusNotFound, w.Code)
+}
+
+// TestPipesHandler_Execute_ObjectParamRejected: a JSON object has no scalar SQL
+// form and is refused with a 400 before any query runs (#317).
+func TestPipesHandler_Execute_ObjectParamRejected(t *testing.T) {
+	t.Parallel()
+	store := pipes.NewMemoryStore(
+		&pipes.NamedQuery{Name: "by_col", SQL: "SELECT * FROM clicks WHERE col = {{p}}"},
+	)
+	h := NewPipesHandler(store, policy.NewMemoryStore(&policy.Policy{}), nil, nil, 0, testutil.NopLogger())
+
+	w := httptest.NewRecorder()
+	body := map[string]any{"p": map[string]any{"k": "v"}}
+	r := pipesRequest(t, http.MethodPost, "/v1/pipes/by_col/execute", "by_col", body)
+	r = r.WithContext(auth.WithRole(r.Context(), "admin"))
+
+	h.Execute(w, r)
+
+	assert.Equal(t, http.StatusBadRequest, w.Code)
+	assert.Contains(t, w.Body.String(), "unsupported parameter type object")
+	testutil.AssertJSONErrorResponse(t, w)
+}
+
+// TestPipesHandler_Execute_TypeMismatchRejected: a value that violates the
+// declared parameter type is a 400, honoring the declared-type contract (#317).
+func TestPipesHandler_Execute_TypeMismatchRejected(t *testing.T) {
+	t.Parallel()
+	store := pipes.NewMemoryStore(
+		&pipes.NamedQuery{
+			Name:       "top",
+			SQL:        "SELECT * FROM clicks LIMIT {{limit}}",
+			Parameters: []pipes.ParamDef{{Name: "limit", Type: "number", Required: true}},
+		},
+	)
+	h := NewPipesHandler(store, policy.NewMemoryStore(&policy.Policy{}), nil, nil, 0, testutil.NopLogger())
+
+	w := httptest.NewRecorder()
+	body := map[string]any{"limit": []any{"99"}}
+	r := pipesRequest(t, http.MethodPost, "/v1/pipes/top/execute", "top", body)
+	r = r.WithContext(auth.WithRole(r.Context(), "admin"))
+
+	h.Execute(w, r)
+
+	assert.Equal(t, http.StatusBadRequest, w.Code)
+	assert.Contains(t, w.Body.String(), "expected number, got array")
+	testutil.AssertJSONErrorResponse(t, w)
+}
+
 // TestPipesHandler_Execute_NoAllowedRoles_AdminAllowed: the privileged built-in
 // roles bypass the allowlist, so admin can run a pipe with no allowed_roles.
 func TestPipesHandler_Execute_NoAllowedRoles_AdminAllowed(t *testing.T) {