1use elasticctl_core::{Error, ErrorKind, Result, Transport};
2use serde_json::{Value, json};
3
4pub const SEARCH_PATH: &str = "/api/detection_engine/signals/search";
5pub const STATUS_PATH: &str = "/api/detection_engine/signals/status";
6pub const TAGS_PATH: &str = "/api/detection_engine/signals/tags";
7pub const ASSIGNEES_PATH: &str = "/api/detection_engine/signals/assignees";
8
9#[derive(Debug, Clone, Copy, PartialEq, Eq)]
10pub enum AlertStatus {
11 Open,
12 Acknowledged,
13 Closed,
14}
15
16impl AlertStatus {
17 pub fn as_str(self) -> &'static str {
18 match self {
19 AlertStatus::Open => "open",
20 AlertStatus::Acknowledged => "acknowledged",
21 AlertStatus::Closed => "closed",
22 }
23 }
24
25 pub fn verb(self) -> &'static str {
26 match self {
27 AlertStatus::Open => "Open",
28 AlertStatus::Acknowledged => "Acknowledge",
29 AlertStatus::Closed => "Close",
30 }
31 }
32
33 pub fn parse(s: &str) -> Result<AlertStatus> {
34 match s {
35 "open" => Ok(AlertStatus::Open),
36 "acknowledged" => Ok(AlertStatus::Acknowledged),
37 "closed" => Ok(AlertStatus::Closed),
38 other => Err(Error::new(
39 ErrorKind::Error,
40 format!("unknown alert status '{other}': expected open, acknowledged, or closed"),
41 )),
42 }
43 }
44}
45
46#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
47pub enum Conflicts {
48 #[default]
49 Abort,
50 Proceed,
51}
52
53impl Conflicts {
54 pub fn as_str(self) -> &'static str {
55 match self {
56 Conflicts::Abort => "abort",
57 Conflicts::Proceed => "proceed",
58 }
59 }
60
61 pub fn parse(s: &str) -> Result<Conflicts> {
62 match s {
63 "abort" => Ok(Conflicts::Abort),
64 "proceed" => Ok(Conflicts::Proceed),
65 other => Err(Error::new(
66 ErrorKind::Error,
67 format!("unknown conflicts mode '{other}': expected abort or proceed"),
68 )),
69 }
70 }
71}
72
73#[derive(Debug, Clone, PartialEq)]
74pub struct AlertHit {
75 pub id: String,
76 pub index: Option<String>,
77 pub source: Value,
78 pub sort: Option<Vec<Value>>,
79}
80
81#[derive(Debug, Clone, PartialEq)]
82pub struct AlertPage {
83 pub hits: Vec<AlertHit>,
84 pub total: Option<u64>,
85}
86
87pub fn decode_page(value: &Value) -> Result<AlertPage> {
88 let hits = value
89 .pointer("/hits/hits")
90 .and_then(Value::as_array)
91 .ok_or_else(|| {
92 Error::new(
93 ErrorKind::Http,
94 "decoding alerts response field `hits.hits`",
95 )
96 })?;
97 let mut out = Vec::with_capacity(hits.len());
98 for hit in hits {
99 let id = hit
100 .get("_id")
101 .and_then(Value::as_str)
102 .ok_or_else(|| Error::new(ErrorKind::Http, "decoding alert hit field `_id`"))?
103 .to_string();
104 let index = hit.get("_index").and_then(Value::as_str).map(str::to_owned);
105 let source = hit
106 .get("_source")
107 .filter(|s| s.is_object())
108 .cloned()
109 .ok_or_else(|| Error::new(ErrorKind::Http, "decoding alert hit field `_source`"))?;
110 let sort = hit.get("sort").and_then(Value::as_array).cloned();
111 out.push(AlertHit {
112 id,
113 index,
114 source,
115 sort,
116 });
117 }
118 let total = value.pointer("/hits/total/value").and_then(Value::as_u64);
119 Ok(AlertPage { hits: out, total })
120}
121
122pub async fn search(t: &Transport, body: &Value) -> Result<AlertPage> {
123 decode_page(&t.post(SEARCH_PATH, Some(body)).await?)
124}
125
126pub async fn search_all(
127 t: &Transport,
128 query: &Value,
129 sort: &Value,
130 limit: Option<usize>,
131) -> Result<Vec<AlertHit>> {
132 search_all_with_page_size(t, query, sort, limit, 1000).await
133}
134
135pub async fn search_all_with_page_size(
136 t: &Transport,
137 query: &Value,
138 sort: &Value,
139 limit: Option<usize>,
140 page_size: usize,
141) -> Result<Vec<AlertHit>> {
142 let mut all = Vec::new();
143 let mut search_after: Option<Vec<Value>> = None;
144 loop {
145 let mut body = json!({
146 "query": query,
147 "sort": sort,
148 "size": page_size,
149 });
150 if let Some(sa) = &search_after {
151 body["search_after"] = json!(sa);
152 }
153 let page = search(t, &body).await?;
154 let short_page = page.hits.len() < page_size;
155 let last_sort = page.hits.last().and_then(|h| h.sort.clone());
156 all.extend(page.hits);
157 if let Some(limit) = limit
158 && all.len() >= limit
159 {
160 all.truncate(limit);
161 return Ok(all);
162 }
163 if short_page {
164 return Ok(all);
165 }
166 match last_sort {
167 Some(sa) => search_after = Some(sa),
168 None => return Ok(all),
169 }
170 }
171}
172
173#[derive(Debug, Clone, PartialEq, serde::Serialize)]
174pub struct SignalsOutcome {
175 pub total: u64,
176 pub updated: u64,
177 pub version_conflicts: u64,
178 pub noops: u64,
179 pub failures: Vec<Value>,
180}
181
182pub fn decode_outcome(value: &Value) -> Result<SignalsOutcome> {
183 let counter = |name: &str| {
184 value.get(name).and_then(Value::as_u64).ok_or_else(|| {
185 Error::new(
186 ErrorKind::Http,
187 format!("decoding signals outcome field `{name}`"),
188 )
189 })
190 };
191 Ok(SignalsOutcome {
192 total: counter("total")?,
193 updated: counter("updated")?,
194 version_conflicts: counter("version_conflicts")?,
195 noops: counter("noops")?,
196 failures: value
197 .get("failures")
198 .and_then(Value::as_array)
199 .cloned()
200 .ok_or_else(|| {
201 Error::new(ErrorKind::Http, "decoding signals outcome field `failures`")
202 })?,
203 })
204}
205
206pub async fn status_by_ids(
207 t: &Transport,
208 ids: &[String],
209 status: AlertStatus,
210 reason: Option<&str>,
211) -> Result<SignalsOutcome> {
212 let mut body = json!({ "signal_ids": ids, "status": status.as_str() });
213 if let Some(r) = reason {
214 body["reason"] = json!(r);
215 }
216 decode_outcome(&t.post(STATUS_PATH, Some(&body)).await?)
217}
218
219pub async fn status_by_query(
220 t: &Transport,
221 query: &Value,
222 status: AlertStatus,
223 conflicts: Conflicts,
224 reason: Option<&str>,
225) -> Result<SignalsOutcome> {
226 let mut body = json!({
227 "query": query,
228 "status": status.as_str(),
229 "conflicts": conflicts.as_str(),
230 });
231 if let Some(r) = reason {
232 body["reason"] = json!(r);
233 }
234 decode_outcome(&t.post(STATUS_PATH, Some(&body)).await?)
235}
236
237pub async fn set_tags(
238 t: &Transport,
239 ids: &[String],
240 add: &[String],
241 remove: &[String],
242) -> Result<SignalsOutcome> {
243 let body = json!({
244 "ids": ids,
245 "tags": { "tags_to_add": add, "tags_to_remove": remove },
246 });
247 decode_outcome(&t.post(TAGS_PATH, Some(&body)).await?)
248}
249
250pub async fn set_assignees(
251 t: &Transport,
252 ids: &[String],
253 add: &[String],
254 remove: &[String],
255) -> Result<SignalsOutcome> {
256 let body = json!({
257 "ids": ids,
258 "assignees": { "add": add, "remove": remove },
259 });
260 decode_outcome(&t.post(ASSIGNEES_PATH, Some(&body)).await?)
261}
262
263#[derive(Debug)]
264pub struct ActivityPage {
265 pub total: u64,
266 pub buckets: Vec<(String, u64)>,
267 pub other: u64,
268 pub hits: Vec<AlertHit>,
269}
270pub async fn search_activity(t: &Transport, body: &Value) -> Result<ActivityPage> {
271 let v = t.post(SEARCH_PATH, Some(body)).await?;
272 let invalid = || Error::new(ErrorKind::Http, "invalid alert activity response");
273 if v.pointer("/hits/total/relation").and_then(Value::as_str) != Some("eq") {
274 return Err(invalid());
275 }
276 let total = v
277 .pointer("/hits/total/value")
278 .and_then(Value::as_u64)
279 .ok_or_else(invalid)?;
280 let status = v.pointer("/aggregations/status").ok_or_else(invalid)?;
281 let buckets = status
282 .get("buckets")
283 .and_then(Value::as_array)
284 .ok_or_else(invalid)?
285 .iter()
286 .map(|b| {
287 Ok((
288 b.get("key")
289 .and_then(Value::as_str)
290 .ok_or_else(invalid)?
291 .to_owned(),
292 b.get("doc_count")
293 .and_then(Value::as_u64)
294 .ok_or_else(invalid)?,
295 ))
296 })
297 .collect::<Result<_>>()?;
298 let other = status
299 .get("sum_other_doc_count")
300 .and_then(Value::as_u64)
301 .ok_or_else(invalid)?;
302 Ok(ActivityPage {
303 total,
304 buckets,
305 other,
306 hits: decode_page(&v)?.hits,
307 })
308}