Dynamic Pipeline Sources🔗

Dynamic pipeline sources let a processing pipeline pull values from external systems (files, local commands, HTTP, NATS) at load time and inject them through the standard Sigma vars + value_placeholders mechanism. This page documents the full source specification, every source type, the four data formats, the three extract languages, the five refresh policies, the three error policies, and every resource limit the runtime enforces.

For an introduction to the feature see Processing Pipelines: dynamic pipelines. For end-to-end testing see pipeline resolve. For runtime metrics see Prometheus metrics: dynamic pipeline sources.

Source declaration🔗

Dynamic sources are declared in standalone YAML files and loaded into the daemon with the repeatable --source flag. Each file has a top-level sources: block; every entry is a YAML mapping with the schema documented below.

# sources.yml
sources:
  - id: employee_directory
    type: file
    path: ./data/employees.json
    format: json
  - id: kev_catalog
    type: http
    url: https://www.cisa.gov/sites/default/files/feeds/known_exploited_vulnerabilities.json
    format: json
    extract: ".vulnerabilities"
    refresh: 1h

Pass it to the daemon:

rsigma engine daemon -r rules/ -p pipeline.yml --source sources.yml

A directory path loads all *.yml/*.yaml files in it, sorted alphabetically:

rsigma engine daemon -r rules/ -p pipeline.yml --source sources.d/

The flag is repeatable, so you can load from multiple files and directories:

rsigma engine daemon -r rules/ -p pipeline.yml \
    --source infra-sources.yml \
    --source threat-intel-sources.yml

Schema🔗

Every entry in the sources: list takes the same shape:

sources:
  - id: <source-id>           # required, used in ${source.<id>} refs
    type: <file|http|command|nats>
    # type-specific fields…
    format: <json|yaml|lines|csv>
    extract: <expression>     # optional
    refresh: <once|<duration>|watch|push|on_demand>
    required: <true|false>    # default true
    timeout: <duration>       # default 30s for http/command
    on_error: <use_cached|fail|use_default>
    default: <value>          # required if on_error=use_default
    max_body_size: <bytes>    # default 10485760 (10 MiB)
    max_stdout: <bytes>       # command type only

The full Rust type lives at rsigma_eval::pipeline::sources::DynamicSource. The parser is at rsigma_eval::pipeline::parsing.

Collision semantics🔗

Source IDs must be unique across every --source file (and across every directory the flag expands into). If the same ID appears in two different declaration sites the daemon exits at startup with an error naming both file paths, so there is exactly one canonical declaration site per source ID.

Source types🔗

file🔗

Reads a local file, parses it according to format, applies extract if set, and returns the result.

- id: field_config
  type: file
  path: /etc/rsigma/fields.json
  format: json
  refresh: watch

refresh: watch is only valid for file sources (uses notify). For other refresh policies, file behaves like the others.

http🔗

GET (or other method) request, response body parsed and optionally extracted. Uses reqwest.

- id: ip_blocklist
  type: http
  url: https://feeds.example.com/blocklist.json
  format: json
  extract: ".ips"
  method: GET                  # default
  headers:                     # optional
    Authorization: "Bearer ${env:FEED_TOKEN}"
  timeout: 10s                 # default 30s
  refresh: 300s
  on_error: use_cached

command🔗

Runs a local executable, captures stdout, parses it according to format. Useful for shelling out to an inventory tool, a script that queries an internal API with credentials only the host has access to, or a generator that produces transformation YAML on demand.

- id: enrichment_rules
  type: command
  command: ["/usr/local/bin/generate-transformations", "--format", "json"]
  format: json
  refresh: once
  timeout: 5s

The runtime additionally caps stderr at 64 KiB regardless of max_stdout. Stderr is logged on failure but not parsed.

nats🔗

Subscribes to a NATS subject and updates the source value with each message. Requires the daemon-nats build feature.

- id: live_iocs
  type: nats
  url: nats://nats.internal:4222
  subject: rsigma.iocs.current
  format: json
  refresh: push
  required: false

refresh: push is only valid for NATS sources. Each subject message replaces the source value.

Data formats🔗

Extract languages🔗

The optional extract: filter slices the parsed data after format parsing. Three languages are supported:

Plain-string extract: defaults to jq. Use the object form for explicit selection:

# Shorthand: jq
extract: ".indicators[].ip"

# Explicit jq
extract:
  type: jq
  expr: ".indicators[].ip"

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

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

Refresh policies🔗

refresh: controls how often the source re-fetches.

A <duration> refresh below MIN_REFRESH_INTERVAL (1 second) clamps silently with a runtime warning. Operators wishing to refresh more aggressively than that should use NATS push, on-demand triggers, or rethink the architecture.

Error policies🔗

on_error: controls what happens when a fetch fails (network down, command exits non-zero, parse error, extract returns empty):

The required flag interacts with on_error:

• required: true + on_error: fail -> startup fails; the daemon exits.
• required: true + on_error: use_cached -> startup succeeds if a cached value exists from a prior run (with --state-db); fails otherwise.
• required: false + on_error: fail -> source resolves to null; pipeline continues.

Template substitution🔗

The ${source.<id>} syntax expands ONLY in the vars: block. The expander does NOT substitute references inside typed transformation fields (e.g. add_condition.conditions.X). The supported pattern is to put the resolved value into a vars: entry and reference it from rules via the standard Sigma %name% placeholder, expanded by the value_placeholders transformation:

# sources.yml -- loaded via `--source sources.yml`
sources:
  - id: ip_blocklist
    type: http
    url: …
    extract: ".ips"

# pipeline.yml -- loaded via `-p pipeline.yml`
vars:
  blocklist: "${source.ip_blocklist}"

transformations:
  - type: value_placeholders

# rule
detection:
    selection:
        DestinationIp: '%blocklist%'
    condition: selection

Dot-path indexing into a nested structure works in vars::

vars:
  admin_emails: "${source.env_config.admin_emails}"
  log_index:    "${source.env_config.log_index}"

Inline templates work too (${source.X} as part of a larger string), but they substitute the source's stringified representation, which is rarely what you want for array sources. Whole-value substitution (where ${source.X} is the entire vars: entry) is the safe form: it expands an array source to multiple vars entries that value_placeholders can map onto rule values. For scalar sources, inline templates compose cleanly:

vars:
  greeting: "Hello, ${source.env_config.org_name}!"

Include directives🔗

A source resolving to a JSON array of transformation objects can be inlined via include::

transformations:
  - include: ${source.dynamic_transforms}

Constraints:

• The resolved value must be a JSON array of transformation objects, not a single object.
• Nested includes are rejected (MAX_INCLUDE_DEPTH = 1). If an included fragment itself contains include: directives, expansion fails at startup with a clear error message.
• Remote sources (HTTP, NATS) require --allow-remote-include on the daemon. The default policy restricts include resolution to local sources (file, command) to limit the blast radius of a compromised CDN or NATS broker.

Triggers and hot-reload🔗

The push policy (NATS) updates the source value continuously on each incoming message, without going through the reload pipeline.

Resource limits🔗

Every dynamic source path enforces hard limits to bound resource consumption. Per-source overrides are noted in the table.

Exceeding any limit produces a SourceErrorKind::ResourceLimit failure with a descriptive message. See Security Hardening for the broader catalogue.

See also🔗

• Processing Pipelines: dynamic pipelines for the narrative version.
• pipeline resolve for offline source testing.
• rule validate --resolve-sources for the strict CI gate.
• Prometheus metrics: dynamic pipeline sources for what every successful and failing resolve exposes.
• HTTP API: sources for the daemon control endpoints.
• Security Hardening for every other resource limit the runtime enforces.
• rsigma_runtime::sources source for the implementation.