Skip to main content

elasticctl_api/
rules.rs

1//! Typed wrappers for the detection-engine API.
2//!
3//! Functions use stable `rule_id` values, not volatile server-side `id` values.
4
5use 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
14/// Elasticsearch's `_find` result window. `from + size` cannot exceed this;
15/// 10,001 returns a 400 error.
16///
17/// It bounds `per_page` and the largest `_find` result. Smaller pages cannot
18/// evade the `from + size` limit. A window-sized request read 2,066 rules in
19/// 2.4 seconds; 21 pages of 100 took 8.4–11 seconds.
20const RESULT_WINDOW: u32 = 10_000;
21
22/// Which rules an operation acts on, grouped by source.
23///
24/// The server-side split filters on `alert.attributes.params.immutable`. On
25/// Serverless 9.6.0, it agreed exactly with `params.ruleSource.type` (2,066
26/// prebuilt / 0 custom). Its presence on older versions is unmeasured.
27/// `customized` narrows the prebuilt set to rules edited on the stack.
28#[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    /// The measured server-side filter clause. `None` means no clause: the
40    /// whole corpus. Spec 5.5.
41    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    /// Exact display name, filtered server-side. This takes one request;
59    /// walking 2,066 rules took 8.8 seconds.
60    pub name: Option<String>,
61    /// A raw KQL fragment, combined with the structured filters above.
62    pub query: Option<String>,
63    /// Friendly name-substring plus exact-tag search (spec 4.7). Distinct from
64    /// `name` (exact) and `query` (raw KQL).
65    pub search: Option<String>,
66}
67
68impl RuleFilter {
69    /// Kibana filters saved objects with KQL over `alert.attributes.*`.
70    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
110/// Escape a value for use inside a double-quoted KQL literal.
111///
112/// A quote could otherwise close the literal and make the remaining value KQL,
113/// turning a scoped bulk action into an unscoped action. Escape backslashes
114/// first to avoid double-escaping inserted quote escapes. Shared with the
115/// exceptions vertical: the two filter builders must not diverge, because a
116/// divergence silently matches the wrong objects.
117pub(crate) fn kql_escape(value: &str) -> String {
118    value.replace('\\', "\\\\").replace('"', "\\\"")
119}
120
121/// Escape a value for a wildcard-wrapped KQL substring term (`*<value>*`).
122///
123/// Builds on `kql_escape`, then escapes the KQL wildcards `*` and `?` so a
124/// search text cannot inject its own wildcards and widen the match. Shared
125/// with the exceptions vertical: the two filter builders must not diverge,
126/// because a divergence silently matches the wrong objects.
127pub(crate) fn kql_escape_wildcard(value: &str) -> String {
128    kql_escape(value).replace('*', "\\*").replace('?', "\\?")
129}
130
131/// KQL selecting exactly the given stable rule ids.
132fn 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
158/// Decode a `_find` response into rules and a total. Fixtures use this same
159/// path offline.
160pub 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
213/// The three totals that prove `immutable` divides the complete rule corpus.
214pub(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
226/// Verify that the custom and prebuilt `immutable` filters are disjoint and
227/// exhaustive. `customized` is intentionally absent: it overlaps prebuilt.
228///
229/// This runs only after an otherwise unselected custom or prebuilt read found
230/// no rules. A valid empty slice must still account for the entire corpus with
231/// its opposite source slice.
232pub(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
275/// Detection-rule types used to partition a corpus. Each rule has one
276/// `params.type`, so type slices are disjoint and exhaustive. Tags are not:
277/// a rule can have many tags or none. Measured against 2,066 rules, these seven
278/// type slices sum exactly to the corpus.
279const RULE_TYPES: [&str; 7] = [
280    "query",
281    "eql",
282    "esql",
283    "threshold",
284    "threat_match",
285    "machine_learning",
286    "new_terms",
287];
288
289/// Every rule matching the filter.
290///
291/// Read corpora within the result window in one request. For larger corpora,
292/// partition by rule type, then by `enabled` when needed. Verify the partition
293/// by summing slice totals to the corpus total.
294///
295/// Never return a partial corpus. `state diff` would report unread rules as
296/// locally added, and `state pull` would silently omit them.
297pub 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    // A caller that filtered by type has one requested slice. Only `enabled`
308    // can subdivide it further.
309    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        // The opening request already fetched this selected type. Repeating it
322        // would return the same oversized result before the `enabled` split.
323        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        // A caller that filtered by `enabled` leaves no further partition.
340        // Refuse the slice rather than truncating it to 10,000 rules.
341        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    // The sum verifies exhaustiveness. A newer rule type would otherwise read
362    // as zero, disappear from pulls, and appear remote-only in diffs.
363    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
377/// A server that counts more rules than it serves contradicts itself. A short
378/// list is indistinguishable from rules deleted between count and read.
379fn 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
389/// A slice that exceeds the window after both partitions cannot be served.
390/// Returning its first 10,000 rules would make unread rules look remote-only.
391fn 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
402/// How many `rule_id`s go into one filtered `_find`.
403///
404/// A KQL disjunction grows with the selection, and `--tag` can select
405/// thousands of rules. Chunking keeps each URL below practical limits.
406const ID_CHUNK: usize = 50;
407
408/// The rules carrying exactly these `rule_id`s.
409///
410/// IDs absent from the stack do not return. That is expected for locally added
411/// rules that `state push` creates.
412pub 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    // PATCH accepts `rule_id` directly, avoiding the volatile server `id`.
454    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
490/// Decode a `_bulk_action` response summary, refusing a malformed success body.
491///
492/// A missing or mistyped counter, or a total that contradicts the per-outcome
493/// counters, must fail rather than read as "nothing happened".
494pub 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    // An empty selection must not become an unscoped query for every rule.
547    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
562/// Export every rule, or exactly the named ones.
563///
564/// `None` posts no body for a whole-space export. `Some(ids)` posts `objects`
565/// so a subset export transfers only the selected rules.
566///
567/// The response is a bundle: rules, exception-list containers, exception items,
568/// and a trailer. `_export` appends the exception objects a rule references, so
569/// decoding them to rules only would drop that content.
570pub 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
585/// IDs to check in one `_find`. This keeps large requests below proxy URL
586/// limits while a 40-rule corpus still uses one request.
587const EXISTENCE_CHUNK: usize = 50;
588
589/// Which of these rule ids already exist on the stack.
590///
591/// Check file IDs before upload so existing rules can be skipped. This makes
592/// imports idempotent and lets dry runs distinguish skipped rules.
593pub async fn existing_rule_ids(
594    t: &Transport,
595    rule_ids: &[String],
596) -> Result<std::collections::BTreeSet<String>> {
597    // An empty list must not become an unscoped find for every rule.
598    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
632/// Run a rule against historical data without writing alerts.
633///
634/// The API requires `invocationCount` and `timeframeEnd`. The `logs` array has
635/// errors and warnings for each simulated invocation.
636pub 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    // The preview API rejects fields that identify a saved rule.
644    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
681/// Where a preview's alerts land. Kibana names the alias per space.
682///
683/// The `rules_preview_hits` fixtures record this alias and its filter field.
684/// A response alone cannot distinguish an empty result from a wrong field.
685pub 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    /// One entry for each returned document: `{"_id": ..., "_source": {...}}`.
691    /// Contains the matching alert document, not a projected subset.
692    pub sample: Vec<Value>,
693}
694
695/// Read back what a preview matched.
696///
697/// `rules/preview` returns a `previewId` but no hit count, so search the alerts
698/// it wrote. `ignore_unavailable=true` reports no preview index as zero hits,
699/// not a 404 error.
700pub 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        // Count all matches, not only the default 10,000.
711        "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
723/// Decode a preview-hits response, refusing a malformed success body.
724pub 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
760/// Decode a preview-hits response, tolerating a malformed body as empty.
761///
762/// Fixtures and offline callers use this path; the live `preview_hits` uses the
763/// checked decoder so a malformed success body fails instead of reading as zero.
764pub 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    /// Spec 5.5, measured 2026-08-14.
773    #[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    /// The wildcard pass runs after `kql_escape`, escaping `*` and `?` without
817    /// undoing the backslash and quote escaping.
818    #[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    /// A backslash before a quote tests escape order. Escaping quotes first
826    /// would double the backslash inserted for the quote and corrupt the value.
827    #[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        // Double the backslash, then escape the quote: three backslashes and a
836        // quote.
837        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}