diff --git a/CHANGELOG.md b/CHANGELOG.md index 80eb0928..ad509c1c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -25,6 +25,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Security +- **A role's structured-query resource caps are now enforced server-side by ClickHouse, so a public read can't outrun its budget during a server-side scan/merge/aggregation phase** (`internal/api/ch_settings.go` (new), `internal/api/structured_query.go`, `internal/policy/policy.go`, `internal/policy/scalars.go` (new), `internal/config/config.go`, `internal/query/builder.go`, `cmd/wavehouse/main.go`, `config.yaml`, `clients/ts/src/types.ts`, `docs/src/content/docs/{access-control.mdx,configuration.mdx}`, plus tests in `internal/api/ch_settings_test.go`, `internal/policy/{policy,scalars}_test.go`, `internal/config/config_test.go`, `internal/query/builder_test.go`, `tests/integration/query_limits_test.go` (all new/expanded), `tests/e2e/sdk/query.test.ts`): closes #316. The native ClickHouse connection passed **no per-query `Settings`**, so a read's policy caps bound it only client-side — a Go `context` deadline (which cancels only while the client is reading result blocks) plus a SQL `LIMIT`. Memory and rows scanned were **never bounded server-side**, so a heavy aggregation could allocate gigabytes of state or scan an entire table well within the time budget; and because `clickhouse-go` derives a `max_execution_time` setting from the context deadline only for deadlines `> 1s`, a sub-second time cap reached ClickHouse with no server-side time bound at all. The structured-query path now attaches per-query `Settings` derived from the role's resolved permissions: `max_execution_time` (fractional seconds, emitted explicitly so the sub-second case is enforced), `max_result_rows` + `result_overflow_mode=throw` (defense-in-depth behind the SQL `LIMIT`), `max_rows_to_read` + `read_overflow_mode=throw`, and `max_memory_usage` — so a query that exceeds its budget is rejected by the server (ClickHouse codes 158 / 241) rather than running to completion. **Boundary:** WaveHouse owns the *dynamic, per-role* caps (sent as per-query settings); the *global, static* backstop — which applies to every query including named pipes and raw admin SQL — is configured in **ClickHouse's own settings profiles and quotas** (documented in `configuration.mdx`), composes with the per-role caps, and holds even against a WaveHouse bug. **Schema (pre-launch):** the per-role policy fields are human-readable in / numeric out — `max_execution_time_ms` (int) → **`max_execution_time`** (set as a duration string `"5s"` or a bare ms number; read back as ms), the new **`max_memory_usage`** (set as a size string `"4GiB"` — IEC/SI respected via `dustin/go-humanize`, so `4GB` ≠ `4GiB` — or a bare byte number; read back as bytes), and the new **`max_rows_to_read`** (int), backed by two small `Millis`/`ByteSize` types in the `policy` package. The formerly hard-coded `query.DefaultMaxRows = 10000` result-LIMIT becomes the documented, tunable `query.default_max_rows` config knob (`Build` takes it as a parameter). Raw admin SQL remains unbounded by WaveHouse (governed by ClickHouse). Verified RED before the fix: the handler-level integration test confirmed a capped read returned the full result set when the settings weren't sent. - **The structured-query column allowlist is now a hard cap on every column a query references — closing a fail-open data-exposure family on the primary read path** (`internal/query/builder.go`, `internal/query/errors.go` (new), `internal/policy/policy.go`, `internal/discovery/discovery.go`, `internal/api/structured_query.go`, `docs/src/content/docs/access-control.md`, `docs/src/content/docs/api.md`, plus tests in `internal/query/builder_test.go`, `internal/api/structured_query_test.go`, `internal/policy/policy_test.go`, `internal/discovery/discovery_test.go`, `tests/e2e/sdk/query.test.ts`): closes #223. `POST /v1/query` previously authorized only the columns a caller *explicitly* listed, so several inputs bypassed a role's `allow_columns`/`deny_columns` entirely: - **Omitting `columns` (or sending `[]` / `["*"]`)** made the builder emit `SELECT *`, returning every column — the reported leak (an unauthenticated `{"limit":2}` returned columns hidden from the public role, including raw webhook `payload`). - **`group_by` on a denied column** (`SELECT count(*) … GROUP BY salary`) enumerated that column's distinct values wholesale; a **`filter`** on a denied column turned the returned row count into a value-inference oracle; an **`order_by`** on a denied column leaked ordering. None of these clauses were policy-checked — only the SELECT list and aggregation arguments were. diff --git a/clients/ts/src/types.ts b/clients/ts/src/types.ts index ad45d674..ecea77d7 100644 --- a/clients/ts/src/types.ts +++ b/clients/ts/src/types.ts @@ -212,8 +212,22 @@ export interface RolePermissions { check?: Record; allowed_aggregations?: string[]; denied_aggregations?: string[]; + /** Caps the query result LIMIT. 0 = no limit. */ max_rows?: number; - max_execution_time_ms?: number; + /** + * Max query execution time, enforced server-side by ClickHouse. Set it as a + * duration string ("5s", "500ms") or a number of milliseconds; reads always + * return the number of milliseconds. + */ + max_execution_time?: number | string; + /** Max rows scanned from storage, enforced server-side by ClickHouse. 0 = no limit. */ + max_rows_to_read?: number; + /** + * Max peak query memory, enforced server-side by ClickHouse. Set it as a size + * string ("4GiB", "512MiB") or a number of bytes; reads always return the + * number of bytes. + */ + max_memory_usage?: number | string; } export interface PolicyFilter { diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go index 40ef8308..51b01c08 100644 --- a/cmd/wavehouse/main.go +++ b/cmd/wavehouse/main.go @@ -381,7 +381,7 @@ func run() int { DLQ: dlqHandler, Policy: api.NewPolicyHandler(policyStore), Pipes: api.NewPipesHandler(pipesStore, policyStore, chConn, cache, cfg.ClickHouse.QueryTimeout, logger), - StructuredQuery: api.NewStructuredQueryHandler(chConn, cache, registry, policyStore, cfg.Cache.TimestampBucketSeconds, cfg.ClickHouse.QueryTimeout, logger), + StructuredQuery: api.NewStructuredQueryHandler(chConn, cache, registry, policyStore, cfg.Cache.TimestampBucketSeconds, cfg.ClickHouse.QueryTimeout, cfg.Query.DefaultMaxRows, logger), AuthMW: authMW, PolicyStore: policyStore, Logger: logger, diff --git a/config.yaml b/config.yaml index fefdbbc8..b1504216 100644 --- a/config.yaml +++ b/config.yaml @@ -47,6 +47,13 @@ clickhouse: password: "" query_timeout: 30s +# Query-shaping defaults. Server-wide RESOURCE limits (memory, rows scanned, +# execution time) live in ClickHouse itself — its settings profiles and quotas +# — so they apply to every query uniformly; see docs/configuration. This block +# holds only the result-LIMIT default. +query: + default_max_rows: 10000 # fallback result LIMIT (<=0 falls back to 10000) + mq: gap_window_minutes: 15 max_bytes_gb: 50 diff --git a/docs/src/content/docs/access-control.mdx b/docs/src/content/docs/access-control.mdx index c58cb0dd..e36dc64e 100644 --- a/docs/src/content/docs/access-control.mdx +++ b/docs/src/content/docs/access-control.mdx @@ -293,12 +293,18 @@ A disallowed function returns `403 aggregation "x" not allowed`. This is useful ## Resource limits -Two fields cap the cost of a single structured query. Both must be non-negative; `0` (the default) means "no policy-imposed limit". +Four fields cap the cost of a single structured query for this role. All must be non-negative; `0` / unset means "no role-imposed limit" — the [server-wide ClickHouse limits](#server-wide-limits-live-in-clickhouse) still apply, and for `max_rows` the `query.default_max_rows` result default (10,000) still clamps the `LIMIT`. | Field | Effect | | ----- | ------ | | `max_rows` | Caps the query's `LIMIT`. If the caller asks for more (or omits a limit), the result is clamped to `max_rows`. | -| `max_execution_time_ms` | Caps the ClickHouse execution timeout for the query, applied as the *minimum* of this value and the server's `clickhouse.query_timeout`. | +| `max_execution_time` | Caps the query execution time, applied as the *minimum* of this value and the server's `clickhouse.query_timeout`. | +| `max_rows_to_read` | Caps the rows **scanned from storage** server-side — the lever that stops a full-table scan. The read is rejected once it is exceeded. | +| `max_memory_usage` | Caps **peak query memory** server-side — the lever that stops a heavy aggregation from exhausting the box. The read is rejected once it is exceeded. | + +`max_execution_time`, `max_rows_to_read`, and `max_memory_usage` are enforced **server-side by ClickHouse** (sent as per-query `max_execution_time` / `max_rows_to_read` / `max_memory_usage` settings), so a query can't outrun its budget during a server-side scan, merge, or aggregation phase — not just while the client is reading rows back. `max_rows` is applied as the SQL `LIMIT` (and mirrored server-side as `max_result_rows` for defense-in-depth). + +**`max_execution_time` and `max_memory_usage` accept a human-readable value or a number.** Set `max_execution_time` as a duration string (`"5s"`, `"500ms"`) or a bare number of **milliseconds**; set `max_memory_usage` as a size string (`"4GiB"`, `"512MiB"` — IEC vs SI is respected, so `"4GB"` is 4×109 and `"4GiB"` is 4×230) or a bare number of **bytes**. The API always **returns** them as those numbers (ms and bytes), so SDK consumers read a plain integer. @@ -306,7 +312,9 @@ Two fields cap the cost of a single structured query. Both must be non-negative; select: viewer: max_rows: 1000 - max_execution_time_ms: 5000 + max_execution_time: "5s" + max_rows_to_read: 50000000 # reject a read that scans > 50M rows + max_memory_usage: "4GiB" # peak memory per query ``` @@ -315,7 +323,9 @@ select: "select": { "viewer": { "max_rows": 1000, - "max_execution_time_ms": 5000 + "max_execution_time": "5s", + "max_rows_to_read": 50000000, + "max_memory_usage": "4GiB" } } } @@ -325,20 +335,24 @@ select: These bound a role's blast radius on the cached read path. They do not apply to raw admin SQL, which is unbounded by design (other than the 64 MiB response cap noted in the [API reference](/api)). +### Server-wide limits live in ClickHouse + +These policy fields are **per-role** caps. A **server-wide** backstop — one that applies to *every* query regardless of role, including named pipes and raw admin SQL — is not part of the WaveHouse policy: configure it in ClickHouse's own [settings profiles and quotas](/configuration#server-side-resource-limits), where it is enforced natively and composes with the per-role caps above. That keeps global resource governance in one authoritative place and holds even against paths the policy engine doesn't touch. + ## Where each rule is enforced The same policy drives every data path, but not every field is meaningful on every path: | Surface | Endpoint | Enforced | | ------- | -------- | -------- | -| Structured read | `POST /v1/query?table={table}` | table+role `select` required, then `allow`/`deny_columns`, row `filter`, aggregation rules, `max_rows`, `max_execution_time_ms` | +| Structured read | `POST /v1/query?table={table}` | table+role `select` required, then `allow`/`deny_columns`, row `filter`, aggregation rules, and the per-role `max_rows` / `max_execution_time` / `max_rows_to_read` / `max_memory_usage` caps (over the [ClickHouse server-wide limits](/configuration#server-side-resource-limits)) | | Ingest (write) | `POST /v1/ingest?table={table}` | table+role `insert` required, then `allow`/`deny_columns` and `check` (enforced and auto-injected) | | Live stream | `GET /v1/stream` | table+role `select` required (a table the role can't read is skipped), then denied columns are masked from each event | | Raw SQL | `POST /v1/admin/query` | `admin_role` only — no per-statement policy; the role gate is the entire authorization story | -| Named pipe | `GET/POST /v1/pipes/{name}` | not the policy engine — per-pipe `allowed_roles`; see [Named Pipes](/pipes) | +| Named pipe | `GET/POST /v1/pipes/{name}` | per-pipe `allowed_roles` (not the policy engine; see [Named Pipes](/pipes)). Resource limits come from ClickHouse's [server-wide settings](/configuration#server-side-resource-limits), not per-role policy caps | :::caution[Live streams enforce access, not row filters] -SSE subscribers are checked for table-level `select` permission and have denied columns stripped from each event, but the row-level `filter` predicates and `max_rows`/`max_execution_time_ms` limits are a property of the SQL query path and are **not** applied to the live event stream. If a role must never observe another tenant's rows in real time, don't grant it stream access to a shared table — scope the data at the table level. +SSE subscribers are checked for table-level `select` permission and have denied columns stripped from each event, but the row-level `filter` predicates and the resource limits (`max_rows`, `max_execution_time`, `max_rows_to_read`, `max_memory_usage`) are a property of the SQL query path and are **not** applied to the live event stream. If a role must never observe another tenant's rows in real time, don't grant it stream access to a shared table — scope the data at the table level. ::: ## Managing the policy @@ -415,7 +429,7 @@ tables: _eq: "{{ jwt.app_metadata.tenant_id }}" denied_aggregations: ["quantile", "median"] max_rows: 1000 - max_execution_time_ms: 5000 + max_execution_time: "5s" insert: writer: # Clients send business fields; the policy stamps identity from the token. @@ -444,7 +458,7 @@ tables: }, "denied_aggregations": ["quantile", "median"], "max_rows": 1000, - "max_execution_time_ms": 5000 + "max_execution_time": "5s" } }, "insert": { @@ -486,8 +500,10 @@ Per-role permissions (`tables..select.` and `.insert.`): | `check` | map | insert | Required insert values (`_eq` only). Enforced if present in the body, auto-injected if absent. Supports templating. | | `allowed_aggregations` | string[] | select | Allowlist of aggregation functions. Empty = all (minus denied). Case-insensitive. | | `denied_aggregations` | string[] | select | Blocklist of aggregation functions. Always wins. | -| `max_rows` | int | select | Caps the query `LIMIT`. `0` = no limit. Must be non-negative. | -| `max_execution_time_ms` | int | select | Caps the query timeout (min with `clickhouse.query_timeout`). `0` = no limit. Must be non-negative. | +| `max_rows` | int | select | Caps the query `LIMIT`. `0` = no limit (the `query.default_max_rows` config default applies). Must be non-negative. | +| `max_execution_time` | duration or ms | select | Caps the query timeout (min with `clickhouse.query_timeout`). Set as `"5s"`/`"500ms"` or a number of ms; returned as ms. `0` = no limit. Must be non-negative. | +| `max_rows_to_read` | int | select | Caps rows **scanned** server-side (ClickHouse `max_rows_to_read`); the read is rejected once exceeded. `0` = no role limit. Must be non-negative. | +| `max_memory_usage` | size or bytes | select | Caps peak query memory server-side (ClickHouse `max_memory_usage`). Set as `"4GiB"`/`"512MiB"` or a number of bytes; returned as bytes. `0` = no role limit. Must be non-negative. | ## See also diff --git a/docs/src/content/docs/api.md b/docs/src/content/docs/api.md index 902f8163..df7a9958 100644 --- a/docs/src/content/docs/api.md +++ b/docs/src/content/docs/api.md @@ -438,7 +438,7 @@ Every column the query references — in `columns`, an aggregation argument, `fi | `filters` | object[] | No | WHERE conditions (`column`, `op`, `value`). Ops: eq, neq, gt, gte, lt, lte, in, like. | | `group_by` | string[] | No | GROUP BY columns. | | `order_by` | object[] | No | ORDER BY clauses (`column`, `dir`). | -| `limit` | int | No | Max rows. Omitted or above 10,000 → silently capped at 10,000 (`DefaultMaxRows`); a policy `max_rows` can lower it further (see [Access Control](/access-control#resource-limits)). | +| `limit` | int | No | Max rows. Omitted or above the configured `query.default_max_rows` (default 10,000) → silently capped at that value; a policy `max_rows` can lower it further (see [Access Control](/access-control#resource-limits)). | | `time_range` | object | No | Time window (`column`, `since`, `until`). `since`/`until` accept RFC3339 or Go-duration relative values ("1h", "30m", "7d", "2w" — day and week suffixes expand to hours). Relative values mean that long *ago*. The window applies only when `column` and `since` are set — an `until` without `since` is ignored. | :::note[Identifier names] diff --git a/docs/src/content/docs/configuration.mdx b/docs/src/content/docs/configuration.mdx index 9210fd5d..e27b2329 100644 --- a/docs/src/content/docs/configuration.mdx +++ b/docs/src/content/docs/configuration.mdx @@ -49,7 +49,55 @@ The server speaks **plain HTTP** — there is no inbound-TLS setting. Terminate | `clickhouse.database` | `WH_CH_DATABASE` | `default` | Database name. Tables are discovered from this database. | | `clickhouse.username` | `WH_CH_USERNAME` | `default` | Authentication username. | | `clickhouse.password` | `WH_CH_PASSWORD` | *(empty)* | Authentication password. | -| `clickhouse.query_timeout` | `WH_CH_QUERY_TIMEOUT` | `30s` | Maximum allowed execution time for ClickHouse queries | +| `clickhouse.query_timeout` | `WH_CH_QUERY_TIMEOUT` | `30s` | Maximum wall-clock time WaveHouse waits for a read. It bounds the client context deadline, from which the driver derives a server-side `max_execution_time`. | + +### Query + +| YAML Key | Env Var | Default | Description | +| --- | --- | ------- | ----------- | +| `query.default_max_rows` | `WH_QUERY_DEFAULT_MAX_ROWS` | `10000` | Fallback result `LIMIT` for a structured query when the caller and policy specify none. `0` falls back to the built-in default; a negative value is rejected at startup. | + +This is a result-**shaping** default, not a resource limit. Server-wide resource limits (memory, rows scanned, execution time) belong in ClickHouse — see [Server-side resource limits](#server-side-resource-limits) below. + +### Server-side resource limits + +WaveHouse enforces a role's **per-role** resource caps (from the [access-control policy](/access-control#resource-limits)) by attaching them to each query as ClickHouse settings. **Server-wide** limits — the backstop that applies to *every* query regardless of role, including raw admin SQL — are configured in **ClickHouse itself**, via its [settings profiles](https://clickhouse.com/docs/operations/settings/settings-profiles) and [quotas](https://clickhouse.com/docs/operations/quotas). This keeps one authoritative place for global governance, has ClickHouse enforce it natively (defense-in-depth, even against a WaveHouse bug), and lets you use standard ClickHouse operations. + +Set the backstop on the profile of the ClickHouse user WaveHouse connects as (`clickhouse.username`). For example, in `users.xml`: + +```xml + + + + + 4000000000 + 30 + 1000000000 + + + + 8000000000 + + + + + + + + + 3600 + 10000000000 + + + + +``` + +:::caution[How the two layers compose] +WaveHouse's per-role caps are sent as per-query `SETTINGS` on its connection, so they **compose** with the ClickHouse profile — a per-role cap *tightens* within the profile's ceiling, and a `` block bounds how far any setting can move. But if the profile marks a setting `readonly` (or `` disallows changing it), ClickHouse will **reject** WaveHouse's per-query override and the query fails. So keep the settings WaveHouse manages (`max_memory_usage`, `max_execution_time`, `max_rows_to_read`, `max_result_rows`) **changeable** for its user — use a `` constraint, not `readonly`, if you want a hard ceiling. +::: ### Schema Discovery @@ -172,6 +220,9 @@ clickhouse: password: "" query_timeout: 30s +query: + default_max_rows: 10000 + mq: gap_window_minutes: 15 max_bytes_gb: 50 @@ -238,6 +289,8 @@ WH_CH_USERNAME=default WH_CH_PASSWORD= WH_CH_QUERY_TIMEOUT=30s +WH_QUERY_DEFAULT_MAX_ROWS=10000 + WH_MQ_GAP_WINDOW_MINUTES=15 WH_MQ_MAX_BYTES_GB=50 diff --git a/docs/src/content/docs/sdk/queries.md b/docs/src/content/docs/sdk/queries.md index a240693a..7eead8ba 100644 --- a/docs/src/content/docs/sdk/queries.md +++ b/docs/src/content/docs/sdk/queries.md @@ -188,7 +188,7 @@ clicks.select('page').count('*', 'total').orderBy('total', 'desc') clicks.select().limit(100) ``` -If no limit is specified, `QueryBuilder.DEFAULT_LIMIT` (1000) is applied automatically to prevent unbounded result sets. The server also enforces a maximum of 10,000 rows (`DefaultMaxRows`). +If no limit is specified, `QueryBuilder.DEFAULT_LIMIT` (1000) is applied automatically to prevent unbounded result sets. The server also enforces the configured maximum (`query.default_max_rows`, default 10,000 rows). #### `.timeRange(column, since, until?)` diff --git a/go.mod b/go.mod index b43d0819..7a1c7d32 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( github.com/MicahParks/keyfunc/v3 v3.8.0 github.com/cockroachdb/pebble v1.1.5 github.com/dgraph-io/ristretto/v2 v2.4.0 + github.com/dustin/go-humanize v1.0.1 github.com/go-chi/chi/v5 v5.3.0 github.com/golang-jwt/jwt/v5 v5.3.1 github.com/google/uuid v1.6.0 @@ -104,7 +105,6 @@ require ( github.com/dnephin/pflag v1.0.7 // indirect github.com/docker/go-connections v0.6.0 // indirect github.com/docker/go-units v0.5.0 // indirect - github.com/dustin/go-humanize v1.0.1 // indirect github.com/ebitengine/purego v0.10.0 // indirect github.com/erikgeiser/coninput v0.0.0-20211004153227-1c3628e74d0f // indirect github.com/fatih/color v1.18.0 // indirect diff --git a/internal/api/ch_settings.go b/internal/api/ch_settings.go new file mode 100644 index 00000000..9f04a8b8 --- /dev/null +++ b/internal/api/ch_settings.go @@ -0,0 +1,64 @@ +package api + +import ( + "time" + + "github.com/ClickHouse/clickhouse-go/v2" +) + +// chQueryLimits is the per-request resource budget a single read runs under, +// taken from the role's resolved policy caps. A zero field means "no limit" for +// that dimension and is omitted from the settings. Server-wide backstops are +// ClickHouse's job (settings profiles / quotas), not WaveHouse's — so an admin, +// whose policy resolves to no caps, sends no settings here and is bounded only +// by ClickHouse's own config. +type chQueryLimits struct { + // ExecutionTime is the wall-clock budget, emitted as max_execution_time in + // fractional seconds. clickhouse-go already derives max_execution_time from + // the context deadline, but only for deadlines > 1s — so a sub-second cap + // would otherwise reach the server with no time bound, and a context cancel + // can't interrupt an already-running server-side phase. Emitting it + // explicitly closes that hole; for >1s budgets the driver overwrites it with + // deadline+5s, a fine backstop. + ExecutionTime time.Duration + // MaxResultRows caps rows RETURNED (max_result_rows + result_overflow_mode= + // throw) — defense-in-depth behind the SQL LIMIT the structured builder + // applies; not used on the pipe path (a pipe may legitimately return many). + MaxResultRows int + // MaxRowsToRead caps rows SCANNED from storage (max_rows_to_read + + // read_overflow_mode=throw) — the lever that stops a full-table scan. + MaxRowsToRead int64 + // MaxMemoryBytes caps peak query memory (max_memory_usage) — the lever that + // stops a heavy aggregation from exhausting the box. + MaxMemoryBytes int64 +} + +// chReadSettings builds the per-query ClickHouse Settings that enforce a read's +// resource budget SERVER-SIDE, so it can't outrun the budget during a +// server-side scan / merge / aggregation phase (#316). Without these, the only +// budget reaching ClickHouse is whatever clickhouse-go derives from the context +// deadline — which never bounds memory or rows scanned. Returns nil when no cap +// applies, so the caller can skip wrapping the context. +func chReadSettings(l chQueryLimits) clickhouse.Settings { + settings := clickhouse.Settings{} + if l.ExecutionTime > 0 { + // Fractional seconds — ClickHouse accepts them, preserving a sub-second + // cap that a whole-second representation would round away. + settings["max_execution_time"] = l.ExecutionTime.Seconds() + } + if l.MaxResultRows > 0 { + settings["max_result_rows"] = l.MaxResultRows + settings["result_overflow_mode"] = "throw" + } + if l.MaxRowsToRead > 0 { + settings["max_rows_to_read"] = l.MaxRowsToRead + settings["read_overflow_mode"] = "throw" + } + if l.MaxMemoryBytes > 0 { + settings["max_memory_usage"] = l.MaxMemoryBytes + } + if len(settings) == 0 { + return nil + } + return settings +} diff --git a/internal/api/ch_settings_test.go b/internal/api/ch_settings_test.go new file mode 100644 index 00000000..930b5eec --- /dev/null +++ b/internal/api/ch_settings_test.go @@ -0,0 +1,116 @@ +package api + +import ( + "testing" + "time" +) + +// TestChReadSettings locks the resource-budget → ClickHouse-setting mapping that +// enforces caps server-side (#316). It is the regression guard for the mapping +// itself; the handler-level enforcement (settings actually reach ClickHouse and +// are honored) is proven by the integration + e2e suites. +func TestChReadSettings(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + limits chQueryLimits + // want is the exact settings map expected; nil means chReadSettings + // must return nil (no caps → no context wrapping). + want map[string]any + }{ + { + name: "no caps set", + limits: chQueryLimits{}, + want: nil, + }, + { + name: "sub-second execution time is a fractional max_execution_time", + limits: chQueryLimits{ExecutionTime: 500 * time.Millisecond}, + // The driver only auto-derives max_execution_time for deadlines > 1s, + // so a 500ms cap MUST be emitted explicitly or it reaches CH unbounded. + want: map[string]any{"max_execution_time": 0.5}, + }, + { + name: "multi-second execution time", + limits: chQueryLimits{ExecutionTime: 3 * time.Second}, + want: map[string]any{"max_execution_time": 3.0}, + }, + { + name: "max_result_rows caps result rows with throw mode", + limits: chQueryLimits{MaxResultRows: 1000}, + want: map[string]any{ + "max_result_rows": 1000, + "result_overflow_mode": "throw", + }, + }, + { + name: "max_rows_to_read caps rows scanned with throw mode", + limits: chQueryLimits{MaxRowsToRead: 1_000_000}, + want: map[string]any{ + "max_rows_to_read": int64(1_000_000), + "read_overflow_mode": "throw", + }, + }, + { + name: "max_memory_usage caps peak query memory", + limits: chQueryLimits{MaxMemoryBytes: 4 << 30}, // 4 GiB > int32 + want: map[string]any{"max_memory_usage": int64(4 << 30)}, + }, + { + name: "all caps together", + limits: chQueryLimits{ + ExecutionTime: 2 * time.Second, + MaxResultRows: 500, + MaxRowsToRead: 2_000_000, + MaxMemoryBytes: 8 << 30, + }, + want: map[string]any{ + "max_execution_time": 2.0, + "max_result_rows": 500, + "result_overflow_mode": "throw", + "max_rows_to_read": int64(2_000_000), + "read_overflow_mode": "throw", + "max_memory_usage": int64(8 << 30), + }, + }, + { + name: "zero caps are omitted even when others are set", + limits: chQueryLimits{MaxRowsToRead: 42}, + want: map[string]any{ + "max_rows_to_read": int64(42), + "read_overflow_mode": "throw", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + got := chReadSettings(tt.limits) + + if tt.want == nil { + if got != nil { + t.Fatalf("expected nil settings, got %#v", got) + } + return + } + if got == nil { + t.Fatalf("expected settings %#v, got nil", tt.want) + } + if len(got) != len(tt.want) { + t.Fatalf("settings key count mismatch: got %#v, want %#v", map[string]any(got), tt.want) + } + for k, wantV := range tt.want { + gotV, ok := got[k] + if !ok { + t.Errorf("missing setting %q (got %#v)", k, map[string]any(got)) + continue + } + if gotV != wantV { + t.Errorf("setting %q = %#v (%T), want %#v (%T)", k, gotV, gotV, wantV, wantV) + } + } + }) + } +} diff --git a/internal/api/structured_query.go b/internal/api/structured_query.go index 0d2d0cb2..3564cf35 100644 --- a/internal/api/structured_query.go +++ b/internal/api/structured_query.go @@ -8,6 +8,7 @@ import ( "net/http" "time" + "github.com/ClickHouse/clickhouse-go/v2" "github.com/ClickHouse/clickhouse-go/v2/lib/driver" "github.com/Wave-RF/WaveHouse/internal/auth" "github.com/Wave-RF/WaveHouse/internal/cache" @@ -26,6 +27,7 @@ type StructuredQueryHandler struct { BucketSecs int sf singleflight.Group maxQueryTimeout time.Duration + defaultMaxRows int logger *slog.Logger // maxRequestBytes optionally overrides the default inbound request body @@ -43,6 +45,7 @@ func NewStructuredQueryHandler( policyStore *policy.Store, bucketSecs int, queryTimeout time.Duration, + defaultMaxRows int, logger *slog.Logger, ) *StructuredQueryHandler { return &StructuredQueryHandler{ @@ -52,6 +55,7 @@ func NewStructuredQueryHandler( PolicyStore: policyStore, BucketSecs: bucketSecs, maxQueryTimeout: queryTimeout, + defaultMaxRows: defaultMaxRows, logger: logger, } } @@ -110,7 +114,7 @@ func (h *StructuredQueryHandler) Handle(w http.ResponseWriter, r *http.Request) // clause (columns, aggregations, filters, group_by, order_by, time_range) can // skip the check. A policy denial returns a typed error we map to 403; a // malformed query maps to 400. - result, err := query.Build(table, &sq, schema, perms, h.BucketSecs) + result, err := query.Build(table, &sq, schema, perms, h.BucketSecs, h.defaultMaxRows) if err != nil { // A query that selects nothing — no columns, no aggregations, no // select_all — is a request for no data, not an error: return an empty @@ -166,13 +170,32 @@ func (h *StructuredQueryHandler) Handle(w http.ResponseWriter, r *http.Request) // Execute with singleflight. v, err, _ := h.sf.Do(cacheKey, func() (interface{}, error) { timeout := h.maxQueryTimeout - if perms.MaxExecutionTimeMs > 0 { - timeout = min(time.Duration(perms.MaxExecutionTimeMs)*time.Millisecond, timeout) + if perms.MaxExecutionTime > 0 { + timeout = min(perms.MaxExecutionTime.Duration(), timeout) } queryCtx, cancel := context.WithTimeout(r.Context(), timeout) defer cancel() + // Enforce the role's resource caps server-side, not just via the client + // context deadline (#316). The settings ride on the query context, so they + // reach ClickHouse for this query only. Server-wide backstops are + // ClickHouse's job (settings profiles / quotas); a role with no caps (e.g. + // admin) sends nothing here. An explicit max_execution_time is sent only + // when the role set a time cap; otherwise the context deadline (= + // query_timeout) is the time bound the driver derives. + limits := chQueryLimits{ + MaxResultRows: perms.MaxRows, + MaxRowsToRead: perms.MaxRowsToRead, + MaxMemoryBytes: perms.MaxMemoryUsage.Bytes(), + } + if perms.MaxExecutionTime > 0 { + limits.ExecutionTime = timeout + } + if settings := chReadSettings(limits); settings != nil { + queryCtx = clickhouse.Context(queryCtx, clickhouse.WithSettings(settings)) + } + start := time.Now() rows, err := executeCHQuery(queryCtx, h.CHConn, result.SQL, result.Params) diff --git a/internal/api/structured_query_test.go b/internal/api/structured_query_test.go index e1dbd5a3..ba13cad3 100644 --- a/internal/api/structured_query_test.go +++ b/internal/api/structured_query_test.go @@ -40,7 +40,7 @@ func newStructuredQueryHandler() *StructuredQueryHandler { }, }, }) - return NewStructuredQueryHandler(nil, nil, reg, nil, 60, 5*time.Second, testutil.NopLogger()) + return NewStructuredQueryHandler(nil, nil, reg, nil, 60, 5*time.Second, 0, testutil.NopLogger()) } func TestStructuredQuery_MissingTable(t *testing.T) { @@ -292,7 +292,7 @@ func newCapturingHandler(t *testing.T, conn driver.Conn, p *policy.Policy) *Stru }, }, }) - return NewStructuredQueryHandler(conn, nil, reg, policy.NewMemoryStore(p), 60, 5*time.Second, testutil.NopLogger()) + return NewStructuredQueryHandler(conn, nil, reg, policy.NewMemoryStore(p), 60, 5*time.Second, 0, testutil.NopLogger()) } func viewerRequest(t *testing.T, sq query.StructuredQuery) *http.Request { diff --git a/internal/config/config.go b/internal/config/config.go index b150941b..8e7b7466 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -29,6 +29,20 @@ type Config struct { Pipes Pipes `yaml:"pipes"` OTel OTel `yaml:"otel"` Prometheus Prometheus `yaml:"prometheus"` + Query Query `yaml:"query"` +} + +// Query holds query-shaping defaults. Server-wide *resource* limits (memory, +// rows scanned, execution time) deliberately live in ClickHouse itself — its +// settings profiles and quotas, see docs/configuration — so they apply +// uniformly to every query (including raw admin SQL) and compose with the +// per-role caps WaveHouse adds via per-query settings. This block holds only +// the result-shaping default that is genuinely WaveHouse's to own. +type Query struct { + // DefaultMaxRows is the result LIMIT applied to a structured query when the + // caller and policy specify none — the visible, tunable form of what used to + // be the hard-coded query.DefaultMaxRows. 0 falls back to that constant. + DefaultMaxRows int `yaml:"default_max_rows" env:"WH_QUERY_DEFAULT_MAX_ROWS" env-default:"10000"` } // OTel configures the OpenTelemetry pipeline. `enabled` is the master switch; @@ -187,6 +201,14 @@ func (c *Config) Validate() error { return fmt.Errorf("clickhouse.query_timeout must be > 0, got %s", c.ClickHouse.QueryTimeout) } + // query.default_max_rows is the fallback result LIMIT. 0 (or a directly-built + // config that omits it) means "use the built-in query.DefaultMaxRows" — the + // builder substitutes the constant for any non-positive value — so only a + // negative value is an error. + if c.Query.DefaultMaxRows < 0 { + return fmt.Errorf("query.default_max_rows must be non-negative, got %d", c.Query.DefaultMaxRows) + } + if c.MQ.GapWindowMinutes < 0 { return fmt.Errorf("mq.gap_window_minutes must be non-negative") } diff --git a/internal/config/config_test.go b/internal/config/config_test.go index df0fbba7..2507792c 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -67,6 +67,43 @@ auth: assert.Equal(t, "test-secret", cfg.Auth.JWTSecret) } +func TestLoad_QueryLimits_Defaults(t *testing.T) { + t.Parallel() + cfg, err := Load("nonexistent.yaml") + require.NoError(t, err) + // Only the result-LIMIT default lives in WaveHouse config now; server-wide + // resource limits (memory, rows scanned, time) are ClickHouse's job. + assert.Equal(t, 10000, cfg.Query.DefaultMaxRows) +} + +func TestLoad_QueryLimits_FromYAML(t *testing.T) { + t.Parallel() + dir := t.TempDir() + yamlContent := ` +query: + default_max_rows: 25000 +` + path := filepath.Join(dir, "config.yaml") + require.NoError(t, os.WriteFile(path, []byte(yamlContent), 0o600)) + + cfg, err := Load(path) + require.NoError(t, err) + assert.Equal(t, 25000, cfg.Query.DefaultMaxRows) +} + +func TestValidate_NegativeQueryDefaultMaxRows(t *testing.T) { + t.Parallel() + cfg := Config{ + Server: Server{Port: 8080}, + ClickHouse: ClickHouse{HTTPScheme: "http", QueryTimeout: 30 * time.Second}, + Schema: Schema{RefreshInterval: 60}, + Query: Query{DefaultMaxRows: -1}, + } + err := cfg.Validate() + require.Error(t, err) + assert.Contains(t, err.Error(), "default_max_rows") +} + func TestLoad_EnvOverridesYAML(t *testing.T) { dir := t.TempDir() yamlContent := ` diff --git a/internal/policy/policy.go b/internal/policy/policy.go index 4a935a25..eea6a9ab 100644 --- a/internal/policy/policy.go +++ b/internal/policy/policy.go @@ -38,7 +38,17 @@ type RolePermissions struct { AllowedAggregations []string `json:"allowed_aggregations,omitempty" yaml:"allowed_aggregations,omitempty"` DeniedAggregations []string `json:"denied_aggregations,omitempty" yaml:"denied_aggregations,omitempty"` MaxRows int `json:"max_rows,omitempty" yaml:"max_rows,omitempty"` - MaxExecutionTimeMs int `json:"max_execution_time_ms,omitempty" yaml:"max_execution_time_ms,omitempty"` + // MaxExecutionTime, MaxRowsToRead, and MaxMemoryUsage are enforced + // server-side by ClickHouse (the max_execution_time / max_rows_to_read / + // max_memory_usage settings, #316), not just as a client-side context + // deadline. They cap wall-clock time, rows scanned, and peak query memory so + // a heavy aggregation can't exhaust the box within the time budget. + // MaxExecutionTime and MaxMemoryUsage are human-readable scalars ("5s", + // "4GiB") to match clickhouse.query_timeout; MaxRowsToRead is a plain count + // (int64, since it can exceed 2^31 on a large table). + MaxExecutionTime Millis `json:"max_execution_time,omitempty" yaml:"max_execution_time,omitempty"` + MaxRowsToRead int64 `json:"max_rows_to_read,omitempty" yaml:"max_rows_to_read,omitempty"` + MaxMemoryUsage ByteSize `json:"max_memory_usage,omitempty" yaml:"max_memory_usage,omitempty"` } // Filter represents a single comparison operation. @@ -61,7 +71,9 @@ type ResolvedPermissions struct { AllowedAggregations []string DeniedAggregations []string MaxRows int - MaxExecutionTimeMs int + MaxExecutionTime Millis + MaxRowsToRead int64 + MaxMemoryUsage ByteSize } // claimTemplateRe matches {{ jwt.claim.path }} templates. @@ -153,7 +165,9 @@ func Evaluate(p *Policy, role, table, operation string, claims map[string]any) * AllowedAggregations: perms.AllowedAggregations, DeniedAggregations: perms.DeniedAggregations, MaxRows: perms.MaxRows, - MaxExecutionTimeMs: perms.MaxExecutionTimeMs, + MaxExecutionTime: perms.MaxExecutionTime, + MaxRowsToRead: perms.MaxRowsToRead, + MaxMemoryUsage: perms.MaxMemoryUsage, } // Resolve filters into WHERE clause. A bind-unsafe filter column can't be @@ -419,8 +433,14 @@ func validateRolePerms(table, op, role string, perms RolePermissions) error { if perms.MaxRows < 0 { return fmt.Errorf("table %q, op %q, role %q: max_rows must be non-negative", table, op, role) } - if perms.MaxExecutionTimeMs < 0 { - return fmt.Errorf("table %q, op %q, role %q: max_execution_time_ms must be non-negative", table, op, role) + if perms.MaxExecutionTime < 0 { + return fmt.Errorf("table %q, op %q, role %q: max_execution_time must be non-negative", table, op, role) + } + if perms.MaxRowsToRead < 0 { + return fmt.Errorf("table %q, op %q, role %q: max_rows_to_read must be non-negative", table, op, role) + } + if perms.MaxMemoryUsage < 0 { + return fmt.Errorf("table %q, op %q, role %q: max_memory_usage must be non-negative", table, op, role) } // Filter and check column names are interpolated into SQL (backtick-quoted) at // query time, so a '?' in one would shift clickhouse-go's positional value diff --git a/internal/policy/policy_test.go b/internal/policy/policy_test.go index d484eb44..43ff6b5c 100644 --- a/internal/policy/policy_test.go +++ b/internal/policy/policy_test.go @@ -3,6 +3,7 @@ package policy import ( "fmt" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -168,7 +169,9 @@ func TestEvaluate_AggregationLimits(t *testing.T) { AllowedAggregations: []string{"count", "sum"}, DeniedAggregations: []string{"avg"}, MaxRows: 1000, - MaxExecutionTimeMs: 5000, + MaxExecutionTime: Millis(5000), + MaxRowsToRead: 2_000_000, + MaxMemoryUsage: 4 << 30, }, }, }, @@ -179,7 +182,12 @@ func TestEvaluate_AggregationLimits(t *testing.T) { assert.Equal(t, []string{"count", "sum"}, perms.AllowedAggregations) assert.Equal(t, []string{"avg"}, perms.DeniedAggregations) assert.Equal(t, 1000, perms.MaxRows) - assert.Equal(t, 5000, perms.MaxExecutionTimeMs) + // The server-side resource caps (#316) must survive Evaluate so the query + // path can turn them into ClickHouse settings. + assert.Equal(t, Millis(5000), perms.MaxExecutionTime) + assert.Equal(t, 5*time.Second, perms.MaxExecutionTime.Duration()) + assert.Equal(t, int64(2_000_000), perms.MaxRowsToRead) + assert.Equal(t, ByteSize(4<<30), perms.MaxMemoryUsage) } func TestIsColumnAllowed(t *testing.T) { @@ -449,13 +457,41 @@ func TestValidate(t *testing.T) { Tables: map[string]TablePolicy{ "clicks": { Insert: map[string]RolePermissions{ - "user": {MaxExecutionTimeMs: -500}, + "user": {MaxExecutionTime: Millis(-500)}, }, }, }, }, wantErr: true, - wantMsg: "max_execution_time_ms", + wantMsg: "max_execution_time", + }, + { + name: "negative max_rows_to_read", + policy: &Policy{ + Tables: map[string]TablePolicy{ + "clicks": { + Select: map[string]RolePermissions{ + "viewer": {MaxRowsToRead: -1}, + }, + }, + }, + }, + wantErr: true, + wantMsg: "max_rows_to_read", + }, + { + name: "negative max_memory_usage", + policy: &Policy{ + Tables: map[string]TablePolicy{ + "clicks": { + Select: map[string]RolePermissions{ + "viewer": {MaxMemoryUsage: -1}, + }, + }, + }, + }, + wantErr: true, + wantMsg: "max_memory_usage", }, { name: "empty role key rejected", diff --git a/internal/policy/scalars.go b/internal/policy/scalars.go new file mode 100644 index 00000000..5ff3ca44 --- /dev/null +++ b/internal/policy/scalars.go @@ -0,0 +1,140 @@ +package policy + +import ( + "bytes" + "encoding/json" + "fmt" + "math" + "strconv" + "strings" + "time" + + "github.com/dustin/go-humanize" + "gopkg.in/yaml.v3" +) + +// Millis and ByteSize are the human-friendly-in / number-out value types for the +// policy's resource caps. On the way IN (a config file or a hand-crafted API +// body), they accept either a readable string ("10s", "4GiB") or a bare number +// in the canonical unit. On the way OUT (GET /v1/admin/policy and any read-back) +// they marshal as that bare number — they implement no Marshaler, so the default +// integer encoding applies — so SDKs consume a plain int and never reimplement +// the humanization. The canonical units are milliseconds (time) and bytes +// (memory); ClickHouse receives those numbers directly, so its own size/duration +// syntax never leaks to policy authors. + +// Millis is a duration stored as whole milliseconds. Input accepts a Go duration +// string ("10s", "500ms", "1m30s") or a bare integer count of milliseconds. +type Millis int64 + +// Duration returns the value as a time.Duration. +func (m Millis) Duration() time.Duration { return time.Duration(m) * time.Millisecond } + +func (m *Millis) parse(s string) error { + s = strings.TrimSpace(s) + if s == "" { + *m = 0 + return nil + } + d, err := time.ParseDuration(s) + if err != nil { + // A unitless string is taken as milliseconds, mirroring the bare-number form. + if n, convErr := strconv.ParseInt(s, 10, 64); convErr == nil { + *m = Millis(n) + return nil + } + return fmt.Errorf("invalid duration %q (use %q, %q, or a number of milliseconds): %w", s, "5s", "500ms", err) + } + ms := d.Milliseconds() + if d != 0 && ms == 0 { + // Fail closed: a sub-millisecond cap must not round to 0 (which would read + // as "no limit"). Milliseconds() truncates toward zero, so this also + // catches negative sub-ms values like "-500us" that would otherwise slip + // past the negative-cap validation downstream. + return fmt.Errorf("duration %q is below the 1ms resolution of a resource cap", s) + } + *m = Millis(ms) + return nil +} + +// UnmarshalJSON accepts a duration string or a bare millisecond count. +func (m *Millis) UnmarshalJSON(data []byte) error { + data = bytes.TrimSpace(data) + if len(data) > 0 && data[0] == '"' { + var s string + if err := json.Unmarshal(data, &s); err != nil { + return err + } + return m.parse(s) + } + var n int64 + if err := json.Unmarshal(data, &n); err != nil { + return fmt.Errorf("max_execution_time must be a duration string (%q) or a number of milliseconds: %w", "5s", err) + } + *m = Millis(n) + return nil +} + +// UnmarshalYAML accepts either a string or a bare numeric scalar (yaml.v3 hands +// the scalar's text to us either way). A non-scalar node (mapping/sequence) is +// rejected rather than parsed: its empty Value would read as 0 and silently +// disable the cap. +func (m *Millis) UnmarshalYAML(value *yaml.Node) error { + if value.Kind != yaml.ScalarNode { + return fmt.Errorf("max_execution_time must be a scalar duration or millisecond number") + } + return m.parse(value.Value) +} + +// ByteSize is a byte count. Input accepts a size string ("4GiB", "512MiB", with +// SI vs IEC distinguished — "4GB" is 4×10^9, "4GiB" is 4×2^30) or a bare integer +// count of bytes. +type ByteSize int64 + +// Bytes returns the value as a raw byte count. +func (b ByteSize) Bytes() int64 { return int64(b) } + +func (b *ByteSize) parse(s string) error { + s = strings.TrimSpace(s) + if s == "" { + *b = 0 + return nil + } + v, err := humanize.ParseBytes(s) + if err != nil { + return fmt.Errorf("invalid byte size %q (use %q, %q, or a number of bytes): %w", s, "512MiB", "4GiB", err) + } + if v > math.MaxInt64 { + return fmt.Errorf("byte size %q is too large", s) + } + *b = ByteSize(v) + return nil +} + +// UnmarshalJSON accepts a size string or a bare byte count. +func (b *ByteSize) UnmarshalJSON(data []byte) error { + data = bytes.TrimSpace(data) + if len(data) > 0 && data[0] == '"' { + var s string + if err := json.Unmarshal(data, &s); err != nil { + return err + } + return b.parse(s) + } + var n int64 + if err := json.Unmarshal(data, &n); err != nil { + return fmt.Errorf("max_memory_usage must be a size string (%q) or a number of bytes: %w", "4GiB", err) + } + *b = ByteSize(n) + return nil +} + +// UnmarshalYAML accepts either a string or a bare numeric scalar. A non-scalar +// node (mapping/sequence) is rejected rather than parsed: its empty Value would +// read as 0 and silently disable the cap. +func (b *ByteSize) UnmarshalYAML(value *yaml.Node) error { + if value.Kind != yaml.ScalarNode { + return fmt.Errorf("max_memory_usage must be a scalar size or byte number") + } + return b.parse(value.Value) +} diff --git a/internal/policy/scalars_test.go b/internal/policy/scalars_test.go new file mode 100644 index 00000000..669f16a3 --- /dev/null +++ b/internal/policy/scalars_test.go @@ -0,0 +1,150 @@ +package policy + +import ( + "encoding/json" + "testing" + + "gopkg.in/yaml.v3" +) + +// TestMillis_FlexibleInput pins the parse-friendly-in / number-out contract for +// the duration cap: a string ("10s") or a bare number (milliseconds) on input, +// the canonical millisecond integer on output. +func TestMillis_FlexibleInput(t *testing.T) { + t.Parallel() + tests := []struct { + name string + json string + want Millis + wantErr bool + }{ + {name: "duration string seconds", json: `"10s"`, want: 10_000}, + {name: "duration string ms", json: `"500ms"`, want: 500}, + {name: "duration string compound", json: `"1m30s"`, want: 90_000}, + {name: "bare number is milliseconds", json: `5000`, want: 5000}, + {name: "unitless string is milliseconds", json: `"5000"`, want: 5000}, + {name: "empty string is zero", json: `""`, want: 0}, + {name: "sub-millisecond is rejected", json: `"500us"`, wantErr: true}, + {name: "garbage string is rejected", json: `"nonsense"`, wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + var m Millis + err := json.Unmarshal([]byte(tt.json), &m) + if tt.wantErr { + if err == nil { + t.Fatalf("expected error for %s, got %v", tt.json, m) + } + return + } + if err != nil { + t.Fatalf("unexpected error for %s: %v", tt.json, err) + } + if m != tt.want { + t.Errorf("Millis(%s) = %d, want %d", tt.json, m, tt.want) + } + // Output is always the canonical number, never a string, so SDKs + // consume a plain int. + out, err := json.Marshal(m) + if err != nil { + t.Fatal(err) + } + if got := string(out); got[0] == '"' { + t.Errorf("Millis marshalled as a string %s, want a number", got) + } + }) + } +} + +func TestByteSize_FlexibleInput(t *testing.T) { + t.Parallel() + tests := []struct { + name string + json string + want ByteSize + wantErr bool + }{ + {name: "IEC binary", json: `"4GiB"`, want: 4 << 30}, + {name: "SI decimal differs from IEC", json: `"4GB"`, want: 4_000_000_000}, + {name: "mebibytes", json: `"512MiB"`, want: 512 << 20}, + {name: "bare number is bytes", json: `4294967296`, want: 4 << 30}, + {name: "unitless string is bytes", json: `"4294967296"`, want: 4 << 30}, + {name: "empty string is zero", json: `""`, want: 0}, + {name: "garbage is rejected", json: `"nonsense"`, wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + var b ByteSize + err := json.Unmarshal([]byte(tt.json), &b) + if tt.wantErr { + if err == nil { + t.Fatalf("expected error for %s, got %d", tt.json, b) + } + return + } + if err != nil { + t.Fatalf("unexpected error for %s: %v", tt.json, err) + } + if b != tt.want { + t.Errorf("ByteSize(%s) = %d, want %d", tt.json, b, tt.want) + } + out, err := json.Marshal(b) + if err != nil { + t.Fatal(err) + } + if got := string(out); got[0] == '"' { + t.Errorf("ByteSize marshalled as a string %s, want a number", got) + } + }) + } +} + +// TestScalars_YAML confirms the YAML bootstrap path accepts the same string or +// bare-number forms (yaml.v3 hands the scalar text to UnmarshalYAML either way) +// and fails closed on inputs that would otherwise silently disable a cap: a +// non-scalar node (empty Value reads as 0) or a negative sub-millisecond +// duration (truncates toward 0). +func TestScalars_YAML(t *testing.T) { + t.Parallel() + tests := []struct { + name string + yaml string + wantT Millis + wantM ByteSize + wantErr bool + }{ + {name: "string form", yaml: "t: 10s\nm: 4GiB\n", wantT: 10_000, wantM: 4 << 30}, + {name: "bare-number form", yaml: "t: 5000\nm: 4294967296\n", wantT: 5000, wantM: 4 << 30}, + {name: "negative sub-millisecond duration is rejected", yaml: "t: -500us\nm: 4GiB\n", wantErr: true}, + {name: "mapping node for duration is rejected", yaml: "t:\n nested: 1\nm: 4GiB\n", wantErr: true}, + {name: "sequence node for byte size is rejected", yaml: "t: 10s\nm:\n - 1\n", wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + type doc struct { + T Millis `yaml:"t"` + M ByteSize `yaml:"m"` + } + var d doc + err := yaml.Unmarshal([]byte(tt.yaml), &d) + if tt.wantErr { + if err == nil { + t.Fatalf("expected error for %q, got T=%d M=%d", tt.yaml, d.T, d.M.Bytes()) + } + return + } + if err != nil { + t.Fatalf("unexpected error for %q: %v", tt.yaml, err) + } + if d.T != tt.wantT { + t.Errorf("T = %d, want %d", d.T, tt.wantT) + } + if d.M != tt.wantM { + t.Errorf("M = %d, want %d", d.M, tt.wantM) + } + }) + } +} diff --git a/internal/query/builder.go b/internal/query/builder.go index b487ec8b..cbd7da63 100644 --- a/internal/query/builder.go +++ b/internal/query/builder.go @@ -12,8 +12,11 @@ import ( "github.com/Wave-RF/WaveHouse/internal/policy" ) -// DefaultMaxRows is applied when no explicit LIMIT is specified and no policy MaxRows is set. -// This prevents unbounded queries from consuming excessive memory. +// DefaultMaxRows is the fallback result LIMIT applied when no explicit LIMIT is +// specified and no policy MaxRows is set — preventing an unbounded read. It is +// the default value of the operator-facing query.default_max_rows config +// knob (passed into Build), and the guard Build falls back to when that knob is +// misconfigured to 0. const DefaultMaxRows = 10000 // BuildResult holds the generated SQL and bound parameters. @@ -42,7 +45,7 @@ type BuildResult struct { // role's allow/deny set); an explicit Columns list projects exactly those (where // "*" is a literal column name, not a wildcard); the two are mutually exclusive. // A query with neither, and no aggregations, selects nothing → ErrEmptyProjection. -func Build(table string, q *StructuredQuery, schema *discovery.TableSchema, perms *policy.ResolvedPermissions, bucketSeconds int) (*BuildResult, error) { +func Build(table string, q *StructuredQuery, schema *discovery.TableSchema, perms *policy.ResolvedPermissions, bucketSeconds, defaultMaxRows int) (*BuildResult, error) { if chsql.BindUnsafe(table) { return nil, fmt.Errorf("unsupported table name (contains '?'): %s", table) } @@ -120,11 +123,18 @@ func Build(table string, q *StructuredQuery, schema *discovery.TableSchema, perm sql += " ORDER BY " + strings.Join(orderParts, ", ") } - // LIMIT — apply explicit or default maximum. - if q.Limit > 0 && q.Limit <= DefaultMaxRows { + // LIMIT — apply the caller's explicit limit, capped at the configured + // default maximum (query.default_max_rows). A misconfigured + // non-positive default falls back to the DefaultMaxRows constant so a read + // can never be left unbounded or clamped to LIMIT 0. + maxRows := defaultMaxRows + if maxRows <= 0 { + maxRows = DefaultMaxRows + } + if q.Limit > 0 && q.Limit <= maxRows { sql += fmt.Sprintf(" LIMIT %d", q.Limit) } else { - sql += fmt.Sprintf(" LIMIT %d", DefaultMaxRows) + sql += fmt.Sprintf(" LIMIT %d", maxRows) } return &BuildResult{SQL: sql, Params: params}, nil diff --git a/internal/query/builder_test.go b/internal/query/builder_test.go index 157aec17..16e4e056 100644 --- a/internal/query/builder_test.go +++ b/internal/query/builder_test.go @@ -28,7 +28,7 @@ func testSchema() *discovery.TableSchema { func TestBuild_SimpleSelect(t *testing.T) { t.Parallel() sq := &StructuredQuery{Columns: []string{"page", "count"}, Limit: 10} - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Equal(t, "SELECT `page`, `count` FROM `clicks` LIMIT 10", result.SQL) assert.Empty(t, result.Params) @@ -37,7 +37,7 @@ func TestBuild_SimpleSelect(t *testing.T) { func TestBuild_SelectStar(t *testing.T) { t.Parallel() // SelectAll (not omitted columns) is what produces a full-row read. - result, err := Build("clicks", &StructuredQuery{SelectAll: true}, testSchema(), nil, 0) + result, err := Build("clicks", &StructuredQuery{SelectAll: true}, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Equal(t, "SELECT * FROM `clicks` LIMIT 10000", result.SQL) } @@ -57,7 +57,7 @@ func TestBuild_EmptyProjection(t *testing.T) { for name, sq := range cases { t.Run(name, func(t *testing.T) { t.Parallel() - _, err := Build("clicks", sq, testSchema(), nil, 0) + _, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.ErrorIs(t, err, ErrEmptyProjection) }) } @@ -67,7 +67,7 @@ func TestBuild_EmptyProjection(t *testing.T) { // ambiguous → ErrColumnsAndSelectAll (handler maps it to 400). func TestBuild_ColumnsAndSelectAll(t *testing.T) { t.Parallel() - _, err := Build("clicks", &StructuredQuery{Columns: Columns{"page"}, SelectAll: true}, testSchema(), nil, 0) + _, err := Build("clicks", &StructuredQuery{Columns: Columns{"page"}, SelectAll: true}, testSchema(), nil, 0, DefaultMaxRows) require.ErrorIs(t, err, ErrColumnsAndSelectAll) } @@ -76,13 +76,13 @@ func TestBuild_ColumnsAndSelectAll(t *testing.T) { func TestBuild_LiteralStarColumn(t *testing.T) { t.Parallel() // Not in the schema → unknown column. - _, err := Build("clicks", &StructuredQuery{Columns: Columns{"*"}}, testSchema(), nil, 0) + _, err := Build("clicks", &StructuredQuery{Columns: Columns{"*"}}, testSchema(), nil, 0, DefaultMaxRows) require.Error(t, err) assert.Contains(t, err.Error(), "unknown column") // Present in the schema → selected as the quoted literal `*`, never a wildcard. starSchema := &discovery.TableSchema{Name: "t", Columns: []discovery.Column{{Name: "*", Type: "String"}}} - result, err := Build("t", &StructuredQuery{Columns: Columns{"*"}}, starSchema, nil, 0) + result, err := Build("t", &StructuredQuery{Columns: Columns{"*"}}, starSchema, nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Equal(t, "SELECT `*` FROM `t` LIMIT 10000", result.SQL) } @@ -93,7 +93,7 @@ func TestBuild_WithAggregation(t *testing.T) { Aggregations: []Aggregation{{Fn: "count", Column: "*", Alias: "total"}}, GroupBy: []string{"page"}, } - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, "count(*) AS `total`") assert.Contains(t, result.SQL, "GROUP BY `page`") @@ -120,7 +120,7 @@ func TestBuild_AllFilterOperators(t *testing.T) { Columns: []string{"page"}, Filters: []Filter{{Column: "page", Op: tt.op, Value: "test"}}, } - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, tt.want) }) @@ -133,7 +133,7 @@ func TestBuild_InFilter(t *testing.T) { Columns: []string{"page"}, Filters: []Filter{{Column: "page", Op: "in", Value: []any{"/home", "/about"}}}, } - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, "`page` IN (?,?)") assert.Len(t, result.Params, 2) @@ -145,14 +145,14 @@ func TestBuild_OrderBy(t *testing.T) { Columns: []string{"page", "count"}, OrderBy: []OrderClause{{Column: "count", Dir: "desc"}, {Column: "page", Dir: "asc"}}, } - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, "ORDER BY `count` DESC, `page` ASC") } func TestBuild_UnknownColumn(t *testing.T) { t.Parallel() - _, err := Build("clicks", &StructuredQuery{Columns: []string{"nonexistent"}}, testSchema(), nil, 0) + _, err := Build("clicks", &StructuredQuery{Columns: []string{"nonexistent"}}, testSchema(), nil, 0, DefaultMaxRows) assert.Error(t, err) assert.Contains(t, err.Error(), "unknown column") } @@ -160,7 +160,7 @@ func TestBuild_UnknownColumn(t *testing.T) { func TestBuild_InvalidAggFn(t *testing.T) { t.Parallel() sq := &StructuredQuery{Aggregations: []Aggregation{{Fn: "drop_table", Column: "count"}}} - _, err := Build("clicks", sq, testSchema(), nil, 0) + _, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) assert.Error(t, err) assert.Contains(t, err.Error(), "unsupported aggregation") } @@ -171,7 +171,7 @@ func TestBuild_TimeRange(t *testing.T) { Columns: []string{"page"}, TimeRange: &TimeRange{Column: "ts", Since: "2024-01-01T00:00:00Z"}, } - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, "`ts` >= ?") assert.Len(t, result.Params, 1) @@ -240,7 +240,7 @@ func TestIsValidAggFn(t *testing.T) { func TestBuild_DefaultMaxRows_Applied(t *testing.T) { t.Parallel() sq := &StructuredQuery{Columns: []string{"page"}} // Limit: 0. - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, fmt.Sprintf("LIMIT %d", DefaultMaxRows)) } @@ -248,7 +248,7 @@ func TestBuild_DefaultMaxRows_Applied(t *testing.T) { func TestBuild_LimitExceedsDefaultMaxRows_Capped(t *testing.T) { t.Parallel() sq := &StructuredQuery{Columns: []string{"page"}, Limit: DefaultMaxRows + 1} - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, fmt.Sprintf("LIMIT %d", DefaultMaxRows)) assert.NotContains(t, result.SQL, fmt.Sprintf("LIMIT %d", DefaultMaxRows+1)) @@ -257,18 +257,49 @@ func TestBuild_LimitExceedsDefaultMaxRows_Capped(t *testing.T) { func TestBuild_LimitWithinRange_Respected(t *testing.T) { t.Parallel() sq := &StructuredQuery{Columns: []string{"page"}, Limit: 50} - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, "LIMIT 50") } +// TestBuild_ConfigurableDefaultMaxRows pins that the default LIMIT is the value +// the caller passes (the query.default_max_rows knob), both as the +// no-limit fallback and as the ceiling an over-large request is clamped to — +// and that a non-positive value falls back to the DefaultMaxRows constant so a +// read is never left unbounded or clamped to LIMIT 0. +func TestBuild_ConfigurableDefaultMaxRows(t *testing.T) { + t.Parallel() + tests := []struct { + name string + limit int + defaultMaxRows int + wantLimit int + }{ + {name: "custom default applied when no limit", limit: 0, defaultMaxRows: 250, wantLimit: 250}, + {name: "request capped at custom default", limit: 999, defaultMaxRows: 250, wantLimit: 250}, + {name: "request under custom default respected", limit: 100, defaultMaxRows: 250, wantLimit: 100}, + {name: "custom default above the constant is honored", limit: 0, defaultMaxRows: 50000, wantLimit: 50000}, + {name: "zero falls back to the constant", limit: 0, defaultMaxRows: 0, wantLimit: DefaultMaxRows}, + {name: "negative falls back to the constant", limit: 0, defaultMaxRows: -1, wantLimit: DefaultMaxRows}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + sq := &StructuredQuery{Columns: []string{"page"}, Limit: tt.limit} + result, err := Build("clicks", sq, testSchema(), nil, 0, tt.defaultMaxRows) + require.NoError(t, err) + assert.Contains(t, result.SQL, fmt.Sprintf("LIMIT %d", tt.wantLimit)) + }) + } +} + func TestBuild_InvalidFilterColumn(t *testing.T) { t.Parallel() sq := &StructuredQuery{ Columns: []string{"page"}, Filters: []Filter{{Column: "nonexistent", Op: "eq", Value: "x"}}, } - _, err := Build("clicks", sq, testSchema(), nil, 0) + _, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.Error(t, err) assert.Contains(t, err.Error(), "unknown column") } @@ -279,7 +310,7 @@ func TestBuild_InvalidGroupByColumn(t *testing.T) { Columns: []string{"page"}, GroupBy: []string{"nonexistent"}, } - _, err := Build("clicks", sq, testSchema(), nil, 0) + _, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.Error(t, err) assert.Contains(t, err.Error(), "unknown column") } @@ -433,7 +464,7 @@ func TestBuild_FilterWithTimestampValue(t *testing.T) { OrderBy: []OrderClause{{Column: "received_timestamp", Dir: "desc"}}, Limit: 3, } - result, err := Build("events", sq, schema, nil, 0) + result, err := Build("events", sq, schema, nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, "`received_timestamp` < ?") require.Len(t, result.Params, 1) @@ -445,7 +476,7 @@ func TestBuild_FilterWithTimestampValue(t *testing.T) { func TestBuild_TableNameWithBacktick(t *testing.T) { t.Parallel() sq := &StructuredQuery{Columns: []string{"page"}, Limit: 10} - result, err := Build("my`table", sq, testSchema(), nil, 0) + result, err := Build("my`table", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) // ClickHouse's canonical escaping (per SHOW CREATE) is backslash, not // backtick-doubling: an embedded ` becomes \`. The column is quoted too. @@ -500,7 +531,7 @@ func TestBuild_InvalidColumns(t *testing.T) { tt := tt t.Run(tt.name, func(t *testing.T) { t.Parallel() - _, err := Build("clicks", tt.sq, testSchema(), nil, 0) + _, err := Build("clicks", tt.sq, testSchema(), nil, 0, DefaultMaxRows) require.Error(t, err) assert.Contains(t, err.Error(), tt.wantErr) }) @@ -513,7 +544,7 @@ func TestBuild_TimeRange_SinceOnly(t *testing.T) { Columns: []string{"page"}, TimeRange: &TimeRange{Column: "ts", Since: "2024-01-01T00:00:00Z", Until: ""}, } - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, "`ts` >= ?") assert.NotContains(t, result.SQL, "`ts` <= ?") @@ -529,7 +560,7 @@ func TestBuild_TimeRange_ClickHouseDateTimeFormat(t *testing.T) { Until: "2024-01-02T03:04:05Z", }, } - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) require.Len(t, result.Params, 2) assert.Equal(t, "2024-01-01 00:00:00", result.Params[0]) @@ -548,7 +579,7 @@ func TestBuild_FilterUnsupportedOp(t *testing.T) { // Unsupported operations should gracefully be ignored by filterToSQL Filters: []Filter{{Column: "page", Op: "magic", Value: "val"}}, } - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) assert.Error(t, err) assert.Nil(t, result) } @@ -560,7 +591,7 @@ func TestBuild_FilterInOp_InvalidValueType(t *testing.T) { // 'in' operator requires an array ([]any) value. A scalar string shouldn't panic. Filters: []Filter{{Column: "page", Op: "in", Value: "not-an-array"}}, } - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) assert.Error(t, err) assert.Nil(t, result) } @@ -593,7 +624,7 @@ func TestBuild_AuthorizesEveryClause(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - _, err := Build("clicks", tt.sq, testSchema(), perms, 0) + _, err := Build("clicks", tt.sq, testSchema(), perms, 0, DefaultMaxRows) var fce *ForbiddenColumnError require.ErrorAs(t, err, &fce, "denied column in %s must be rejected", tt.name) assert.Equal(t, denied, fce.Column) @@ -613,7 +644,7 @@ func TestBuild_AllowsAuthorizedColumnsInEveryClause(t *testing.T) { OrderBy: []OrderClause{{Column: "ts", Dir: "asc"}}, TimeRange: &TimeRange{Column: "ts", Since: "1h"}, } - _, err := Build("clicks", sq, testSchema(), perms, 0) + _, err := Build("clicks", sq, testSchema(), perms, 0, DefaultMaxRows) require.NoError(t, err) } @@ -665,7 +696,7 @@ func TestBuild_SelectAllProjection(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - result, err := Build("clicks", &StructuredQuery{SelectAll: true}, testSchema(), tt.perms, 0) + result, err := Build("clicks", &StructuredQuery{SelectAll: true}, testSchema(), tt.perms, 0, DefaultMaxRows) if tt.wantErr != nil { require.ErrorIs(t, err, tt.wantErr) assert.Nil(t, result) @@ -685,7 +716,7 @@ func TestBuild_ForbiddenAggregation(t *testing.T) { t.Parallel() perms := &policy.ResolvedPermissions{Allowed: true, AllowColumns: []string{"count"}, DeniedAggregations: []string{"sum"}} sq := &StructuredQuery{Aggregations: []Aggregation{{Fn: "sum", Column: "count", Alias: "total"}}} - _, err := Build("clicks", sq, testSchema(), perms, 0) + _, err := Build("clicks", sq, testSchema(), perms, 0, DefaultMaxRows) var fae *ForbiddenAggregationError require.ErrorAs(t, err, &fae) assert.Equal(t, "sum", fae.Fn) @@ -704,7 +735,7 @@ func TestBuild_OrderByAliasSkipsColumnPolicy(t *testing.T) { GroupBy: []string{"page"}, OrderBy: []OrderClause{{Column: "n", Dir: "desc"}}, } - result, err := Build("clicks", sq, testSchema(), perms, 0) + result, err := Build("clicks", sq, testSchema(), perms, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, "ORDER BY `n` DESC") } @@ -716,7 +747,7 @@ func TestBuild_CountStarWithoutReadableColumns(t *testing.T) { t.Parallel() perms := &policy.ResolvedPermissions{Allowed: true, AllowColumns: []string{"nonexistent"}} sq := &StructuredQuery{Aggregations: []Aggregation{{Fn: "count", Column: "*", Alias: "n"}}} - result, err := Build("clicks", sq, testSchema(), perms, 0) + result, err := Build("clicks", sq, testSchema(), perms, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, "count(*) AS `n`") } @@ -744,7 +775,7 @@ func TestBuild_AggregationAliasQuotedAndContained(t *testing.T) { t.Run(alias, func(t *testing.T) { t.Parallel() sq := &StructuredQuery{Aggregations: []Aggregation{{Fn: "count", Column: "*", Alias: alias}}} - result, err := Build("clicks", sq, testSchema(), nil, 0) + result, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoError(t, err) assert.Contains(t, result.SQL, "AS "+chsql.QuoteIdent(alias), "alias must be emitted as one backtick-quoted, escaped token") @@ -757,7 +788,7 @@ func TestBuild_AggregationAliasQuotedAndContained(t *testing.T) { func TestBuild_RejectsBindUnsafeAlias(t *testing.T) { t.Parallel() sq := &StructuredQuery{Aggregations: []Aggregation{{Fn: "count", Column: "*", Alias: "we?ird"}}} - _, err := Build("clicks", sq, testSchema(), nil, 0) + _, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.Error(t, err) assert.Contains(t, err.Error(), "unsupported aggregation alias") } @@ -817,7 +848,7 @@ func TestBuild_PermissiveColumnNames_AcceptedInEveryClause(t *testing.T) { "time_range": {Columns: []string{"id"}, TimeRange: &TimeRange{Column: col, Since: "1h"}}, } for clause, sq := range clauses { - _, err := Build("weird", sq, schema, nil, 0) + _, err := Build("weird", sq, schema, nil, 0, DefaultMaxRows) require.NoErrorf(t, err, "%s clause must accept legal ClickHouse column %q", clause, col) } }) @@ -841,7 +872,7 @@ func TestBuild_PermissiveAliases_Accepted(t *testing.T) { t.Run(alias, func(t *testing.T) { t.Parallel() sq := &StructuredQuery{Aggregations: []Aggregation{{Fn: "count", Column: "*", Alias: alias}}} - _, err := Build("clicks", sq, testSchema(), nil, 0) + _, err := Build("clicks", sq, testSchema(), nil, 0, DefaultMaxRows) require.NoErrorf(t, err, "alias %q must be accepted — containment is an escaping concern, not a rejection", alias) }) } diff --git a/tests/e2e/sdk/query.test.ts b/tests/e2e/sdk/query.test.ts index f7e56e16..144bab2d 100644 --- a/tests/e2e/sdk/query.test.ts +++ b/tests/e2e/sdk/query.test.ts @@ -419,7 +419,7 @@ describe("Query", () => { }); }); - it("enforces max_execution_time_ms policy limit", async () => { + it("enforces max_execution_time policy limit", async () => { const admin = adminClient(); const currentPolicyRes = await admin.policy.get(); @@ -432,7 +432,7 @@ describe("Query", () => { select: { viewer: { allow_columns: ["*"], - max_execution_time_ms: 1, // 1 millisecond limit + max_execution_time: "1ms", // human-readable duration }, }, }, @@ -469,4 +469,35 @@ describe("Query", () => { await admin.policy.set(currentPolicyRes.data!); } }); + + // ── #316: row/memory caps are enforced by ClickHouse server-side ──────────── + // + // Before #316 the policy's resource caps reached ClickHouse only as a client + // context deadline — never as max_rows_to_read / max_memory_usage — so a + // capped role's query returned the full result set instead of being rejected. + // These drive the real public path (SDK → WaveHouse → ClickHouse) under a + // viewer policy whose cap is impossibly small, and assert the server rejects + // the read (500 carrying the ClickHouse limit error). Unlike the + // execution-time race above, both are deterministic: a full scan always blows + // past a 1-row / 1-byte budget on the first attempt. The unique event_id + // filter keeps each query's SQL out of the shared result cache, so a cached + // success can't mask a broken cap. + + it("enforces max_rows_to_read policy limit server-side (#316)", async () => { + await withViewerSelect({ allow_columns: ["*"], max_rows_to_read: 1 }, async () => { + const result = await wh.from(T.clicks).selectAll().where("event_id", "=", testId()).fetch(); + expect(result.error).not.toBeNull(); + expect(result.error!.status).toBe(500); + }); + }); + + it("enforces max_memory_usage policy limit server-side (#316)", async () => { + // A bare number is bytes, so 1 = a 1-byte cap (also exercises the + // number-input form alongside the "1ms" string above). + await withViewerSelect({ allow_columns: ["*"], max_memory_usage: 1 }, async () => { + const result = await wh.from(T.clicks).selectAll().where("event_id", "=", testId()).fetch(); + expect(result.error).not.toBeNull(); + expect(result.error!.status).toBe(500); + }); + }); }); diff --git a/tests/integration/identifier_roundtrip_test.go b/tests/integration/identifier_roundtrip_test.go index dad1869a..9df59084 100644 --- a/tests/integration/identifier_roundtrip_test.go +++ b/tests/integration/identifier_roundtrip_test.go @@ -148,7 +148,7 @@ func TestIntegration_WeirdColumnNamesRoundTrip(t *testing.T) { res, err := query.Build(table, &query.StructuredQuery{ Columns: []string{col}, Filters: []query.Filter{{Column: "id", Op: "eq", Value: "row1"}}, - }, schema, nil, 0) + }, schema, nil, 0, query.DefaultMaxRows) require.NoErrorf(t, err, "builder must accept legal ClickHouse column %q", col) assert.Equalf(t, want, queryOneString(t, res), @@ -182,7 +182,7 @@ func TestIntegration_WeirdTableNamesRoundTrip(t *testing.T) { res, err := query.Build(table, &query.StructuredQuery{ Columns: []string{"val"}, Filters: []query.Filter{{Column: "id", Op: "eq", Value: "row1"}}, - }, schema, nil, 0) + }, schema, nil, 0, query.DefaultMaxRows) require.NoError(t, err) assert.Equalf(t, want, queryOneString(t, res), @@ -207,7 +207,7 @@ func TestIntegration_WeirdEverythingCombined(t *testing.T) { res, err := query.Build(table, &query.StructuredQuery{ Aggregations: []query.Aggregation{{Fn: "max", Column: "weird.metric", Alias: "naïve max"}}, Filters: []query.Filter{{Column: "id", Op: "eq", Value: "row1"}}, - }, schema, nil, 0) + }, schema, nil, 0, query.DefaultMaxRows) require.NoError(t, err, "builder must accept weird table + weird agg column + weird alias") assert.Equal(t, "7", queryOneString(t, res), @@ -232,7 +232,7 @@ func TestIntegration_AliasInjectionContained(t *testing.T) { const evil = "c FROM system.tables; --" res, err := query.Build(table, &query.StructuredQuery{ Aggregations: []query.Aggregation{{Fn: "count", Column: "*", Alias: evil}}, - }, schema, nil, 0) + }, schema, nil, 0, query.DefaultMaxRows) require.NoError(t, err, "a permissive alias must be accepted, not rejected") ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) @@ -262,13 +262,13 @@ func TestIntegration_StarColumnSemantics(t *testing.T) { "discovery should surface a real column literally named *") // columns:["*"] reads the literal column named "*" — not all columns. - res, err := query.Build(table, &query.StructuredQuery{Columns: query.Columns{"*"}}, schema, nil, 0) + res, err := query.Build(table, &query.StructuredQuery{Columns: query.Columns{"*"}}, schema, nil, 0, query.DefaultMaxRows) require.NoError(t, err) assert.Equalf(t, "star-value", queryOneString(t, res), `columns:["*"] must read the literal column named *: %s`, res.SQL) // select_all is the all-columns wildcard (returns every column, including *). - res, err = query.Build(table, &query.StructuredQuery{SelectAll: true}, schema, nil, 0) + res, err = query.Build(table, &query.StructuredQuery{SelectAll: true}, schema, nil, 0, query.DefaultMaxRows) require.NoError(t, err) ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() diff --git a/tests/integration/query_limits_test.go b/tests/integration/query_limits_test.go new file mode 100644 index 00000000..b7b58e17 --- /dev/null +++ b/tests/integration/query_limits_test.go @@ -0,0 +1,118 @@ +//go:build integration + +package tests + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/Wave-RF/WaveHouse/internal/api" + "github.com/Wave-RF/WaveHouse/internal/auth" + "github.com/Wave-RF/WaveHouse/internal/policy" + "github.com/Wave-RF/WaveHouse/internal/testutil" +) + +// TestStructuredQuery_ResourceCapsEnforcedServerSide is the executable proof +// for #316: a role's policy resource caps reach ClickHouse as per-query +// settings and are enforced SERVER-SIDE, so a structured read can't outrun its +// budget during a scan / aggregation phase. Before the fix the handler sent no +// Settings, so every case below returned 200 with the full result set — the +// caps were latent. The control case ("no cap") shares the exact query shape, +// so a rejection in the capped cases is attributable to the cap, not a broken +// query. +// +// It drives the real StructuredQueryHandler (not the shared admin-stamped +// server, which bypasses all policy) against the package's real ClickHouse, as +// a non-admin `viewer` whose policy carries the cap under test. (Server-wide +// resource backstops are ClickHouse's job — its settings profiles / quotas — +// not WaveHouse's, so there's nothing global to assert here; this proves the +// per-role caps that ARE WaveHouse's to enforce.) +func TestStructuredQuery_ResourceCapsEnforcedServerSide(t *testing.T) { + e := env(t) + + // A handful of rows: enough that a max_rows_to_read=1 cap is exceeded by a + // full scan, small enough that the uncapped control returns them all. + const seededRows = 25 + table := createTable(t, "id String, page String, n UInt32", "ORDER BY id") + seedCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + for i := 0; i < seededRows; i++ { + require.NoError(t, e.chConn.Exec(seedCtx, + fmt.Sprintf("INSERT INTO `%s` (id, page, n) VALUES (?, ?, ?)", table), + fmt.Sprintf("row-%02d", i), "/p", uint32(i), + ), "seed row") + } + + tests := []struct { + name string + perms policy.RolePermissions // viewer's per-table select caps + wantStatus int + wantBodyHas string // substring required in the response body + }{ + { + // Control: identical query, no resource cap → full result set. + name: "no cap returns all rows", + perms: policy.RolePermissions{AllowColumns: []string{"*"}}, + wantStatus: http.StatusOK, + wantBodyHas: `"row-24"`, + }, + { + // max_rows_to_read bounds rows SCANNED — the lever that stops a + // full-table scan. A 25-row scan blows past a cap of 1. ClickHouse + // error code 158 == TOO_MANY_ROWS (the native driver surfaces the + // numeric code, not the HTTP interface's symbolic suffix). + name: "per-role max_rows_to_read is enforced (code 158 TOO_MANY_ROWS)", + perms: policy.RolePermissions{AllowColumns: []string{"*"}, MaxRowsToRead: 1}, + wantStatus: http.StatusInternalServerError, + wantBodyHas: "code: 158", + }, + { + // max_memory_usage bounds peak query memory — the lever that stops + // a heavy aggregation from exhausting the box. A 1-byte cap is below + // the floor any query allocates. Code 241 == MEMORY_LIMIT_EXCEEDED. + // (ByteSize literal 1 == 1 byte.) + name: "per-role max_memory_usage is enforced (code 241 MEMORY_LIMIT_EXCEEDED)", + perms: policy.RolePermissions{AllowColumns: []string{"*"}, MaxMemoryUsage: 1}, + wantStatus: http.StatusInternalServerError, + wantBodyHas: "code: 241", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // Fresh handler + policy per case. Cache is nil so every case + // actually executes against ClickHouse (no cross-case cache hit + // masking enforcement). defaultMaxRows 0 falls back to the builder's + // constant. singleflight's zero value is ready to use. + store := policy.NewMemoryStore(&policy.Policy{ + AdminRole: "admin", + Tables: map[string]policy.TablePolicy{ + table: {Select: map[string]policy.RolePermissions{"viewer": tt.perms}}, + }, + }) + h := api.NewStructuredQueryHandler( + e.chConn, nil, e.registry, store, 60, 30*time.Second, 0, testutil.NopLogger(), + ) + + req := httptest.NewRequest(http.MethodPost, + "/v1/query?table="+table, strings.NewReader(`{"select_all":true}`)) + req = req.WithContext(auth.WithRole(req.Context(), "viewer")) + rec := httptest.NewRecorder() + + h.Handle(rec, req) + + body := rec.Body.String() + require.Equal(t, tt.wantStatus, rec.Code, + "unexpected status; body: %s", body) + assert.Contains(t, body, tt.wantBodyHas) + }) + } +}