elasticctl_api/fleet/
agent_policy_ops.rs

1//! Agent-policy selection, normalization, portability, planning, and apply.
2
3use crate::content_codec::{self, ContentFormat};
4use crate::fleet::agent_policies::{
5    self, AGENTLESS_FIELD, AgentPolicyDetail, AgentPolicySpec, AgentPolicySummary, ENVIRONMENT_IDS,
6    PLATFORM_FLAGS,
7};
8use crate::ops::{ExportOutcome, MutationPlan};
9use elasticctl_core::{Error, ErrorKind, Result, Transport};
10use serde::Serialize;
11use serde_json::{Map, Value, json};
12use std::collections::{BTreeMap, BTreeSet};
13use std::path::Path;
14
15mod artifact;
16mod delete;
17mod import;
18pub use artifact::{export, validate};
19pub use delete::{
20    AgentPolicyDeletePlan, AgentPolicyDeleteReport, AgentPolicyDeleteTarget, apply_delete,
21    plan_delete,
22};
23pub use import::{
24    AgentPolicyImportPlan, AgentPolicyImportReport, apply_import, build_replace_body, plan_import,
25};
26
27const PAGE_SIZE: u64 = 1000;
28
29#[derive(Debug, Clone, Default, PartialEq, Eq)]
30pub struct AgentPolicyFilter {
31    pub search: Option<String>,
32    pub limit: Option<usize>,
33}
34
35#[derive(Debug, Clone, PartialEq, Serialize)]
36pub struct AgentPolicyList {
37    pub total: u64,
38    pub agent_policies: Vec<AgentPolicySummary>,
39    pub truncated: bool,
40}
41
42#[derive(Debug, Clone, PartialEq)]
43struct ResolvedAgentPolicy {
44    summary: AgentPolicySummary,
45    item: Map<String, Value>,
46}
47
48/// A single live read reduced to what planning needs.
49#[derive(Debug, Clone, PartialEq)]
50pub struct LiveAgentPolicy {
51    pub spec: AgentPolicySpec,
52    pub agents: u64,
53    pub attached: Vec<String>,
54}
55
56/// The narrow parent facts an integration-policy operation needs to compare
57/// attachment, namespace, ownership, and blast radius. This deliberately does
58/// not apply agent-policy portability checks: environment references are not
59/// integration-parent facts.
60#[derive(Debug, Clone, PartialEq, Eq)]
61pub(crate) struct AgentPolicyParentSnapshot {
62    pub id: String,
63    pub name: String,
64    pub namespace: String,
65    pub agents: u64,
66    pub attached_integrations: Vec<String>,
67    pub platform_owned: bool,
68    pub protected: bool,
69}
70
71/// Collect every page in the measured deterministic order, then sort by id.
72pub async fn collect(transport: &Transport) -> Result<Vec<Map<String, Value>>> {
73    let mut page_number = 1;
74    let mut total = None;
75    let mut items = Vec::new();
76    let mut ids = BTreeSet::new();
77    loop {
78        let page = agent_policies::list_page(transport, page_number).await?;
79        if page.page != page_number || page.per_page != PAGE_SIZE {
80            return Err(http(
81                "decoding agent policies list: unexpected page metadata",
82            ));
83        }
84        match total {
85            Some(total) if total != page.total => {
86                return Err(http(
87                    "decoding agent policies list: total changed while paging",
88                ));
89            }
90            Some(_) => {}
91            None => total = Some(page.total),
92        }
93        let page_len = page.items.len() as u64;
94        for item in page.items {
95            let id = item
96                .get("id")
97                .and_then(Value::as_str)
98                .filter(|id| !id.is_empty())
99                .ok_or_else(|| http("decoding agent policies list: item without id"))?
100                .to_owned();
101            if !ids.insert(id.clone()) {
102                return Err(http(format!(
103                    "decoding agent policies list: duplicate agent policy id '{id}'"
104                )));
105            }
106            items.push(item);
107        }
108        let expected = total.expect("set from the first page");
109        if items.len() as u64 >= expected {
110            break;
111        }
112        if page_len != PAGE_SIZE {
113            return Err(http(
114                "decoding agent policies list: page was short before total",
115            ));
116        }
117        page_number += 1;
118    }
119    if items.len() as u64 > total.unwrap_or(0) {
120        return Err(http(
121            "decoding agent policies list: returned more items than total",
122        ));
123    }
124    items.sort_by(|left, right| left["id"].as_str().cmp(&right["id"].as_str()));
125    Ok(items)
126}
127
128pub async fn list_op(transport: &Transport, filter: &AgentPolicyFilter) -> Result<AgentPolicyList> {
129    let items = collect(transport).await?;
130    let total = items.len() as u64;
131    let needle = filter.search.as_ref().map(|search| search.to_lowercase());
132    let mut rows = Vec::new();
133    for item in &items {
134        let summary = AgentPolicySummary::from_item(item)?;
135        let keep = needle.as_ref().is_none_or(|needle| {
136            summary.id.to_lowercase().contains(needle)
137                || summary.name.to_lowercase().contains(needle)
138        });
139        if keep {
140            rows.push(summary);
141        }
142    }
143    let limit = filter.limit.unwrap_or(usize::MAX);
144    let truncated = rows.len() > limit;
145    rows.truncate(limit);
146    Ok(AgentPolicyList {
147        total,
148        agent_policies: rows,
149        truncated,
150    })
151}
152
153/// Stable id first through the single-object route; exact name second.
154pub async fn resolve(transport: &Transport, selector: &str) -> Result<AgentPolicySummary> {
155    Ok(resolve_item(transport, selector).await?.summary)
156}
157
158/// Resolve a selector and retain the single-object response needed by callers
159/// that require populated agent and integration facts.
160async fn resolve_item(transport: &Transport, selector: &str) -> Result<ResolvedAgentPolicy> {
161    match agent_policies::get(transport, selector).await {
162        Ok(policy) => {
163            return Ok(ResolvedAgentPolicy {
164                summary: AgentPolicySummary::from_item(&policy.item)?,
165                item: policy.item,
166            });
167        }
168        Err(error) if error.kind == ErrorKind::NotFound => {}
169        Err(error) => return Err(error),
170    }
171    let items = collect(transport).await?;
172    let matches: Vec<AgentPolicySummary> = items
173        .iter()
174        .filter(|item| item.get("name").and_then(Value::as_str) == Some(selector))
175        .map(AgentPolicySummary::from_item)
176        .collect::<Result<_>>()?;
177    match matches.as_slice() {
178        [] => Err(Error::new(
179            ErrorKind::NotFound,
180            format!("no agent policy with id or name '{selector}'"),
181        )),
182        [one] => {
183            let policy = agent_policies::get(transport, &one.id).await?;
184            Ok(ResolvedAgentPolicy {
185                summary: one.clone(),
186                item: policy.item,
187            })
188        }
189        many => Err(Error::new(
190            ErrorKind::Conflict,
191            format!(
192                "agent policy '{selector}' is ambiguous: {}",
193                many.iter()
194                    .map(|row| row.id.as_str())
195                    .collect::<Vec<_>>()
196                    .join(", ")
197            ),
198        )),
199    }
200}
201
202pub async fn get_op(transport: &Transport, selector: &str) -> Result<AgentPolicyDetail> {
203    let ResolvedAgentPolicy { summary, item } = resolve_item(transport, selector).await?;
204    let agents = required_agents(&item, &summary.id)?;
205    let attached_integrations = attached_integration_ids(&item, &summary.id)?;
206    let status = match item.get("status") {
207        None | Some(Value::Null) => None,
208        Some(Value::String(status)) => Some(status.clone()),
209        Some(_) => {
210            return Err(http(format!(
211                "decoding agent policy '{}': status must be a string or null",
212                summary.id
213            )));
214        }
215    };
216    let blocked_by = portability_reasons(&item, transport.space())?
217        .into_iter()
218        .map(str::to_owned)
219        .collect();
220    Ok(AgentPolicyDetail {
221        id: summary.id,
222        name: summary.name,
223        namespace: summary.namespace,
224        description: summary.description,
225        agents,
226        status,
227        attached_integrations,
228        blocked_by,
229    })
230}
231
232/// True when a boolean platform flag is true or `agentless` is non-null.
233/// Used only by `--all-custom` filtering;
234/// `normalize` still refuses these policies when selected explicitly.
235pub fn is_platform_owned(item: &Map<String, Value>) -> Result<bool> {
236    for flag in PLATFORM_FLAGS {
237        if optional_server_bool(item, flag)? == Some(true) {
238            return Ok(true);
239        }
240    }
241    match item.get(AGENTLESS_FIELD) {
242        None | Some(Value::Null) => Ok(false),
243        Some(Value::Object(_)) => Ok(true),
244        Some(_) => Err(http(
245            "decoding agent policy: agentless must be an object or null",
246        )),
247    }
248}
249
250/// Read one agent-policy parent while retaining only integration-operation
251/// facts. Missing `agents` is a Fleet privilege error; all other malformed
252/// known parent fields are malformed HTTP responses.
253pub(crate) async fn read_parent_snapshot(
254    transport: &Transport,
255    id: &str,
256) -> Result<AgentPolicyParentSnapshot> {
257    let policy = agent_policies::get(transport, id).await?;
258    let item = &policy.item;
259    let returned_id = item
260        .get("id")
261        .and_then(Value::as_str)
262        .filter(|value| !value.trim().is_empty())
263        .ok_or_else(|| http("decoding agent policy parent: id must be a non-empty string"))?;
264    if returned_id != id {
265        return Err(http(format!(
266            "decoding agent policy parent: expected id '{id}', got '{returned_id}'"
267        )));
268    }
269    let name = item
270        .get("name")
271        .and_then(Value::as_str)
272        .filter(|value| !value.trim().is_empty())
273        .ok_or_else(|| {
274            http(format!(
275                "decoding agent policy '{id}': name must be a non-empty string"
276            ))
277        })?
278        .to_owned();
279    let namespace = item
280        .get("namespace")
281        .and_then(Value::as_str)
282        .filter(|value| !value.trim().is_empty())
283        .ok_or_else(|| {
284            http(format!(
285                "decoding agent policy '{id}': namespace must be a non-empty string"
286            ))
287        })?
288        .to_owned();
289    Ok(AgentPolicyParentSnapshot {
290        id: returned_id.to_owned(),
291        name,
292        namespace,
293        agents: required_agents(item, id)?,
294        attached_integrations: attached_integration_ids(item, id)?,
295        platform_owned: is_platform_owned(item)?,
296        protected: optional_server_bool(item, "is_protected")?.unwrap_or(false),
297    })
298}
299
300const PORTABLE_OPTIONAL: [&str; 13] = [
301    "description",
302    "inactivity_timeout",
303    "unenroll_timeout",
304    "monitoring_enabled",
305    "agent_features",
306    "global_data_tags",
307    "advanced_settings",
308    "overrides",
309    "keep_monitoring_alive",
310    "monitoring_pprof_enabled",
311    "monitoring_http",
312    "monitoring_diagnostics",
313    "namespace",
314];
315
316/// Live top-level fields normalization removes or refuses as server-owned or
317/// derived, per spec 5.2: audit and saved-object identity, agent and
318/// version-condition facts, populated `package_policies`, the platform and
319/// portability-refusal fields, and the active space's `space_ids`. Sorted.
320/// A live field outside this list and the portable set is `unsupported`.
321const REMOVED_FIELDS: [&str; 31] = [
322    "agentless",
323    "agents",
324    "agents_per_version",
325    "created_at",
326    "created_by",
327    "data_output_id",
328    "download_source_id",
329    "fips_agents",
330    "fleet_server_host_id",
331    "has_agent_version_conditions",
332    "has_fleet_server",
333    "is_default",
334    "is_default_fleet_server",
335    "is_managed",
336    "is_preconfigured",
337    "is_protected",
338    "is_verifier",
339    "min_agent_version",
340    "monitoring_output_id",
341    "package_agent_version_conditions",
342    "package_policies",
343    "required_versions",
344    "revision",
345    "schema_version",
346    "space_ids",
347    "status",
348    "supports_agentless",
349    "unprivileged_agents",
350    "updated_at",
351    "updated_by",
352    "version",
353];
354
355/// Convert a live policy into its filled portable form, or refuse it.
356pub fn normalize(item: &Map<String, Value>, active_space: &str) -> Result<AgentPolicySpec> {
357    let id = item
358        .get("id")
359        .and_then(Value::as_str)
360        .ok_or_else(|| http("decoding agent policy: expected string id"))?;
361    let reasons = portability_reasons(item, active_space)?;
362    if !reasons.is_empty() {
363        return Err(Error::new(
364            ErrorKind::Unsupported,
365            format!(
366                "agent policy '{id}' is not portable: {}",
367                reasons.into_iter().collect::<Vec<_>>().join(", ")
368            ),
369        ));
370    }
371
372    let mut portable = Map::new();
373    for key in ["id", "name"] {
374        if let Some(value) = item.get(key) {
375            portable.insert(key.to_string(), value.clone());
376        }
377    }
378    for key in PORTABLE_OPTIONAL {
379        if let Some(value) = item.get(key)
380            && !value.is_null()
381        {
382            portable.insert(key.to_string(), value.clone());
383        }
384    }
385    let known: BTreeSet<&str> = ["id", "name"]
386        .into_iter()
387        .chain(PORTABLE_OPTIONAL)
388        .chain(REMOVED_FIELDS)
389        .collect();
390    let unknown: BTreeSet<&str> = item
391        .keys()
392        .map(String::as_str)
393        .filter(|key| !known.contains(key))
394        .collect();
395    if let Some(first) = unknown.into_iter().next() {
396        return Err(Error::new(
397            ErrorKind::Unsupported,
398            format!("agent policy '{id}' carries unknown field '{first}'"),
399        ));
400    }
401    AgentPolicySpec::try_from(Value::Object(portable))
402        .map_err(|error| http(format!("decoding agent policy '{id}': {}", error.message)))
403}
404
405fn portability_reasons(
406    item: &Map<String, Value>,
407    active_space: &str,
408) -> Result<BTreeSet<&'static str>> {
409    let active = if active_space.is_empty() {
410        "default"
411    } else {
412        active_space
413    };
414    let mut reasons = BTreeSet::new();
415    for flag in PLATFORM_FLAGS.into_iter().chain(["is_protected"]) {
416        if optional_server_bool(item, flag)? == Some(true) {
417            reasons.insert(flag);
418        }
419    }
420    match item.get(AGENTLESS_FIELD) {
421        None | Some(Value::Null) => {}
422        Some(Value::Object(_)) => {
423            reasons.insert(AGENTLESS_FIELD);
424        }
425        Some(_) => {
426            return Err(http(
427                "decoding agent policy: agentless must be an object or null",
428            ));
429        }
430    }
431    for field in ENVIRONMENT_IDS {
432        match item.get(field) {
433            None | Some(Value::Null) => {}
434            Some(Value::String(value)) if !value.trim().is_empty() => {
435                reasons.insert(field);
436            }
437            Some(_) => {
438                return Err(http(format!(
439                    "decoding agent policy: {field} must be a non-empty string or null"
440                )));
441            }
442        }
443    }
444    match item.get("required_versions") {
445        None | Some(Value::Null) => {}
446        Some(Value::Array(_)) => {
447            reasons.insert("required_versions");
448        }
449        Some(_) => {
450            return Err(http(
451                "decoding agent policy: required_versions must be an array or null",
452            ));
453        }
454    }
455    match item.get("space_ids") {
456        None | Some(Value::Null) => {}
457        Some(Value::Array(spaces)) => {
458            let decoded =
459                spaces
460                    .iter()
461                    .map(|space| {
462                        space.as_str().filter(|space| !space.is_empty()).ok_or_else(|| {
463                        http("decoding agent policy: space_ids must contain non-empty strings")
464                    })
465                    })
466                    .collect::<Result<Vec<_>>>()?;
467            if decoded.iter().any(|space| *space != active) {
468                reasons.insert("space_ids");
469            }
470        }
471        Some(_) => {
472            return Err(http(
473                "decoding agent policy: space_ids must be an array or null",
474            ));
475        }
476    }
477    Ok(reasons)
478}
479
480fn optional_server_bool(item: &Map<String, Value>, field: &str) -> Result<Option<bool>> {
481    match item.get(field) {
482        None | Some(Value::Null) => Ok(None),
483        Some(Value::Bool(value)) => Ok(Some(*value)),
484        Some(_) => Err(http(format!(
485            "decoding agent policy: {field} must be a boolean or null"
486        ))),
487    }
488}
489
490/// Read one policy and reduce it to the facts planning compares and rechecks.
491pub(crate) async fn read_live(transport: &Transport, id: &str) -> Result<LiveAgentPolicy> {
492    let policy = agent_policies::get(transport, id).await?;
493    live_from_policy(&policy, id, transport.space())
494}
495
496/// Reduce an already-read policy to the facts planning compares and rechecks,
497/// without a further route call. Shared by `read_live` and `plan_import`'s
498/// conversion of a raw existing snapshot, which must defer this call (and its
499/// `normalize` refusal) until the row is known to need it: a skipped or
500/// conflicting existing policy is never normalized.
501fn live_from_policy(
502    policy: &agent_policies::AgentPolicy,
503    id: &str,
504    active_space: &str,
505) -> Result<LiveAgentPolicy> {
506    live_from_item(&policy.item, id, active_space)
507}
508
509fn live_from_item(
510    item: &Map<String, Value>,
511    id: &str,
512    active_space: &str,
513) -> Result<LiveAgentPolicy> {
514    let spec = normalize(item, active_space)?;
515    if spec.id != id {
516        return Err(http(format!(
517            "decoding agent policy: expected id '{id}', got '{}'",
518            spec.id
519        )));
520    }
521    let agents = required_agents(item, id)?;
522    let attached = attached_integration_ids(item, id)?;
523    Ok(LiveAgentPolicy {
524        spec,
525        agents,
526        attached,
527    })
528}
529
530/// `package_policies` is a list of ids or of populated objects carrying `id`.
531fn attached_integration_ids(item: &Map<String, Value>, id: &str) -> Result<Vec<String>> {
532    let entries = item
533        .get("package_policies")
534        .ok_or_else(|| {
535            http(format!(
536                "decoding agent policy '{id}': missing package_policies"
537            ))
538        })?
539        .as_array()
540        .ok_or_else(|| {
541            http(format!(
542                "decoding agent policy '{id}': package_policies must be an array"
543            ))
544        })?;
545    let mut ids = Vec::with_capacity(entries.len());
546    for entry in entries {
547        let attached_id = match entry {
548            Value::String(attached_id) => Some(attached_id.as_str()),
549            Value::Object(object) => object.get("id").and_then(Value::as_str),
550            _ => None,
551        }
552        .filter(|attached_id| !attached_id.is_empty())
553        .ok_or_else(|| {
554            http(format!(
555                "decoding agent policy '{id}': package_policies entry without id"
556            ))
557        })?;
558        ids.push(attached_id.to_owned());
559    }
560    ids.sort();
561    if ids.windows(2).any(|ids| ids[0] == ids[1]) {
562        return Err(http(format!(
563            "decoding agent policy '{id}': duplicate package_policies id"
564        )));
565    }
566    Ok(ids)
567}
568
569/// Kibana populates `agents` only for a caller with Fleet agents read, so an
570/// absent field is a privilege gap, not a malformed response.
571fn required_agents(item: &Map<String, Value>, id: &str) -> Result<u64> {
572    match item.get("agents") {
573        None => Err(Error::new(
574            ErrorKind::Permission,
575            format!(
576                "agent policy '{id}' has no agents count; the API key lacks the Fleet agents read privilege"
577            ),
578        )),
579        Some(value) => value.as_u64().ok_or_else(|| {
580            http(format!(
581                "decoding agent policy '{id}': agents must be an unsigned integer"
582            ))
583        }),
584    }
585}
586
587fn http(message: impl Into<String>) -> Error {
588    Error::new(ErrorKind::Http, message)
589}
elasticctl_api/fleet/agent_policy_ops.rs

elasticctl_api/fleet/
agent_policy_ops.rs
