From 3571e173b745ce5274defcceb27795376c0f2e72 Mon Sep 17 00:00:00 2001 From: Derek Date: Sun, 15 Nov 2026 11:06:57 +1100 Subject: [PATCH] fix(rules): keep a rule's condition and preview it before saving The engine stores structured conditions so the console's rule builder can reopen them for editing. - A rule takes either a structured condition or raw SQL. The engine compiles the condition to SQL and stores both. A rule saved as SQL has no condition. - POST /api/v1/rules/preview runs the same checks with the same organisation restrictions as saving, without writing anything. For a condition that saving would reject, it returns 200 with the reasons. - The condition operators are the source form's six plus lt, lte, gt, gte and in. The RuleCondition schema in openapi.json lists them with each operator's value type and the nesting limit. - GET /api/v1/hunts takes rule= to list the hunts that run a rule. - source-detections.md showed a rule as a where_clause file while the API takes a full SELECT. It now documents stored rules and conditions. Every text operand goes through quote_literal. Each field is split into quoted identifier parts. A field uses the matching source-table column, or a path inside _json read as the operand's type. Compilation fails when ClickHouse is configured but cannot be read. --- docs/control-plane/ui-api-guide.md | 1 + docs/data-plane/source-detections.md | 43 +- docs/reference/api-route-map.md | 7 +- openapi/openapi.json | 429 +++++++++++++- scripts/route-plan.toml | 1 + src/dfe_engine/api/v1/hunts.py | 8 +- src/dfe_engine/api/v1/rules.py | 219 ++++++-- src/dfe_engine/hunts/rule_condition.py | 404 ++++++++++++++ src/dfe_engine/hunts/rule_creation_service.py | 111 +++- src/dfe_engine/hunts/rule_model.py | 31 +- tests/fixtures/api_conformance_allow.txt | 2 + tests/unit/test_api/test_route_refusals.py | 5 + tests/unit/test_api/test_rules_condition.py | 522 ++++++++++++++++++ tests/unit/test_hunts/test_rule_condition.py | 348 ++++++++++++ 14 files changed, 2059 insertions(+), 72 deletions(-) create mode 100644 src/dfe_engine/hunts/rule_condition.py create mode 100644 tests/unit/test_api/test_rules_condition.py create mode 100644 tests/unit/test_hunts/test_rule_condition.py diff --git a/docs/control-plane/ui-api-guide.md b/docs/control-plane/ui-api-guide.md index b20d0a81c..d90332773 100644 --- a/docs/control-plane/ui-api-guide.md +++ b/docs/control-plane/ui-api-guide.md @@ -250,6 +250,7 @@ es.addEventListener('complete', (e) => { | POST | `/orgs/{org}/ingest-credentials/{credential}/rotate` | Mint a new token in place of the old one, returned once. A rotate that loses a race answers the same: `412` on an `If-Match` it sent, else `409 conflict` | | POST | `/rules` | Create hunt rule | | POST | `/rules/validate` | Validate rule | +| POST | `/rules/preview` | Compile a rule condition and check it as a save would, writing nothing | | POST | `/transforms/compile` | Compile VRL/CEL transform | | POST | `/transforms/test` | Test transform against sample data | | GET | `/system/version` | Stack, engine, ui and every pinned app's version (the console footer) | diff --git a/docs/data-plane/source-detections.md b/docs/data-plane/source-detections.md index 247410392..9d0f67b26 100644 --- a/docs/data-plane/source-detections.md +++ b/docs/data-plane/source-detections.md @@ -14,18 +14,51 @@ declared. The source itself is [source.md](source.md). ## Rules -A rule is one detection condition on a source's table: a SQL WHERE fragment, a -CEL filter the engine transpiles to one, or both ANDed together. +A rule is one detection condition on a source's table. `POST /api/v1/rules` takes it as a full `SELECT ... FROM . WHERE ...` in `user_sql` (the FROM names the source table), or as a structured `condition` on a `source`, and never both. A CEL filter may be ANDed in beside either. The engine stores the WHERE it extracts, the source table and the original SQL, which is the file the hunt runner reads. ```yaml # rules/brute_force_login.yaml -- the filename stem is the rule's id -rule_id: brute_force_login -name: Brute Force Login Attempt +display_name: Brute Force Login Attempt severity: high -source_table: winaudit # In the data database unless source_db names another +source_db: dfe +source_table: winaudit where_clause: "event_id = 4625" +original_sql: SELECT * FROM dfe.winaudit WHERE event_id = 4625 ``` +### Conditions + +A `condition` is one test on a field, or `all` or `any` of several, nested to `x-dfe-max-nesting` groups. `negate` inverts a test or a group. The engine compiles it to `original_sql` and `where_clause` and stores it beside them, so `GET /api/v1/rules/{name}` returns the `condition` and the console reopens its builder on it. A rule saved as SQL has no condition, and a `PUT` with SQL drops one. + +```json +{"all": [{"field": "user.name", "value": "bob"}, + {"field": "attempts", "operator": "gte", "value": 5}]} +``` + +| Operator | Value | +|---|---| +| `equals` (the default), `not_equals` | text or a number | +| `includes`, `starts_with`, `ends_with` | text, not empty | +| `lt`, `lte`, `gt`, `gte` | a number, or an ISO 8601 date or date-time (UTC unless it names a zone) | +| `in` | a list of texts or of numbers, at most 500 | +| `exists` | none | + +The `RuleCondition` schema in `openapi.json` publishes the operators, the kind of value each takes (`x-dfe-operand-kinds`) and the nesting limit. + +A field is a column of the source table when the table has one by that name, otherwise a path inside `_json` (`_json.` in front is accepted). A path is read as the type of its operand. A row without the field fails every test but `exists`, and `negate` does not change that, so `exists` negated finds the rows that lack it. Every value is quoted and every field name is split into identifier parts, so no operand reaches the SQL as text. The hunt window bounds time for every rule, so a condition cannot compare `_timestamp` or the other time columns with `lt`, `lte`, `gt` or `gte`, and a save or preview says so. A text operand cannot hold `{window}` or `{unseen}`, which the runner fills in wherever they sit. A condition is always `source_type` `raw`. + +### Preview + +`POST /api/v1/rules/preview` takes `{source, condition, cel_filter?, cost_window_minutes?}` and runs what a save runs under the same org fence, then stops. Send `name` when editing a rule so the count follows the hunts that run it. A condition the save would refuse still answers 200, with the reasons in `sql_errors` or `matches_everything`. + +```json +{"user_sql": "SELECT * FROM `dfe`.`winaudit` WHERE ...", "source_db": "dfe", + "source_table": "winaudit", "where_clause": "...", "sql_errors": [], + "matches_everything": null, "cost_estimate": {"estimated_rows": 2, "band": "ok"}} +``` + +`GET /api/v1/hunts?rule=` lists the hunts whose `rules` include exactly that name. + ## Hunts A hunt groups rules, names the customers whose rows they read, and carries the diff --git a/docs/reference/api-route-map.md b/docs/reference/api-route-map.md index 5b59d3946..ea5b6855d 100644 --- a/docs/reference/api-route-map.md +++ b/docs/reference/api-route-map.md @@ -12,9 +12,9 @@ One row per operation the engine serves today. **New** is where it goes, and its | merged into | 9 | Folded into the operation **New** names | | unchanged | 42 | Same method and path | | frozen | 54 | Kept as it is | -| decide | 129 | Not placed yet: see its question | +| decide | 130 | Not placed yet: see its question | | removed | 0 | Gone, with nothing in its place | -| total | 321 | | +| total | 322 | | ## Questions @@ -23,7 +23,7 @@ One row per operation the engine serves today. **New** is where it goes, and its | [apps-depth](#apps-depth) | 22 | | [groups-scope](#groups-scope) | 11 | | [compute-helpers](#compute-helpers) | 10 | -| [detection-rules](#detection-rules) | 7 | +| [detection-rules](#detection-rules) | 8 | | [hunts-orgs](#hunts-orgs) | 7 | | [synthetic-data](#synthetic-data) | 7 | | [queries](#queries) | 6 | @@ -373,6 +373,7 @@ Detection rules (hunt SQL and CEL): `rules` is kept for routing rules under `sou | `GET /api/v1/rules` | - | - | - | - | decide | ui 3, infra 1 | | | `POST /api/v1/rules` | - | - | - | - | decide | ui 3, infra 1 | | | `POST /api/v1/rules/from-hyperdx` | - | - | - | - | decide | 0 | | +| `POST /api/v1/rules/preview` | - | - | - | - | decide | 0 | | | `POST /api/v1/rules/validate` | - | - | - | - | decide | ui 2 | | | `DELETE /api/v1/rules/{name}` | - | - | - | - | decide | ui 3 | | | `GET /api/v1/rules/{name}` | - | - | - | - | decide | ui 3 | | diff --git a/openapi/openapi.json b/openapi/openapi.json index b08f887ec..def1498b2 100644 --- a/openapi/openapi.json +++ b/openapi/openapi.json @@ -12078,7 +12078,7 @@ "Rules" ], "summary": "Create Rule", - "description": "Create a new hunt rule via RuleCreationService.\n\nThe service sanitizes the SQL, applies CEL->SQL transpilation,\nvalidates column references, and optionally estimates query cost. A rule\nwhose SQL the hunt runner could not compile is refused with 422. The cost\npreview counts only the orgs the hunts running the rule read, cut to the orgs\nits author may read.\n\nA production+team write is routed to a review branch instead of the branch the\nrunner git-syncs, and the ``X-DFE-Review-Required`` header says so.", + "description": "Create a new hunt rule via RuleCreationService.\n\nThe rule is written as SQL or as a structured ``condition``, which the engine\ncompiles to the SQL and stores with it. The service sanitizes the SQL,\napplies CEL->SQL transpilation, validates column references, and optionally\nestimates query cost. A rule whose SQL the hunt runner could not compile is\nrefused with 422. The cost preview counts only the orgs the hunts running the\nrule read, cut to the orgs its author may read.\n\nA production+team write is routed to a review branch instead of the branch the\nrunner git-syncs, and the ``X-DFE-Review-Required`` header says so.", "operationId": "rules.create", "requestBody": { "required": true, @@ -12348,6 +12348,98 @@ } } }, + "/api/v1/rules/preview": { + "post": { + "tags": [ + "Rules" + ], + "summary": "Preview Rule", + "description": "Compile a condition and check it exactly as a save would, writing nothing.\n\nRuns the creation pipeline of ``POST /rules`` (compile, parse, the forbidden-keyword\nand column checks, the matches-everything check and the alert-volume preview) under the\nsame org fence, and stops before the save. A condition the save would refuse comes back\nwith 200 and its reasons in ``sql_errors`` or ``matches_everything``, so a builder can\nshow them as the author edits. A body that is not a condition at all is 422.", + "operationId": "rules.preview", + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RulePreviewRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/RulePreviewResponse" + } + } + } + }, + "401": { + "description": "No valid session: the credentials are missing, invalid or expired, or the account behind them is disabled or blocked", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + } + }, + "403": { + "description": "Refused: the session lacks the action this route checks, or its account must replace an issued password first. A SCIM route answers a refused action in the RFC 7644 error envelope", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + } + }, + "default": { + "description": "Problem details (RFC 9457)", + "content": { + "application/problem+json": { + "schema": { + "$ref": "#/components/schemas/Problem" + } + } + } + } + }, + "security": [ + { + "BearerAuth": [] + } + ], + "x-scalo": { + "group": "rules", + "resource": "rules", + "hierarchy": [ + "rules" + ], + "verb": "preview", + "read_only": false, + "cli": { + "hidden": false, + "aliases": [], + "positional": [] + }, + "action": "rule:validate" + } + } + }, "/api/v1/rules/{name}": { "get": { "tags": [ @@ -12445,7 +12537,7 @@ "Rules" ], "summary": "Update Rule", - "description": "Replace a detection rule (re-runs creation pipeline, preserves created_at).\n\nA replacement the hunt runner could not compile is refused with 422 and the\nstored rule is left as it was. A production+team write is routed to a review\nbranch instead of the branch the runner git-syncs, and the\n``X-DFE-Review-Required`` header says so.", + "description": "Replace a detection rule (re-runs creation pipeline, preserves created_at).\n\nThe replacement is written as SQL or as a ``condition``, as on create, and a\nrule replaced with SQL no longer carries a condition.\n\nA replacement the hunt runner could not compile is refused with 422 and the\nstored rule is left as it was. A production+team write is routed to a review\nbranch instead of the branch the runner git-syncs, and the\n``X-DFE-Review-Required`` header says so.", "operationId": "rules.apply", "parameters": [ { @@ -14677,6 +14769,24 @@ }, "description": "Case-insensitive search in name, display_name, customers, rules, source_table, target_table" }, + { + "name": "rule", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "description": "Only the hunts whose rules include this rule, by its exact name", + "title": "Rule" + }, + "description": "Only the hunts whose rules include this rule, by its exact name" + }, { "name": "page_size", "in": "query", @@ -48131,6 +48241,105 @@ "title": "RuleBlockResponse", "description": "One block of the receiver's rule set: a source's rows, a profile's, or the default." }, + "RuleCondition": { + "properties": { + "field": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Field", + "description": "Field a test reads: a column of the source table, or a dotted path inside the record as it arrived (e.g. 'tags.collector.type', or '_json.tags.collector.type'). Each part is letters, digits, '_' and '-', starting with a letter or '_'. Absent on a group" + }, + "operator": { + "anyOf": [ + { + "type": "string", + "enum": [ + "equals", + "not_equals", + "exists", + "includes", + "starts_with", + "ends_with", + "lt", + "lte", + "gt", + "gte", + "in" + ] + }, + { + "type": "null" + } + ], + "title": "Operator", + "description": "How a test compares ``field`` to ``value``: equals (the default for a test), not_equals, exists, includes, starts_with, ends_with, lt, lte, gt, gte, in. Absent on a group", + "x-dfe-operand-kinds": { + "ends_with": "text", + "equals": "text_or_number", + "exists": "none", + "gt": "number_or_date", + "gte": "number_or_date", + "in": "list", + "includes": "text", + "lt": "number_or_date", + "lte": "number_or_date", + "not_equals": "text_or_number", + "starts_with": "text" + } + }, + "value": { + "anyOf": [ + { + "$ref": "#/components/schemas/Scalar" + }, + { + "items": { + "$ref": "#/components/schemas/Scalar" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Value", + "description": "Operand for the operator, by its kind: none for exists, text for includes, starts_with and ends_with, text or a number for equals and not_equals, a number or an ISO 8601 date or date-time for lt, lte, gt and gte, a list of texts or of numbers for in" + }, + "negate": { + "type": "boolean", + "title": "Negate", + "description": "Invert this test or group", + "default": false + }, + "all": { + "items": { + "$ref": "#/components/schemas/RuleCondition" + }, + "type": "array", + "title": "All", + "description": "A group that holds when every part holds" + }, + "any": { + "items": { + "$ref": "#/components/schemas/RuleCondition" + }, + "type": "array", + "title": "Any", + "description": "A group that holds when at least one part holds" + } + }, + "additionalProperties": false, + "type": "object", + "title": "RuleCondition", + "description": "One test on a field, or ``all`` or ``any`` of several, as the rule builder writes it.\n\nA test compares one ``field`` with an ``operator``. ``all`` holds when every\npart does and ``any`` when at least one does. Each part is a condition of its\nown, so groups nest, and ``negate`` inverts the test or group it sits on.", + "x-dfe-max-nesting": 4 + }, "RuleCreateRequest": { "properties": { "display_name": { @@ -48152,10 +48361,28 @@ "default": "medium" }, "user_sql": { - "type": "string", - "minLength": 1, + "anyOf": [ + { + "type": "string", + "minLength": 1 + }, + { + "type": "null" + } + ], "title": "User Sql", - "description": "Full detection query: SELECT ... FROM .
WHERE ... (the FROM names the source table)" + "description": "Full detection query: SELECT ... FROM .
WHERE ... (the FROM names the source table). Give this or ``condition``, not both. A rule saved this way has no ``condition``" + }, + "condition": { + "anyOf": [ + { + "$ref": "#/components/schemas/RuleCondition" + }, + { + "type": "null" + } + ], + "description": "The rule's structured condition, which the engine compiles to its SQL and stores with it, so the builder reopens on it. Reads the table of ``source``. Give this or ``user_sql``, not both" }, "cel_filter": { "anyOf": [ @@ -48231,7 +48458,6 @@ }, "type": "object", "required": [ - "user_sql", "name" ], "title": "RuleCreateRequest", @@ -48525,6 +48751,146 @@ "title": "RuleOrderResponse", "description": "The field's blocks as the reorder leaves them." }, + "RulePreviewRequest": { + "properties": { + "source": { + "type": "string", + "title": "Source", + "description": "Source label whose table the condition reads" + }, + "condition": { + "$ref": "#/components/schemas/RuleCondition" + }, + "cel_filter": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Cel Filter", + "description": "CEL expression ANDed in, as on save" + }, + "cost_window_minutes": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "title": "Cost Window Minutes", + "description": "Lookback window in minutes the volume preview counts over. Absent or 0 uses the deployment's preview window." + }, + "name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Name", + "description": "The rule being edited. The volume preview then counts the orgs of the hunts that run it and of the hunt it names, as its save would" + }, + "hunt_name": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Hunt Name", + "description": "The hunt the rule names as its own, which the count follows. Defaults to the stored rule's" + } + }, + "type": "object", + "required": [ + "source", + "condition" + ], + "title": "RulePreviewRequest", + "description": "A condition to compile and check as a save would, without saving it." + }, + "RulePreviewResponse": { + "properties": { + "user_sql": { + "type": "string", + "title": "User Sql", + "description": "The rule SQL the condition compiles to, empty when it does not" + }, + "source_db": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Source Db" + }, + "source_table": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Source Table" + }, + "where_clause": { + "type": "string", + "title": "Where Clause", + "description": "The detection WHERE the rule would store" + }, + "sql_errors": { + "items": { + "$ref": "#/components/schemas/SqlValidationError" + }, + "type": "array", + "title": "Sql Errors", + "description": "Why a save would be refused as invalid" + }, + "matches_everything": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Matches Everything", + "description": "Why a save would be refused for matching every event, when it would" + }, + "cost_estimate": { + "anyOf": [ + { + "$ref": "#/components/schemas/CostEstimate" + }, + { + "type": "null" + } + ] + } + }, + "type": "object", + "required": [ + "user_sql", + "where_clause" + ], + "title": "RulePreviewResponse", + "description": "What a save of the condition would compile to and find, and nothing written." + }, "RuleResponse": { "properties": { "name": { @@ -48578,6 +48944,17 @@ ], "title": "Cel Filter" }, + "condition": { + "anyOf": [ + { + "$ref": "#/components/schemas/RuleCondition" + }, + { + "type": "null" + } + ], + "description": "The structured condition the SQL was compiled from, absent for a rule written as SQL" + }, "original_sql": { "type": "string", "title": "Original Sql" @@ -48730,10 +49107,28 @@ "default": "medium" }, "user_sql": { - "type": "string", - "minLength": 1, + "anyOf": [ + { + "type": "string", + "minLength": 1 + }, + { + "type": "null" + } + ], "title": "User Sql", - "description": "Full detection query: SELECT ... FROM .
WHERE ... (the FROM names the source table)" + "description": "Full detection query: SELECT ... FROM .
WHERE ... (the FROM names the source table). Give this or ``condition``, not both. A rule saved this way has no ``condition``" + }, + "condition": { + "anyOf": [ + { + "$ref": "#/components/schemas/RuleCondition" + }, + { + "type": "null" + } + ], + "description": "The rule's structured condition, which the engine compiles to its SQL and stores with it, so the builder reopens on it. Reads the table of ``source``. Give this or ``user_sql``, not both" }, "cel_filter": { "anyOf": [ @@ -48791,9 +49186,6 @@ } }, "type": "object", - "required": [ - "user_sql" - ], "title": "RuleUpdateRequest", "description": "Update an existing hunt rule (re-runs creation pipeline, and ``source_type`` is not accepted)." }, @@ -49243,6 +49635,19 @@ "title": "ScaffoldRequest", "description": "Body for a starter SELECT over a table's columns." }, + "Scalar": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "integer" + }, + { + "type": "number" + } + ] + }, "ScalingRequest": { "properties": { "replica_count": { diff --git a/scripts/route-plan.toml b/scripts/route-plan.toml index c27aee0b0..dc1328120 100644 --- a/scripts/route-plan.toml +++ b/scripts/route-plan.toml @@ -200,6 +200,7 @@ routes = [ { old = "GET /api/v1/rules", status = "decide", note = "detection-rules" }, { old = "POST /api/v1/rules", status = "decide", note = "detection-rules" }, { old = "POST /api/v1/rules/from-hyperdx", status = "decide", note = "detection-rules" }, + { old = "POST /api/v1/rules/preview", status = "decide", note = "detection-rules" }, { old = "POST /api/v1/rules/validate", status = "decide", note = "detection-rules" }, { old = "DELETE /api/v1/rules/{name}", status = "decide", note = "detection-rules" }, { old = "GET /api/v1/rules/{name}", status = "decide", note = "detection-rules" }, diff --git a/src/dfe_engine/api/v1/hunts.py b/src/dfe_engine/api/v1/hunts.py index 00b93c2b3..984a79203 100644 --- a/src/dfe_engine/api/v1/hunts.py +++ b/src/dfe_engine/api/v1/hunts.py @@ -471,6 +471,10 @@ def list_hunts( "source_table, target_table" ), ), + rule: str | None = Query( + None, + description="Only the hunts whose rules include this rule, by its exact name", + ), ) -> Page[HuntSummary]: """List persisted hunt configurations, each with its run state, paginated. @@ -479,6 +483,8 @@ def list_hunts( (rows the last run wrote). Next due is derived, not stored. """ raw = registry.list_hunts() + if rule is not None: + raw = [row for row in raw if rule in (row.get("rules") or [])] raw = apply_search( raw, search, @@ -489,7 +495,7 @@ def list_hunts( status = _run_status_for([row["name"] for row in raw], now) fraction = settings.hunts.phase_spread_fraction summaries = [_hunt_row_to_summary(row, status.get(row["name"]), now, fraction) for row in raw] - return paginate(summaries, query, bound={"search": search}) + return paginate(summaries, query, bound={"search": search, "rule": rule}) @router.post( diff --git a/src/dfe_engine/api/v1/rules.py b/src/dfe_engine/api/v1/rules.py index 777427f62..fc210f966 100644 --- a/src/dfe_engine/api/v1/rules.py +++ b/src/dfe_engine/api/v1/rules.py @@ -6,6 +6,7 @@ PUT /api/v1/rules/{name} -> Update rule DELETE /api/v1/rules/{name} -> Delete rule POST /api/v1/rules/validate -> Validate SQL/CEL without creating +POST /api/v1/rules/preview -> Compile a condition and check it as a save would, saving nothing """ import functools @@ -13,7 +14,7 @@ from typing import Annotated, Any, Literal from fastapi import APIRouter, Depends, Query, Request, Response -from pydantic import BaseModel, Field +from pydantic import BaseModel, Field, model_validator from scalo.api import ListQuery, Page, ProblemError, list_query, paginate, sort_by_order from scalo.concurrency import run_blocking @@ -35,6 +36,7 @@ from dfe_engine.git_identity import git_author from dfe_engine.hunts.customers import hunt_tenant_ids from dfe_engine.hunts.hunt_config_registry import HuntConfigNotFoundError, default_display_name +from dfe_engine.hunts.rule_condition import RuleCondition from dfe_engine.hunts.rule_creation_service import CostEstimate from dfe_engine.hunts.rule_guard import source_label from dfe_engine.hunts.rule_model import Severity @@ -70,11 +72,21 @@ class _RuleWriteFields(BaseModel): description="Human-readable label (defaults from ``name`` when omitted on create)", ) severity: Severity = Field(default="medium", description="low|medium|high|critical") - user_sql: str = Field( + user_sql: str | None = Field( + default=None, min_length=1, description=( "Full detection query: SELECT ... FROM .
WHERE ... " - "(the FROM names the source table)" + "(the FROM names the source table). Give this or ``condition``, not both. " + "A rule saved this way has no ``condition``" + ), + ) + condition: RuleCondition | None = Field( + default=None, + description=( + "The rule's structured condition, which the engine compiles to its SQL and stores " + "with it, so the builder reopens on it. Reads the table of ``source``. " + "Give this or ``user_sql``, not both" ), ) cel_filter: str | None = Field(default=None, description="CEL expression filter") @@ -95,6 +107,12 @@ class _RuleWriteFields(BaseModel): ), ) + @model_validator(mode="after") + def _sql_or_condition(self) -> _RuleWriteFields: + if (self.user_sql is None) == (self.condition is None): + raise ValueError("give 'user_sql' or 'condition', one of the two") + return self + class RuleCreateRequest(_RuleWriteFields): """Create a hunt rule via RuleCreationService.""" @@ -106,6 +124,14 @@ class RuleCreateRequest(_RuleWriteFields): description="'raw' (plain SQL) or 'hyperdx' (Search saved search format)", ) + @model_validator(mode="after") + def _condition_names_its_source(self) -> RuleCreateRequest: + if self.condition is not None and not self.source: + raise ValueError("a condition reads a source's table, so name the source") + if self.condition is not None and self.source_type != "raw": + raise ValueError("a condition is compiled by the engine, so its source_type is 'raw'") + return self + class RuleUpdateRequest(_RuleWriteFields): """Update an existing hunt rule (re-runs creation pipeline, and ``source_type`` is not accepted).""" @@ -134,6 +160,51 @@ class SqlValidationResponse(BaseModel): errors: list[SqlValidationError] = Field(default_factory=list) +class RulePreviewRequest(BaseModel): + """A condition to compile and check as a save would, without saving it.""" + + source: str = Field(description="Source label whose table the condition reads") + condition: RuleCondition + cel_filter: str | None = Field(default=None, description="CEL expression ANDed in, as on save") + cost_window_minutes: int | None = Field( + default=None, + description=( + "Lookback window in minutes the volume preview counts over. Absent or 0 uses the " + "deployment's preview window." + ), + ) + name: str | None = Field( + default=None, + description=( + "The rule being edited. The volume preview then counts the orgs of the hunts " + "that run it and of the hunt it names, as its save would" + ), + ) + hunt_name: str | None = Field( + default=None, + description="The hunt the rule names as its own, which the count follows. Defaults to the stored rule's", + ) + + +class RulePreviewResponse(BaseModel): + """What a save of the condition would compile to and find, and nothing written.""" + + user_sql: str = Field( + description="The rule SQL the condition compiles to, empty when it does not" + ) + source_db: str | None = None + source_table: str | None = None + where_clause: str = Field(description="The detection WHERE the rule would store") + sql_errors: list[SqlValidationError] = Field( + default_factory=list, description="Why a save would be refused as invalid" + ) + matches_everything: str | None = Field( + default=None, + description="Why a save would be refused for matching every event, when it would", + ) + cost_estimate: CostEstimate | None = None + + class RuleResponse(BaseModel): """A detection rule as the API returns it.""" @@ -144,6 +215,10 @@ class RuleResponse(BaseModel): source_table: str | None = None where_clause: str cel_filter: str | None = None + condition: RuleCondition | None = Field( + default=None, + description="The structured condition the SQL was compiled from, absent for a rule written as SQL", + ) original_sql: str hunt_name: str | None = None source: str | None = None @@ -274,27 +349,22 @@ def create_rule( ): """Create a new hunt rule via RuleCreationService. - The service sanitizes the SQL, applies CEL->SQL transpilation, - validates column references, and optionally estimates query cost. A rule - whose SQL the hunt runner could not compile is refused with 422. The cost - preview counts only the orgs the hunts running the rule read, cut to the orgs - its author may read. + The rule is written as SQL or as a structured ``condition``, which the engine + compiles to the SQL and stores with it. The service sanitizes the SQL, + applies CEL->SQL transpilation, validates column references, and optionally + estimates query cost. A rule whose SQL the hunt runner could not compile is + refused with 422. The cost preview counts only the orgs the hunts running the + rule read, cut to the orgs its author may read. A production+team write is routed to a review branch instead of the branch the runner git-syncs, and the ``X-DFE-Review-Required`` header says so. """ - from dfe_engine.hunts.rule_creation_service import ( - RuleCreateRequest as SvcRequest, - ) - from dfe_engine.hunts.rule_creation_service import ( - RuleCreationService, - ) - from dfe_engine.settings import get_clickhouse_config + from dfe_engine.hunts.rule_creation_service import RuleCreateRequest as SvcRequest if registry.name_exists(body.name): raise ProblemError(409, f"Rule '{body.name}' already exists", code="conflict") - service = RuleCreationService(ch_config=get_clickhouse_config(settings), hunts=settings.hunts) + service = _rule_service(settings) display = ( body.display_name if body.display_name is not None else default_display_name(body.name) ) @@ -304,6 +374,7 @@ def create_rule( severity=body.severity, source_type=body.source_type, user_sql=body.user_sql, + condition=body.condition, cel_filter=body.cel_filter, hunt_name=body.hunt_name, source=body.source, @@ -363,18 +434,12 @@ def _create_rule_from_sql( hunts: OptionalHuntConfigReg, ) -> RuleFromHyperdxResponse: """The deploy-repo and ClickHouse half of ``create_rule_from_hyperdx``, on a worker thread.""" - from dfe_engine.hunts.rule_creation_service import ( - RuleCreateRequest as SvcRequest, - ) - from dfe_engine.hunts.rule_creation_service import ( - RuleCreationService, - ) - from dfe_engine.settings import get_clickhouse_config + from dfe_engine.hunts.rule_creation_service import RuleCreateRequest as SvcRequest rule_id = _unique_rule_id(registry, body.saved_search_name) display = body.saved_search_name or default_display_name(rule_id) - service = RuleCreationService(ch_config=get_clickhouse_config(settings), hunts=settings.hunts) + service = _rule_service(settings) svc_request = SvcRequest( name=display, severity=body.severity, @@ -414,10 +479,7 @@ def validate_rule_sql( Returns any syntax errors or validation issues found. """ - from dfe_engine.hunts.rule_creation_service import RuleCreationService - from dfe_engine.settings import get_clickhouse_config - - service = RuleCreationService(ch_config=get_clickhouse_config(settings), hunts=settings.hunts) + service = _rule_service(settings) errors = service.validate_sql(body.sql) return SqlValidationResponse( @@ -433,6 +495,58 @@ def validate_rule_sql( ) +@router.post( + "/preview", + response_model=RulePreviewResponse, + dependencies=[Depends(require_action(scopes_dict["rule_validate"]))], +) +def preview_rule( + body: RulePreviewRequest, + user: CurrentUser, + settings: Settings, + registry: RuleReg, + request: Request, + hunts: OptionalHuntConfigReg, +) -> RulePreviewResponse: + """Compile a condition and check it exactly as a save would, writing nothing. + + Runs the creation pipeline of ``POST /rules`` (compile, parse, the forbidden-keyword + and column checks, the matches-everything check and the alert-volume preview) under the + same org fence, and stops before the save. A condition the save would refuse comes back + with 200 and its reasons in ``sql_errors`` or ``matches_everything``, so a builder can + show them as the author edits. A body that is not a condition at all is 422. + """ + from dfe_engine.hunts.rule_creation_service import RuleCreateRequest as SvcRequest + + rule_id = body.name or "preview" + hunt_name = body.hunt_name + if hunt_name is None and body.name: + # An update keeps the stored rule's hunt when the body names none, so the preview does. + try: + hunt_name = registry.get(body.name).hunt_name + except RuleNotFoundError: + pass + svc_request = SvcRequest( + name=rule_id, + source=body.source, + condition=body.condition, + cel_filter=body.cel_filter, + hunt_name=hunt_name, + cost_window_minutes=body.cost_window_minutes, + ) + fence = _preview_tenant_ids(request, user, hunts, body.name or "", hunt_name) + result = _rule_service(settings).create_rule(svc_request, rule_id, tenant_ids=fence) + return RulePreviewResponse( + user_sql=result.rule.original_sql, + source_db=result.rule.source_db, + source_table=result.rule.source_table, + where_clause=result.rule.where_clause, + sql_errors=_save_blockers(result), + matches_everything=result.matches_everything, + cost_estimate=result.cost_estimate, + ) + + @router.get( "/{name}", response_model=RuleResponse, @@ -466,6 +580,9 @@ def update_rule( ): """Replace a detection rule (re-runs creation pipeline, preserves created_at). + The replacement is written as SQL or as a ``condition``, as on create, and a + rule replaced with SQL no longer carries a condition. + A replacement the hunt runner could not compile is refused with 422 and the stored rule is left as it was. A production+team write is routed to a review branch instead of the branch the runner git-syncs, and the @@ -476,15 +593,9 @@ def update_rule( except RuleNotFoundError: raise ProblemError(404, f"Rule '{name}' not found", code="not_found") from None - from dfe_engine.hunts.rule_creation_service import ( - RuleCreateRequest as SvcRequest, - ) - from dfe_engine.hunts.rule_creation_service import ( - RuleCreationService, - ) - from dfe_engine.settings import get_clickhouse_config + from dfe_engine.hunts.rule_creation_service import RuleCreateRequest as SvcRequest - service = RuleCreationService(ch_config=get_clickhouse_config(settings), hunts=settings.hunts) + service = _rule_service(settings) effective_display = body.display_name if body.display_name is not None else existing.name effective_source = body.source if body.source is not None else existing.source effective_hunt = body.hunt_name if body.hunt_name is not None else existing.hunt_name @@ -493,6 +604,7 @@ def update_rule( severity=body.severity, source_type="raw", user_sql=body.user_sql, + condition=body.condition, cel_filter=body.cel_filter, hunt_name=effective_hunt, source=effective_source, @@ -544,6 +656,18 @@ def delete_rule( # -- Helpers -------------------------------------------------- +def _rule_service(settings: Settings): + """The creation service every rule check and save runs through.""" + from dfe_engine.hunts.rule_creation_service import RuleCreationService + from dfe_engine.settings import get_clickhouse_config + + return RuleCreationService( + ch_config=get_clickhouse_config(settings), + hunts=settings.hunts, + data_database=settings.clickhouse.effective_data_database, + ) + + def _preview_tenant_ids( request: Request, user: CurrentUser, hunts: Any, rule: str, hunt_name: str | None ) -> list[str] | None: @@ -586,10 +710,7 @@ def _preview_tenant_ids( def _sql_errors_for_original_sql(settings: Settings, original_sql: str) -> list[SqlValidationError]: if not original_sql.strip(): return [] - from dfe_engine.hunts.rule_creation_service import RuleCreationService - from dfe_engine.settings import get_clickhouse_config - - service = RuleCreationService(ch_config=get_clickhouse_config(settings), hunts=settings.hunts) + service = _rule_service(settings) return _map_sql_errors(service.validate_sql(original_sql)) @@ -602,6 +723,7 @@ def _rule_to_response(rule, *, sql_errors: list[SqlValidationError] | None = Non source_table=rule.source_table, where_clause=rule.where_clause, cel_filter=rule.cel_filter, + condition=rule.condition, original_sql=rule.original_sql, hunt_name=rule.hunt_name, source=rule.source, @@ -622,19 +744,26 @@ def _map_sql_errors(errors) -> list[SqlValidationError]: ] -def _refuse_uncompilable(result) -> None: - """Refuse, before anything is written, a rule the hunt runner cannot compile. +def _save_blockers(result) -> list[SqlValidationError]: + """Why a save of the checked rule is refused as invalid, empty when it is not. SQL that does not validate is refused on its SQL errors. SQL that validates can still leave nothing to run, such as a search with no filter, which the rule's own execution check reports. - - Raises: - ProblemError: 422 ``invalid_sql``, every error in ``sql_errors``. """ sql_errors = _map_sql_errors(result.sql_errors or []) if not sql_errors: sql_errors = [SqlValidationError(message=m) for m in result.rule.validate_rule()] + return sql_errors + + +def _refuse_uncompilable(result) -> None: + """Refuse, before anything is written, a rule the hunt runner cannot compile. + + Raises: + ProblemError: 422 ``invalid_sql``, every error in ``sql_errors``. + """ + sql_errors = _save_blockers(result) if not sql_errors: return raise ProblemError( diff --git a/src/dfe_engine/hunts/rule_condition.py b/src/dfe_engine/hunts/rule_condition.py new file mode 100644 index 000000000..87e780838 --- /dev/null +++ b/src/dfe_engine/hunts/rule_condition.py @@ -0,0 +1,404 @@ +# Project: dfe-engine +# File: hunts/rule_condition.py +# Purpose: A rule's structured condition and the WHERE clause it compiles to +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED +"""A rule's condition as data, so the console can reopen the builder that wrote it. + +:class:`RuleCondition` has the match shape a source form uses (a test on a field, or +``all`` and ``any`` groups of them, with ``negate``), widened to the comparisons and +the list test a detection needs. :func:`compile_condition` turns it into the WHERE +fragment the rule stores beside it. Every text operand is quoted with +:func:`~dfe_engine.clickhouse.quoting.quote_literal`, every field is split into +identifier parts that are each quoted with ``quote_identifier``, and numbers are +rendered from Python numbers, so no operand or field name reaches the SQL as text +the caller chose. + +A field is a column of the source table when the table has one by that name, +otherwise a path inside the ``_json`` column, which is how the receiver reads a +field too. A path is read as the type of its operand, the way a CEL rule reads +one, because a JSON path holds a different type from row to row. + +A test on a missing value is neither true nor false in ClickHouse, so every test +but ``exists`` fails for a row without the field, and ``negate`` does not turn +that into a match. ``exists`` negated is how to find rows that lack a field. +""" + +import math +import re +from collections.abc import Callable, Collection +from datetime import UTC, datetime +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, StrictFloat, StrictInt, model_validator + +from dfe_engine.clickhouse.quoting import quote_identifier, quote_literal +from dfe_engine.services.schema.json_promotion_service import JSON_COLUMN +from dfe_engine.source.models import MATCH_NESTING_LIMIT + +RuleConditionOperator = Literal[ + "equals", + "not_equals", + "exists", + "includes", + "starts_with", + "ends_with", + "lt", + "lte", + "gt", + "gte", + "in", +] + +# What each operator's ``value`` holds, published beside the operator list so the +# console offers the right input for it. +OPERAND_KINDS: dict[str, str] = { + "equals": "text_or_number", + "not_equals": "text_or_number", + "exists": "none", + "includes": "text", + "starts_with": "text", + "ends_with": "text", + "lt": "number_or_date", + "lte": "number_or_date", + "gt": "number_or_date", + "gte": "number_or_date", + "in": "list", +} + +_COMPARISONS = { + "equals": "=", + "not_equals": "!=", + "lt": "<", + "lte": "<=", + "gt": ">", + "gte": ">=", +} +_ORDERINGS = frozenset({"lt", "lte", "gt", "gte"}) +_TEXT_TESTS = frozenset({"includes", "starts_with", "ends_with"}) + +MAX_TESTS = 100 +"""The most field tests one condition holds, which bounds the SQL it compiles to.""" + +MAX_LIST_VALUES = 500 +"""The most values an ``in`` test lists.""" + +MAX_OPERAND_LENGTH = 1024 +"""The longest text operand.""" + +RESERVED_TOKENS = ("{window}", "{unseen}", "{timestamp_condition}") +"""Placeholders the hunt runner replaces in a rule's text wherever they sit, a quoted operand included.""" + +MAX_FIELD_PARTS = 16 +"""The most dotted parts one field names.""" + +INT_MIN = -(2**63) +INT_MAX = 2**64 - 1 +"""The integers a typed read can hold: Int64 below, UInt64 above.""" + +_FIELD_PART = re.compile(r"[A-Za-z_][A-Za-z0-9_-]*") +_JSON_PREFIX = f"{JSON_COLUMN}." + +type Scalar = str | StrictInt | StrictFloat + + +class RuleConditionError(ValueError): + """A condition that cannot compile against its source.""" + + +class RuleCondition(BaseModel): + """One test on a field, or ``all`` or ``any`` of several, as the rule builder writes it. + + A test compares one ``field`` with an ``operator``. ``all`` holds when every + part does and ``any`` when at least one does. Each part is a condition of its + own, so groups nest, and ``negate`` inverts the test or group it sits on. + """ + + # One schema for request and response, which the pruning serializer would otherwise split. + # The nesting limit is published so the console reads it instead of restating it. + model_config = ConfigDict( + extra="forbid", + json_schema_mode_override="validation", + json_schema_extra={"x-dfe-max-nesting": MATCH_NESTING_LIMIT}, + ) + + field: str | None = Field( + default=None, + description=( + "Field a test reads: a column of the source table, or a dotted path inside the " + "record as it arrived (e.g. 'tags.collector.type', or '_json.tags.collector.type'). " + "Each part is letters, digits, '_' and '-', starting with a letter or '_'. " + "Absent on a group" + ), + ) + operator: RuleConditionOperator | None = Field( + default=None, + description=( + "How a test compares ``field`` to ``value``: equals (the default for a test), " + "not_equals, exists, includes, starts_with, ends_with, lt, lte, gt, gte, in. " + "Absent on a group" + ), + json_schema_extra={"x-dfe-operand-kinds": {**OPERAND_KINDS}}, + ) + value: Scalar | list[Scalar] | None = Field( + default=None, + description=( + "Operand for the operator, by its kind: none for exists, text for includes, " + "starts_with and ends_with, text or a number for equals and not_equals, a number " + "or an ISO 8601 date or date-time for lt, lte, gt and gte, a list of texts or of " + "numbers for in" + ), + ) + negate: bool = Field(default=False, description="Invert this test or group") + all: list[RuleCondition] = Field( + default_factory=list, description="A group that holds when every part holds" + ) + any: list[RuleCondition] = Field( + default_factory=list, description="A group that holds when at least one part holds" + ) + + @model_validator(mode="before") + @classmethod + def _equals_by_default(cls, data: Any) -> Any: + """A test that names no operator compares with ``equals``.""" + if isinstance(data, dict) and data.get("field") is not None and not data.get("operator"): + return {**data, "operator": "equals"} + return data + + @model_validator(mode="after") + def _one_shape(self) -> RuleCondition: + if self.all and self.any: + raise ValueError("a condition takes all or any, not both: nest one group in the other") + if self.all or self.any: + stray = [k for k in ("field", "operator", "value") if getattr(self, k) is not None] + if stray: + raise ValueError(f"{', '.join(stray)} do not apply beside all or any") + if self.depth() > MATCH_NESTING_LIMIT: + raise ValueError(f"a condition nests at most {MATCH_NESTING_LIMIT} groups deep") + if self.test_count() > MAX_TESTS: + raise ValueError(f"a condition holds at most {MAX_TESTS} tests") + return self + if self.field is None: + raise ValueError("a condition tests a field, or groups tests under all or any") + _check_field(self.field) + if self.operator is None: + raise ValueError("a test names an operator") + _check_operand(self.operator, self.value) + return self + + @property + def is_test(self) -> bool: + """Whether this is one test on a field rather than a group.""" + return not (self.all or self.any) + + @property + def parts(self) -> list[RuleCondition]: + """A group's parts, empty for a test.""" + return self.all or self.any + + def depth(self) -> int: + """How many groups nest here: 0 for a test.""" + if self.is_test: + return 0 + return 1 + max(part.depth() for part in self.parts) + + def test_count(self) -> int: + """How many field tests this holds.""" + if self.is_test: + return 1 + return sum(part.test_count() for part in self.parts) + + +def _check_field(field: str) -> None: + parts = field.split(".") + if len(parts) > MAX_FIELD_PARTS or not all(_FIELD_PART.fullmatch(p) for p in parts): + raise ValueError( + f"field {field!r} is not a column or a dotted path: each part is letters, " + "digits, '_' and '-', starting with a letter or '_', " + f"at most {MAX_FIELD_PARTS} parts" + ) + + +def _check_scalar(value: Scalar, operator: str) -> None: + if isinstance(value, str): + if len(value) > MAX_OPERAND_LENGTH: + raise ValueError( + f"an operand for {operator!r} is at most {MAX_OPERAND_LENGTH} characters" + ) + if "\x00" in value: + raise ValueError(f"an operand for {operator!r} holds no NUL character") + if reserved := next((token for token in RESERVED_TOKENS if token in value), None): + raise ValueError( + f"an operand for {operator!r} holds no {reserved}, which the runner fills in" + ) + elif not math.isfinite(value): + raise ValueError(f"an operand for {operator!r} is a finite number") + elif isinstance(value, int) and not INT_MIN <= value <= INT_MAX: + raise ValueError(f"an integer operand for {operator!r} fits in 64 bits") + + +def _check_operand(operator: str, value: Scalar | list[Scalar] | None) -> None: + """Hold ``value`` to the kind ``operator`` takes. + + Raises: + ValueError: ``value`` is not what ``operator`` compares against. + """ + kind = OPERAND_KINDS[operator] + if kind == "none": + if value is not None: + raise ValueError(f"{operator!r} compares nothing, so it takes no value") + return + if value is None: + raise ValueError(f"value is required when operator is {operator!r}") + if kind == "list": + _check_list(value) + return + if isinstance(value, list): + raise ValueError(f"{operator!r} takes one value, and a list is for 'in'") + _check_scalar(value, operator) + if kind == "text" and not isinstance(value, str): + raise ValueError(f"{operator!r} takes text") + if kind == "text" and not value: + raise ValueError( + f"{operator!r} takes text to look for, and an empty one matches every value" + ) + if kind == "number_or_date" and isinstance(value, str): + instant_literal(value) + + +def _check_list(value: Scalar | list[Scalar]) -> None: + if not isinstance(value, list): + raise ValueError("'in' takes a list of values") + if not value: + raise ValueError("'in' takes at least one value") + if len(value) > MAX_LIST_VALUES: + raise ValueError(f"'in' takes at most {MAX_LIST_VALUES} values") + if len({isinstance(item, str) for item in value}) > 1: + raise ValueError("an 'in' list holds texts or numbers, not both") + for item in value: + _check_scalar(item, "in") + + +def instant_literal(value: str) -> str: + """The ClickHouse expression for an ISO 8601 date or date-time, read as UTC when it names no zone. + + Args: + value: ``2026-01-31``, ``2026-01-31T09:30:00``, or the same with ``Z`` or an offset. + + Returns: + A ``toDateTime64`` call at millisecond precision. + + Raises: + ValueError: ``value`` is not an ISO 8601 date or date-time. + """ + try: + parsed = datetime.fromisoformat(value.strip()) + except ValueError: + raise ValueError(f"{value!r} is not a number or an ISO 8601 date or date-time") from None + parsed = parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed.astimezone(UTC) + stamp = parsed.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] + return f"toDateTime64({quote_literal(stamp)}, 3, 'UTC')" + + +def _operand(value: Scalar) -> str: + if isinstance(value, str): + return quote_literal(value) + return str(value) if isinstance(value, int) else repr(value) + + +def _json_path(path: str) -> str: + return f"{JSON_COLUMN}." + ".".join(quote_identifier(part) for part in path.split(".")) + + +def _accessor(field: str, columns: Collection[str] | None) -> tuple[str, bool]: + """The SQL that reads ``field``, and whether it reads a path inside ``_json``. + + Args: + field: A validated field. + columns: The source table's columns, or None when they are not known, in + which case only a name starting with ``_`` is taken for a column. + + Raises: + RuleConditionError: The field names a column wrongly. + """ + if field == JSON_COLUMN: + raise RuleConditionError(f"name a path inside {JSON_COLUMN}, as {JSON_COLUMN}.") + if field.startswith(_JSON_PREFIX): + return _json_path(field[len(_JSON_PREFIX) :]), True + first, _, rest = field.partition(".") + is_column = first.startswith("_") if columns is None else first in columns + if columns is not None and first.startswith("_") and not is_column: + raise RuleConditionError(f"the source table has no column {first!r}") + if not is_column: + return _json_path(field), True + if rest: + raise RuleConditionError( + f"{first!r} is a column and holds one value, so {field!r} names nothing inside it" + ) + return quote_identifier(field), False + + +def _typed_reads(path: str, sample: Scalar) -> list[str]: + """The typed reads of a JSON path a test on ``sample`` holds in, in the order they are tried.""" + if isinstance(sample, str): + return [f"{path}.:String"] + return [f"{path}.:{kind}" for kind in ("Int64", "UInt64", "Float64")] + + +def _over_reads(sql: str, in_json: bool, sample: Scalar, render: Callable[[str], str]) -> str: + """``render`` applied to a column, or to each typed read of a JSON path, first non-null wins.""" + if not in_json: + return render(sql) + reads = _typed_reads(sql, sample) + if len(reads) == 1: + return render(reads[0]) + return "coalesce(" + ", ".join(render(read) for read in reads) + ")" + + +def _compile_test( + field: str, operator: str, value: Scalar | list[Scalar] | None, columns: Collection[str] | None +) -> str: + sql, in_json = _accessor(field, columns) + if operator == "exists": + return f"isNotNull({sql}) AND notEmpty(toString({sql}))" + if operator in _TEXT_TESTS: + subject = f"{sql}.:String" if in_json else f"toString({sql})" + text = quote_literal(str(value)) + if operator == "includes": + return f"position({subject}, {text}) > 0" + function = "startsWith" if operator == "starts_with" else "endsWith" + return f"{function}({subject}, {text})" + if operator == "in" and isinstance(value, list): + listed = ", ".join(_operand(item) for item in value) + return _over_reads(sql, in_json, value[0], lambda read: f"{read} IN ({listed})") + if value is None or isinstance(value, list): + raise RuleConditionError(f"operator {operator!r} takes a single value") + comparison = _COMPARISONS[operator] + if operator in _ORDERINGS and isinstance(value, str): + subject = f"parseDateTime64BestEffortOrNull({sql}.:String, 3, 'UTC')" if in_json else sql + return f"{subject} {comparison} {instant_literal(value)}" + operand = _operand(value) + return _over_reads(sql, in_json, value, lambda read: f"{read} {comparison} {operand}") + + +def compile_condition(condition: RuleCondition, columns: Collection[str] | None = None) -> str: + """The WHERE fragment ``condition`` stands for. + + Args: + condition: A validated condition. + columns: The source table's columns, or None when they are not known. + + Returns: + A boolean SQL expression over the source table, each test and group bracketed. + + Raises: + RuleConditionError: A field names a column the table lacks, or a path inside one. + """ + if condition.field is not None and condition.operator is not None: + text = f"({_compile_test(condition.field, condition.operator, condition.value, columns)})" + else: + joiner = " AND " if condition.all else " OR " + text = "(" + joiner.join(compile_condition(part, columns) for part in condition.parts) + ")" + return f"NOT {text}" if condition.negate else text diff --git a/src/dfe_engine/hunts/rule_creation_service.py b/src/dfe_engine/hunts/rule_creation_service.py index ad2c3e01b..9251358d7 100644 --- a/src/dfe_engine/hunts/rule_creation_service.py +++ b/src/dfe_engine/hunts/rule_creation_service.py @@ -47,8 +47,10 @@ from ..clickhouse.errors import ErrorCategory, parse_code, wrap_ch_error from ..clickhouse.quoting import plain_source_name, quote_identifier +from ..schema.applier import SchemaApplyError, live_columns from ..settings import DetectionGuardSettings, HuntsSettings from .hdx_sanitizer import REFUSED_CLAUSES, HdxSanitizeError, HdxSanitizer, HdxSanitizeResult +from .rule_condition import RuleCondition, RuleConditionError, compile_condition from .rule_guard import ( VolumeBand, VolumeMeasure, @@ -71,8 +73,10 @@ class RuleCreateRequest(BaseModel): """Input for the rule creation pipeline. - At least one of ``user_sql`` or ``cel_filter`` must be provided. - ``source_type`` controls whether HyperDX sanitization is applied. + At least one of ``user_sql``, ``condition`` or ``cel_filter`` must be provided, + and ``user_sql`` and ``condition`` are two ways to write the same thing, so not + both. A ``condition`` reads the table of ``source``. ``source_type`` controls + whether HyperDX sanitization is applied. """ name: str = Field(..., description="Human-readable rule name") @@ -82,6 +86,9 @@ class RuleCreateRequest(BaseModel): description="SQL origin: 'raw' for clean SQL, 'hyperdx' for HyperDX-generated", ) user_sql: str | None = Field(default=None, description="User-supplied SQL SELECT") + condition: RuleCondition | None = Field( + default=None, description="Structured condition, compiled to the rule's SQL" + ) cel_filter: str | None = Field(default=None, description="CEL expression for query pushdown") hunt_name: str | None = Field(default=None, description="Parent hunt name") source: str | None = Field( @@ -101,8 +108,14 @@ class RuleCreateRequest(BaseModel): @model_validator(mode="after") def _require_sql_or_cel(self) -> RuleCreateRequest: - if not self.user_sql and not self.cel_filter: - raise ValueError("At least one of 'user_sql' or 'cel_filter' must be provided.") + if not self.user_sql and not self.cel_filter and self.condition is None: + raise ValueError( + "At least one of 'user_sql', 'condition' or 'cel_filter' must be provided." + ) + if self.user_sql and self.condition is not None: + raise ValueError("Give 'user_sql' or 'condition', not both.") + if self.condition is not None and self.source_type != "raw": + raise ValueError("A condition is compiled by the engine, so its source_type is 'raw'.") return self @@ -331,6 +344,8 @@ class RuleCreationService: Args: rewriter: Custom RuleRewriter instance (optional). hdx_sanitizer: Custom HdxSanitizer instance (optional). + data_database: The database a source's table is read from when a rule + is written as a condition. ch_config: ClickHouse config. With it, the parse check and the volume preview use the engine's pooled ClickHouse client. hunts: Hunt settings: the detection guard's thresholds and the load-time @@ -344,12 +359,14 @@ def __init__( *, rewriter: RuleRewriter | None = None, hdx_sanitizer: HdxSanitizer | None = None, + data_database: str | None = None, ch_config: dict[str, Any] | None = None, hunts: HuntsSettings | None = None, ch_client: ChQueryClient | None = None, ) -> None: self._rewriter = rewriter or RuleRewriter() self._hdx_sanitizer = hdx_sanitizer or HdxSanitizer() + self._data_database = data_database self._ch_config = ch_config self._hunts = hunts or HuntsSettings() self._ch_client = ch_client @@ -381,6 +398,12 @@ def create_rule( RuleCreateResult with rule, validation errors, the match-everything verdict, cost estimate, and AI context. """ + if request.condition is not None: + try: + compiled = self.condition_sql(request.source, request.condition) + except RuleConditionError as exc: + return self._refused_condition(request, rule_id, str(exc)) + request = request.model_copy(update={"user_sql": compiled}) sanitize_summary: dict[str, Any] = {} clean_sql = request.user_sql refusal: SqlValidationError | None = None @@ -415,6 +438,7 @@ def create_rule( severity=request.severity, user_sql=clean_sql or request.user_sql, cel_filter=request.cel_filter, + condition=request.condition, hunt_name=request.hunt_name, source=request.source, ) @@ -431,6 +455,7 @@ def create_rule( name=request.name, severity=request.severity, cel_filter=request.cel_filter, + condition=request.condition, original_sql=clean_sql or request.user_sql or "", hunt_name=request.hunt_name, source=request.source, @@ -494,6 +519,84 @@ def validate_sql(self, sql: str) -> list[SqlValidationError]: """Standalone SQL validation without rule creation.""" return self._validate_sql_syntax(strip_time_placeholder(sql)) + def condition_sql(self, source: str | None, condition: RuleCondition) -> str: + """The rule SQL a condition on ``source`` compiles to. + + The condition reads the source's table in the data database. A field is a + column of that table when ClickHouse reports one by that name, else a path + inside its ``_json`` column. + + Args: + source: The source's label, which is also its table's name. + condition: A validated condition. + + Returns: + ``SELECT * FROM .
WHERE ``, every name quoted. + + Raises: + RuleConditionError: No source or data database to read, a source that + is not a table label, or a field that does not fit the table. + """ + if not source: + raise RuleConditionError("a condition reads a source's table, so name the source") + if not self._data_database: + raise RuleConditionError("no data database is configured to read the source from") + try: + database, table = plain_source_name(source) + except ValueError as exc: + raise RuleConditionError(f"source {source!r} is not a table label: {exc}") from None + if database: + raise RuleConditionError(f"source {source!r} is a label, not a .
name") + where = compile_condition(condition, self._table_columns(self._data_database, table)) + target = f"{quote_identifier(self._data_database)}.{quote_identifier(table)}" + sql = f"SELECT * FROM {target} WHERE {where}" # noqa: S608 - quoted names, and operands quoted by compile_condition + # What the stored condition says must be what the runner runs, so a rule read of the SQL may not change it. + parsed = self._rewriter.parse_user_sql(sql) + if parsed.where_clause != where: + if parsed.stripped_time_bounds: + raise RuleConditionError( + "the hunt window bounds time for every rule, so a condition cannot compare a " + f"time column with lt, lte, gt or gte: {', '.join(parsed.stripped_time_bounds)}" + ) + raise RuleConditionError("reading the compiled condition as a rule changed it") + return sql + + def _table_columns(self, database: str, table: str) -> frozenset[str] | None: + """The table's column names, or None when no ClickHouse is configured or the table is absent. + + Raises: + RuleConditionError: ClickHouse is configured and could not be read, so a + column cannot be told from a JSON path and compiling would guess. + """ + client = self._query_client() + if client is None: + return None + try: + columns = live_columns(client=client, database=database, table=table) + except SchemaApplyError as exc: + logger.warning(f"columns of {database}.{table} unavailable: {exc}") + raise RuleConditionError( + f"ClickHouse could not be read for the columns of {database}.{table}, so a " + "field cannot be told from a JSON path. Retry shortly" + ) from None + return frozenset(columns) or None + + @staticmethod + def _refused_condition( + request: RuleCreateRequest, rule_id: str, message: str + ) -> RuleCreateResult: + """The result for a condition that does not compile, which the caller refuses on its error.""" + rule = Rule( + rule_id=rule_id, + name=request.name, + severity=request.severity, + cel_filter=request.cel_filter, + condition=request.condition, + hunt_name=request.hunt_name, + source=request.source, + ) + return RuleCreateResult(rule=rule, sql_errors=[SqlValidationError(message=message)]) + # -- Internal methods ---------------------------------- def _validate_sql_syntax(self, sql: str) -> list[SqlValidationError]: diff --git a/src/dfe_engine/hunts/rule_model.py b/src/dfe_engine/hunts/rule_model.py index e2f17b15b..5a8111573 100644 --- a/src/dfe_engine/hunts/rule_model.py +++ b/src/dfe_engine/hunts/rule_model.py @@ -41,8 +41,16 @@ from datetime import UTC, datetime from typing import TYPE_CHECKING, Annotated, Any -from pydantic import AfterValidator, BaseModel, Field, field_validator, model_validator - +from pydantic import ( + AfterValidator, + BaseModel, + Field, + field_serializer, + field_validator, + model_validator, +) + +from .rule_condition import RuleCondition from .rule_rewriter import ParsedRule, RuleRewriter if TYPE_CHECKING: @@ -100,6 +108,10 @@ class RuleCreate(BaseModel): default=None, description="CEL expression for query pushdown (transpiled to SQL WHERE)", ) + condition: RuleCondition | None = Field( + default=None, + description="The structured condition ``user_sql`` was compiled from, kept so a builder can reopen it", + ) hunt_name: str | None = Field( default=None, description="Parent hunt name (optional -- may be assigned later)", @@ -155,6 +167,13 @@ class Rule(BaseModel): default=None, description="Original CEL expression (before transpilation)", ) + condition: RuleCondition | None = Field( + default=None, + description=( + "The structured condition ``original_sql`` was compiled from. None for a " + "rule written as SQL" + ), + ) had_select_star: bool = Field(default=False, description="Whether user used SELECT *") original_sql: str = Field(default="", description="Original user SQL") hunt_name: str | None = Field(default=None, description="Parent hunt name") @@ -182,6 +201,13 @@ class Rule(BaseModel): description="Creation timestamp (ISO 8601)", ) + @field_serializer("condition") + def _store_condition(self, condition: RuleCondition | None) -> dict[str, Any] | None: + """Write only what the builder set, so the stored YAML stays short.""" + if condition is None: + return None + return condition.model_dump(exclude_defaults=True, exclude_none=True) + @classmethod def from_create( cls, @@ -237,6 +263,7 @@ def from_create( source_table=parsed.source_table, where_clause=where_clause, cel_filter=create.cel_filter, + condition=create.condition, had_select_star=parsed.had_select_star, original_sql=parsed.original_sql, hunt_name=create.hunt_name, diff --git a/tests/fixtures/api_conformance_allow.txt b/tests/fixtures/api_conformance_allow.txt index b7067e757..b8d9c881d 100644 --- a/tests/fixtures/api_conformance_allow.txt +++ b/tests/fixtures/api_conformance_allow.txt @@ -82,6 +82,7 @@ dry-run POST /api/v1/queries/raw dry-run POST /api/v1/queries/views/{label}/execute dry-run POST /api/v1/rules dry-run POST /api/v1/rules/from-hyperdx +dry-run POST /api/v1/rules/preview dry-run POST /api/v1/rules/validate dry-run POST /api/v1/sample dry-run POST /api/v1/schemas/definitions/derived/{schema_path} @@ -324,6 +325,7 @@ hierarchy POST /api/v1/queries/raw hierarchy POST /api/v1/queries/views/{label}/execute hierarchy POST /api/v1/rules hierarchy POST /api/v1/rules/from-hyperdx +hierarchy POST /api/v1/rules/preview hierarchy POST /api/v1/rules/validate hierarchy POST /api/v1/sample hierarchy POST /api/v1/schemas/definitions/derived/{schema_path} diff --git a/tests/unit/test_api/test_route_refusals.py b/tests/unit/test_api/test_route_refusals.py index cbd2c1a8c..e8c40c206 100644 --- a/tests/unit/test_api/test_route_refusals.py +++ b/tests/unit/test_api/test_route_refusals.py @@ -55,6 +55,11 @@ {"name": "viewer_rule", "severity": "low", "user_sql": "event_type = 'login'"}, ), ("post", "/api/v1/rules/validate", {"sql": "event_type = 'login'"}), + ( + "post", + "/api/v1/rules/preview", + {"source": "windows-audit", "condition": {"field": "host", "value": "a"}}, + ), ("post", "/api/v1/services/seed", None), ("post", "/api/v1/synthetic-data/lookalike", {"rows": [{"a": 1}], "count": 1}), ("get", "/api/v1/system/defaults/drift", None), diff --git a/tests/unit/test_api/test_rules_condition.py b/tests/unit/test_api/test_rules_condition.py new file mode 100644 index 000000000..92c6ed7e4 --- /dev/null +++ b/tests/unit/test_api/test_rules_condition.py @@ -0,0 +1,522 @@ +# Project: dfe-engine +# File: tests/unit/test_api/test_rules_condition.py +# Purpose: A rule written as a condition: create, update, read back, and the preview +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED +"""A rule stores its structured condition beside the SQL it compiles to. + +The preview runs the save's own checks and writes nothing, and a hunt list filters +by the rule it runs. +""" + +from types import SimpleNamespace + +import pytest + +from dfe_engine.api.deps import get_rule_registry +from dfe_engine.hunts.rule_creation_service import RuleCreationService +from dfe_engine.hunts.rule_model import Rule + +RULES = "/api/v1/rules" +HUNTS = "/api/v1/hunts" +ORGS = "/api/v1/orgs" + +_CONDITION = { + "all": [ + {"field": "user.name", "value": "bob"}, + {"field": "attempts", "operator": "gte", "value": 5}, + ] +} +_COLUMNS = [ + ("_org_id", "String"), + ("_uuid", "UUID"), + ("_timestamp", "DateTime64(3)"), + ("host", "String"), + ("_json", "JSON"), +] + + +class _ClickHouse: + """Reports the source's columns, parses every rule, and counts 100 events with 2 matches.""" + + def __init__(self) -> None: + self.previews: list[str] = [] + self.columns_readable = True + + def command(self, *_args, **_kwargs) -> None: + """Answer the parse and plan checks.""" + + def query(self, sql: str, **_kwargs) -> SimpleNamespace: + """Answer the column read, and record and count a preview.""" + if "system.columns" in sql: + if not self.columns_readable: + raise ConnectionError("clickhouse is down") + return SimpleNamespace(result_rows=_COLUMNS) + self.previews.append(sql) + return SimpleNamespace(result_rows=[(100, 2)]) + + +@pytest.fixture +def clickhouse(monkeypatch) -> _ClickHouse: + recording = _ClickHouse() + monkeypatch.setattr(RuleCreationService, "_query_client", lambda _self: recording) + return recording + + +def _rule(name: str = "bob_attempts", **fields) -> dict: + return {"name": name, "severity": "high", "source": "windows-audit", **fields} + + +def _created(client, headers, **fields): + resp = client.post(RULES, json=_rule(condition=_CONDITION, **fields), headers=headers) + assert resp.status_code == 201, resp.text + return resp.json() + + +class TestCreateWithACondition: + def test_the_engine_compiles_it_and_stores_both(self, client, admin_headers, clickhouse): + created = _created(client, admin_headers) + + rule = created["rule"] + assert rule["condition"]["all"][0] == { + "field": "user.name", + "operator": "equals", + "value": "bob", + "negate": False, + "all": [], + "any": [], + } + assert rule["original_sql"].startswith("SELECT * FROM `dfe`.`windows-audit` WHERE ") + assert rule["source_db"] == "dfe" + assert rule["source_table"] == "windows-audit" + assert "_json.`user`.`name`.:String = 'bob'" in rule["where_clause"] + assert created["cost_estimate"]["estimated_rows"] == 2 + + def test_it_reads_back_as_written(self, client, admin_headers, clickhouse): + written = _created(client, admin_headers)["rule"]["condition"] + + read = client.get(f"{RULES}/bob_attempts", headers=admin_headers).json() + + assert read["condition"] == written + assert "attempts" in read["where_clause"] + + def test_the_stored_condition_holds_only_what_was_set(self): + rule = Rule( + rule_id="r", + name="r", + condition={"all": [{"field": "a", "value": "x"}, {"field": "b", "operator": "exists"}]}, + ) + + assert rule.model_dump()["condition"] == { + "all": [ + {"field": "a", "operator": "equals", "value": "x"}, + {"field": "b", "operator": "exists"}, + ] + } + assert Rule.model_validate(rule.model_dump()).condition == rule.condition + + def test_a_field_is_a_column_where_the_table_has_one(self, client, admin_headers, clickhouse): + body = _rule(condition={"field": "host", "operator": "in", "value": ["a", "b"]}) + + resp = client.post(RULES, json=body, headers=admin_headers) + + assert resp.status_code == 201, resp.text + assert resp.json()["rule"]["where_clause"] == "(`host` IN ('a', 'b'))" + + def test_a_rule_written_as_sql_has_no_condition(self, client, admin_headers): + sql = "SELECT * FROM dfe.main WHERE severity = 'high'" + + resp = client.post(RULES, json=_rule(user_sql=sql), headers=admin_headers) + + assert resp.status_code == 201, resp.text + assert resp.json()["rule"]["condition"] is None + + +class TestUpdateWithACondition: + def test_a_condition_is_replaced_by_another(self, client, admin_headers, clickhouse): + _created(client, admin_headers) + replacement = {"field": "host", "value": "alpha"} + + resp = client.put( + f"{RULES}/bob_attempts", json=_rule(condition=replacement), headers=admin_headers + ) + + assert resp.status_code == 200, resp.text + assert resp.json()["rule"]["condition"]["field"] == "host" + assert resp.json()["rule"]["where_clause"] == "(`host` = 'alpha')" + + def test_sql_replaces_a_condition_and_drops_it(self, client, admin_headers, clickhouse): + _created(client, admin_headers) + sql = "SELECT * FROM dfe.main WHERE severity = 'high'" + + resp = client.put(f"{RULES}/bob_attempts", json=_rule(user_sql=sql), headers=admin_headers) + + assert resp.status_code == 200, resp.text + assert ( + client.get(f"{RULES}/bob_attempts", headers=admin_headers).json()["condition"] is None + ) + + def test_a_condition_reads_the_source_the_rule_already_has( + self, client, admin_headers, clickhouse + ): + sql = "SELECT * FROM dfe.main WHERE severity = 'high'" + client.post(RULES, json=_rule("by_sql", user_sql=sql), headers=admin_headers) + body = {"name": "by_sql", "condition": {"field": "host", "value": "alpha"}} + + resp = client.put(f"{RULES}/by_sql", json=body, headers=admin_headers) + + assert resp.status_code == 200, resp.text + assert resp.json()["rule"]["source_table"] == "windows-audit" + + +class TestARefusedWriteSavesNothing: + @pytest.mark.parametrize( + ("fields", "code", "mentions"), + [ + ( + {"user_sql": "SELECT 1 FROM dfe.main WHERE x = 1"}, + "validation_error", + "one of the two", + ), + ({"condition": None}, "validation_error", "one of the two"), + ({"source": None}, "validation_error", "name the source"), + ({"source": "dfe.windows-audit"}, "invalid_sql", "a label, not a .
"), + ({"source": "a b"}, "invalid_sql", "is not a table label"), + ( + {"condition": {"field": "_nothing", "value": "x"}}, + "invalid_sql", + "no column '_nothing'", + ), + ( + {"condition": {"field": "host.name", "value": "x"}}, + "invalid_sql", + "names nothing inside it", + ), + ({"condition": {"field": "a b", "value": "x"}}, "validation_error", "not a column"), + ( + {"condition": {"field": "a", "operator": "gt", "value": "yesterday"}}, + "validation_error", + "ISO 8601", + ), + ( + {"condition": {"field": "_uuid", "operator": "exists"}}, + "rule_matches_everything", + "every event", + ), + ( + {"condition": {"field": "_timestamp", "operator": "gt", "value": "2026-01-01"}}, + "invalid_sql", + "bounds time for every rule", + ), + ( + { + "condition": { + "all": [ + {"field": "host", "value": "a"}, + {"field": "_timestamp", "operator": "lte", "value": 1700000000}, + ] + } + }, + "invalid_sql", + "bounds time for every rule", + ), + ({"source_type": "hyperdx"}, "validation_error", "source_type is 'raw'"), + ( + {"condition": {"field": "a", "value": "{window}"}}, + "validation_error", + "which the runner fills in", + ), + ], + ids=[ + "sql and condition", + "neither", + "no source", + "source with a database", + "source that is no label", + "unknown header column", + "path inside a plain column", + "field that is no path", + "date that is no date", + "condition that matches everything", + "time column against a date", + "time column against a number in a group", + "hyperdx source type", + "operand carrying a runner placeholder", + ], + ) + def test_create(self, client, admin_headers, clickhouse, fields, code, mentions): + body = _rule(condition=_CONDITION) + body.update(fields) + + resp = client.post(RULES, json=body, headers=admin_headers) + + assert resp.status_code == 422, resp.text + assert resp.json()["code"] == code + assert mentions in resp.text + assert client.get(f"{RULES}/bob_attempts", headers=admin_headers).status_code == 404 + + def test_a_condition_is_not_guessed_when_the_columns_cannot_be_read( + self, client, admin_headers, clickhouse + ): + clickhouse.columns_readable = False + + resp = client.post(RULES, json=_rule(condition=_CONDITION), headers=admin_headers) + + assert resp.status_code == 422, resp.text + assert "ClickHouse could not be read for the columns of dfe.windows-audit" in resp.text + assert client.get(f"{RULES}/bob_attempts", headers=admin_headers).status_code == 404 + + def test_an_update_keeps_the_stored_rule(self, client, admin_headers, clickhouse): + before = _created(client, admin_headers)["rule"] + body = _rule(condition={"field": "_nothing", "value": "x"}) + + resp = client.put(f"{RULES}/bob_attempts", json=body, headers=admin_headers) + + assert resp.status_code == 422, resp.text + assert ( + client.get(f"{RULES}/bob_attempts", headers=admin_headers).json()["condition"] + == (before["condition"]) + ) + + def test_a_condition_with_no_source_anywhere_is_refused_on_update( + self, client, admin_headers, clickhouse + ): + get_rule_registry().save( + Rule(rule_id="bare", name="bare", where_clause="x = 1", source_table="main") + ) + + resp = client.put( + f"{RULES}/bare", + json={"name": "bare", "condition": {"field": "host", "value": "a"}}, + headers=admin_headers, + ) + + assert resp.status_code == 422, resp.text + assert "name the source" in resp.text + + +class TestPreview: + def test_it_answers_what_a_save_would_and_writes_nothing( + self, client, admin_headers, clickhouse + ): + resp = client.post( + f"{RULES}/preview", + json={"source": "windows-audit", "condition": _CONDITION}, + headers=admin_headers, + ) + + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["user_sql"].startswith("SELECT * FROM `dfe`.`windows-audit` WHERE ") + assert (body["source_db"], body["source_table"]) == ("dfe", "windows-audit") + assert "_json.`user`.`name`.:String = 'bob'" in body["where_clause"] + assert body["sql_errors"] == [] + assert body["matches_everything"] is None + assert body["cost_estimate"]["estimated_rows"] == 2 + assert client.get(RULES, headers=admin_headers).json()["items"] == [] + + def test_it_compiles_what_a_save_stores(self, client, admin_headers, clickhouse): + previewed = client.post( + f"{RULES}/preview", + json={"source": "windows-audit", "condition": _CONDITION}, + headers=admin_headers, + ).json() + + saved = _created(client, admin_headers)["rule"] + + assert previewed["user_sql"] == saved["original_sql"] + assert previewed["where_clause"] == saved["where_clause"] + + def test_a_cel_filter_is_anded_in_as_on_save(self, client, admin_headers, clickhouse): + body = { + "source": "windows-audit", + "condition": {"field": "host", "value": "a"}, + "cel_filter": "amount > 10", + } + + previewed = client.post(f"{RULES}/preview", json=body, headers=admin_headers).json() + + assert previewed["where_clause"] == "((`host` = 'a')) AND (amount > 10)" + assert previewed["sql_errors"] == [] + + @pytest.mark.parametrize( + ("body", "in_errors"), + [ + ({"source": "windows-audit", "condition": {"field": "_nothing", "value": "x"}}, True), + ({"source": "dfe.windows-audit", "condition": {"field": "a", "value": "x"}}, True), + ( + { + "source": "windows-audit", + "condition": {"field": "a", "value": "x"}, + "cel_filter": "amount >", + }, + True, + ), + ], + ids=["unknown column", "source with a database", "cel that does not parse"], + ) + def test_a_condition_a_save_refuses_is_answered_with_its_reasons( + self, client, admin_headers, clickhouse, body, in_errors + ): + resp = client.post(f"{RULES}/preview", json=body, headers=admin_headers) + + assert resp.status_code == 200, resp.text + assert bool(resp.json()["sql_errors"]) is in_errors + assert resp.json()["cost_estimate"] is None + + def test_a_condition_that_matches_everything_says_why(self, client, admin_headers, clickhouse): + body = {"source": "windows-audit", "condition": {"field": "_uuid", "operator": "exists"}} + + resp = client.post(f"{RULES}/preview", json=body, headers=admin_headers) + + assert resp.status_code == 200, resp.text + assert "every event" in resp.json()["matches_everything"] + assert resp.json()["cost_estimate"] is None + + def test_a_time_bound_a_save_would_drop_is_reported_not_previewed( + self, client, admin_headers, clickhouse + ): + body = { + "source": "windows-audit", + "condition": {"field": "_timestamp", "operator": "gt", "value": "2026-01-01"}, + } + + resp = client.post(f"{RULES}/preview", json=body, headers=admin_headers) + + assert resp.status_code == 200, resp.text + assert "bounds time for every rule" in resp.json()["sql_errors"][0]["message"] + assert resp.json()["cost_estimate"] is None + + @pytest.mark.parametrize( + "body", + [ + {"source": "windows-audit"}, + {"condition": {"field": "a", "value": "x"}}, + { + "source": "windows-audit", + "condition": {"field": "a", "operator": "regex", "value": "x"}, + }, + {"source": "windows-audit", "condition": {"field": "a", "not": True, "value": "x"}}, + {"source": "windows-audit", "condition": {"field": "a b", "value": "x"}}, + {"source": "windows-audit", "condition": {"all": []}}, + ], + ids=[ + "no condition", + "no source", + "operator no rule takes", + "unknown key", + "field that is no path", + "empty group", + ], + ) + def test_a_body_that_is_no_condition_is_422(self, client, admin_headers, body): + resp = client.post(f"{RULES}/preview", json=body, headers=admin_headers) + + assert resp.status_code == 422, resp.text + assert resp.json()["code"] == "validation_error" + + def test_it_counts_the_orgs_of_the_hunts_that_run_the_rule( + self, client, admin_headers, clickhouse + ): + org = client.post(ORGS, json={"name": "acme", "org_ids": ["acme-1"]}, headers=admin_headers) + assert org.status_code == 201, org.text + hunt = { + "name": "windows_hunt", + "cron": "*/5 * * * *", + "rules": ["bob_attempts"], + "global_source_table_name": "dfe.windows-audit", + "global_target_table_name": "dfe.detection", + "customers": ["acme"], + } + assert client.post(HUNTS, json=hunt, headers=admin_headers).status_code == 201 + + def counted(**extra) -> int: + clickhouse.previews.clear() + body = {"source": "windows-audit", "condition": _CONDITION, **extra} + assert ( + client.post(f"{RULES}/preview", json=body, headers=admin_headers).status_code == 200 + ) + return sum(sql.count("_org_id IN ('acme-1')") for sql in clickhouse.previews) + + assert counted(name="bob_attempts") == 2 + assert counted(hunt_name="windows_hunt") == 2 + assert counted() == 0 + + def test_an_edit_counts_the_hunt_the_stored_rule_names(self, client, admin_headers, clickhouse): + org = client.post(ORGS, json={"name": "acme", "org_ids": ["acme-1"]}, headers=admin_headers) + assert org.status_code == 201, org.text + hunt = { + "name": "windows_hunt", + "cron": "*/5 * * * *", + "rules": ["another_rule"], + "global_source_table_name": "dfe.windows-audit", + "global_target_table_name": "dfe.detection", + "customers": ["acme"], + } + assert client.post(HUNTS, json=hunt, headers=admin_headers).status_code == 201 + _created(client, admin_headers, hunt_name="windows_hunt") + clickhouse.previews.clear() + + body = {"source": "windows-audit", "condition": _CONDITION, "name": "bob_attempts"} + assert client.post(f"{RULES}/preview", json=body, headers=admin_headers).status_code == 200 + preview_sql = clickhouse.previews[-1] + clickhouse.previews.clear() + update = _rule(condition=_CONDITION) + update.pop("name") + assert ( + client.put(f"{RULES}/bob_attempts", json=update, headers=admin_headers).status_code + == 200 + ) + + assert preview_sql.count("_org_id IN ('acme-1')") == 2 + assert preview_sql == clickhouse.previews[-1] + + +class TestHuntsByRule: + @pytest.fixture + def hunts(self, client, admin_headers) -> None: + org = client.post(ORGS, json={"name": "acme", "org_ids": ["acme-1"]}, headers=admin_headers) + assert org.status_code == 201, org.text + base = { + "cron": "*/5 * * * *", + "customers": ["acme"], + "global_target_table_name": "dfe.detection", + } + for name, rules in { + "hunt_a": ["alpha", "beta"], + "hunt_b": ["beta"], + "hunt_c": ["alpha_two"], + }.items(): + resp = client.post( + HUNTS, + json={**base, "name": name, "rules": rules, "global_source_table_name": "dfe.main"}, + headers=admin_headers, + ) + assert resp.status_code == 201, resp.text + + @pytest.mark.parametrize( + ("rule", "expected"), + [ + ("alpha", ["hunt_a"]), + ("beta", ["hunt_a", "hunt_b"]), + ("alpha_two", ["hunt_c"]), + ("alp", []), + ("ALPHA", []), + ("nothing", []), + ("", []), + ], + ) + def test_the_list_keeps_the_hunts_that_run_exactly_that_rule( + self, client, admin_headers, hunts, rule, expected + ): + resp = client.get(HUNTS, params={"rule": rule}, headers=admin_headers) + + assert resp.status_code == 200, resp.text + assert sorted(item["name"] for item in resp.json()["items"]) == expected + + def test_without_the_filter_every_hunt_is_listed(self, client, admin_headers, hunts): + resp = client.get(HUNTS, headers=admin_headers) + + assert len(resp.json()["items"]) == 3 diff --git a/tests/unit/test_hunts/test_rule_condition.py b/tests/unit/test_hunts/test_rule_condition.py new file mode 100644 index 000000000..e9879593c --- /dev/null +++ b/tests/unit/test_hunts/test_rule_condition.py @@ -0,0 +1,348 @@ +# Project: dfe-engine +# File: tests/unit/test_hunts/test_rule_condition.py +# Purpose: A rule condition's shape rules and the WHERE it compiles to +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED +"""RuleCondition holds the builder's input to one shape, and compiles to a quoted WHERE. + +The compiled text was run against ClickHouse 26.3 over a table with a ``_json`` column +for the cases in ``COMPILED``, and each matched the rows its test describes. +""" + +import pytest +import sqlglot +from pydantic import ValidationError + +from dfe_engine.hunts.rule_condition import ( + MAX_LIST_VALUES, + MAX_TESTS, + OPERAND_KINDS, + RuleCondition, + RuleConditionError, + compile_condition, + instant_literal, +) +from dfe_engine.hunts.rule_rewriter import RuleRewriter +from dfe_engine.source.models import MATCH_NESTING_LIMIT + +COLUMNS = frozenset({"_org_id", "_timestamp", "event_id", "host", "_json"}) + + +def _compile(body: dict, columns=COLUMNS) -> str: + return compile_condition(RuleCondition.model_validate(body), columns) + + +COMPILED = { + "a JSON path equals text, read as a String": ( + {"field": "user.name", "value": "bob"}, + "(_json.`user`.`name`.:String = 'bob')", + ), + "not_equals is a != test, so a row without the field is not matched": ( + {"field": "user.name", "operator": "not_equals", "value": "bob"}, + "(_json.`user`.`name`.:String != 'bob')", + ), + "a JSON path compared with a number is tried as each numeric type": ( + {"field": "n", "operator": "gt", "value": 5}, + "(coalesce(_json.`n`.:Int64 > 5, _json.`n`.:UInt64 > 5, _json.`n`.:Float64 > 5))", + ), + "a column compared with a number is read as it is": ( + {"field": "event_id", "operator": "lt", "value": 3}, + "(`event_id` < 3)", + ), + "a JSON path against a date is parsed, in UTC": ( + {"field": "ts", "operator": "gte", "value": "2026-01-01"}, + "(parseDateTime64BestEffortOrNull(_json.`ts`.:String, 3, 'UTC') " + ">= toDateTime64('2026-01-01 00:00:00.000', 3, 'UTC'))", + ), + "a column against a date-time with an offset is moved to UTC": ( + {"field": "_timestamp", "operator": "gt", "value": "2026-01-02T10:00:00+10:00"}, + "(`_timestamp` > toDateTime64('2026-01-02 00:00:00.000', 3, 'UTC'))", + ), + "in on a column lists its texts": ( + {"field": "host", "operator": "in", "value": ["alpha", "beta"]}, + "(`host` IN ('alpha', 'beta'))", + ), + "in on a JSON path with numbers is tried as each numeric type": ( + {"field": "n", "operator": "in", "value": [3, 12]}, + "(coalesce(_json.`n`.:Int64 IN (3, 12), _json.`n`.:UInt64 IN (3, 12), " + "_json.`n`.:Float64 IN (3, 12)))", + ), + "starts_with on a JSON path reads a String": ( + {"field": "msg", "operator": "starts_with", "value": "hel"}, + "(startsWith(_json.`msg`.:String, 'hel'))", + ), + "ends_with on a column is taken over its text": ( + {"field": "host", "operator": "ends_with", "value": "ma"}, + "(endsWith(toString(`host`), 'ma'))", + ), + "includes on a column is taken over its text": ( + {"field": "event_id", "operator": "includes", "value": "3"}, + "(position(toString(`event_id`), '3') > 0)", + ), + "exists negated finds rows that lack the field": ( + {"field": "msg", "operator": "exists", "negate": True}, + "NOT (isNotNull(_json.`msg`) AND notEmpty(toString(_json.`msg`)))", + ), + "an explicit _json prefix is the same path as a bare one": ( + {"field": "_json.a-b.c", "value": "x"}, + "(_json.`a-b`.`c`.:String = 'x')", + ), + "a header column is a column": ( + {"field": "_org_id", "value": "acme-1"}, + "(`_org_id` = 'acme-1')", + ), + "groups nest, and negate turns a group into none-of": ( + { + "all": [ + {"field": "host", "value": "alpha"}, + { + "any": [ + {"field": "event_id", "operator": "lt", "value": 3}, + {"field": "event_id", "operator": "gt", "value": 9}, + ], + "negate": True, + }, + ] + }, + "((`host` = 'alpha') AND NOT ((`event_id` < 3) OR (`event_id` > 9)))", + ), +} + + +@pytest.mark.parametrize(("body", "sql"), COMPILED.values(), ids=list(COMPILED)) +def test_compiles_to_the_where(body, sql): + assert _compile(body) == sql + + +# sqlglot has no typed JSON read (``.:String``), so those cases were run on ClickHouse itself. +PARSEABLE = {name: case for name, case in COMPILED.items() if ".:" not in case[1]} + + +@pytest.mark.parametrize(("body", "sql"), PARSEABLE.values(), ids=list(PARSEABLE)) +def test_the_where_parses_as_clickhouse(body, sql): + """The text is one boolean expression ClickHouse's grammar reads, as sqlglot reads it.""" + sqlglot.parse_one(f"SELECT 1 FROM t WHERE {sql}", read="clickhouse") + + +def test_a_field_is_a_column_only_when_the_table_has_it(): + """With the columns unknown only a name starting with an underscore is a column.""" + body = {"field": "host", "value": "alpha"} + + assert _compile(body, columns={"host"}) == "(`host` = 'alpha')" + assert _compile(body, columns=None) == "(_json.`host`.:String = 'alpha')" + assert _compile({"field": "_org_id", "value": "a"}, columns=None) == "(`_org_id` = 'a')" + + +COMPILE_REFUSED = { + "a header column the table lacks": ( + {"field": "_nothing", "value": "x"}, + "no column '_nothing'", + ), + "a path inside a plain column": ( + {"field": "host.name", "value": "x"}, + "names nothing inside it", + ), + "the bare JSON column": ({"field": "_json", "operator": "exists"}, "name a path inside _json"), +} + + +@pytest.mark.parametrize(("body", "message"), COMPILE_REFUSED.values(), ids=list(COMPILE_REFUSED)) +def test_a_field_that_does_not_fit_the_table_is_refused(body, message): + with pytest.raises(RuleConditionError, match=message): + _compile(body) + + +HOSTILE_TEXT = [ + "x' OR 1=1 --", + "x\\' OR 1=1 --", + "x\\", + "'); DROP TABLE dfe.main; --", + "`; SELECT 1 --", + "a\nb", + "%_*?", + "\u2019 '", +] + + +@pytest.mark.parametrize("text", HOSTILE_TEXT) +@pytest.mark.parametrize( + "operator", ["equals", "not_equals", "includes", "starts_with", "ends_with"] +) +def test_an_operand_never_leaves_its_quotes(operator, text): + """Every quote and backslash in a text operand is doubled, so one literal holds it all.""" + sql = _compile({"field": "host", "operator": operator, "value": text}) + + statement = sqlglot.parse_one(f"SELECT 1 FROM t WHERE {sql}", read="clickhouse") + literals = [lit.this for lit in statement.find_all(sqlglot.exp.Literal) if lit.is_string] + assert literals == [text] + + +@pytest.mark.parametrize("text", HOSTILE_TEXT) +def test_a_listed_operand_never_leaves_its_quotes(text): + sql = _compile({"field": "host", "operator": "in", "value": [text, "b"]}) + + statement = sqlglot.parse_one(f"SELECT 1 FROM t WHERE {sql}", read="clickhouse") + literals = [lit.this for lit in statement.find_all(sqlglot.exp.Literal) if lit.is_string] + assert literals == [text, "b"] + + +SHAPE_REFUSED = { + "an unknown key such as not": ({"field": "a", "not": True, "value": "x"}, "not"), + "always, which a rule never takes": ({"field": "a", "operator": "always"}, "operator"), + "an operator no test has": ({"field": "a", "operator": "regex", "value": "x"}, "operator"), + "a field with a space": ({"field": "a b", "value": "x"}, "not a column or a dotted path"), + "a field with a backtick": ({"field": "a`b", "value": "x"}, "not a column or a dotted path"), + "a field with an empty part": ( + {"field": "a..b", "value": "x"}, + "not a column or a dotted path", + ), + "a field starting with a digit": ( + {"field": "1a", "value": "x"}, + "not a column or a dotted path", + ), + "a field too deep": ({"field": ".".join("a" * 17), "value": "x"}, "at most 16 parts"), + "a test with no field": ({"operator": "equals", "value": "x"}, "tests a field"), + "a test with no value": ({"field": "a", "operator": "equals"}, "value is required"), + "an empty text to look for": ( + {"field": "a", "operator": "includes", "value": ""}, + "matches every value", + ), + "exists with a value": ({"field": "a", "operator": "exists", "value": "x"}, "takes no value"), + "includes with a number": ({"field": "a", "operator": "includes", "value": 5}, "takes text"), + "a comparison against text that is no date": ( + {"field": "a", "operator": "gt", "value": "yesterday"}, + "ISO 8601", + ), + "a comparison against a boolean": ({"field": "a", "operator": "lt", "value": True}, None), + "equals against a list": ( + {"field": "a", "operator": "equals", "value": ["a"]}, + "takes one value", + ), + "in against a scalar": ({"field": "a", "operator": "in", "value": "a"}, "takes a list"), + "in against an empty list": ({"field": "a", "operator": "in", "value": []}, "at least one"), + "in against texts and numbers": ( + {"field": "a", "operator": "in", "value": ["a", 1]}, + "not both", + ), + "in against too many values": ( + {"field": "a", "operator": "in", "value": list(range(MAX_LIST_VALUES + 1))}, + "at most 500", + ), + "a number that is not finite": ( + {"field": "a", "operator": "gt", "value": float("inf")}, + "finite", + ), + "an integer past 64 bits": ( + {"field": "a", "operator": "gt", "value": 2**64}, + "fits in 64 bits", + ), + "a text with a NUL": ({"field": "a", "value": "a\x00b"}, "NUL"), + "a text carrying a runner placeholder": ( + {"field": "a", "value": "x{window}y"}, + "{window}, which the runner fills in", + ), + "a text carrying the retired time placeholder": ( + {"field": "a", "value": "{timestamp_condition}"}, + "{timestamp_condition}", + ), + "a text too long": ({"field": "a", "value": "a" * 1025}, "at most 1024"), + "all and any together": ( + {"all": [{"field": "a", "value": "x"}], "any": [{"field": "b", "value": "y"}]}, + "not both", + ), + "a group that also tests a field": ( + {"all": [{"field": "a", "value": "x"}], "field": "b", "operator": "equals", "value": "y"}, + "do not apply beside all or any", + ), + "an empty group": ({"all": []}, "tests a field"), + "groups past the nesting limit": ( + None, + f"at most {MATCH_NESTING_LIMIT} groups deep", + ), + "more tests than the limit": ( + {"all": [{"field": "a", "value": "x"}] * (MAX_TESTS + 1)}, + f"at most {MAX_TESTS} tests", + ), +} + + +def _nested(depth: int) -> dict: + body: dict = {"field": "a", "value": "x"} + for _ in range(depth): + body = {"all": [body]} + return body + + +@pytest.mark.parametrize(("body", "message"), SHAPE_REFUSED.values(), ids=list(SHAPE_REFUSED)) +def test_a_condition_the_builder_cannot_write_is_refused(body, message): + if body is None: + body = _nested(MATCH_NESTING_LIMIT + 1) + + with pytest.raises(ValidationError) as refused: + RuleCondition.model_validate(body) + + if message is not None: + assert message in str(refused.value) + + +def test_groups_at_the_nesting_limit_are_accepted(): + assert RuleCondition.model_validate(_nested(MATCH_NESTING_LIMIT)).depth() == MATCH_NESTING_LIMIT + + +def test_a_test_with_no_operator_is_an_equals(): + assert RuleCondition.model_validate({"field": "a", "value": "x"}).operator == "equals" + + +def test_every_operator_names_the_kind_of_value_it_takes(): + """The published list is the operators the model accepts, no more and no fewer.""" + schema = RuleCondition.model_json_schema()["$defs"]["RuleCondition"]["properties"]["operator"] + accepted = next(branch["enum"] for branch in schema["anyOf"] if "enum" in branch) + + assert set(accepted) == set(OPERAND_KINDS) + assert schema["x-dfe-operand-kinds"] == OPERAND_KINDS + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + ("2026-01-31", "toDateTime64('2026-01-31 00:00:00.000', 3, 'UTC')"), + ("2026-01-31T09:30:15.123456", "toDateTime64('2026-01-31 09:30:15.123', 3, 'UTC')"), + ("2026-01-31T09:30:00Z", "toDateTime64('2026-01-31 09:30:00.000', 3, 'UTC')"), + ("2026-01-31T09:30:00-05:00", "toDateTime64('2026-01-31 14:30:00.000', 3, 'UTC')"), + ], +) +def test_an_instant_is_read_as_utc(value, expected): + assert instant_literal(value) == expected + + +@pytest.mark.parametrize("value", ["", "now()", "2026-13-01", "1 OR 1=1", "2026-01-01'); --"]) +def test_an_instant_that_is_not_iso_8601_is_refused(value): + with pytest.raises(ValueError, match="ISO 8601"): + instant_literal(value) + + +# The one compiled case the rule reader changes: it takes a time bound out of a rule's filter. +_TIME_BOUND = "a column against a date-time with an offset is moved to UTC" +READ_BACK = {name: case for name, case in COMPILED.items() if name != _TIME_BOUND} + + +@pytest.mark.parametrize(("body", "sql"), READ_BACK.values(), ids=list(READ_BACK)) +def test_reading_the_compiled_sql_as_a_rule_gives_the_same_where(body, sql): + """What the rule stores is what the condition says, so a save cannot widen it.""" + where = _compile(body) + + parsed = RuleRewriter().parse_user_sql(f"SELECT * FROM `dfe`.`cisco-ios` WHERE {where}") + + assert parsed.where_clause == where + assert (parsed.source_db, parsed.source_table) == ("dfe", "cisco-ios") + + +def test_the_one_case_the_rule_reader_changes_is_a_time_bound(): + where = _compile(COMPILED[_TIME_BOUND][0]) + + parsed = RuleRewriter().parse_user_sql(f"SELECT * FROM `dfe`.`cisco-ios` WHERE {where}") + + assert parsed.where_clause == "" + assert parsed.stripped_time_bounds