Notification & automation flows (n8n/Node-RED style: trigger → filter → action) #24

New issue

Open

opened 2026-06-30 13:51:09 +02:00 by jonas · 0 comments

jonas commented

2026-06-30 13:51:09 +02:00

Owner

Goal

A notification / automation flow builder, conceptually like n8n / Node-RED: users compose triggers → filters (AND/OR) → mapping → actions, and bind them to events in SPAM. v1 actions: Slack / in-web notification / webhook. The model must be extensible so later we can add richer action nodes (base64 decode, small Python scripts, linting/scanning steps, etc.).

Filters must be bindable to clusters, images, and repos (the same scopes SPAM already filters on).

What already exists (foundation)

Event backbone: Postgres NOTIFY/LISTEN pub/sub on channel spam_events (api/internal/events/notify.go), already feeding SSE streams (/api/app/stream, /api/runs/...). This is the natural trigger source.
In-web notifications page: web/src/routes/(app)/notifications/+page.svelte — currently placeholders only (hardcoded alerts, "Delivery settings" stubs for push/email/webhook). No real delivery backend.
Job system: internal/jobs (QUEUED→RUNNING→SUCCEEDED/FAILED/RETRY, SELECT … FOR UPDATE SKIP LOCKED) — the right executor for action runs (esp. future code-running nodes that need sandboxing like internal/runner).
ACL scopes: acl.Provider already filters repos / images / clusters — reuse for "this rule applies to assets in scope X".
No workflow/condition engine exists today — this is greenfield.

Proposed architecture

Data model (new `internal/automation` package)

flow — {id, name, enabled, owner_user_id, created_at}
flow_trigger — {flow_id, event_type} where event_type ∈ event taxonomy below
flow_filter — a tree (group node with AND/OR + leaf predicates). Leaf predicate: {field, op, value} e.g. severity >= HIGH, kev == true, cluster.slug in [...], image.repository matches …, repo.id in [...], internet_exposed == true, tier == fix_now. Store as JSONB for flexibility.
flow_action — ordered list: {type, config}. v1 types: slack, web_notification, webhook. config is per-type JSONB (Slack channel/webhook URL + template; webhook URL + headers + HMAC secret via secrets AES-GCM).
flow_run — execution record per fired event (for observability/retry), tied to a job.
web_notification — delivered in-app notifications (replaces the placeholder data) {user_id/scope, title, body, severity, read_at, source_flow_id, deep_link}.

Event taxonomy (triggers)

Define a stable internal event vocabulary emitted via events.NotifyEvent():

finding.new, finding.escalated (e.g. CVE → KEV / now internet-exposed / tier→fix_now)
asset.tier_changed, secret.detected
scan.completed, scan.failed
suppression.expired (ties into triage epic)
cluster.offline / cluster.online

Events must carry enough context (asset_type/id, cluster, severity, vuln_id, tier) for filters to evaluate without extra DB joins where possible.

Execution

Listener consumes spam_events → matches enabled flows whose trigger == event_type.
Evaluate filter tree against the event payload (+ ACL scope intersection).
For matches, enqueue a FLOW_ACTION job per action (retry/backoff for free via job framework).
Action workers deliver (Slack/webhook HTTP, or insert web_notification).

Extensibility seam

Action is an interface (Action.Execute(ctx, event, config) error) with a registry. v1 registers slack/web/webhook. Later nodes (base64 transform, Python script, lint/scan) register the same way; untrusted code-running nodes must run in the sandboxed K8s runner (internal/runner), never in-process. Mapping/transform nodes can chain by passing transformed payload to the next node.

Edge cases / decisions

Multi-replica fan-out. Only one replica should fire each event's actions — use job-queue claim (SKIP LOCKED) so the listener enqueues and any worker executes exactly once. No in-memory subscriptions.
Loops / runaway. A flow whose action causes an event that re-triggers it → cap depth / dedup per (flow, event) within a window.
Delivery reliability. Webhooks: retries, exponential backoff, dead-letter after N; per-flow failure surfaced in flow_run.
Secret handling. Slack tokens / webhook secrets encrypted (secrets pkg), never returned in plaintext via API.
Noise control. Debounce / digest (batch many findings into one Slack message); rate caps per flow.
Authorization. A flow can only act on events within its owner's ACL scope; webhook destinations may need an allowlist to prevent SSRF/exfil.
Filter on derived state. Some filters (tier, internet_exposed) come from asset_risk — ensure the event payload or a cheap lookup carries them at fire time.

Tasks

Define event taxonomy + enrich events.NotifyEvent emissions across scan/triage/secret/cluster paths
Models: flow / trigger / filter-tree / action / flow_run / web_notification (+ migration)
Filter-tree evaluator (AND/OR + leaf ops) with cluster/image/repo predicates + ACL intersection
Action interface + registry; v1 slack, web_notification, webhook with retry/DLQ via jobs
Wire real in-app notifications page to web_notification (replace placeholders)
Flow builder UI (visual graph optional in v1; form-based trigger/filter/action acceptable first)
Admin: webhook destination allowlist, secret storage, per-flow rate limits
(Later) transform/code-running action nodes via sandboxed runner

Relates to: triage epic (suppression.expired, finding.escalated events), public scan API (scan.completed).

## Goal A **notification / automation flow** builder, conceptually like n8n / Node-RED: users compose **triggers → filters (AND/OR) → mapping → actions**, and bind them to events in SPAM. v1 actions: **Slack / in-web notification / webhook**. The model must be extensible so later we can add richer action nodes (base64 decode, small Python scripts, linting/scanning steps, etc.). Filters must be **bindable to clusters, images, and repos** (the same scopes SPAM already filters on). ## What already exists (foundation) - **Event backbone:** Postgres `NOTIFY/LISTEN` pub/sub on channel `spam_events` (`api/internal/events/notify.go`), already feeding SSE streams (`/api/app/stream`, `/api/runs/...`). This is the natural **trigger source**. - **In-web notifications page:** `web/src/routes/(app)/notifications/+page.svelte` — currently **placeholders only** (hardcoded alerts, "Delivery settings" stubs for push/email/webhook). No real delivery backend. - **Job system:** `internal/jobs` (QUEUED→RUNNING→SUCCEEDED/FAILED/RETRY, `SELECT … FOR UPDATE SKIP LOCKED`) — the right executor for action runs (esp. future code-running nodes that need sandboxing like `internal/runner`). - **ACL scopes:** `acl.Provider` already filters repos / images / clusters — reuse for "this rule applies to assets in scope X". - **No workflow/condition engine exists today** — this is greenfield. ## Proposed architecture ### Data model (new `internal/automation` package) - `flow` — `{id, name, enabled, owner_user_id, created_at}` - `flow_trigger` — `{flow_id, event_type}` where `event_type` ∈ event taxonomy below - `flow_filter` — a tree (group node with `AND`/`OR` + leaf predicates). Leaf predicate: `{field, op, value}` e.g. `severity >= HIGH`, `kev == true`, `cluster.slug in [...]`, `image.repository matches …`, `repo.id in [...]`, `internet_exposed == true`, `tier == fix_now`. Store as JSONB for flexibility. - `flow_action` — ordered list: `{type, config}`. v1 types: `slack`, `web_notification`, `webhook`. `config` is per-type JSONB (Slack channel/webhook URL + template; webhook URL + headers + HMAC secret via `secrets` AES-GCM). - `flow_run` — execution record per fired event (for observability/retry), tied to a job. - `web_notification` — delivered in-app notifications (replaces the placeholder data) `{user_id/scope, title, body, severity, read_at, source_flow_id, deep_link}`. ### Event taxonomy (triggers) Define a stable internal event vocabulary emitted via `events.NotifyEvent()`: - `finding.new`, `finding.escalated` (e.g. CVE → KEV / now internet-exposed / tier→fix_now) - `asset.tier_changed`, `secret.detected` - `scan.completed`, `scan.failed` - `suppression.expired` (ties into triage epic) - `cluster.offline` / `cluster.online` Events must carry enough context (asset_type/id, cluster, severity, vuln_id, tier) for filters to evaluate **without extra DB joins** where possible. ### Execution 1. Listener consumes `spam_events` → matches enabled flows whose trigger == event_type. 2. Evaluate filter tree against the event payload (+ ACL scope intersection). 3. For matches, enqueue a `FLOW_ACTION` job per action (retry/backoff for free via job framework). 4. Action workers deliver (Slack/webhook HTTP, or insert `web_notification`). ### Extensibility seam Action is an interface (`Action.Execute(ctx, event, config) error`) with a registry. v1 registers `slack`/`web`/`webhook`. Later nodes (base64 transform, Python script, lint/scan) register the same way; **untrusted code-running nodes must run in the sandboxed K8s runner** (`internal/runner`), never in-process. Mapping/transform nodes can chain by passing transformed payload to the next node. ## Edge cases / decisions 1. **Multi-replica fan-out.** Only one replica should fire each event's actions — use job-queue claim (`SKIP LOCKED`) so the listener enqueues and any worker executes exactly once. No in-memory subscriptions. 2. **Loops / runaway.** A flow whose action causes an event that re-triggers it → cap depth / dedup per (flow, event) within a window. 3. **Delivery reliability.** Webhooks: retries, exponential backoff, dead-letter after N; per-flow failure surfaced in `flow_run`. 4. **Secret handling.** Slack tokens / webhook secrets encrypted (`secrets` pkg), never returned in plaintext via API. 5. **Noise control.** Debounce / digest (batch many findings into one Slack message); rate caps per flow. 6. **Authorization.** A flow can only act on events within its owner's ACL scope; webhook destinations may need an allowlist to prevent SSRF/exfil. 7. **Filter on derived state.** Some filters (tier, internet_exposed) come from `asset_risk` — ensure the event payload or a cheap lookup carries them at fire time. ## Tasks - [ ] Define event taxonomy + enrich `events.NotifyEvent` emissions across scan/triage/secret/cluster paths - [ ] Models: flow / trigger / filter-tree / action / flow_run / web_notification (+ migration) - [ ] Filter-tree evaluator (AND/OR + leaf ops) with cluster/image/repo predicates + ACL intersection - [ ] Action interface + registry; v1 `slack`, `web_notification`, `webhook` with retry/DLQ via jobs - [ ] Wire real in-app notifications page to `web_notification` (replace placeholders) - [ ] Flow builder UI (visual graph optional in v1; form-based trigger/filter/action acceptable first) - [ ] Admin: webhook destination allowlist, secret storage, per-flow rate limits - [ ] (Later) transform/code-running action nodes via sandboxed runner **Relates to:** triage epic (`suppression.expired`, `finding.escalated` events), public scan API (`scan.completed`).

jonas added the

labels

2026-06-30 13:51:09 +02:00

jonas referenced this issue

2026-06-30 15:53:42 +02:00

Dependency maintenance intelligence: detect stale/abandoned libs, per-library threat view, deepsearch harness #28