1use crate::codec::{self, Bundle};
6use crate::model::Rule;
7use crate::normalize;
8use elasticctl_core::{Error, ErrorKind, Feature, Result, Transport, urlencode};
9use serde::{Deserialize, Serialize};
10use serde_json::{Value, json};
11
12const BASE: &str = "/api/detection_engine/rules";
13
14const RESULT_WINDOW: u32 = 10_000;
21
22#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
29#[serde(rename_all = "lowercase")]
30pub enum RuleSource {
31 Custom,
32 Customized,
33 Prebuilt,
34 #[default]
35 All,
36}
37
38impl RuleSource {
39 pub fn clause(&self) -> Option<&'static str> {
42 match self {
43 RuleSource::Custom => Some("alert.attributes.params.immutable: false"),
44 RuleSource::Prebuilt => Some("alert.attributes.params.immutable: true"),
45 RuleSource::Customized => Some("alert.attributes.params.ruleSource.isCustomized: true"),
46 RuleSource::All => None,
47 }
48 }
49}
50
51#[derive(Debug, Clone, Default)]
52pub struct RuleFilter {
53 pub source: RuleSource,
54 pub enabled: Option<bool>,
55 pub rule_type: Option<String>,
56 pub severity: Option<String>,
57 pub tag: Option<String>,
58 pub name: Option<String>,
61 pub query: Option<String>,
63}
64
65impl RuleFilter {
66 pub fn to_kql(&self) -> Option<String> {
68 let mut parts: Vec<String> = Vec::new();
69 if let Some(clause) = self.source.clause() {
70 parts.push(clause.to_string());
71 }
72 if let Some(v) = self.enabled {
73 parts.push(format!("alert.attributes.enabled: {v}"));
74 }
75 if let Some(v) = &self.name {
76 parts.push(format!("alert.attributes.name: \"{}\"", kql_escape(v)));
77 }
78 if let Some(v) = &self.rule_type {
79 parts.push(format!(
80 "alert.attributes.params.type: \"{}\"",
81 kql_escape(v)
82 ));
83 }
84 if let Some(v) = &self.severity {
85 parts.push(format!(
86 "alert.attributes.params.severity: \"{}\"",
87 kql_escape(v)
88 ));
89 }
90 if let Some(v) = &self.tag {
91 parts.push(format!("alert.attributes.tags: \"{}\"", kql_escape(v)));
92 }
93 if let Some(v) = &self.query {
94 parts.push(v.clone());
95 }
96 (!parts.is_empty()).then(|| parts.join(" AND "))
97 }
98}
99
100pub(crate) fn kql_escape(value: &str) -> String {
108 value.replace('\\', "\\\\").replace('"', "\\\"")
109}
110
111fn rule_id_query(rule_ids: &[String]) -> String {
113 rule_ids
114 .iter()
115 .map(|id| format!("alert.attributes.params.ruleId: \"{}\"", kql_escape(id)))
116 .collect::<Vec<_>>()
117 .join(" OR ")
118}
119
120pub async fn find_page(
121 t: &Transport,
122 filter: &RuleFilter,
123 page: u32,
124 per_page: u32,
125) -> Result<(Vec<Rule>, u64)> {
126 if filter.source != RuleSource::All {
127 t.require_feature(Feature::RuleSourceScoping).await?;
128 }
129 let mut path = format!("{BASE}/_find?page={page}&per_page={per_page}");
130 if let Some(kql) = filter.to_kql() {
131 path.push_str(&format!("&filter={}", urlencode(&kql)));
132 }
133
134 let body = t.get(&path).await?;
135 decode_find(&body)
136}
137
138pub fn decode_find(body: &Value) -> Result<(Vec<Rule>, u64)> {
141 let map = body
142 .as_object()
143 .ok_or_else(|| find_error("response", "must be a JSON object"))?;
144 let data = map
145 .get("data")
146 .and_then(Value::as_array)
147 .ok_or_else(|| find_error("data", "must be an array"))?;
148 let total = required_u64(map, "total")?;
149 let page = required_u64(map, "page")?;
150 if page == 0 {
151 return Err(find_error("page", "must be greater than zero"));
152 }
153 let per_page = required_u64(map, "perPage")?;
154 if per_page == 0 {
155 return Err(find_error("perPage", "must be greater than zero"));
156 }
157
158 let data_len = data.len() as u64;
159 if data_len > total {
160 return Err(find_error(
161 "data.len() > total",
162 format!("{data_len} returned records exceed total {total}"),
163 ));
164 }
165 if data_len > per_page {
166 return Err(find_error(
167 "data.len() > perPage",
168 format!("{data_len} returned records exceed perPage {per_page}"),
169 ));
170 }
171
172 let rules = data
173 .iter()
174 .cloned()
175 .map(Rule::from_value)
176 .collect::<Result<Vec<_>>>()?;
177 Ok((rules, total))
178}
179
180fn find_error(field: &str, detail: impl std::fmt::Display) -> Error {
181 Error::new(
182 ErrorKind::Http,
183 format!("decoding rule _find response field {field}: {detail}"),
184 )
185}
186
187fn required_u64(map: &serde_json::Map<String, Value>, field: &str) -> Result<u64> {
188 map.get(field)
189 .and_then(Value::as_u64)
190 .ok_or_else(|| find_error(field, "must be an unsigned integer"))
191}
192
193pub(crate) struct SourceTotals {
195 pub custom: u64,
196 pub prebuilt: u64,
197 pub all: u64,
198}
199
200impl SourceTotals {
201 fn is_exhaustive(&self) -> bool {
202 self.custom.checked_add(self.prebuilt) == Some(self.all)
203 }
204}
205
206pub(crate) async fn verify_source_partition(t: &Transport) -> Result<SourceTotals> {
213 let (_, custom) = find_page(
214 t,
215 &RuleFilter {
216 source: RuleSource::Custom,
217 ..Default::default()
218 },
219 1,
220 1,
221 )
222 .await?;
223 let (_, prebuilt) = find_page(
224 t,
225 &RuleFilter {
226 source: RuleSource::Prebuilt,
227 ..Default::default()
228 },
229 1,
230 1,
231 )
232 .await?;
233 let (_, all) = find_page(t, &RuleFilter::default(), 1, 1).await?;
234
235 let totals = SourceTotals {
236 custom,
237 prebuilt,
238 all,
239 };
240 if !totals.is_exhaustive() {
241 return Err(Error::new(
242 ErrorKind::Unsupported,
243 format!(
244 "the immutable source partition is not exhaustive: custom={}, \
245 prebuilt={}, all={}. The field may be absent on this stack; \
246 re-run with --source all to read the corpus.",
247 totals.custom, totals.prebuilt, totals.all
248 ),
249 ));
250 }
251
252 Ok(totals)
253}
254
255const RULE_TYPES: [&str; 7] = [
260 "query",
261 "eql",
262 "esql",
263 "threshold",
264 "threat_match",
265 "machine_learning",
266 "new_terms",
267];
268
269pub async fn find_all(t: &Transport, filter: &RuleFilter) -> Result<Vec<Rule>> {
278 let (rules, total) = find_page(t, filter, 1, RESULT_WINDOW).await?;
279
280 if total <= u64::from(RESULT_WINDOW) {
281 if (rules.len() as u64) < total {
282 return Err(short_read(total, rules.len()));
283 }
284 return Ok(rules);
285 }
286
287 let types: Vec<&str> = match &filter.rule_type {
290 Some(t) => vec![t.as_str()],
291 None => RULE_TYPES.to_vec(),
292 };
293
294 let mut collected: Vec<Rule> = Vec::new();
295 let mut summed: u64 = 0;
296
297 for rule_type in types {
298 let mut type_filter = filter.clone();
299 type_filter.rule_type = Some(rule_type.to_string());
300
301 let slice_total = if filter.rule_type.is_some() {
304 total
305 } else {
306 let (slice_rules, slice_total) = find_page(t, &type_filter, 1, RESULT_WINDOW).await?;
307
308 if slice_total <= u64::from(RESULT_WINDOW) {
309 if (slice_rules.len() as u64) < slice_total {
310 return Err(short_read(slice_total, slice_rules.len()));
311 }
312 summed += slice_total;
313 collected.extend(slice_rules);
314 continue;
315 }
316 slice_total
317 };
318
319 if filter.enabled.is_some() {
322 return Err(oversized(slice_total));
323 }
324
325 for enabled in [true, false] {
326 let mut enabled_filter = type_filter.clone();
327 enabled_filter.enabled = Some(enabled);
328 let (enabled_rules, enabled_total) =
329 find_page(t, &enabled_filter, 1, RESULT_WINDOW).await?;
330 if enabled_total > u64::from(RESULT_WINDOW) {
331 return Err(oversized(enabled_total));
332 }
333 if (enabled_rules.len() as u64) < enabled_total {
334 return Err(short_read(enabled_total, enabled_rules.len()));
335 }
336 summed += enabled_total;
337 collected.extend(enabled_rules);
338 }
339 }
340
341 if summed != total {
344 return Err(Error::new(
345 ErrorKind::Http,
346 format!(
347 "the server counted {total} rules across the corpus but the type slices \
348 sum to {summed}. Refusing a partial corpus: a rule type added by a newer \
349 stack version would otherwise read as zero in every pull and diff."
350 ),
351 ));
352 }
353
354 Ok(collected)
355}
356
357fn short_read(counted: u64, returned: usize) -> Error {
360 Error::new(
361 ErrorKind::Http,
362 format!(
363 "the server counted {counted} rules and returned {returned}. Refusing a partial \
364 corpus: a short read is indistinguishable from rules having been deleted."
365 ),
366 )
367}
368
369fn oversized(count: u64) -> Error {
372 Error::new(
373 ErrorKind::Unsupported,
374 format!(
375 "{count} rules match, more than the {RESULT_WINDOW} a single search can return \
376 even after partitioning by type and enabled. Narrow the selection with a \
377 filter or a tag."
378 ),
379 )
380}
381
382const ID_CHUNK: usize = 50;
387
388pub async fn find_by_rule_ids(t: &Transport, rule_ids: &[String]) -> Result<Vec<Rule>> {
393 let mut found = Vec::with_capacity(rule_ids.len());
394 for chunk in rule_ids.chunks(ID_CHUNK) {
395 let filter = RuleFilter {
396 query: Some(rule_id_query(chunk)),
397 ..Default::default()
398 };
399 let (rules, _) = find_page(t, &filter, 1, RESULT_WINDOW).await?;
400 found.extend(rules);
401 }
402 Ok(found)
403}
404
405pub async fn get(t: &Transport, rule_id: &str) -> Result<Rule> {
406 let body = t
407 .get(&format!("{BASE}?rule_id={}", urlencode(rule_id)))
408 .await?;
409 Rule::from_value(body)
410}
411
412pub async fn create(t: &Transport, rule: &Rule) -> Result<Rule> {
413 let mut payload = rule.clone();
414 normalize::strip_volatile(&mut payload);
415 let response = t
416 .post(BASE, Some(&Value::Object(payload.as_map().clone())))
417 .await?;
418 Rule::from_value(response)
419}
420
421pub async fn update(t: &Transport, rule: &Rule) -> Result<Rule> {
422 let mut payload = rule.clone();
423 normalize::strip_volatile(&mut payload);
424 let response = t
425 .put(BASE, &Value::Object(payload.as_map().clone()))
426 .await?;
427 Rule::from_value(response)
428}
429
430pub async fn patch(t: &Transport, rule_id: &str, patch: &Value) -> Result<Rule> {
431 let mut body = patch.as_object().cloned().unwrap_or_default();
432 body.insert("rule_id".into(), json!(rule_id));
433 let response = t.patch(BASE, &Value::Object(body)).await?;
435 Rule::from_value(response)
436}
437
438pub async fn delete(t: &Transport, rule_id: &str) -> Result<Rule> {
439 let body = t
440 .delete(&format!("{BASE}?rule_id={}", urlencode(rule_id)))
441 .await?;
442 Rule::from_value(body)
443}
444
445#[derive(Debug, Clone, Copy, PartialEq, Eq)]
446pub enum BulkAction {
447 Enable,
448 Disable,
449 Delete,
450}
451
452impl BulkAction {
453 pub fn as_str(&self) -> &'static str {
454 match self {
455 Self::Enable => "enable",
456 Self::Disable => "disable",
457 Self::Delete => "delete",
458 }
459 }
460}
461
462#[derive(Debug, Clone, PartialEq, Eq, Default)]
463pub struct BulkOutcome {
464 pub succeeded: u64,
465 pub failed: u64,
466 pub skipped: u64,
467 pub total: u64,
468}
469
470pub fn decode_bulk_outcome(body: &Value) -> Result<BulkOutcome> {
475 let map = body
476 .as_object()
477 .ok_or_else(|| bulk_error("response", "must be a JSON object"))?;
478 let summary = map
479 .get("attributes")
480 .and_then(|attributes| attributes.get("summary"))
481 .and_then(Value::as_object)
482 .ok_or_else(|| bulk_error("attributes.summary", "must be an object"))?;
483 let succeeded = bulk_u64(summary, "succeeded")?;
484 let failed = bulk_u64(summary, "failed")?;
485 let skipped = bulk_u64(summary, "skipped")?;
486 let total = bulk_u64(summary, "total")?;
487 let applied = succeeded
488 .checked_add(failed)
489 .and_then(|count| count.checked_add(skipped))
490 .ok_or_else(|| bulk_error("total", "counter sum overflows u64"))?;
491 if applied != total {
492 return Err(bulk_error(
493 "total",
494 format!(
495 "{total} does not equal succeeded {succeeded} + failed {failed} + skipped {skipped}"
496 ),
497 ));
498 }
499 Ok(BulkOutcome {
500 succeeded,
501 failed,
502 skipped,
503 total,
504 })
505}
506
507fn bulk_error(field: &str, detail: impl std::fmt::Display) -> Error {
508 Error::new(
509 ErrorKind::Http,
510 format!("decoding rule _bulk_action response field {field}: {detail}"),
511 )
512}
513
514fn bulk_u64(map: &serde_json::Map<String, Value>, field: &str) -> Result<u64> {
515 map.get(field)
516 .and_then(Value::as_u64)
517 .ok_or_else(|| bulk_error(field, "must be an unsigned integer"))
518}
519
520pub async fn bulk_by_rule_ids(
521 t: &Transport,
522 action: BulkAction,
523 rule_ids: &[String],
524 dry_run: bool,
525) -> Result<BulkOutcome> {
526 if rule_ids.is_empty() {
528 return Ok(BulkOutcome::default());
529 }
530
531 let path = if dry_run {
532 format!("{BASE}/_bulk_action?dry_run=true")
533 } else {
534 format!("{BASE}/_bulk_action")
535 };
536 let body = json!({ "action": action.as_str(), "query": rule_id_query(rule_ids) });
537
538 let response = t.post(&path, Some(&body)).await?;
539 decode_bulk_outcome(&response)
540}
541
542pub async fn export(t: &Transport, rule_ids: Option<&[String]>) -> Result<Bundle> {
551 let body = rule_ids.map(|ids| {
552 json!({
553 "objects": ids
554 .iter()
555 .map(|id| json!({"rule_id": id}))
556 .collect::<Vec<_>>()
557 })
558 });
559 let text = t
560 .post_text(&format!("{BASE}/_export"), body.as_ref())
561 .await?;
562 codec::decode_bundle(&text)
563}
564
565const EXISTENCE_CHUNK: usize = 50;
568
569pub async fn existing_rule_ids(
574 t: &Transport,
575 rule_ids: &[String],
576) -> Result<std::collections::BTreeSet<String>> {
577 let mut found = std::collections::BTreeSet::new();
579 if rule_ids.is_empty() {
580 return Ok(found);
581 }
582
583 for chunk in rule_ids.chunks(EXISTENCE_CHUNK) {
584 let path = format!(
585 "{BASE}/_find?page=1&per_page={}&filter={}",
586 chunk.len(),
587 urlencode(&rule_id_query(chunk))
588 );
589 let (rules, _) = decode_find(&t.get(&path).await?)?;
590 for r in rules {
591 if let Ok(id) = r.rule_id() {
592 found.insert(id.to_string());
593 }
594 }
595 }
596
597 Ok(found)
598}
599
600pub async fn import(t: &Transport, ndjson: &str, overwrite: bool) -> Result<Value> {
601 t.post_multipart_ndjson(&format!("{BASE}/_import?overwrite={overwrite}"), ndjson)
602 .await
603}
604
605#[derive(Debug, Clone, Default, PartialEq)]
606pub struct PreviewResult {
607 pub preview_id: Option<String>,
608 pub errors: Vec<String>,
609 pub warnings: Vec<String>,
610}
611
612pub async fn preview(
617 t: &Transport,
618 rule: &Rule,
619 invocation_count: u32,
620 timeframe_end: &str,
621) -> Result<PreviewResult> {
622 let mut body = rule.as_map().clone();
623 for k in [
625 "rule_id",
626 "id",
627 "immutable",
628 "rule_source",
629 "revision",
630 "version",
631 ] {
632 body.remove(k);
633 }
634 body.insert("invocationCount".into(), json!(invocation_count));
635 body.insert("timeframeEnd".into(), json!(timeframe_end));
636
637 let response = t
638 .post(&format!("{BASE}/preview"), Some(&Value::Object(body)))
639 .await?;
640
641 let collect = |key: &str| -> Vec<String> {
642 response["logs"]
643 .as_array()
644 .map(|logs| {
645 logs.iter()
646 .filter_map(|l| l.get(key)?.as_array())
647 .flatten()
648 .filter_map(|v| v.as_str().map(str::to_owned))
649 .collect()
650 })
651 .unwrap_or_default()
652 };
653
654 Ok(PreviewResult {
655 preview_id: response["previewId"].as_str().map(str::to_owned),
656 errors: collect("errors"),
657 warnings: collect("warnings"),
658 })
659}
660
661pub const PREVIEW_ALERTS_INDEX_PREFIX: &str = ".preview.alerts-security.alerts-";
666
667#[derive(Debug, Clone, Default, PartialEq)]
668pub struct PreviewHits {
669 pub total: u64,
670 pub sample: Vec<Value>,
673}
674
675pub async fn preview_hits(
681 t: &Transport,
682 space: &str,
683 preview_id: &str,
684 sample: usize,
685) -> Result<PreviewHits> {
686 let space = if space.is_empty() { "default" } else { space };
687 let index = urlencode(&format!("{PREVIEW_ALERTS_INDEX_PREFIX}{space}"));
688 let body = json!({
689 "size": sample,
690 "track_total_hits": true,
692 "query": {"term": {"kibana.alert.rule.uuid": preview_id}},
693 "sort": [{"@timestamp": {"order": "desc"}}]
694 });
695
696 let response = t
697 .post_absolute_es(&format!("/{index}/_search?ignore_unavailable=true"), &body)
698 .await?;
699
700 decode_preview_hits_checked(&response)
701}
702
703pub fn decode_preview_hits_checked(response: &Value) -> Result<PreviewHits> {
705 let map = response
706 .as_object()
707 .ok_or_else(|| preview_error("response", "must be a JSON object"))?;
708 let hits = map
709 .get("hits")
710 .and_then(Value::as_object)
711 .ok_or_else(|| preview_error("hits", "must be an object"))?;
712 let total = hits
713 .get("total")
714 .and_then(|total| total.get("value"))
715 .and_then(Value::as_u64)
716 .ok_or_else(|| preview_error("hits.total.value", "must be an unsigned integer"))?;
717 let sample = hits
718 .get("hits")
719 .and_then(Value::as_array)
720 .ok_or_else(|| preview_error("hits.hits", "must be an array"))?
721 .iter()
722 .map(|h| {
723 json!({
724 "_id": h.get("_id").cloned().unwrap_or(Value::Null),
725 "_source": h.get("_source").cloned().unwrap_or(Value::Null),
726 })
727 })
728 .collect();
729
730 Ok(PreviewHits { total, sample })
731}
732
733fn preview_error(field: &str, detail: impl std::fmt::Display) -> Error {
734 Error::new(
735 ErrorKind::Http,
736 format!("decoding preview-hits response field {field}: {detail}"),
737 )
738}
739
740pub fn decode_preview_hits(response: &Value) -> PreviewHits {
745 decode_preview_hits_checked(response).unwrap_or_default()
746}
747
748#[cfg(test)]
749mod tests {
750 use super::*;
751
752 #[test]
754 fn each_source_maps_to_its_measured_filter() {
755 assert_eq!(
756 RuleSource::Custom.clause(),
757 Some("alert.attributes.params.immutable: false")
758 );
759 assert_eq!(
760 RuleSource::Customized.clause(),
761 Some("alert.attributes.params.ruleSource.isCustomized: true")
762 );
763 assert_eq!(RuleSource::All.clause(), None, "all adds no clause");
764 }
765
766 #[test]
767 fn a_source_clause_combines_with_other_filters() {
768 let f = RuleFilter {
769 source: RuleSource::Custom,
770 tag: Some("prod".into()),
771 ..Default::default()
772 };
773 let kql = f.to_kql().unwrap();
774 assert!(kql.contains("immutable: false"), "{kql}");
775 assert!(kql.contains("prod"), "{kql}");
776 assert!(
777 kql.contains(" AND "),
778 "clauses combine, they do not replace: {kql}"
779 );
780 }
781
782 #[test]
783 fn kql_escape_doubles_a_lone_backslash() {
784 assert_eq!(kql_escape("a\\b"), "a\\\\b");
785 }
786
787 #[test]
788 fn kql_escape_escapes_a_lone_quote() {
789 let mut expected = String::from("a");
790 expected.push('\\');
791 expected.push('"');
792 expected.push('b');
793 assert_eq!(kql_escape("a\"b"), expected);
794 }
795
796 #[test]
799 fn kql_escape_orders_backslash_before_quote() {
800 let mut input = String::new();
801 input.push('\\');
802 input.push('"');
803
804 let escaped = kql_escape(&input);
805
806 let mut expected = "\\".repeat(3);
809 expected.push('"');
810 assert_eq!(escaped, expected);
811 }
812
813 #[test]
814 fn rule_id_query_escapes_a_quote_in_the_id() {
815 let mut id = String::from("x");
816 id.push('"');
817 id.push('y');
818
819 let q = rule_id_query(&[id]);
820
821 let mut expected = String::from("alert.attributes.params.ruleId: \"x");
822 expected.push('\\');
823 expected.push('"');
824 expected.push_str("y\"");
825 assert_eq!(q, expected);
826 assert!(
827 !q.contains(" OR "),
828 "a single id must produce exactly one clause: {q}"
829 );
830 }
831
832 #[test]
833 fn rule_id_query_neutralizes_a_kql_injection_payload() {
834 let payload = "x\" or alert.attributes.enabled: true or \"";
835 let q = rule_id_query(&[payload.to_string()]);
836 assert!(
837 !q.contains("\" or alert.attributes.enabled: true or \""),
838 "the injected quote must not close the literal: {q}"
839 );
840 }
841
842 #[test]
843 fn to_kql_escapes_a_quote_in_the_tag() {
844 let f = RuleFilter {
845 tag: Some("a\"b".into()),
846 ..Default::default()
847 };
848 let kql = f.to_kql().unwrap();
849
850 let mut expected = String::from("alert.attributes.tags: \"a");
851 expected.push('\\');
852 expected.push('"');
853 expected.push_str("b\"");
854 assert_eq!(kql, expected);
855 }
856}