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 pub search: Option<String>,
66}
67
68impl RuleFilter {
69 pub fn to_kql(&self) -> Option<String> {
71 let mut parts: Vec<String> = Vec::new();
72 if let Some(clause) = self.source.clause() {
73 parts.push(clause.to_string());
74 }
75 if let Some(v) = self.enabled {
76 parts.push(format!("alert.attributes.enabled: {v}"));
77 }
78 if let Some(v) = &self.name {
79 parts.push(format!("alert.attributes.name: \"{}\"", kql_escape(v)));
80 }
81 if let Some(v) = &self.rule_type {
82 parts.push(format!(
83 "alert.attributes.params.type: \"{}\"",
84 kql_escape(v)
85 ));
86 }
87 if let Some(v) = &self.severity {
88 parts.push(format!(
89 "alert.attributes.params.severity: \"{}\"",
90 kql_escape(v)
91 ));
92 }
93 if let Some(v) = &self.tag {
94 parts.push(format!("alert.attributes.tags: \"{}\"", kql_escape(v)));
95 }
96 if let Some(v) = &self.search {
97 parts.push(format!(
98 "(alert.attributes.name: \"*{}*\" OR alert.attributes.tags: \"{}\")",
99 kql_escape_wildcard(v),
100 kql_escape(v)
101 ));
102 }
103 if let Some(v) = &self.query {
104 parts.push(v.clone());
105 }
106 (!parts.is_empty()).then(|| parts.join(" AND "))
107 }
108}
109
110pub(crate) fn kql_escape(value: &str) -> String {
118 value.replace('\\', "\\\\").replace('"', "\\\"")
119}
120
121pub(crate) fn kql_escape_wildcard(value: &str) -> String {
128 kql_escape(value).replace('*', "\\*").replace('?', "\\?")
129}
130
131fn rule_id_query(rule_ids: &[String]) -> String {
133 rule_ids
134 .iter()
135 .map(|id| format!("alert.attributes.params.ruleId: \"{}\"", kql_escape(id)))
136 .collect::<Vec<_>>()
137 .join(" OR ")
138}
139
140pub async fn find_page(
141 t: &Transport,
142 filter: &RuleFilter,
143 page: u32,
144 per_page: u32,
145) -> Result<(Vec<Rule>, u64)> {
146 if filter.source != RuleSource::All {
147 t.require_feature(Feature::RuleSourceScoping).await?;
148 }
149 let mut path = format!("{BASE}/_find?page={page}&per_page={per_page}");
150 if let Some(kql) = filter.to_kql() {
151 path.push_str(&format!("&filter={}", urlencode(&kql)));
152 }
153
154 let body = t.get(&path).await?;
155 decode_find(&body)
156}
157
158pub fn decode_find(body: &Value) -> Result<(Vec<Rule>, u64)> {
161 let map = body
162 .as_object()
163 .ok_or_else(|| find_error("response", "must be a JSON object"))?;
164 let data = map
165 .get("data")
166 .and_then(Value::as_array)
167 .ok_or_else(|| find_error("data", "must be an array"))?;
168 let total = required_u64(map, "total")?;
169 let page = required_u64(map, "page")?;
170 if page == 0 {
171 return Err(find_error("page", "must be greater than zero"));
172 }
173 let per_page = required_u64(map, "perPage")?;
174 if per_page == 0 {
175 return Err(find_error("perPage", "must be greater than zero"));
176 }
177
178 let data_len = data.len() as u64;
179 if data_len > total {
180 return Err(find_error(
181 "data.len() > total",
182 format!("{data_len} returned records exceed total {total}"),
183 ));
184 }
185 if data_len > per_page {
186 return Err(find_error(
187 "data.len() > perPage",
188 format!("{data_len} returned records exceed perPage {per_page}"),
189 ));
190 }
191
192 let rules = data
193 .iter()
194 .cloned()
195 .map(Rule::from_value)
196 .collect::<Result<Vec<_>>>()?;
197 Ok((rules, total))
198}
199
200fn find_error(field: &str, detail: impl std::fmt::Display) -> Error {
201 Error::new(
202 ErrorKind::Http,
203 format!("decoding rule _find response field {field}: {detail}"),
204 )
205}
206
207fn required_u64(map: &serde_json::Map<String, Value>, field: &str) -> Result<u64> {
208 map.get(field)
209 .and_then(Value::as_u64)
210 .ok_or_else(|| find_error(field, "must be an unsigned integer"))
211}
212
213pub(crate) struct SourceTotals {
215 pub custom: u64,
216 pub prebuilt: u64,
217 pub all: u64,
218}
219
220impl SourceTotals {
221 fn is_exhaustive(&self) -> bool {
222 self.custom.checked_add(self.prebuilt) == Some(self.all)
223 }
224}
225
226pub(crate) async fn verify_source_partition(t: &Transport) -> Result<SourceTotals> {
233 let (_, custom) = find_page(
234 t,
235 &RuleFilter {
236 source: RuleSource::Custom,
237 ..Default::default()
238 },
239 1,
240 1,
241 )
242 .await?;
243 let (_, prebuilt) = find_page(
244 t,
245 &RuleFilter {
246 source: RuleSource::Prebuilt,
247 ..Default::default()
248 },
249 1,
250 1,
251 )
252 .await?;
253 let (_, all) = find_page(t, &RuleFilter::default(), 1, 1).await?;
254
255 let totals = SourceTotals {
256 custom,
257 prebuilt,
258 all,
259 };
260 if !totals.is_exhaustive() {
261 return Err(Error::new(
262 ErrorKind::Unsupported,
263 format!(
264 "the immutable source partition is not exhaustive: custom={}, \
265 prebuilt={}, all={}. The field may be absent on this stack; \
266 re-run with --source all to read the corpus.",
267 totals.custom, totals.prebuilt, totals.all
268 ),
269 ));
270 }
271
272 Ok(totals)
273}
274
275const RULE_TYPES: [&str; 7] = [
280 "query",
281 "eql",
282 "esql",
283 "threshold",
284 "threat_match",
285 "machine_learning",
286 "new_terms",
287];
288
289pub async fn find_all(t: &Transport, filter: &RuleFilter) -> Result<Vec<Rule>> {
298 let (rules, total) = find_page(t, filter, 1, RESULT_WINDOW).await?;
299
300 if total <= u64::from(RESULT_WINDOW) {
301 if (rules.len() as u64) < total {
302 return Err(short_read(total, rules.len()));
303 }
304 return Ok(rules);
305 }
306
307 let types: Vec<&str> = match &filter.rule_type {
310 Some(t) => vec![t.as_str()],
311 None => RULE_TYPES.to_vec(),
312 };
313
314 let mut collected: Vec<Rule> = Vec::new();
315 let mut summed: u64 = 0;
316
317 for rule_type in types {
318 let mut type_filter = filter.clone();
319 type_filter.rule_type = Some(rule_type.to_string());
320
321 let slice_total = if filter.rule_type.is_some() {
324 total
325 } else {
326 let (slice_rules, slice_total) = find_page(t, &type_filter, 1, RESULT_WINDOW).await?;
327
328 if slice_total <= u64::from(RESULT_WINDOW) {
329 if (slice_rules.len() as u64) < slice_total {
330 return Err(short_read(slice_total, slice_rules.len()));
331 }
332 summed += slice_total;
333 collected.extend(slice_rules);
334 continue;
335 }
336 slice_total
337 };
338
339 if filter.enabled.is_some() {
342 return Err(oversized(slice_total));
343 }
344
345 for enabled in [true, false] {
346 let mut enabled_filter = type_filter.clone();
347 enabled_filter.enabled = Some(enabled);
348 let (enabled_rules, enabled_total) =
349 find_page(t, &enabled_filter, 1, RESULT_WINDOW).await?;
350 if enabled_total > u64::from(RESULT_WINDOW) {
351 return Err(oversized(enabled_total));
352 }
353 if (enabled_rules.len() as u64) < enabled_total {
354 return Err(short_read(enabled_total, enabled_rules.len()));
355 }
356 summed += enabled_total;
357 collected.extend(enabled_rules);
358 }
359 }
360
361 if summed != total {
364 return Err(Error::new(
365 ErrorKind::Http,
366 format!(
367 "the server counted {total} rules across the corpus but the type slices \
368 sum to {summed}. Refusing a partial corpus: a rule type added by a newer \
369 stack version would otherwise read as zero in every pull and diff."
370 ),
371 ));
372 }
373
374 Ok(collected)
375}
376
377fn short_read(counted: u64, returned: usize) -> Error {
380 Error::new(
381 ErrorKind::Http,
382 format!(
383 "the server counted {counted} rules and returned {returned}. Refusing a partial \
384 corpus: a short read is indistinguishable from rules having been deleted."
385 ),
386 )
387}
388
389fn oversized(count: u64) -> Error {
392 Error::new(
393 ErrorKind::Unsupported,
394 format!(
395 "{count} rules match, more than the {RESULT_WINDOW} a single search can return \
396 even after partitioning by type and enabled. Narrow the selection with a \
397 filter or a tag."
398 ),
399 )
400}
401
402const ID_CHUNK: usize = 50;
407
408pub async fn find_by_rule_ids(t: &Transport, rule_ids: &[String]) -> Result<Vec<Rule>> {
413 let mut found = Vec::with_capacity(rule_ids.len());
414 for chunk in rule_ids.chunks(ID_CHUNK) {
415 let filter = RuleFilter {
416 query: Some(rule_id_query(chunk)),
417 ..Default::default()
418 };
419 let (rules, _) = find_page(t, &filter, 1, RESULT_WINDOW).await?;
420 found.extend(rules);
421 }
422 Ok(found)
423}
424
425pub async fn get(t: &Transport, rule_id: &str) -> Result<Rule> {
426 let body = t
427 .get(&format!("{BASE}?rule_id={}", urlencode(rule_id)))
428 .await?;
429 Rule::from_value(body)
430}
431
432pub async fn create(t: &Transport, rule: &Rule) -> Result<Rule> {
433 let mut payload = rule.clone();
434 normalize::strip_volatile(&mut payload);
435 let response = t
436 .post(BASE, Some(&Value::Object(payload.as_map().clone())))
437 .await?;
438 Rule::from_value(response)
439}
440
441pub async fn update(t: &Transport, rule: &Rule) -> Result<Rule> {
442 let mut payload = rule.clone();
443 normalize::strip_volatile(&mut payload);
444 let response = t
445 .put(BASE, &Value::Object(payload.as_map().clone()))
446 .await?;
447 Rule::from_value(response)
448}
449
450pub async fn patch(t: &Transport, rule_id: &str, patch: &Value) -> Result<Rule> {
451 let mut body = patch.as_object().cloned().unwrap_or_default();
452 body.insert("rule_id".into(), json!(rule_id));
453 let response = t.patch(BASE, &Value::Object(body)).await?;
455 Rule::from_value(response)
456}
457
458pub async fn delete(t: &Transport, rule_id: &str) -> Result<Rule> {
459 let body = t
460 .delete(&format!("{BASE}?rule_id={}", urlencode(rule_id)))
461 .await?;
462 Rule::from_value(body)
463}
464
465#[derive(Debug, Clone, Copy, PartialEq, Eq)]
466pub enum BulkAction {
467 Enable,
468 Disable,
469 Delete,
470}
471
472impl BulkAction {
473 pub fn as_str(&self) -> &'static str {
474 match self {
475 Self::Enable => "enable",
476 Self::Disable => "disable",
477 Self::Delete => "delete",
478 }
479 }
480}
481
482#[derive(Debug, Clone, PartialEq, Eq, Default)]
483pub struct BulkOutcome {
484 pub succeeded: u64,
485 pub failed: u64,
486 pub skipped: u64,
487 pub total: u64,
488}
489
490pub fn decode_bulk_outcome(body: &Value) -> Result<BulkOutcome> {
495 let map = body
496 .as_object()
497 .ok_or_else(|| bulk_error("response", "must be a JSON object"))?;
498 let summary = map
499 .get("attributes")
500 .and_then(|attributes| attributes.get("summary"))
501 .and_then(Value::as_object)
502 .ok_or_else(|| bulk_error("attributes.summary", "must be an object"))?;
503 let succeeded = bulk_u64(summary, "succeeded")?;
504 let failed = bulk_u64(summary, "failed")?;
505 let skipped = bulk_u64(summary, "skipped")?;
506 let total = bulk_u64(summary, "total")?;
507 let applied = succeeded
508 .checked_add(failed)
509 .and_then(|count| count.checked_add(skipped))
510 .ok_or_else(|| bulk_error("total", "counter sum overflows u64"))?;
511 if applied != total {
512 return Err(bulk_error(
513 "total",
514 format!(
515 "{total} does not equal succeeded {succeeded} + failed {failed} + skipped {skipped}"
516 ),
517 ));
518 }
519 Ok(BulkOutcome {
520 succeeded,
521 failed,
522 skipped,
523 total,
524 })
525}
526
527fn bulk_error(field: &str, detail: impl std::fmt::Display) -> Error {
528 Error::new(
529 ErrorKind::Http,
530 format!("decoding rule _bulk_action response field {field}: {detail}"),
531 )
532}
533
534fn bulk_u64(map: &serde_json::Map<String, Value>, field: &str) -> Result<u64> {
535 map.get(field)
536 .and_then(Value::as_u64)
537 .ok_or_else(|| bulk_error(field, "must be an unsigned integer"))
538}
539
540pub async fn bulk_by_rule_ids(
541 t: &Transport,
542 action: BulkAction,
543 rule_ids: &[String],
544 dry_run: bool,
545) -> Result<BulkOutcome> {
546 if rule_ids.is_empty() {
548 return Ok(BulkOutcome::default());
549 }
550
551 let path = if dry_run {
552 format!("{BASE}/_bulk_action?dry_run=true")
553 } else {
554 format!("{BASE}/_bulk_action")
555 };
556 let body = json!({ "action": action.as_str(), "query": rule_id_query(rule_ids) });
557
558 let response = t.post(&path, Some(&body)).await?;
559 decode_bulk_outcome(&response)
560}
561
562pub async fn export(t: &Transport, rule_ids: Option<&[String]>) -> Result<Bundle> {
571 let body = rule_ids.map(|ids| {
572 json!({
573 "objects": ids
574 .iter()
575 .map(|id| json!({"rule_id": id}))
576 .collect::<Vec<_>>()
577 })
578 });
579 let text = t
580 .post_text(&format!("{BASE}/_export"), body.as_ref())
581 .await?;
582 codec::decode_bundle(&text)
583}
584
585const EXISTENCE_CHUNK: usize = 50;
588
589pub async fn existing_rule_ids(
594 t: &Transport,
595 rule_ids: &[String],
596) -> Result<std::collections::BTreeSet<String>> {
597 let mut found = std::collections::BTreeSet::new();
599 if rule_ids.is_empty() {
600 return Ok(found);
601 }
602
603 for chunk in rule_ids.chunks(EXISTENCE_CHUNK) {
604 let path = format!(
605 "{BASE}/_find?page=1&per_page={}&filter={}",
606 chunk.len(),
607 urlencode(&rule_id_query(chunk))
608 );
609 let (rules, _) = decode_find(&t.get(&path).await?)?;
610 for r in rules {
611 if let Ok(id) = r.rule_id() {
612 found.insert(id.to_string());
613 }
614 }
615 }
616
617 Ok(found)
618}
619
620pub async fn import(t: &Transport, ndjson: &str, overwrite: bool) -> Result<Value> {
621 t.post_multipart_ndjson(&format!("{BASE}/_import?overwrite={overwrite}"), ndjson)
622 .await
623}
624
625#[derive(Debug, Clone, Default, PartialEq)]
626pub struct PreviewResult {
627 pub preview_id: Option<String>,
628 pub errors: Vec<String>,
629 pub warnings: Vec<String>,
630}
631
632pub async fn preview(
637 t: &Transport,
638 rule: &Rule,
639 invocation_count: u32,
640 timeframe_end: &str,
641) -> Result<PreviewResult> {
642 let mut body = rule.as_map().clone();
643 for k in [
645 "rule_id",
646 "id",
647 "immutable",
648 "rule_source",
649 "revision",
650 "version",
651 ] {
652 body.remove(k);
653 }
654 body.insert("invocationCount".into(), json!(invocation_count));
655 body.insert("timeframeEnd".into(), json!(timeframe_end));
656
657 let response = t
658 .post(&format!("{BASE}/preview"), Some(&Value::Object(body)))
659 .await?;
660
661 let collect = |key: &str| -> Vec<String> {
662 response["logs"]
663 .as_array()
664 .map(|logs| {
665 logs.iter()
666 .filter_map(|l| l.get(key)?.as_array())
667 .flatten()
668 .filter_map(|v| v.as_str().map(str::to_owned))
669 .collect()
670 })
671 .unwrap_or_default()
672 };
673
674 Ok(PreviewResult {
675 preview_id: response["previewId"].as_str().map(str::to_owned),
676 errors: collect("errors"),
677 warnings: collect("warnings"),
678 })
679}
680
681pub const PREVIEW_ALERTS_INDEX_PREFIX: &str = ".preview.alerts-security.alerts-";
686
687#[derive(Debug, Clone, Default, PartialEq)]
688pub struct PreviewHits {
689 pub total: u64,
690 pub sample: Vec<Value>,
693}
694
695pub async fn preview_hits(
701 t: &Transport,
702 space: &str,
703 preview_id: &str,
704 sample: usize,
705) -> Result<PreviewHits> {
706 let space = if space.is_empty() { "default" } else { space };
707 let index = urlencode(&format!("{PREVIEW_ALERTS_INDEX_PREFIX}{space}"));
708 let body = json!({
709 "size": sample,
710 "track_total_hits": true,
712 "query": {"term": {"kibana.alert.rule.uuid": preview_id}},
713 "sort": [{"@timestamp": {"order": "desc"}}]
714 });
715
716 let response = t
717 .post_absolute_es(&format!("/{index}/_search?ignore_unavailable=true"), &body)
718 .await?;
719
720 decode_preview_hits_checked(&response)
721}
722
723pub fn decode_preview_hits_checked(response: &Value) -> Result<PreviewHits> {
725 let map = response
726 .as_object()
727 .ok_or_else(|| preview_error("response", "must be a JSON object"))?;
728 let hits = map
729 .get("hits")
730 .and_then(Value::as_object)
731 .ok_or_else(|| preview_error("hits", "must be an object"))?;
732 let total = hits
733 .get("total")
734 .and_then(|total| total.get("value"))
735 .and_then(Value::as_u64)
736 .ok_or_else(|| preview_error("hits.total.value", "must be an unsigned integer"))?;
737 let sample = hits
738 .get("hits")
739 .and_then(Value::as_array)
740 .ok_or_else(|| preview_error("hits.hits", "must be an array"))?
741 .iter()
742 .map(|h| {
743 json!({
744 "_id": h.get("_id").cloned().unwrap_or(Value::Null),
745 "_source": h.get("_source").cloned().unwrap_or(Value::Null),
746 })
747 })
748 .collect();
749
750 Ok(PreviewHits { total, sample })
751}
752
753fn preview_error(field: &str, detail: impl std::fmt::Display) -> Error {
754 Error::new(
755 ErrorKind::Http,
756 format!("decoding preview-hits response field {field}: {detail}"),
757 )
758}
759
760pub fn decode_preview_hits(response: &Value) -> PreviewHits {
765 decode_preview_hits_checked(response).unwrap_or_default()
766}
767
768#[cfg(test)]
769mod tests {
770 use super::*;
771
772 #[test]
774 fn each_source_maps_to_its_measured_filter() {
775 assert_eq!(
776 RuleSource::Custom.clause(),
777 Some("alert.attributes.params.immutable: false")
778 );
779 assert_eq!(
780 RuleSource::Customized.clause(),
781 Some("alert.attributes.params.ruleSource.isCustomized: true")
782 );
783 assert_eq!(RuleSource::All.clause(), None, "all adds no clause");
784 }
785
786 #[test]
787 fn a_source_clause_combines_with_other_filters() {
788 let f = RuleFilter {
789 source: RuleSource::Custom,
790 tag: Some("prod".into()),
791 ..Default::default()
792 };
793 let kql = f.to_kql().unwrap();
794 assert!(kql.contains("immutable: false"), "{kql}");
795 assert!(kql.contains("prod"), "{kql}");
796 assert!(
797 kql.contains(" AND "),
798 "clauses combine, they do not replace: {kql}"
799 );
800 }
801
802 #[test]
803 fn kql_escape_doubles_a_lone_backslash() {
804 assert_eq!(kql_escape("a\\b"), "a\\\\b");
805 }
806
807 #[test]
808 fn kql_escape_escapes_a_lone_quote() {
809 let mut expected = String::from("a");
810 expected.push('\\');
811 expected.push('"');
812 expected.push('b');
813 assert_eq!(kql_escape("a\"b"), expected);
814 }
815
816 #[test]
819 fn kql_escape_wildcard_escapes_kql_wildcards_after_kql_escape() {
820 assert_eq!(kql_escape_wildcard("*"), "\\*");
821 assert_eq!(kql_escape_wildcard("?"), "\\?");
822 assert_eq!(kql_escape_wildcard("a\\b\"c"), kql_escape("a\\b\"c"));
823 }
824
825 #[test]
828 fn kql_escape_orders_backslash_before_quote() {
829 let mut input = String::new();
830 input.push('\\');
831 input.push('"');
832
833 let escaped = kql_escape(&input);
834
835 let mut expected = "\\".repeat(3);
838 expected.push('"');
839 assert_eq!(escaped, expected);
840 }
841
842 #[test]
843 fn rule_id_query_escapes_a_quote_in_the_id() {
844 let mut id = String::from("x");
845 id.push('"');
846 id.push('y');
847
848 let q = rule_id_query(&[id]);
849
850 let mut expected = String::from("alert.attributes.params.ruleId: \"x");
851 expected.push('\\');
852 expected.push('"');
853 expected.push_str("y\"");
854 assert_eq!(q, expected);
855 assert!(
856 !q.contains(" OR "),
857 "a single id must produce exactly one clause: {q}"
858 );
859 }
860
861 #[test]
862 fn rule_id_query_neutralizes_a_kql_injection_payload() {
863 let payload = "x\" or alert.attributes.enabled: true or \"";
864 let q = rule_id_query(&[payload.to_string()]);
865 assert!(
866 !q.contains("\" or alert.attributes.enabled: true or \""),
867 "the injected quote must not close the literal: {q}"
868 );
869 }
870
871 #[test]
872 fn to_kql_escapes_a_quote_in_the_tag() {
873 let f = RuleFilter {
874 tag: Some("a\"b".into()),
875 ..Default::default()
876 };
877 let kql = f.to_kql().unwrap();
878
879 let mut expected = String::from("alert.attributes.tags: \"a");
880 expected.push('\\');
881 expected.push('"');
882 expected.push_str("b\"");
883 assert_eq!(kql, expected);
884 }
885}