1mod preview_execution;
6pub use preview_execution::{
7 PREVIEW_ALERTS_INDEX_PREFIX, PreviewHits, PreviewResult, decode_preview_hits,
8 decode_preview_hits_checked, preview, preview_hits,
9};
10
11use crate::codec::{self, Bundle};
12use crate::model::Rule;
13use crate::normalize;
14use elasticctl_core::{Error, ErrorKind, Feature, Result, Transport, urlencode};
15use serde::{Deserialize, Serialize};
16use serde_json::{Value, json};
17
18const BASE: &str = "/api/detection_engine/rules";
19
20const RESULT_WINDOW: u32 = 10_000;
27
28#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
35#[serde(rename_all = "lowercase")]
36pub enum RuleSource {
37 Custom,
38 Customized,
39 Prebuilt,
40 #[default]
41 All,
42}
43
44impl RuleSource {
45 pub fn clause(&self) -> Option<&'static str> {
48 match self {
49 RuleSource::Custom => Some("alert.attributes.params.immutable: false"),
50 RuleSource::Prebuilt => Some("alert.attributes.params.immutable: true"),
51 RuleSource::Customized => Some("alert.attributes.params.ruleSource.isCustomized: true"),
52 RuleSource::All => None,
53 }
54 }
55}
56
57#[derive(Debug, Clone, Default)]
58pub struct RuleFilter {
59 pub source: RuleSource,
60 pub enabled: Option<bool>,
61 pub rule_type: Option<String>,
62 pub severity: Option<String>,
63 pub tag: Option<String>,
64 pub name: Option<String>,
67 pub query: Option<String>,
69 pub search: Option<String>,
72}
73
74impl RuleFilter {
75 pub fn to_kql(&self) -> Option<String> {
77 let mut parts: Vec<String> = Vec::new();
78 if let Some(clause) = self.source.clause() {
79 parts.push(clause.to_string());
80 }
81 if let Some(v) = self.enabled {
82 parts.push(format!("alert.attributes.enabled: {v}"));
83 }
84 if let Some(v) = &self.name {
85 parts.push(format!("alert.attributes.name: \"{}\"", kql_escape(v)));
86 }
87 if let Some(v) = &self.rule_type {
88 parts.push(format!(
89 "alert.attributes.params.type: \"{}\"",
90 kql_escape(v)
91 ));
92 }
93 if let Some(v) = &self.severity {
94 parts.push(format!(
95 "alert.attributes.params.severity: \"{}\"",
96 kql_escape(v)
97 ));
98 }
99 if let Some(v) = &self.tag {
100 parts.push(format!("alert.attributes.tags: \"{}\"", kql_escape(v)));
101 }
102 if let Some(v) = &self.search {
103 parts.push(format!(
104 "(alert.attributes.name: \"*{}*\" OR alert.attributes.tags: \"{}\")",
105 kql_escape_wildcard(v),
106 kql_escape(v)
107 ));
108 }
109 if let Some(v) = &self.query {
110 parts.push(v.clone());
111 }
112 (!parts.is_empty()).then(|| parts.join(" AND "))
113 }
114}
115
116pub(crate) fn kql_escape(value: &str) -> String {
124 value.replace('\\', "\\\\").replace('"', "\\\"")
125}
126
127pub(crate) fn kql_escape_wildcard(value: &str) -> String {
134 kql_escape(value).replace('*', "\\*").replace('?', "\\?")
135}
136
137fn rule_id_query(rule_ids: &[String]) -> String {
139 rule_ids
140 .iter()
141 .map(|id| format!("alert.attributes.params.ruleId: \"{}\"", kql_escape(id)))
142 .collect::<Vec<_>>()
143 .join(" OR ")
144}
145
146pub async fn find_page(
147 t: &Transport,
148 filter: &RuleFilter,
149 page: u32,
150 per_page: u32,
151) -> Result<(Vec<Rule>, u64)> {
152 find(t, filter, page, per_page, "").await
153}
154
155pub async fn find_first_by_name(
158 t: &Transport,
159 filter: &RuleFilter,
160 per_page: u32,
161) -> Result<(Vec<Rule>, u64)> {
162 let (rules, total) = find(t, filter, 1, per_page, "&sort_field=name&sort_order=asc").await?;
163 if (rules.len() as u64) < total.min(u64::from(per_page)) {
164 return Err(short_read(total, rules.len()));
165 }
166 Ok((rules, total))
167}
168
169async fn find(
170 t: &Transport,
171 filter: &RuleFilter,
172 page: u32,
173 per_page: u32,
174 sort: &str,
175) -> Result<(Vec<Rule>, u64)> {
176 if filter.source != RuleSource::All {
177 t.require_feature(Feature::RuleSourceScoping).await?;
178 }
179 let mut path = format!("{BASE}/_find?page={page}&per_page={per_page}{sort}");
180 if let Some(kql) = filter.to_kql() {
181 path.push_str(&format!("&filter={}", urlencode(&kql)));
182 }
183
184 let body = t.get(&path).await?;
185 decode_find(&body)
186}
187
188pub fn decode_find(body: &Value) -> Result<(Vec<Rule>, u64)> {
191 let map = body
192 .as_object()
193 .ok_or_else(|| find_error("response", "must be a JSON object"))?;
194 let data = map
195 .get("data")
196 .and_then(Value::as_array)
197 .ok_or_else(|| find_error("data", "must be an array"))?;
198 let total = required_u64(map, "total")?;
199 let page = required_u64(map, "page")?;
200 if page == 0 {
201 return Err(find_error("page", "must be greater than zero"));
202 }
203 let per_page = required_u64(map, "perPage")?;
204 if per_page == 0 {
205 return Err(find_error("perPage", "must be greater than zero"));
206 }
207
208 let data_len = data.len() as u64;
209 if data_len > total {
210 return Err(find_error(
211 "data.len() > total",
212 format!("{data_len} returned records exceed total {total}"),
213 ));
214 }
215 if data_len > per_page {
216 return Err(find_error(
217 "data.len() > perPage",
218 format!("{data_len} returned records exceed perPage {per_page}"),
219 ));
220 }
221
222 let rules = data
223 .iter()
224 .cloned()
225 .map(Rule::from_value)
226 .collect::<Result<Vec<_>>>()?;
227 Ok((rules, total))
228}
229
230fn find_error(field: &str, detail: impl std::fmt::Display) -> Error {
231 Error::new(
232 ErrorKind::Http,
233 format!("decoding rule _find response field {field}: {detail}"),
234 )
235}
236
237fn required_u64(map: &serde_json::Map<String, Value>, field: &str) -> Result<u64> {
238 map.get(field)
239 .and_then(Value::as_u64)
240 .ok_or_else(|| find_error(field, "must be an unsigned integer"))
241}
242
243pub(crate) struct SourceTotals {
245 pub custom: u64,
246 pub prebuilt: u64,
247 pub all: u64,
248}
249
250impl SourceTotals {
251 fn is_exhaustive(&self) -> bool {
252 self.custom.checked_add(self.prebuilt) == Some(self.all)
253 }
254}
255
256pub(crate) async fn verify_source_partition(t: &Transport) -> Result<SourceTotals> {
263 let (_, custom) = find_page(
264 t,
265 &RuleFilter {
266 source: RuleSource::Custom,
267 ..Default::default()
268 },
269 1,
270 1,
271 )
272 .await?;
273 let (_, prebuilt) = find_page(
274 t,
275 &RuleFilter {
276 source: RuleSource::Prebuilt,
277 ..Default::default()
278 },
279 1,
280 1,
281 )
282 .await?;
283 let (_, all) = find_page(t, &RuleFilter::default(), 1, 1).await?;
284
285 let totals = SourceTotals {
286 custom,
287 prebuilt,
288 all,
289 };
290 if !totals.is_exhaustive() {
291 return Err(Error::new(
292 ErrorKind::Unsupported,
293 format!(
294 "the immutable source partition is not exhaustive: custom={}, \
295 prebuilt={}, all={}. The field may be absent on this stack; \
296 re-run with --source all to read the corpus.",
297 totals.custom, totals.prebuilt, totals.all
298 ),
299 ));
300 }
301
302 Ok(totals)
303}
304
305const RULE_TYPES: [&str; 7] = [
310 "query",
311 "eql",
312 "esql",
313 "threshold",
314 "threat_match",
315 "machine_learning",
316 "new_terms",
317];
318
319pub async fn find_all(t: &Transport, filter: &RuleFilter) -> Result<Vec<Rule>> {
328 let (rules, total) = find_page(t, filter, 1, RESULT_WINDOW).await?;
329
330 if total <= u64::from(RESULT_WINDOW) {
331 if (rules.len() as u64) < total {
332 return Err(short_read(total, rules.len()));
333 }
334 return Ok(rules);
335 }
336
337 let types: Vec<&str> = match &filter.rule_type {
340 Some(t) => vec![t.as_str()],
341 None => RULE_TYPES.to_vec(),
342 };
343
344 let mut collected: Vec<Rule> = Vec::new();
345 let mut summed: u64 = 0;
346
347 for rule_type in types {
348 let mut type_filter = filter.clone();
349 type_filter.rule_type = Some(rule_type.to_string());
350
351 let slice_total = if filter.rule_type.is_some() {
354 total
355 } else {
356 let (slice_rules, slice_total) = find_page(t, &type_filter, 1, RESULT_WINDOW).await?;
357
358 if slice_total <= u64::from(RESULT_WINDOW) {
359 if (slice_rules.len() as u64) < slice_total {
360 return Err(short_read(slice_total, slice_rules.len()));
361 }
362 summed += slice_total;
363 collected.extend(slice_rules);
364 continue;
365 }
366 slice_total
367 };
368
369 if filter.enabled.is_some() {
372 return Err(oversized(slice_total));
373 }
374
375 for enabled in [true, false] {
376 let mut enabled_filter = type_filter.clone();
377 enabled_filter.enabled = Some(enabled);
378 let (enabled_rules, enabled_total) =
379 find_page(t, &enabled_filter, 1, RESULT_WINDOW).await?;
380 if enabled_total > u64::from(RESULT_WINDOW) {
381 return Err(oversized(enabled_total));
382 }
383 if (enabled_rules.len() as u64) < enabled_total {
384 return Err(short_read(enabled_total, enabled_rules.len()));
385 }
386 summed += enabled_total;
387 collected.extend(enabled_rules);
388 }
389 }
390
391 if summed != total {
394 return Err(Error::new(
395 ErrorKind::Http,
396 format!(
397 "the server counted {total} rules across the corpus but the type slices \
398 sum to {summed}. Refusing a partial corpus: a rule type added by a newer \
399 stack version would otherwise read as zero in every pull and diff."
400 ),
401 ));
402 }
403
404 Ok(collected)
405}
406
407fn short_read(counted: u64, returned: usize) -> Error {
410 Error::new(
411 ErrorKind::Http,
412 format!(
413 "the server counted {counted} rules and returned {returned}. Refusing a partial \
414 corpus: a short read is indistinguishable from rules having been deleted."
415 ),
416 )
417}
418
419fn oversized(count: u64) -> Error {
422 Error::new(
423 ErrorKind::Unsupported,
424 format!(
425 "{count} rules match, more than the {RESULT_WINDOW} a single search can return \
426 even after partitioning by type and enabled. Narrow the selection with a \
427 filter or a tag."
428 ),
429 )
430}
431
432const ID_CHUNK: usize = 50;
437
438pub async fn find_by_rule_ids(t: &Transport, rule_ids: &[String]) -> Result<Vec<Rule>> {
443 let mut found = Vec::with_capacity(rule_ids.len());
444 for chunk in rule_ids.chunks(ID_CHUNK) {
445 let filter = RuleFilter {
446 query: Some(rule_id_query(chunk)),
447 ..Default::default()
448 };
449 let (rules, _) = find_page(t, &filter, 1, RESULT_WINDOW).await?;
450 found.extend(rules);
451 }
452 Ok(found)
453}
454
455pub async fn get(t: &Transport, rule_id: &str) -> Result<Rule> {
456 let body = t
457 .get(&format!("{BASE}?rule_id={}", urlencode(rule_id)))
458 .await?;
459 Rule::from_value(body)
460}
461
462pub async fn create(t: &Transport, rule: &Rule) -> Result<Rule> {
463 let mut payload = rule.clone();
464 normalize::strip_volatile(&mut payload);
465 let response = t
466 .post(BASE, Some(&Value::Object(payload.as_map().clone())))
467 .await?;
468 Rule::from_value(response)
469}
470
471pub async fn update(t: &Transport, rule: &Rule) -> Result<Rule> {
472 let mut payload = rule.clone();
473 normalize::strip_volatile(&mut payload);
474 let response = t
475 .put(BASE, &Value::Object(payload.as_map().clone()))
476 .await?;
477 Rule::from_value(response)
478}
479
480pub async fn patch(t: &Transport, rule_id: &str, patch: &Value) -> Result<Rule> {
481 let mut body = patch.as_object().cloned().unwrap_or_default();
482 body.insert("rule_id".into(), json!(rule_id));
483 let response = t.patch(BASE, &Value::Object(body)).await?;
485 Rule::from_value(response)
486}
487
488pub async fn delete(t: &Transport, rule_id: &str) -> Result<Rule> {
489 let body = t
490 .delete(&format!("{BASE}?rule_id={}", urlencode(rule_id)))
491 .await?;
492 Rule::from_value(body)
493}
494
495#[derive(Debug, Clone, Copy, PartialEq, Eq)]
496pub enum BulkAction {
497 Enable,
498 Disable,
499 Delete,
500}
501
502impl BulkAction {
503 pub fn as_str(&self) -> &'static str {
504 match self {
505 Self::Enable => "enable",
506 Self::Disable => "disable",
507 Self::Delete => "delete",
508 }
509 }
510}
511
512#[derive(Debug, Clone, PartialEq, Eq, Default)]
513pub struct BulkOutcome {
514 pub succeeded: u64,
515 pub failed: u64,
516 pub skipped: u64,
517 pub total: u64,
518}
519
520pub fn decode_bulk_outcome(body: &Value) -> Result<BulkOutcome> {
525 let map = body
526 .as_object()
527 .ok_or_else(|| bulk_error("response", "must be a JSON object"))?;
528 let summary = map
529 .get("attributes")
530 .and_then(|attributes| attributes.get("summary"))
531 .and_then(Value::as_object)
532 .ok_or_else(|| bulk_error("attributes.summary", "must be an object"))?;
533 let succeeded = bulk_u64(summary, "succeeded")?;
534 let failed = bulk_u64(summary, "failed")?;
535 let skipped = bulk_u64(summary, "skipped")?;
536 let total = bulk_u64(summary, "total")?;
537 let applied = succeeded
538 .checked_add(failed)
539 .and_then(|count| count.checked_add(skipped))
540 .ok_or_else(|| bulk_error("total", "counter sum overflows u64"))?;
541 if applied != total {
542 return Err(bulk_error(
543 "total",
544 format!(
545 "{total} does not equal succeeded {succeeded} + failed {failed} + skipped {skipped}"
546 ),
547 ));
548 }
549 Ok(BulkOutcome {
550 succeeded,
551 failed,
552 skipped,
553 total,
554 })
555}
556
557fn bulk_error(field: &str, detail: impl std::fmt::Display) -> Error {
558 Error::new(
559 ErrorKind::Http,
560 format!("decoding rule _bulk_action response field {field}: {detail}"),
561 )
562}
563
564fn bulk_u64(map: &serde_json::Map<String, Value>, field: &str) -> Result<u64> {
565 map.get(field)
566 .and_then(Value::as_u64)
567 .ok_or_else(|| bulk_error(field, "must be an unsigned integer"))
568}
569
570pub async fn bulk_by_rule_ids(
571 t: &Transport,
572 action: BulkAction,
573 rule_ids: &[String],
574 dry_run: bool,
575) -> Result<BulkOutcome> {
576 if rule_ids.is_empty() {
578 return Ok(BulkOutcome::default());
579 }
580
581 let path = if dry_run {
582 format!("{BASE}/_bulk_action?dry_run=true")
583 } else {
584 format!("{BASE}/_bulk_action")
585 };
586 let body = json!({ "action": action.as_str(), "query": rule_id_query(rule_ids) });
587
588 let response = t.post(&path, Some(&body)).await?;
589 decode_bulk_outcome(&response)
590}
591
592pub async fn export(t: &Transport, rule_ids: Option<&[String]>) -> Result<Bundle> {
601 let body = rule_ids.map(|ids| {
602 json!({
603 "objects": ids
604 .iter()
605 .map(|id| json!({"rule_id": id}))
606 .collect::<Vec<_>>()
607 })
608 });
609 let text = t
610 .post_text(&format!("{BASE}/_export"), body.as_ref())
611 .await?;
612 codec::decode_bundle(&text)
613}
614
615const EXISTENCE_CHUNK: usize = 50;
618
619pub async fn existing_rule_ids(
624 t: &Transport,
625 rule_ids: &[String],
626) -> Result<std::collections::BTreeSet<String>> {
627 let mut found = std::collections::BTreeSet::new();
629 if rule_ids.is_empty() {
630 return Ok(found);
631 }
632
633 for chunk in rule_ids.chunks(EXISTENCE_CHUNK) {
634 let path = format!(
635 "{BASE}/_find?page=1&per_page={}&filter={}",
636 chunk.len(),
637 urlencode(&rule_id_query(chunk))
638 );
639 let (rules, _) = decode_find(&t.get(&path).await?)?;
640 for r in rules {
641 if let Ok(id) = r.rule_id() {
642 found.insert(id.to_string());
643 }
644 }
645 }
646
647 Ok(found)
648}
649
650pub async fn import(t: &Transport, ndjson: &str, overwrite: bool) -> Result<Value> {
651 t.post_multipart_ndjson(&format!("{BASE}/_import?overwrite={overwrite}"), ndjson)
652 .await
653}
654
655#[cfg(test)]
656mod tests {
657 use super::*;
658
659 #[test]
661 fn each_source_maps_to_its_measured_filter() {
662 assert_eq!(
663 RuleSource::Custom.clause(),
664 Some("alert.attributes.params.immutable: false")
665 );
666 assert_eq!(
667 RuleSource::Customized.clause(),
668 Some("alert.attributes.params.ruleSource.isCustomized: true")
669 );
670 assert_eq!(RuleSource::All.clause(), None, "all adds no clause");
671 }
672
673 #[test]
674 fn a_source_clause_combines_with_other_filters() {
675 let f = RuleFilter {
676 source: RuleSource::Custom,
677 tag: Some("prod".into()),
678 ..Default::default()
679 };
680 let kql = f.to_kql().unwrap();
681 assert!(kql.contains("immutable: false"), "{kql}");
682 assert!(kql.contains("prod"), "{kql}");
683 assert!(
684 kql.contains(" AND "),
685 "clauses combine, they do not replace: {kql}"
686 );
687 }
688
689 #[test]
690 fn kql_escape_doubles_a_lone_backslash() {
691 assert_eq!(kql_escape("a\\b"), "a\\\\b");
692 }
693
694 #[test]
695 fn kql_escape_escapes_a_lone_quote() {
696 let mut expected = String::from("a");
697 expected.push('\\');
698 expected.push('"');
699 expected.push('b');
700 assert_eq!(kql_escape("a\"b"), expected);
701 }
702
703 #[test]
706 fn kql_escape_wildcard_escapes_kql_wildcards_after_kql_escape() {
707 assert_eq!(kql_escape_wildcard("*"), "\\*");
708 assert_eq!(kql_escape_wildcard("?"), "\\?");
709 assert_eq!(kql_escape_wildcard("a\\b\"c"), kql_escape("a\\b\"c"));
710 }
711
712 #[test]
715 fn kql_escape_orders_backslash_before_quote() {
716 let mut input = String::new();
717 input.push('\\');
718 input.push('"');
719
720 let escaped = kql_escape(&input);
721
722 let mut expected = "\\".repeat(3);
725 expected.push('"');
726 assert_eq!(escaped, expected);
727 }
728
729 #[test]
730 fn rule_id_query_escapes_a_quote_in_the_id() {
731 let mut id = String::from("x");
732 id.push('"');
733 id.push('y');
734
735 let q = rule_id_query(&[id]);
736
737 let mut expected = String::from("alert.attributes.params.ruleId: \"x");
738 expected.push('\\');
739 expected.push('"');
740 expected.push_str("y\"");
741 assert_eq!(q, expected);
742 assert!(
743 !q.contains(" OR "),
744 "a single id must produce exactly one clause: {q}"
745 );
746 }
747
748 #[test]
749 fn rule_id_query_neutralizes_a_kql_injection_payload() {
750 let payload = "x\" or alert.attributes.enabled: true or \"";
751 let q = rule_id_query(&[payload.to_string()]);
752 assert!(
753 !q.contains("\" or alert.attributes.enabled: true or \""),
754 "the injected quote must not close the literal: {q}"
755 );
756 }
757
758 #[test]
759 fn to_kql_escapes_a_quote_in_the_tag() {
760 let f = RuleFilter {
761 tag: Some("a\"b".into()),
762 ..Default::default()
763 };
764 let kql = f.to_kql().unwrap();
765
766 let mut expected = String::from("alert.attributes.tags: \"a");
767 expected.push('\\');
768 expected.push('"');
769 expected.push_str("b\"");
770 assert_eq!(kql, expected);
771 }
772}