mod create;
mod recheck;
pub use create::{RuleCreateOutcome, RuleCreatePlan, apply_create, plan_create};
pub use recheck::observe_rules;
mod observed;
pub use observed::plan_set_enabled_observed;
use crate::codec::{self, Format};
use crate::model::{ListKey, Rule, exception_refs, server_defaults};
use crate::normalize;
use crate::ops::{DeleteOutcome, ExportOutcome, ImportPlan, ImportReport, MutationPlan};
use crate::rules::{self, BulkAction, RuleFilter, RuleSource};
use crate::selection;
use elasticctl_core::{Error, ErrorKind, Result, Transport};
use serde::Serialize;
use serde_json::{Value, json};
use std::path::Path;
const EXCEPTION_POINTER_PLACEHOLDER: &str = "00000000-0000-0000-0000-000000000000";
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct RuleListReport {
pub total: usize,
pub rules: Vec<Rule>,
}
#[derive(Debug, Clone)]
pub struct RulePage {
pub total: u64,
pub rules: Vec<Rule>,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct SetEnabledOutcome {
pub applied: bool,
pub succeeded: u64,
pub failed: u64,
pub skipped: u64,
pub total: u64,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct RuleValidation {
pub rule_id: String,
pub name: String,
#[serde(rename = "type")]
pub rule_type: String,
pub defaults_applied: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct ValidateReport {
pub valid: bool,
pub count: usize,
pub rules: Vec<RuleValidation>,
}
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct PreviewReport {
pub rule: String,
pub preview_id: Option<String>,
pub invocations: u32,
pub hits: Option<u64>,
pub errors: Vec<String>,
pub warnings: Vec<String>,
pub hits_error: Option<String>,
pub sample: Vec<Value>,
}
pub async fn list(t: &Transport, filter: &RuleFilter) -> Result<RuleListReport> {
let rules = rules::find_all(t, filter).await?;
if rules.is_empty() && is_unselected_source_query(filter) {
rules::verify_source_partition(t).await?;
}
let total = rules.len();
Ok(RuleListReport { total, rules })
}
pub async fn list_page(t: &Transport, filter: &RuleFilter, limit: u32) -> Result<RulePage> {
let (rules, total) = rules::find_first_by_name(t, filter, limit).await?;
if rules.is_empty() && is_unselected_source_query(filter) {
rules::verify_source_partition(t).await?;
}
Ok(RulePage { total, rules })
}
fn is_unselected_source_query(filter: &RuleFilter) -> bool {
matches!(filter.source, RuleSource::Custom | RuleSource::Prebuilt)
&& filter.enabled.is_none()
&& filter.rule_type.is_none()
&& filter.severity.is_none()
&& filter.tag.is_none()
&& filter.name.is_none()
&& filter.query.is_none()
&& filter.search.is_none()
&& filter.technique.is_none()
&& filter.tactic.is_none()
}
pub async fn get_one(t: &Transport, selector: &str) -> Result<Rule> {
let (_, rule) = selection::to_rule(t, selector).await?;
Ok(normalize::canonical(&rule))
}
pub fn validate(path: &Path) -> Result<ValidateReport> {
let body = std::fs::read_to_string(path)
.map_err(|e| Error::new(ErrorKind::Error, format!("reading {}: {e}", path.display())))?;
let rules = match Format::from_path(path) {
Format::Yaml => codec::decode_yaml(&body)?,
Format::Ndjson => codec::decode_ndjson(&body)?.0,
};
let defaults = server_defaults();
let mut reports = Vec::with_capacity(rules.len());
let mut failures = Vec::new();
for (i, r) in rules.iter().enumerate() {
match r.rule_id() {
Ok(rule_id) => {
let mut applied: Vec<String> = defaults
.keys()
.filter(|k| !r.as_map().contains_key(*k))
.cloned()
.collect();
applied.sort();
reports.push(RuleValidation {
rule_id: rule_id.to_string(),
name: r.name().to_string(),
rule_type: r.rule_type().to_string(),
defaults_applied: applied,
});
}
Err(e) => failures.push(format!("rule at index {i}: {}", e.message)),
}
}
if !failures.is_empty() {
return Err(Error::new(ErrorKind::Error, failures.join("; ")));
}
Ok(ValidateReport {
valid: true,
count: rules.len(),
rules: reports,
})
}
async fn resolve_targets(t: &Transport, selectors: &[String]) -> Result<Vec<(String, Rule)>> {
let mut out = Vec::with_capacity(selectors.len());
for s in selectors {
out.push(selection::to_rule(t, s).await?);
}
Ok(out)
}
pub async fn plan_set_enabled(
t: &Transport,
selectors: &[String],
enable: bool,
) -> Result<MutationPlan> {
Ok(plan_set_enabled_observed(t, selectors, enable).await?.0)
}
pub async fn apply_set_enabled(
t: &Transport,
plan: &MutationPlan,
enable: bool,
) -> Result<SetEnabledOutcome> {
let action = if enable {
BulkAction::Enable
} else {
BulkAction::Disable
};
let o = rules::bulk_by_rule_ids(t, action, &plan.targets, false).await?;
Ok(SetEnabledOutcome {
applied: true,
succeeded: o.succeeded,
failed: o.failed,
skipped: o.skipped,
total: o.total,
})
}
pub async fn plan_delete(t: &Transport, selectors: &[String]) -> Result<MutationPlan> {
let resolved = resolve_targets(t, selectors).await?;
let preview_details = resolved
.iter()
.map(|(id, r)| format!("{id} {}", r.name()))
.collect();
Ok(MutationPlan {
preview_action: format!("Delete {} rule(s)", resolved.len()),
preview_details,
targets: resolved.into_iter().map(|(id, _)| id).collect(),
})
}
pub async fn apply_delete(t: &Transport, plan: &MutationPlan) -> Result<DeleteOutcome> {
let mut deleted = Vec::new();
let mut failed = Vec::new();
for id in &plan.targets {
match rules::delete(t, id).await {
Ok(_) => deleted.push(json!({"rule_id": id})),
Err(e) => failed.push(json!({"rule_id": id, "error": e.message})),
}
}
Ok(DeleteOutcome {
applied: true,
deleted,
failed,
total: plan.targets.len(),
})
}
pub async fn export_rules(
t: &Transport,
selectors: &[String],
tag: Option<&str>,
source: RuleSource,
format: Format,
) -> Result<ExportOutcome> {
let selection: Option<Vec<String>> =
if selectors.is_empty() && tag.is_none() && source != RuleSource::All {
let scoped = rules::find_all(
t,
&RuleFilter {
source,
..Default::default()
},
)
.await?;
if scoped.is_empty() {
if matches!(source, RuleSource::Custom | RuleSource::Prebuilt) {
rules::verify_source_partition(t).await?;
}
return Ok(ExportOutcome {
body: String::new(),
exported: 0,
missing: Vec::new(),
});
}
Some(
scoped
.iter()
.filter_map(|r| r.rule_id().ok().map(str::to_owned))
.collect(),
)
} else {
selection::resolve(t, selectors, tag, None, &[], "export").await?
};
let mut bundle = rules::export(t, selection.as_deref()).await?;
for r in &mut bundle.rules {
*r = normalize::canonical(r);
}
normalize::sort_rules(&mut bundle.rules);
let body = match format {
Format::Yaml => {
if !bundle.lists.is_empty() || !bundle.items.is_empty() {
return Err(Error::new(
ErrorKind::Unsupported,
format!(
"this export carries {} exception list(s) and {} item(s), which the \
YAML format cannot represent; re-run with --format-file ndjson",
bundle.lists.len(),
bundle.items.len()
),
));
}
codec::encode_yaml(&bundle.rules)?
}
Format::Ndjson => codec::encode_bundle(&bundle)?,
};
let missing = bundle
.summary
.as_ref()
.map(|s| s.missing_rules.clone())
.unwrap_or_default();
Ok(ExportOutcome {
body,
exported: bundle.rules.len() as u64,
missing,
})
}
pub async fn plan_import(
t: Option<&Transport>,
path: &Path,
overwrite: bool,
skip_existing: bool,
) -> Result<ImportPlan> {
let body = std::fs::read_to_string(path)
.map_err(|e| Error::new(ErrorKind::Error, format!("reading {}: {e}", path.display())))?;
let format = Format::from_path(path);
let mut bundle = match format {
Format::Yaml => codec::Bundle {
rules: codec::decode_yaml(&body)?,
..Default::default()
},
Format::Ndjson => codec::decode_bundle(&body)?,
};
let total = bundle.rules.len();
let mut skipped: Vec<Value> = Vec::new();
if skip_existing {
let t = t.ok_or_else(|| {
Error::new(ErrorKind::Error, "import --skip-existing needs a transport")
})?;
let ids: Vec<String> = bundle
.rules
.iter()
.filter_map(|r| r.rule_id().ok().map(str::to_owned))
.collect();
let existing = rules::existing_rule_ids(t, &ids).await?;
let mut keep = Vec::with_capacity(bundle.rules.len());
for rule in std::mem::take(&mut bundle.rules) {
match rule.rule_id() {
Ok(id) if existing.contains(id) => {
skipped.push(json!({"rule_id": id, "reason": "exists"}));
}
_ => keep.push(rule),
}
}
bundle.rules = keep;
}
if format == Format::Ndjson {
retain_referenced_exception_objects(&mut bundle);
}
add_upload_pointer_placeholders(&mut bundle);
let mut details: Vec<String> = bundle
.rules
.iter()
.map(|r| format!("{} {} import", r.rule_id().unwrap_or(""), r.name()))
.collect();
details.extend(skipped.iter().map(|s| {
format!(
"{} skip (already exists)",
s["rule_id"].as_str().unwrap_or("")
)
}));
let qualifier = if overwrite {
", overwriting existing".to_string()
} else if skip_existing && !skipped.is_empty() {
format!(", skipping {} that already exist", skipped.len())
} else {
String::new()
};
let preview = MutationPlan {
preview_action: format!(
"Import {} rule(s) from {}{qualifier}",
bundle.rules.len(),
path.display()
),
preview_details: details,
targets: bundle
.rules
.iter()
.filter_map(|r| r.rule_id().ok().map(str::to_owned))
.collect(),
};
let ndjson = match format {
Format::Yaml => codec::encode_ndjson(&bundle.rules)?,
Format::Ndjson => codec::encode_bundle(&bundle)?,
};
Ok(ImportPlan {
preview,
ndjson,
total,
skipped,
})
}
fn retain_referenced_exception_objects(bundle: &mut codec::Bundle) {
let wanted: std::collections::BTreeSet<ListKey> = bundle
.rules
.iter()
.flat_map(exception_refs)
.map(|reference| ListKey {
list_id: reference.list_id,
namespace_type: reference.namespace_type,
})
.collect();
bundle
.lists
.retain(|list| list.key().is_ok_and(|key| wanted.contains(&key)));
bundle.items.retain(|item| {
item.list_id().is_ok_and(|list_id| {
wanted.contains(&ListKey {
list_id: list_id.to_string(),
namespace_type: item.namespace_type().to_string(),
})
})
});
}
fn add_upload_pointer_placeholders(bundle: &mut codec::Bundle) {
for rule in &mut bundle.rules {
let Some(Value::Array(references)) = rule.as_map_mut().get_mut("exceptions_list") else {
continue;
};
for reference in references {
let Value::Object(reference) = reference else {
continue;
};
if reference.get("list_id").is_some_and(Value::is_string) {
reference.insert(
"id".to_string(),
Value::String(EXCEPTION_POINTER_PLACEHOLDER.to_string()),
);
}
}
}
}
pub async fn apply_import(t: &Transport, ndjson: &str, overwrite: bool) -> Result<ImportReport> {
if ndjson.is_empty() {
return Ok(ImportReport {
succeeded: json!(0),
failed: json!([]),
});
}
let response = rules::import(t, ndjson, overwrite).await?;
crate::ops::decode_import_report(&response, "rules")
}
async fn fetch_hits(
transport: &Transport,
space: &str,
preview_id: &str,
sample: usize,
) -> Result<rules::PreviewHits> {
let first = rules::preview_hits(transport, space, preview_id, sample).await?;
if first.total > 0 {
return Ok(first);
}
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
rules::preview_hits(transport, space, preview_id, sample).await
}
pub async fn preview_rule(
t: &Transport,
source: &str,
invocations: u32,
sample: u32,
space: &str,
) -> Result<PreviewReport> {
let path = Path::new(source);
let rule = if path.exists() {
let body = std::fs::read_to_string(path).map_err(|e| {
Error::new(ErrorKind::Error, format!("reading {}: {e}", path.display()))
})?;
let rules = match Format::from_path(path) {
Format::Yaml => codec::decode_yaml(&body)?,
Format::Ndjson => codec::decode_ndjson(&body)?.0,
};
rules.into_iter().next().ok_or_else(|| {
Error::new(
ErrorKind::Error,
format!("{} contains no rules", path.display()),
)
})?
} else {
selection::to_rule(t, source).await?.1
};
let timeframe_end = now_rfc3339();
let result = rules::preview(t, &rule, invocations, &timeframe_end).await?;
let (hits, hits_error, sample_hits) = match &result.preview_id {
None => (
None,
Some("the server returned no preview_id".to_string()),
Vec::new(),
),
Some(preview_id) => match fetch_hits(t, space, preview_id, sample as usize).await {
Ok(h) => (Some(h.total), None, h.sample),
Err(e) => (None, Some(e.message), Vec::new()),
},
};
Ok(PreviewReport {
rule: rule.name().to_string(),
preview_id: result.preview_id,
invocations,
hits,
errors: result.errors,
warnings: result.warnings,
hits_error,
sample: sample_hits,
})
}
fn now_rfc3339() -> String {
use std::time::{SystemTime, UNIX_EPOCH};
let secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
let days = secs / 86_400;
let rem = secs % 86_400;
let (y, m, d) = civil_from_days(days as i64);
format!(
"{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}.000Z",
rem / 3600,
(rem % 3600) / 60,
rem % 60
)
}
fn civil_from_days(z: i64) -> (i64, u32, u32) {
let z = z + 719_468;
let era = z.div_euclid(146_097);
let doe = z.rem_euclid(146_097);
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
let y = yoe + era * 400;
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
let mp = (5 * doy + 2) / 153;
let d = (doy - (153 * mp + 2) / 5 + 1) as u32;
let m = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
(if m <= 2 { y + 1 } else { y }, m, d)
}
#[cfg(test)]
mod date_tests {
use super::*;
#[test]
fn civil_from_days_matches_independently_computed_epoch_days() {
let cases = [
(0, (1970, 1, 1), "epoch"),
(19782, (2024, 2, 29), "leap day"),
(11017, (2000, 3, 1), "century leap year (2000 % 400 == 0)"),
(20818, (2026, 12, 31), "year end"),
(20819, (2027, 1, 1), "year rollover"),
(47541, (2100, 3, 1), "century non-leap (2100 % 400 != 0)"),
(
47540,
(2100, 2, 28),
"day before the century non-leap rollover",
),
];
for (day, expected, label) in cases {
assert_eq!(civil_from_days(day), expected, "{label}: day {day}");
}
}
#[test]
fn now_rfc3339_matches_the_shape_the_api_requires() {
let s = now_rfc3339();
let bytes = s.as_bytes();
assert_eq!(s.len(), 24, "{s}");
assert!(bytes[4] == b'-' && bytes[7] == b'-', "{s}");
assert_eq!(bytes[10], b'T', "{s}");
assert!(bytes[13] == b':' && bytes[16] == b':', "{s}");
assert_eq!(bytes[19], b'.', "{s}");
assert_eq!(&s[20..], "000Z", "{s}");
assert!(
s[..19]
.chars()
.enumerate()
.all(|(i, c)| { matches!(i, 4 | 7 | 10 | 13 | 16) || c.is_ascii_digit() }),
"every non-separator position must be a digit: {s}"
);
}
}