elasticctl_api/fleet/integration_policy_ops/

normalization.rs

use super::common::*;

const PORTABLE_OPTIONAL: [&str; 6] = [
    "description",
    "namespace",
    "vars",
    "var_group_selections",
    "condition",
    "additional_datastreams_permissions",
];

const REMOVED_FIELDS: [&str; 19] = [
    "agents",
    "cloud_connector_id",
    "cloud_connector_name",
    "created_at",
    "created_by",
    "elasticsearch",
    "enabled",
    "is_managed",
    "output_id",
    "package_agent_version_condition",
    "policy_id",
    "revision",
    "secret_references",
    "spaceIds",
    "supports_agentless",
    "supports_cloud_connector",
    "updated_at",
    "updated_by",
    "version",
];

/// Build a fresh portable policy from a simplified live Fleet response.
pub fn normalize(item: &Map<String, Value>, active_space: &str) -> Result<IntegrationPolicySpec> {
    let id = required_string(item, "id", "integration policy")?;
    portability_check(item, &id, active_space)?;
    reject_unknown_top_level(item, &id)?;

    let mut portable = Map::new();
    for field in ["id", "name"] {
        if let Some(value) = item.get(field) {
            portable.insert(field.to_owned(), value.clone());
        }
    }
    portable.insert(
        "policy_ids".to_owned(),
        Value::Array(
            read_parents(&id, item)?
                .into_iter()
                .map(Value::String)
                .collect(),
        ),
    );
    portable.insert("package".to_owned(), normalize_package(item, &id)?);
    portable.insert("inputs".to_owned(), normalize_inputs(item, &id)?);
    for field in PORTABLE_OPTIONAL {
        if let Some(value) = item.get(field)
            && !value.is_null()
        {
            portable.insert(field.to_owned(), value.clone());
        }
    }
    IntegrationPolicySpec::try_from(Value::Object(portable)).map_err(|error| {
        http(format!(
            "decoding integration policy '{id}': {}",
            error.message
        ))
    })
}

fn normalize_package(item: &Map<String, Value>, id: &str) -> Result<Value> {
    let package = item
        .get("package")
        .and_then(Value::as_object)
        .ok_or_else(|| {
            http(format!(
                "decoding integration policy '{id}': package must be an object"
            ))
        })?;
    for (field, expected) in [("title", "a string")] {
        if let Some(value) = package.get(field)
            && !value.is_null()
            && !value.is_string()
        {
            return Err(http(format!(
                "decoding integration policy '{id}': package.{field} must be {expected} or null"
            )));
        }
    }
    for field in ["requires_root", "fips_compatible"] {
        if let Some(value) = package.get(field)
            && !value.is_null()
            && !value.is_boolean()
        {
            return Err(http(format!(
                "decoding integration policy '{id}': package.{field} must be a boolean or null"
            )));
        }
    }
    let mut portable = Map::new();
    for field in ["name", "version"] {
        if let Some(value) = package.get(field) {
            portable.insert(field.to_owned(), value.clone());
        }
    }
    let known: BTreeSet<&str> = [
        "name",
        "version",
        "title",
        "requires_root",
        "fips_compatible",
    ]
    .into_iter()
    .collect();
    if let Some(field) = package
        .keys()
        .map(String::as_str)
        .filter(|field| !known.contains(field))
        .min()
    {
        return Err(Error::new(
            ErrorKind::Unsupported,
            format!("integration policy '{id}' carries unknown package field '{field}'"),
        ));
    }
    Ok(Value::Object(portable))
}

fn normalize_inputs(item: &Map<String, Value>, id: &str) -> Result<Value> {
    let inputs = item
        .get("inputs")
        .and_then(Value::as_object)
        .ok_or_else(|| {
            http(format!(
                "decoding integration policy '{id}': inputs must be an object"
            ))
        })?;
    let mut normalized = Map::new();
    for (input_id, input) in inputs {
        normalized.insert(
            input_id.clone(),
            normalize_package_map(input, "compiled_input")?,
        );
    }
    Ok(Value::Object(normalized))
}

/// Keep package-defined input and stream maps open while rebuilding them
/// without Fleet-generated ids or compiled content.
fn normalize_package_map(value: &Value, compiled_field: &str) -> Result<Value> {
    let object = value
        .as_object()
        .ok_or_else(|| http("decoding integration policy: input must be an object"))?;
    let mut normalized = Map::new();
    for (field, value) in object {
        if field == "id" {
            if !value.is_string() {
                return Err(http(
                    "decoding integration policy: generated id must be a string",
                ));
            }
            continue;
        }
        if field == compiled_field {
            if !value.is_object() {
                return Err(http(format!(
                    "decoding integration policy: {compiled_field} must be an object"
                )));
            }
            continue;
        }
        if field == "streams" {
            let streams = value.as_object().ok_or_else(|| {
                http("decoding integration policy: input streams must be an object")
            })?;
            let mut normalized_streams = Map::new();
            for (stream_id, stream) in streams {
                normalized_streams.insert(
                    stream_id.clone(),
                    normalize_package_map(stream, "compiled_stream")?,
                );
            }
            normalized.insert(field.clone(), Value::Object(normalized_streams));
        } else {
            normalized.insert(field.clone(), value.clone());
        }
    }
    Ok(Value::Object(normalized))
}

fn portability_check(item: &Map<String, Value>, id: &str, active_space: &str) -> Result<()> {
    let mut reasons = BTreeSet::new();
    required_true(item, "enabled", id)?;
    if let Some(value) = item.get("elasticsearch")
        && !value.is_object()
    {
        return Err(http(format!(
            "decoding integration policy '{id}': elasticsearch must be an object"
        )));
    }
    if let Some(value) = item.get("package_agent_version_condition")
        && !value.is_null()
        && !value.is_string()
    {
        return Err(http(format!(
            "decoding integration policy '{id}': package_agent_version_condition must be a string or null"
        )));
    }
    for field in [
        "is_managed",
        "supports_agentless",
        "supports_cloud_connector",
    ] {
        if let Some(true) = optional_bool(item, field, id)? {
            reasons.insert(field);
        }
    }
    for field in ["output_id", "cloud_connector_id", "cloud_connector_name"] {
        match item.get(field) {
            None | Some(Value::Null) | Some(Value::Bool(false)) => {}
            Some(Value::String(_)) => {
                reasons.insert(field);
            }
            Some(_) => {
                return Err(http(format!(
                    "decoding integration policy '{id}': {field} must be a string or null"
                )));
            }
        }
    }
    match item.get("secret_references") {
        None | Some(Value::Null) => {}
        Some(Value::Array(references)) if references.is_empty() => {}
        Some(Value::Array(_)) => {
            reasons.insert("secret_references");
        }
        Some(_) => {
            return Err(http(format!(
                "decoding integration policy '{id}': secret_references must be an array or null"
            )));
        }
    }
    let active = if active_space.is_empty() {
        "default"
    } else {
        active_space
    };
    match item.get("spaceIds") {
        None | Some(Value::Null) => {}
        Some(Value::Array(spaces)) => {
            for space in spaces {
                let space = space.as_str().filter(|space| !space.is_empty()).ok_or_else(|| {
                    http(format!(
                        "decoding integration policy '{id}': spaceIds must contain non-empty strings"
                    ))
                })?;
                if space != active {
                    reasons.insert("spaceIds");
                }
            }
        }
        Some(_) => {
            return Err(http(format!(
                "decoding integration policy '{id}': spaceIds must be an array or null"
            )));
        }
    }
    if let Some(policy_id) = item.get("policy_id").filter(|value| !value.is_null()) {
        let policy_id = policy_id
            .as_str()
            .filter(|value| !value.is_empty())
            .ok_or_else(|| {
                http(format!(
                    "decoding integration policy '{id}': policy_id must be a non-empty string"
                ))
            })?;
        let policy_ids = item
            .get("policy_ids")
            .and_then(Value::as_array)
            .ok_or_else(|| {
                http(format!(
                    "decoding integration policy '{id}': policy_ids must be an array"
                ))
            })?;
        if policy_ids.first().and_then(Value::as_str) != Some(policy_id) {
            return Err(http(format!(
                "decoding integration policy '{id}': policy_id must equal policy_ids[0]"
            )));
        }
    }
    if reasons.is_empty() {
        Ok(())
    } else {
        unsupported(format!(
            "integration policy '{id}' is not portable: {}",
            reasons.into_iter().collect::<Vec<_>>().join(", ")
        ))
    }
}

fn required_true(item: &Map<String, Value>, field: &str, id: &str) -> Result<()> {
    match item.get(field) {
        Some(Value::Bool(true)) => Ok(()),
        Some(Value::Bool(false)) => unsupported(format!(
            "integration policy '{id}' is not portable: {field}"
        )),
        _ => Err(http(format!(
            "decoding integration policy '{id}': {field} must be true"
        ))),
    }
}

fn reject_unknown_top_level(item: &Map<String, Value>, id: &str) -> Result<()> {
    let known: BTreeSet<&str> = ["id", "name", "policy_ids", "package", "inputs"]
        .into_iter()
        .chain(PORTABLE_OPTIONAL)
        .chain(REMOVED_FIELDS)
        .collect();
    if let Some(field) = item
        .keys()
        .map(String::as_str)
        .filter(|field| !known.contains(field))
        .min()
    {
        return unsupported(format!(
            "integration policy '{id}' carries unknown field '{field}'"
        ));
    }
    Ok(())
}