rsigma_eval/pipeline/sources.rs
1//! Dynamic source declarations and template references for dynamic Sigma pipelines.
2//!
3//! This module defines the types for declaring external data sources in a pipeline
4//! YAML (`sources` section) and for tracking `${source.*}` template references
5//! found throughout the pipeline.
6
7use std::collections::HashMap;
8use std::path::PathBuf;
9use std::time::Duration;
10
11// =============================================================================
12// Dynamic source declaration
13// =============================================================================
14
15/// A dynamic source declared in the pipeline's `sources` section.
16///
17/// Each source describes how to fetch external data that can be referenced
18/// by `${source.<id>}` expressions anywhere in the pipeline YAML.
19#[derive(Debug, Clone, PartialEq)]
20pub struct DynamicSource {
21 /// Unique identifier for this source, referenced in `${source.<id>}` expressions.
22 pub id: String,
23 /// The type-specific configuration for fetching data.
24 pub source_type: SourceType,
25 /// How often the source data should be refreshed.
26 pub refresh: RefreshPolicy,
27 /// Maximum time to wait for a fetch to complete.
28 pub timeout: Option<Duration>,
29 /// What to do when a fetch fails.
30 pub on_error: ErrorPolicy,
31 /// Whether the daemon must resolve this source before processing events.
32 pub required: bool,
33 /// Fallback value if the source cannot be resolved.
34 pub default: Option<yaml_serde::Value>,
35}
36
37/// Type-specific configuration for a dynamic source.
38#[derive(Debug, Clone, PartialEq)]
39pub enum SourceType {
40 /// Fetch data from an HTTP endpoint.
41 Http {
42 url: String,
43 method: Option<String>,
44 headers: HashMap<String, String>,
45 /// Optional request body sent verbatim after `${VAR}` expansion. When
46 /// set and `method` is unset, the request defaults to `POST`. Pairs
47 /// with query APIs that require a body (GraphQL, `_search`, TheHive 5's
48 /// `/api/v1/query`).
49 body: Option<String>,
50 format: DataFormat,
51 extract: Option<ExtractExpr>,
52 },
53 /// Run a local command and capture its stdout.
54 Command {
55 command: Vec<String>,
56 format: DataFormat,
57 extract: Option<ExtractExpr>,
58 },
59 /// Read data from a local file.
60 File {
61 path: PathBuf,
62 format: DataFormat,
63 extract: Option<ExtractExpr>,
64 },
65 /// Subscribe to a NATS subject for push-based updates.
66 Nats {
67 url: String,
68 subject: String,
69 format: DataFormat,
70 extract: Option<ExtractExpr>,
71 },
72}
73
74/// An extraction expression applied to source data after parsing.
75///
76/// Supports two syntax forms in YAML:
77/// - Plain string: always jq (the common case): `extract: ".emails[]"`
78/// - Structured object: explicit language: `extract: { expr: "$.emails[*]", type: jsonpath }`
79#[derive(Debug, Clone, PartialEq)]
80pub enum ExtractExpr {
81 /// A jq expression (default). Evaluated via jaq.
82 Jq(String),
83 /// A JSONPath expression. Evaluated via serde_json_path.
84 JsonPath(String),
85 /// A CEL (Common Expression Language) expression. Evaluated via cel-interpreter.
86 Cel(String),
87}
88
89/// How often a source should be refreshed.
90#[derive(Debug, Clone, PartialEq)]
91pub enum RefreshPolicy {
92 /// Fetch at startup only, never refresh.
93 Once,
94 /// Re-fetch on a fixed interval.
95 Interval(Duration),
96 /// Watch the file for changes (file sources only).
97 Watch,
98 /// Value updated on each incoming NATS message (NATS sources only).
99 Push,
100 /// Fetch at startup, then only when explicitly triggered via API/signal.
101 OnDemand,
102}
103
104/// What to do when a source fetch fails.
105#[derive(Debug, Clone, Copy, PartialEq, Eq)]
106pub enum ErrorPolicy {
107 /// Use the last successfully fetched value.
108 UseCached,
109 /// Fail the pipeline load (at startup: exit; at runtime: keep previous state).
110 Fail,
111 /// Fall back to the declared `default` value.
112 UseDefault,
113}
114
115/// The format of data returned by a source.
116#[derive(Debug, Clone, Copy, PartialEq, Eq)]
117pub enum DataFormat {
118 /// JSON (parsed with serde_json).
119 Json,
120 /// YAML (parsed with yaml_serde).
121 Yaml,
122 /// One value per line.
123 Lines,
124 /// Comma-separated values.
125 Csv,
126}
127
128// =============================================================================
129// Source references (detected during parsing)
130// =============================================================================
131
132/// A `${source.*}` template reference found in the pipeline YAML.
133#[derive(Debug, Clone, PartialEq)]
134pub struct SourceRef {
135 /// The source ID (first path segment after `source.`).
136 pub source_id: String,
137 /// Optional dot-path into the source data (e.g., `field_mapping` in `${source.env_config.field_mapping}`).
138 pub sub_path: Option<String>,
139 /// Where in the pipeline this reference appears.
140 pub location: RefLocation,
141 /// The raw template string as it appeared in the YAML.
142 pub raw_template: String,
143}
144
145/// Where in the pipeline a source reference appears.
146#[derive(Debug, Clone, PartialEq)]
147pub enum RefLocation {
148 /// In the `vars` section, under the given variable name.
149 Var { var_name: String },
150 /// In a transformation's field value.
151 TransformationField {
152 transform_index: usize,
153 field_name: String,
154 },
155 /// An `include` directive in the transformations list.
156 Include { transform_index: usize },
157}
158
159// =============================================================================
160// Source status (for PipelineState tracking)
161// =============================================================================
162
163/// Resolution status of a dynamic source.
164#[derive(Debug, Clone, Copy, PartialEq, Eq)]
165pub enum SourceStatus {
166 /// Source has not been resolved yet.
167 Pending,
168 /// Source was successfully resolved.
169 Resolved,
170 /// Source resolution failed.
171 Failed,
172}