elasticctl_api/fleet/integration_policy_ops/
normalization.rs

1use super::common::*;
2
3const PORTABLE_OPTIONAL: [&str; 6] = [
4    "description",
5    "namespace",
6    "vars",
7    "var_group_selections",
8    "condition",
9    "additional_datastreams_permissions",
10];
11
12const REMOVED_FIELDS: [&str; 19] = [
13    "agents",
14    "cloud_connector_id",
15    "cloud_connector_name",
16    "created_at",
17    "created_by",
18    "elasticsearch",
19    "enabled",
20    "is_managed",
21    "output_id",
22    "package_agent_version_condition",
23    "policy_id",
24    "revision",
25    "secret_references",
26    "spaceIds",
27    "supports_agentless",
28    "supports_cloud_connector",
29    "updated_at",
30    "updated_by",
31    "version",
32];
33
34pub fn normalize(item: &Map<String, Value>, active_space: &str) -> Result<IntegrationPolicySpec> {
35    let id = required_string(item, "id", "integration policy")?;
36    portability_check(item, &id, active_space)?;
37    reject_unknown_top_level(item, &id)?;
38
39    let mut portable = Map::new();
40    for field in ["id", "name"] {
41        if let Some(value) = item.get(field) {
42            portable.insert(field.to_owned(), value.clone());
43        }
44    }
45    portable.insert(
46        "policy_ids".to_owned(),
47        Value::Array(
48            read_parents(&id, item)?
49                .into_iter()
50                .map(Value::String)
51                .collect(),
52        ),
53    );
54    portable.insert("package".to_owned(), normalize_package(item, &id)?);
55    portable.insert("inputs".to_owned(), normalize_inputs(item, &id)?);
56    for field in PORTABLE_OPTIONAL {
57        if let Some(value) = item.get(field)
58            && !value.is_null()
59        {
60            portable.insert(field.to_owned(), value.clone());
61        }
62    }
63    IntegrationPolicySpec::try_from(Value::Object(portable)).map_err(|error| {
64        http(format!(
65            "decoding integration policy '{id}': {}",
66            error.message
67        ))
68    })
69}
70
71fn normalize_package(item: &Map<String, Value>, id: &str) -> Result<Value> {
72    let package = item
73        .get("package")
74        .and_then(Value::as_object)
75        .ok_or_else(|| {
76            http(format!(
77                "decoding integration policy '{id}': package must be an object"
78            ))
79        })?;
80    for (field, expected) in [("title", "a string")] {
81        if let Some(value) = package.get(field)
82            && !value.is_null()
83            && !value.is_string()
84        {
85            return Err(http(format!(
86                "decoding integration policy '{id}': package.{field} must be {expected} or null"
87            )));
88        }
89    }
90    for field in ["requires_root", "fips_compatible"] {
91        if let Some(value) = package.get(field)
92            && !value.is_null()
93            && !value.is_boolean()
94        {
95            return Err(http(format!(
96                "decoding integration policy '{id}': package.{field} must be a boolean or null"
97            )));
98        }
99    }
100    let mut portable = Map::new();
101    for field in ["name", "version"] {
102        if let Some(value) = package.get(field) {
103            portable.insert(field.to_owned(), value.clone());
104        }
105    }
106    let known: BTreeSet<&str> = [
107        "name",
108        "version",
109        "title",
110        "requires_root",
111        "fips_compatible",
112    ]
113    .into_iter()
114    .collect();
115    if let Some(field) = package
116        .keys()
117        .map(String::as_str)
118        .filter(|field| !known.contains(field))
119        .min()
120    {
121        return Err(Error::new(
122            ErrorKind::Unsupported,
123            format!("integration policy '{id}' carries unknown package field '{field}'"),
124        ));
125    }
126    Ok(Value::Object(portable))
127}
128
129fn normalize_inputs(item: &Map<String, Value>, id: &str) -> Result<Value> {
130    let inputs = item
131        .get("inputs")
132        .and_then(Value::as_object)
133        .ok_or_else(|| {
134            http(format!(
135                "decoding integration policy '{id}': inputs must be an object"
136            ))
137        })?;
138    let mut normalized = Map::new();
139    for (input_id, input) in inputs {
140        normalized.insert(
141            input_id.clone(),
142            normalize_package_map(input, "compiled_input")?,
143        );
144    }
145    Ok(Value::Object(normalized))
146}
147
148fn normalize_package_map(value: &Value, compiled_field: &str) -> Result<Value> {
149    let object = value
150        .as_object()
151        .ok_or_else(|| http("decoding integration policy: input must be an object"))?;
152    let mut normalized = Map::new();
153    for (field, value) in object {
154        if field == "id" {
155            if !value.is_string() {
156                return Err(http(
157                    "decoding integration policy: generated id must be a string",
158                ));
159            }
160            continue;
161        }
162        if field == compiled_field {
163            if !value.is_object() {
164                return Err(http(format!(
165                    "decoding integration policy: {compiled_field} must be an object"
166                )));
167            }
168            continue;
169        }
170        if field == "streams" {
171            let streams = value.as_object().ok_or_else(|| {
172                http("decoding integration policy: input streams must be an object")
173            })?;
174            let mut normalized_streams = Map::new();
175            for (stream_id, stream) in streams {
176                normalized_streams.insert(
177                    stream_id.clone(),
178                    normalize_package_map(stream, "compiled_stream")?,
179                );
180            }
181            normalized.insert(field.clone(), Value::Object(normalized_streams));
182        } else {
183            normalized.insert(field.clone(), value.clone());
184        }
185    }
186    Ok(Value::Object(normalized))
187}
188
189fn portability_check(item: &Map<String, Value>, id: &str, active_space: &str) -> Result<()> {
190    let mut reasons = BTreeSet::new();
191    required_true(item, "enabled", id)?;
192    if let Some(value) = item.get("elasticsearch")
193        && !value.is_object()
194    {
195        return Err(http(format!(
196            "decoding integration policy '{id}': elasticsearch must be an object"
197        )));
198    }
199    if let Some(value) = item.get("package_agent_version_condition")
200        && !value.is_null()
201        && !value.is_string()
202    {
203        return Err(http(format!(
204            "decoding integration policy '{id}': package_agent_version_condition must be a string or null"
205        )));
206    }
207    for field in [
208        "is_managed",
209        "supports_agentless",
210        "supports_cloud_connector",
211    ] {
212        if let Some(true) = optional_bool(item, field, id)? {
213            reasons.insert(field);
214        }
215    }
216    for field in ["output_id", "cloud_connector_id", "cloud_connector_name"] {
217        match item.get(field) {
218            None | Some(Value::Null) | Some(Value::Bool(false)) => {}
219            Some(Value::String(_)) => {
220                reasons.insert(field);
221            }
222            Some(_) => {
223                return Err(http(format!(
224                    "decoding integration policy '{id}': {field} must be a string or null"
225                )));
226            }
227        }
228    }
229    match item.get("secret_references") {
230        None | Some(Value::Null) => {}
231        Some(Value::Array(references)) if references.is_empty() => {}
232        Some(Value::Array(_)) => {
233            reasons.insert("secret_references");
234        }
235        Some(_) => {
236            return Err(http(format!(
237                "decoding integration policy '{id}': secret_references must be an array or null"
238            )));
239        }
240    }
241    let active = if active_space.is_empty() {
242        "default"
243    } else {
244        active_space
245    };
246    match item.get("spaceIds") {
247        None | Some(Value::Null) => {}
248        Some(Value::Array(spaces)) => {
249            for space in spaces {
250                let space = space.as_str().filter(|space| !space.is_empty()).ok_or_else(|| {
251                    http(format!(
252                        "decoding integration policy '{id}': spaceIds must contain non-empty strings"
253                    ))
254                })?;
255                if space != active {
256                    reasons.insert("spaceIds");
257                }
258            }
259        }
260        Some(_) => {
261            return Err(http(format!(
262                "decoding integration policy '{id}': spaceIds must be an array or null"
263            )));
264        }
265    }
266    if let Some(policy_id) = item.get("policy_id").filter(|value| !value.is_null()) {
267        let policy_id = policy_id
268            .as_str()
269            .filter(|value| !value.is_empty())
270            .ok_or_else(|| {
271                http(format!(
272                    "decoding integration policy '{id}': policy_id must be a non-empty string"
273                ))
274            })?;
275        let policy_ids = item
276            .get("policy_ids")
277            .and_then(Value::as_array)
278            .ok_or_else(|| {
279                http(format!(
280                    "decoding integration policy '{id}': policy_ids must be an array"
281                ))
282            })?;
283        if policy_ids.first().and_then(Value::as_str) != Some(policy_id) {
284            return Err(http(format!(
285                "decoding integration policy '{id}': policy_id must equal policy_ids[0]"
286            )));
287        }
288    }
289    if reasons.is_empty() {
290        Ok(())
291    } else {
292        unsupported(format!(
293            "integration policy '{id}' is not portable: {}",
294            reasons.into_iter().collect::<Vec<_>>().join(", ")
295        ))
296    }
297}
298
299fn required_true(item: &Map<String, Value>, field: &str, id: &str) -> Result<()> {
300    match item.get(field) {
301        Some(Value::Bool(true)) => Ok(()),
302        Some(Value::Bool(false)) => unsupported(format!(
303            "integration policy '{id}' is not portable: {field}"
304        )),
305        _ => Err(http(format!(
306            "decoding integration policy '{id}': {field} must be true"
307        ))),
308    }
309}
310
311fn reject_unknown_top_level(item: &Map<String, Value>, id: &str) -> Result<()> {
312    let known: BTreeSet<&str> = ["id", "name", "policy_ids", "package", "inputs"]
313        .into_iter()
314        .chain(PORTABLE_OPTIONAL)
315        .chain(REMOVED_FIELDS)
316        .collect();
317    if let Some(field) = item
318        .keys()
319        .map(String::as_str)
320        .filter(|field| !known.contains(field))
321        .min()
322    {
323        return unsupported(format!(
324            "integration policy '{id}' carries unknown field '{field}'"
325        ));
326    }
327    Ok(())
328}
elasticctl_api/fleet/integration_policy_ops/normalization.rs

elasticctl_api/fleet/integration_policy_ops/
normalization.rs
