Processing Pipelines🔗

Processing pipelines are RSigma's mechanism for transforming Sigma rules before they reach the engine or a backend. They handle the impedance mismatch between how Sigma rules name fields (CommandLine, EventID, User) and how your events actually name fields (process.command_line, winlog.event_id, actor.user.name). Pipelines are pySigma-compatible, so anything you write for pySigma works in RSigma, and most pipelines from the SigmaHQ pySigma backends work as well.

This page covers static pipelines (the bread and butter), builtin pipelines (ecs_windows, sysmon), and dynamic pipelines (an RSigma-only capability that pulls values from external sources at runtime).

What pipelines do🔗

A pipeline is a list of transformations applied to each rule in priority order before the rule is compiled or converted. Common transformations:

• Map field names: CommandLine becomes process.command_line.
• Inject conditions: every process_creation rule gets EventID: 1 added.
• Set logsource: rewrite product: windows to product: windows_sysmon.
• Drop rules that target a different schema.
• Set backend-specific options like postgres.table or index.

All of this happens in memory, once, at engine load time (or whenever rules hot-reload). The compiled or converted output looks as if the rules had been written against your schema directly.

A pipeline YAML file looks like this:

name: My ECS Mapping
priority: 20
transformations:
  - id: ecs_process_fields
    type: field_name_mapping
    mapping:
      CommandLine: process.command_line
      Image: process.executable
      ProcessId: process.pid
      ParentImage: process.parent.executable
    rule_conditions:
      - type: logsource
        product: windows
        category: process_creation
  - id: ecs_user_fields
    type: field_name_mapping
    mapping:
      User: user.name
      TargetUserName: user.target.name
    rule_conditions:
      - type: logsource
        product: windows

priority: 20 controls ordering when multiple pipelines are chained (lower numbers run first). rule_conditions gates each transformation so it only runs against matching rules.

Loading pipelines🔗

Pass -p NAME_OR_PATH to any subcommand that accepts pipelines (engine eval, engine daemon, backend convert, rule validate, rule fields, pipeline resolve). The flag is repeatable:

rsigma engine eval -r rules/ -p ecs_windows -e @events.ndjson
rsigma engine daemon -r rules/ -p ecs_windows -p custom-mappings.yml
rsigma backend convert rules/ -t postgres -p pipelines/ocsf_postgres.yml
rsigma rule validate rules/ -p sysmon

The argument is first checked against builtin pipelines. If no match, it is treated as a file path. Multiple pipelines are sorted by priority and applied in sequence.

Builtin pipelines🔗

RSigma embeds two ready-to-use pipelines in the binary, so common cases need no external file:

rsigma engine eval -r rules/ -p ecs_windows -e '{"process.command_line": "whoami"}'
rsigma engine daemon -r rules/ -p sysmon

Builtin pipelines are not file-watched (they are embedded at compile time). Updating them means upgrading RSigma.

See Builtin Pipelines reference for the complete field maps and conditions inside each one.

Transformations: the 26 types🔗

Pipelines compose 26 transformation types. The most common ones in practice are:

The full list with every field is in the rsigma-eval README. All transformations support the same three-tier condition system below.

Conditions: when does a transformation run?🔗

Every transformation can be gated by one or more conditions at three levels:

Rule conditions🔗

Apply at the rule level. Common types:

transformations:
  - id: drop_aws
    type: rule_failure
    message: "AWS rules not supported in this deployment"
    rule_conditions:
      - type: logsource
        product: aws

Detection item conditions🔗

Apply per detection item:

Field name conditions🔗

Filter by field name:

transformations:
  - id: prefix_security_fields
    type: field_name_prefix
    prefix: "security."
    field_name_conditions:
      - type: include_fields
        fields: ["TargetUserName", "SourceIp"]

Chaining pipelines🔗

Multiple pipelines compose by priority. Lower runs first. Each pipeline carries its own internal state (PipelineState), so values set with set_state in one pipeline are visible only to subsequent transformations within the same pipeline. The CLI does not merge pipelines: each remains separate.

rsigma engine eval -r rules/ \
    -p pipelines/01-windows-base.yml \
    -p pipelines/02-ecs-mapping.yml \
    -p pipelines/03-org-overrides.yml

If two pipelines set the same custom attribute on the same rule, the last one wins.

Custom attributes (rsigma.* and postgres.*)🔗

Transformations can write per-rule attributes that the engine and backends read. The most useful:

Use set_custom_attribute to write them:

transformations:
  - id: keep_full_events_for_brute_force
    type: set_custom_attribute
    attribute: rsigma.include_event
    value: "true"
    rule_conditions:
      - type: rule_attribute
        attribute: id
        value: "brute-force-detection"

See Custom Attributes reference for the full list.

Dynamic pipelines🔗

Static pipelines hardcode every value in YAML. Dynamic pipelines let those values come from external sources at runtime: HTTP APIs, local commands, files, or NATS subjects. This is a capability unique to RSigma. Nothing in pySigma or the SigmaHQ ecosystem matches it.

The use cases are concrete:

• A threat-intel feed publishes IOC lists. Reference them inside an add_condition so detection rules update without rule edits.
• A central config service hands out field mappings per environment. Reference them inside field_name_mapping.
• An on-prem catalog publishes which tables hold which event categories. Reference it inside set_state to route rules to the right postgres.table per logsource.
• A bus broadcasts pipeline updates. Subscribe over NATS and re-resolve on push.

Source declaration🔗

Add a sources section to your pipeline YAML (or, preferably, a standalone source file loaded via --source). Each source has a type, a configuration, an extraction expression, and a refresh policy. Substitution into the pipeline is wired through vars:, which the runtime expands with the resolved data; rules then reference the resulting values via standard %placeholder% syntax handled by the value_placeholders transformation:

name: dynamic_threat_intel
priority: 50
sources:
  - id: ip_blocklist
    type: http
    url: https://feeds.example.com/blocklist.json
    format: json
    extract: ".ips"
    refresh: 300s
    timeout: 10s
    on_error: use_cached
    required: true

  - id: enrichment_rules
    type: command
    command: ["generate-transformations", "--format", "json"]
    format: json
    refresh: once

vars:
  blocklist: "${source.ip_blocklist}"

transformations:
  - id: expand_placeholders
    type: value_placeholders

  - include: ${source.enrichment_rules}

In rules, reference the var with the standard Sigma %name% placeholder:

title: Connection to known-bad IP
logsource:
    category: network_connection
detection:
    selection:
        Action: 'allow'
        DestinationIp: '%blocklist%'
    condition: selection

${source.<id>} substitution applies to vars: entries and to include: directives. Transformation field values such as add_condition.conditions.<field> are parsed as typed structures and do not substitute dynamic sources directly; route lists of values through vars plus value_placeholders as shown above. Single scalar substitutions inside transformation fields (such as set_state.value) follow the same pattern through vars.

Source types🔗

Data formats🔗

Extraction languages🔗

After parsing, an optional extract expression selects a subset of the data. Three languages are supported. The plain-string shorthand is jq:

# jq (default for plain strings)
extract: ".indicators[].ip"

# JSONPath
extract:
  type: jsonpath
  expr: "$.indicators[*].ip"

# CEL (Common Expression Language)
extract:
  type: cel
  expr: "data.indicators.filter(i, i.severity > 7).map(i, i.ip)"

See the Dynamic Sources reference for benchmarks and tradeoffs.

Refresh policies🔗

Error handling🔗

Required sources block the daemon's startup until they resolve. Optional sources (required: false) let the daemon start with a null fallback and retry in the background.

Include directives🔗

A whole block of transformations can be injected from a resolved source:

transformations:
  - include: ${source.dynamic_transforms}

The source must resolve to a JSON array of transformation objects. Nested includes are rejected (max depth 1). Remote sources (HTTP/NATS) require --allow-remote-include on the daemon for security.

Testing dynamic sources offline🔗

rsigma pipeline resolve resolves all sources in a pipeline and prints the result without running the engine. Useful for testing config:

rsigma pipeline resolve -p pipelines/dynamic.yml --pretty
rsigma pipeline resolve -p pipelines/dynamic.yml --source threat_intel
rsigma pipeline resolve -p pipelines/dynamic.yml --dry-run

--dry-run lists each source's type, refresh policy, and required flag without performing the actual fetch. Good for catching config typos before they hit production.

rsigma rule validate --resolve-sources -p pipeline.yml extends validation to also exercise source resolution. Sources must be reachable for validation to pass, so this is the right gate to wire into CI for dynamic pipelines.

Hot-reload and dynamic sources🔗

The daemon's hot-reload mechanism extends to dynamic sources. The triggers:

Security model🔗

Dynamic pipelines can run external commands and reach out over HTTP, so the daemon enforces hard limits:

See Security Hardening reference for the full picture.

OCSF pipelines🔗

Two OCSF (Open Cybersecurity Schema Framework) pipelines are included with the rsigma-convert crate and useful as starting points for PostgreSQL-backed deployments:

Use them as templates and copy/customise for your schema. They're typical examples of field_name_mapping plus set_state plus set_custom_attribute working together.

See also🔗

• Builtin Pipelines reference for the contents of ecs_windows and sysmon.
• Dynamic Sources reference for the full source spec, extract language details, and benchmarks.
• Custom Attributes reference for every rsigma.* and postgres.* knob.
• Security Hardening reference for the resource limits enforced on dynamic sources.
• CLI reference: pipeline resolve for the offline source-testing command.
• Library API: rsigma-eval for embedding the pipeline engine in your own code.