use std::collections::HashMap;
use std::path::PathBuf;
use chrono::{DateTime, Utc};
use rsigma_convert::backends::postgres::PostgresBackend;
pub(crate) const DEFAULT_LIMIT: usize = 1000;
#[derive(Debug, Default, Clone)]
pub(crate) struct HuntWindow {
pub since: Option<DateTime<Utc>>,
pub until: Option<DateTime<Utc>>,
}
impl HuntWindow {
pub(crate) fn validate(&self) -> Result<(), HuntQueryError> {
if let (Some(since), Some(until)) = (&self.since, &self.until)
&& since >= until
{
return Err(HuntQueryError::EmptyWindow {
since: since.to_rfc3339(),
until: until.to_rfc3339(),
});
}
Ok(())
}
}
#[derive(Debug)]
pub(crate) struct HuntQuery {
pub rule_title: String,
pub rule_id: Option<String>,
pub sql: String,
}
#[derive(Debug)]
pub(crate) struct HuntPlan {
pub queries: Vec<HuntQuery>,
pub timestamp_field: String,
pub json_field: Option<String>,
#[cfg(feature = "hunt-postgres")]
pub limit: usize,
}
#[derive(Debug)]
pub(crate) enum HuntQueryError {
CorrelationRejected { titles: Vec<String> },
InvalidIdentifier { option: String, value: String },
EmptyWindow { since: String, until: String },
Conversion(String),
RuleFailures(Vec<String>),
NonSelectOutput { rule_title: String },
InvalidTimeBound { value: String },
}
impl HuntQueryError {
pub(crate) fn message(&self) -> String {
match self {
Self::CorrelationRejected { titles } => format!(
"hunt run supports detection rules only; found {} correlation rule(s): {}. \
Correlation queries return aggregate rows, not events. \
Convert with `rsigma backend convert -t postgres` and run the query manually.",
titles.len(),
titles.join(", ")
),
Self::InvalidIdentifier { option, value } => format!(
"invalid {option} '{value}': expected a plain SQL identifier \
(letters, digits, `_`, `$`, not starting with a digit)"
),
Self::EmptyWindow { since, until } => {
format!("empty hunt window: --since {since} is not before --until {until}")
}
Self::Conversion(msg) => format!("conversion failed: {msg}"),
Self::RuleFailures(lines) => format!(
"{} rule(s) failed to convert:\n {}",
lines.len(),
lines.join("\n ")
),
Self::NonSelectOutput { rule_title } => format!(
"rule '{rule_title}' did not produce a plain SELECT; hunts require the \
backend's default format (a pipeline query_expression_template or custom \
attribute replaced it)"
),
Self::InvalidTimeBound { value } => format!(
"invalid time bound '{value}': expected an RFC 3339 instant or a duration \
like 30m, 12h, 7d"
),
}
}
}
pub(crate) fn parse_time_bound(
value: &str,
now: DateTime<Utc>,
) -> Result<DateTime<Utc>, HuntQueryError> {
if let Ok(ts) = DateTime::parse_from_rfc3339(value) {
return Ok(ts.with_timezone(&Utc));
}
let duration = humantime::parse_duration(value)
.map_err(|_| HuntQueryError::InvalidTimeBound {
value: value.to_string(),
})
.and_then(|d| {
chrono::Duration::from_std(d).map_err(|_| HuntQueryError::InvalidTimeBound {
value: value.to_string(),
})
})?;
Ok(now - duration)
}
fn validate_identifier(option: &str, value: &str) -> Result<(), HuntQueryError> {
let mut chars = value.chars();
let first_ok = chars
.next()
.is_some_and(|c| c.is_ascii_alphabetic() || c == '_');
let rest_ok = chars.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '$');
if first_ok && rest_ok {
Ok(())
} else {
Err(HuntQueryError::InvalidIdentifier {
option: option.to_string(),
value: value.to_string(),
})
}
}
fn timestamp_literal(ts: &DateTime<Utc>) -> String {
format!("'{}'::timestamptz", ts.to_rfc3339())
}
pub(crate) fn wrap_query(
generated: &str,
timestamp_column: &str,
window: &HuntWindow,
limit: usize,
) -> String {
let inner = generated.trim().trim_end_matches(';');
let mut sql = format!("SELECT * FROM ({inner}) AS __hunt");
let mut predicates = Vec::new();
if let Some(since) = &window.since {
predicates.push(format!(
"{timestamp_column} >= {}",
timestamp_literal(since)
));
}
if let Some(until) = &window.until {
predicates.push(format!("{timestamp_column} < {}", timestamp_literal(until)));
}
if !predicates.is_empty() {
sql.push_str(" WHERE ");
sql.push_str(&predicates.join(" AND "));
}
sql.push_str(&format!(" ORDER BY {timestamp_column}"));
if limit > 0 {
sql.push_str(&format!(" LIMIT {limit}"));
}
sql
}
pub(crate) fn build_hunt_plan(
rule_paths: &[PathBuf],
pipeline_paths: &[PathBuf],
options: &HashMap<String, String>,
window: &HuntWindow,
limit: usize,
) -> Result<HuntPlan, HuntQueryError> {
window.validate()?;
let mut collection = crate::load_collection_multi(rule_paths);
if !collection.correlations.is_empty() {
let titles = collection
.correlations
.iter()
.map(|c| c.title.clone())
.collect();
return Err(HuntQueryError::CorrelationRejected { titles });
}
for rule in &mut collection.rules {
rule.fields.clear();
}
let pipelines = crate::load_pipelines(pipeline_paths);
let backend = PostgresBackend::from_options(options);
validate_identifier("timestamp_field", &backend.timestamp_field)?;
if let Some(json_field) = &backend.json_field {
validate_identifier("json_field", json_field)?;
}
let output = rsigma_convert::convert_collection(&backend, &collection, &pipelines, "default")
.map_err(|e| HuntQueryError::Conversion(e.to_string()))?;
if !output.errors.is_empty() {
let lines = output
.errors
.iter()
.map(|(title, err)| format!("{title}: {err}"))
.collect();
return Err(HuntQueryError::RuleFailures(lines));
}
for (rule_title, warning) in output.warnings() {
eprintln!("Warning: rule '{rule_title}': {warning}");
}
let timestamp_column = rsigma_convert::backend::text_escape_and_quote_field(
backend.config,
&backend.timestamp_field,
);
let mut queries = Vec::new();
for result in &output.queries {
for query in &result.queries {
if !query
.trim_start()
.to_ascii_uppercase()
.starts_with("SELECT")
{
return Err(HuntQueryError::NonSelectOutput {
rule_title: result.rule_title.clone(),
});
}
queries.push(HuntQuery {
rule_title: result.rule_title.clone(),
rule_id: result.rule_id.clone(),
sql: wrap_query(query, ×tamp_column, window, limit),
});
}
}
Ok(HuntPlan {
queries,
timestamp_field: backend.timestamp_field.clone(),
json_field: backend.json_field.clone(),
#[cfg(feature = "hunt-postgres")]
limit,
})
}
#[cfg(test)]
mod tests {
use super::*;
fn fixed_now() -> DateTime<Utc> {
DateTime::parse_from_rfc3339("2026-07-09T12:00:00Z")
.unwrap()
.with_timezone(&Utc)
}
fn window(since: Option<&str>, until: Option<&str>) -> HuntWindow {
HuntWindow {
since: since.map(|s| DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc)),
until: until.map(|s| DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc)),
}
}
#[test]
fn parse_rfc3339_instant() {
let ts = parse_time_bound("2026-07-01T00:00:00Z", fixed_now()).unwrap();
assert_eq!(ts.to_rfc3339(), "2026-07-01T00:00:00+00:00");
}
#[test]
fn parse_relative_duration() {
let ts = parse_time_bound("30d", fixed_now()).unwrap();
assert_eq!(ts.to_rfc3339(), "2026-06-09T12:00:00+00:00");
}
#[test]
fn parse_rejects_garbage() {
let err = parse_time_bound("next tuesday", fixed_now()).unwrap_err();
assert!(err.message().contains("invalid time bound 'next tuesday'"));
}
#[test]
fn window_rejects_empty_range() {
let w = window(Some("2026-07-02T00:00:00Z"), Some("2026-07-01T00:00:00Z"));
let err = w.validate().unwrap_err();
assert!(err.message().contains("empty hunt window"));
}
#[test]
fn window_accepts_touching_bounds_excluded_by_check() {
let w = window(Some("2026-07-01T00:00:00Z"), Some("2026-07-02T00:00:00Z"));
w.validate().unwrap();
}
#[test]
fn identifier_validation() {
assert!(validate_identifier("timestamp_field", "time").is_ok());
assert!(validate_identifier("timestamp_field", "_ts1$").is_ok());
assert!(validate_identifier("timestamp_field", "1time").is_err());
assert!(validate_identifier("timestamp_field", "time; DROP TABLE").is_err());
assert!(validate_identifier("timestamp_field", "").is_err());
let err = validate_identifier("json_field", "data col").unwrap_err();
assert!(err.message().contains("invalid json_field 'data col'"));
}
#[test]
fn wrap_since_only() {
let sql = wrap_query(
"SELECT * FROM security_events WHERE id = 1",
"time",
&window(Some("2026-07-01T00:00:00Z"), None),
1000,
);
assert_eq!(
sql,
"SELECT * FROM (SELECT * FROM security_events WHERE id = 1) AS __hunt \
WHERE time >= '2026-07-01T00:00:00+00:00'::timestamptz ORDER BY time LIMIT 1000"
);
}
#[test]
fn wrap_until_only() {
let sql = wrap_query(
"SELECT * FROM security_events WHERE id = 1",
"time",
&window(None, Some("2026-07-02T00:00:00Z")),
1000,
);
assert_eq!(
sql,
"SELECT * FROM (SELECT * FROM security_events WHERE id = 1) AS __hunt \
WHERE time < '2026-07-02T00:00:00+00:00'::timestamptz ORDER BY time LIMIT 1000"
);
}
#[test]
fn wrap_both_bounds() {
let sql = wrap_query(
"SELECT * FROM security_events WHERE id = 1",
"time",
&window(Some("2026-07-01T00:00:00Z"), Some("2026-07-02T00:00:00Z")),
500,
);
assert_eq!(
sql,
"SELECT * FROM (SELECT * FROM security_events WHERE id = 1) AS __hunt \
WHERE time >= '2026-07-01T00:00:00+00:00'::timestamptz \
AND time < '2026-07-02T00:00:00+00:00'::timestamptz ORDER BY time LIMIT 500"
);
}
#[test]
fn wrap_no_bounds_no_limit() {
let sql = wrap_query(
"SELECT * FROM security_events WHERE id = 1;",
"time",
&HuntWindow::default(),
0,
);
assert_eq!(
sql,
"SELECT * FROM (SELECT * FROM security_events WHERE id = 1) AS __hunt ORDER BY time"
);
}
#[test]
fn wrap_custom_timestamp_field() {
let sql = wrap_query(
"SELECT * FROM security_events WHERE id = 1",
"event_time",
&window(Some("2026-07-01T00:00:00Z"), None),
10,
);
assert!(sql.contains("WHERE event_time >="));
assert!(sql.contains("ORDER BY event_time"));
}
fn golden_path(name: &str) -> PathBuf {
std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
.join("tests/golden")
.join(name)
}
fn check_golden(name: &str, actual: &str) {
let path = golden_path(name);
if std::env::var_os("RSIGMA_UPDATE_GOLDEN").is_some() {
std::fs::write(&path, actual)
.unwrap_or_else(|e| panic!("failed to write {}: {e}", path.display()));
return;
}
let expected = std::fs::read_to_string(&path)
.unwrap_or_else(|e| panic!("failed to read {}: {e}", path.display()))
.replace("\r\n", "\n");
assert_eq!(actual, expected, "hunt golden drifted for '{name}'");
}
const FLAT_RULE: &str = r#"
title: Suspicious Process Start
id: 00000000-0000-0000-0000-000000000101
logsource:
category: process_creation
detection:
selection:
Image: /usr/bin/curl
CommandLine|contains: "--insecure"
condition: selection
level: medium
"#;
const FIELDS_RULE: &str = r#"
title: Field Listed Rule
id: 00000000-0000-0000-0000-000000000102
logsource:
category: process_creation
detection:
selection:
Image: /bin/sh
condition: selection
fields:
- Image
- CommandLine
level: low
"#;
fn plan_for(rule_yaml: &str, options: &HashMap<String, String>) -> HuntPlan {
let mut f = tempfile::NamedTempFile::new().unwrap();
std::io::Write::write_all(&mut f, rule_yaml.as_bytes()).unwrap();
let window = window(Some("2026-07-01T00:00:00Z"), Some("2026-07-02T00:00:00Z"));
build_hunt_plan(
std::slice::from_ref(&f.path().to_path_buf()),
&[],
options,
&window,
1000,
)
.unwrap()
}
#[test]
fn golden_flat_mode_sql() {
let plan = plan_for(FLAT_RULE, &HashMap::new());
assert_eq!(plan.timestamp_field, "time");
assert_eq!(plan.json_field, None);
let actual = format!("{};\n", plan.queries[0].sql);
check_golden("hunt_sql_flat.sql", &actual);
}
#[test]
fn golden_jsonb_mode_sql() {
let options: HashMap<String, String> = [
("table".to_string(), "events".to_string()),
("json_field".to_string(), "data".to_string()),
]
.into_iter()
.collect();
let plan = plan_for(FLAT_RULE, &options);
assert_eq!(plan.json_field.as_deref(), Some("data"));
let actual = format!("{};\n", plan.queries[0].sql);
check_golden("hunt_sql_jsonb.sql", &actual);
}
#[test]
fn golden_fields_list_cleared() {
let plan = plan_for(FIELDS_RULE, &HashMap::new());
let actual = format!("{};\n", plan.queries[0].sql);
check_golden("hunt_sql_fields_cleared.sql", &actual);
}
#[test]
fn correlation_rules_are_rejected() {
let yaml = format!(
"{FLAT_RULE}\n---\ntitle: Repeated Hits\ncorrelation:\n type: event_count\n rules:\n - Suspicious Process Start\n group-by:\n - Image\n timespan: 10m\n condition:\n gte: 5\n"
);
let mut f = tempfile::NamedTempFile::new().unwrap();
std::io::Write::write_all(&mut f, yaml.as_bytes()).unwrap();
let err = build_hunt_plan(
&[f.path().to_path_buf()],
&[],
&HashMap::new(),
&HuntWindow::default(),
1000,
)
.unwrap_err();
let msg = err.message();
assert!(msg.contains("detection rules only"), "{msg}");
assert!(msg.contains("Repeated Hits"), "{msg}");
assert!(msg.contains("rsigma backend convert"), "{msg}");
}
}