security.rs

use std::collections::BTreeMap;

use snafu::{ResultExt, Snafu};
use stackable_operator::{
    builder::{
        self,
        pod::{
            PodBuilder,
            container::ContainerBuilder,
            volume::{
                SecretFormat, SecretOperatorVolumeSourceBuilder,
                SecretOperatorVolumeSourceBuilderError, VolumeBuilder,
            },
        },
    },
    crd::listener,
    k8s_openapi::{
        api::core::v1::{ContainerPort, Probe, ServicePort, TCPSocketAction},
        apimachinery::pkg::util::intstr::IntOrString,
    },
    shared::time::Duration,
};

use crate::crd::{
    DruidRole, STACKABLE_TRUST_STORE_PASSWORD,
    authentication::{self, AuthenticationClassesResolved},
    v1alpha1,
};

#[derive(Snafu, Debug)]
pub enum Error {
    #[snafu(display("failed to process authentication class"))]
    InvalidAuthenticationClassConfiguration { source: authentication::Error },

    #[snafu(display("failed to build the Secret operator Volume"))]
    SecretVolumeBuild {
        source: SecretOperatorVolumeSourceBuilderError,
    },

    #[snafu(display("failed to add needed volume"))]
    AddVolume { source: builder::pod::Error },

    #[snafu(display("failed to add needed volumeMount"))]
    AddVolumeMount {
        source: builder::pod::container::Error,
    },
}

/// Helper struct combining TLS settings for server and internal tls with the resolved AuthenticationClasses
pub struct DruidTlsSecurity {
    auth_classes: AuthenticationClassesResolved,
    server_and_internal_secret_class: Option<String>,
}

// Ports
const ENABLE_PLAINTEXT_PORT: &str = "druid.enablePlaintextPort";
const PLAINTEXT_PORT: &str = "druid.plaintextPort";
const ENABLE_TLS_PORT: &str = "druid.enableTlsPort";
const TLS_PORT: &str = "druid.tlsPort";
// Port names
pub const PLAINTEXT_PORT_NAME: &str = "http";
pub const TLS_PORT_NAME: &str = "https";
// Client side (Druid) TLS
const CLIENT_HTTPS_KEY_STORE_PATH: &str = "druid.client.https.keyStorePath";
const CLIENT_HTTPS_KEY_STORE_TYPE: &str = "druid.client.https.keyStoreType";
const CLIENT_HTTPS_KEY_STORE_PASSWORD: &str = "druid.client.https.keyStorePassword";
const CLIENT_HTTPS_TRUST_STORE_PATH: &str = "druid.client.https.trustStorePath";
const CLIENT_HTTPS_TRUST_STORE_TYPE: &str = "druid.client.https.trustStoreType";
const CLIENT_HTTPS_TRUST_STORE_PASSWORD: &str = "druid.client.https.trustStorePassword";
const CLIENT_HTTPS_CERT_ALIAS: &str = "druid.client.https.certAlias";
const CLIENT_HTTPS_VALIDATE_HOST_NAMES: &str = "druid.client.https.validateHostnames";
const CLIENT_HTTPS_KEY_MANAGER_PASSWORD: &str = "druid.client.https.keyManagerPassword";
// Server side TLS
const SERVER_HTTPS_KEY_STORE_PATH: &str = "druid.server.https.keyStorePath";
const SERVER_HTTPS_KEY_STORE_TYPE: &str = "druid.server.https.keyStoreType";
const SERVER_HTTPS_KEY_STORE_PASSWORD: &str = "druid.server.https.keyStorePassword";
const SERVER_HTTPS_TRUST_STORE_PATH: &str = "druid.server.https.trustStorePath";
const SERVER_HTTPS_TRUST_STORE_TYPE: &str = "druid.server.https.trustStoreType";
const SERVER_HTTPS_TRUST_STORE_PASSWORD: &str = "druid.server.https.trustStorePassword";
const SERVER_HTTPS_CERT_ALIAS: &str = "druid.server.https.certAlias";
const SERVER_HTTPS_VALIDATE_HOST_NAMES: &str = "druid.server.https.validateHostnames";
const SERVER_HTTPS_KEY_MANAGER_PASSWORD: &str = "druid.server.https.keyManagerPassword";
const SERVER_HTTPS_REQUIRE_CLIENT_CERTIFICATE: &str = "druid.server.https.requireClientCertificate";
/// The alias of the certificate in the keystore used for TLS stuff.
/// All secret-op generated keystores have one entry with the alias "1".
/// (side node: I think technically they don't have an alias and the JVm counts them, but not sure)
const TLS_ALIAS_NAME: &str = "1";
pub const AUTH_TRUST_STORE_PATH: &str = "druid.auth.basic.ssl.trustStorePath";
pub const AUTH_TRUST_STORE_TYPE: &str = "druid.auth.basic.ssl.trustStoreType";
pub const AUTH_TRUST_STORE_PASSWORD: &str = "druid.auth.basic.ssl.trustStorePassword";
// Misc TLS
pub const TLS_STORE_PASSWORD: &str = "changeit";
pub const TLS_STORE_TYPE: &str = "pkcs12";

// directories
const STACKABLE_MOUNT_TLS_DIR: &str = "/stackable/mount_tls";
pub const STACKABLE_TLS_DIR: &str = "/stackable/tls";

// volume names
const TLS_VOLUME_NAME: &str = "tls";
const TLS_MOUNT_VOLUME_NAME: &str = "tls-mount";

pub const INTERNAL_INITIAL_CLIENT_PASSWORD_ENV: &str = "INTERNAL_INITIAL_CLIENT_PASSWORD";
// It seems this needs to be the same password for Druid to work, so we re-use the existing env variable from above.
pub const ESCALATOR_INTERNAL_CLIENT_PASSWORD_ENV: &str = INTERNAL_INITIAL_CLIENT_PASSWORD_ENV;

impl DruidTlsSecurity {
    #[cfg(test)]
    pub fn new(
        auth_classes: &AuthenticationClassesResolved,
        server_and_internal_secret_class: Option<String>,
    ) -> Self {
        Self {
            auth_classes: auth_classes.clone(),
            server_and_internal_secret_class,
        }
    }

    /// Create a `DruidTlsSecurity` struct from the Druid custom resource and resolve
    /// all provided `AuthenticationClass` references.
    pub fn new_from_druid_cluster(
        druid: &v1alpha1::DruidCluster,
        auth_classes: &AuthenticationClassesResolved,
    ) -> Self {
        DruidTlsSecurity {
            auth_classes: auth_classes.clone(),
            server_and_internal_secret_class: druid
                .spec
                .cluster_config
                .tls
                .as_ref()
                .and_then(|tls| tls.server_and_internal_secret_class.clone()),
        }
    }

    /// Check if TLS encryption is enabled. This could be due to:
    ///
    /// - A provided server `SecretClass`
    /// - A provided client `AuthenticationClass` using tls
    ///
    /// This affects init container commands, Druid configuration, volume mounts
    /// and the Druid client port
    pub fn tls_enabled(&self) -> bool {
        // TODO: This must be adapted if other authentication methods are supported and require TLS
        self.auth_classes.tls_authentication_enabled()
            || self.tls_server_and_internal_secret_class().is_some()
    }

    /// Retrieve an optional TLS secret class for external client -> server and server <-> server communications.
    pub fn tls_server_and_internal_secret_class(&self) -> Option<&str> {
        self.server_and_internal_secret_class.as_deref()
    }

    pub fn container_ports(&self, role: &DruidRole) -> Vec<ContainerPort> {
        self.exposed_ports(role)
            .into_iter()
            .map(|(name, val)| ContainerPort {
                name: Some(name),
                container_port: val.into(),
                protocol: Some("TCP".to_string()),
                ..ContainerPort::default()
            })
            .collect()
    }

    pub fn service_ports(&self, role: &DruidRole) -> Vec<ServicePort> {
        self.exposed_ports(role)
            .into_iter()
            .map(|(name, val)| ServicePort {
                name: Some(name),
                port: val.into(),
                protocol: Some("TCP".to_string()),
                ..ServicePort::default()
            })
            .collect()
    }

    pub fn listener_ports(
        &self,
        role: &DruidRole,
    ) -> Option<Vec<listener::v1alpha1::ListenerPort>> {
        let listener_ports = self
            .exposed_ports(role)
            .into_iter()
            .map(|(name, val)| listener::v1alpha1::ListenerPort {
                name,
                port: val.into(),
                protocol: Some("TCP".to_string()),
            })
            .collect();

        Some(listener_ports)
    }

    fn exposed_ports(&self, role: &DruidRole) -> Vec<(String, u16)> {
        if self.tls_enabled() {
            vec![(TLS_PORT_NAME.to_string(), role.get_https_port())]
        } else {
            vec![(PLAINTEXT_PORT_NAME.to_string(), role.get_http_port())]
        }
    }

    /// Adds required tls volume mounts to image and product container builders
    /// Adds required tls volumes to pod builder
    pub fn add_tls_volume_and_volume_mounts(
        &self,
        prepare: &mut ContainerBuilder,
        druid: &mut ContainerBuilder,
        pod: &mut PodBuilder,
        requested_secret_lifetime: &Duration,
        listener_scope: Option<String>,
    ) -> Result<(), Error> {
        // `ResolvedAuthenticationClasses::validate` already checked that the tls AuthenticationClass
        // uses the same SecretClass as the Druid server itself.
        if let Some(secret_class) = &self.server_and_internal_secret_class {
            let mut secret_volume_source_builder =
                SecretOperatorVolumeSourceBuilder::new(secret_class);

            secret_volume_source_builder
                .with_pod_scope()
                .with_format(SecretFormat::TlsPkcs12)
                .with_tls_pkcs12_password(TLS_STORE_PASSWORD)
                .with_auto_tls_cert_lifetime(*requested_secret_lifetime);

            if let Some(listener_scope) = &listener_scope {
                secret_volume_source_builder.with_listener_volume_scope(listener_scope);
            }

            pod.add_volume(
                VolumeBuilder::new(TLS_MOUNT_VOLUME_NAME)
                    .ephemeral(
                        secret_volume_source_builder
                            .build()
                            .context(SecretVolumeBuildSnafu)?,
                    )
                    .build(),
            )
            .context(AddVolumeSnafu)?;
            prepare
                .add_volume_mount(TLS_MOUNT_VOLUME_NAME, STACKABLE_MOUNT_TLS_DIR)
                .context(AddVolumeMountSnafu)?;
            druid
                .add_volume_mount(TLS_MOUNT_VOLUME_NAME, STACKABLE_MOUNT_TLS_DIR)
                .context(AddVolumeMountSnafu)?;

            pod.add_volume(
                VolumeBuilder::new(TLS_VOLUME_NAME)
                    .with_empty_dir(Option::<&str>::None, None)
                    .build(),
            )
            .context(AddVolumeSnafu)?;

            prepare
                .add_volume_mount(TLS_VOLUME_NAME, STACKABLE_TLS_DIR)
                .context(AddVolumeMountSnafu)?;
            druid
                .add_volume_mount(TLS_VOLUME_NAME, STACKABLE_TLS_DIR)
                .context(AddVolumeMountSnafu)?;
        }
        Ok(())
    }

    fn add_tls_port_config_properties(
        &self,
        config: &mut BTreeMap<String, Option<String>>,
        role: &DruidRole,
    ) {
        // no secure communication
        if !self.tls_enabled() {
            config.insert(ENABLE_PLAINTEXT_PORT.to_string(), Some("true".to_string()));
            config.insert(ENABLE_TLS_PORT.to_string(), Some("false".to_string()));
            config.insert(
                PLAINTEXT_PORT.to_string(),
                Some(role.get_http_port().to_string()),
            );
        }
        // only allow secure communication
        else {
            config.insert(ENABLE_PLAINTEXT_PORT.to_string(), Some("false".to_string()));
            config.insert(ENABLE_TLS_PORT.to_string(), Some("true".to_string()));
            config.insert(
                TLS_PORT.to_string(),
                Some(role.get_https_port().to_string()),
            );
        }
    }

    /// Add required TLS ports, trust/key store properties
    pub fn add_tls_config_properties(
        &self,
        config: &mut BTreeMap<String, Option<String>>,
        role: &DruidRole,
    ) {
        self.add_tls_port_config_properties(config, role);

        if self.tls_enabled() {
            Self::add_tls_encryption_config_properties(config, STACKABLE_TLS_DIR, TLS_ALIAS_NAME);
        }

        if self.auth_classes.tls_authentication_enabled() {
            Self::add_tls_auth_config_properties(config, STACKABLE_TLS_DIR, TLS_ALIAS_NAME);
        }
    }

    fn add_tls_encryption_config_properties(
        config: &mut BTreeMap<String, Option<String>>,
        store_directory: &str,
        store_alias: &str,
    ) {
        // We need a truststore in addition to a keystore here, because server and internal tls
        // can only be enabled/disabled together
        config.insert(
            CLIENT_HTTPS_TRUST_STORE_PATH.to_string(),
            Some(format!("{}/truststore.p12", store_directory)),
        );
        config.insert(
            CLIENT_HTTPS_TRUST_STORE_TYPE.to_string(),
            Some(TLS_STORE_TYPE.to_string()),
        );
        config.insert(
            CLIENT_HTTPS_TRUST_STORE_PASSWORD.to_string(),
            Some(TLS_STORE_PASSWORD.to_string()),
        );

        config.insert(
            SERVER_HTTPS_KEY_STORE_PATH.to_string(),
            Some(format!("{}/keystore.p12", store_directory)),
        );
        config.insert(
            SERVER_HTTPS_KEY_STORE_TYPE.to_string(),
            Some(TLS_STORE_TYPE.to_string()),
        );
        config.insert(
            SERVER_HTTPS_KEY_STORE_PASSWORD.to_string(),
            Some(TLS_STORE_PASSWORD.to_string()),
        );
        config.insert(
            SERVER_HTTPS_CERT_ALIAS.to_string(),
            Some(store_alias.to_string()),
        );

        // We also need to configure the truststore for authentication related stuff,
        // such as verifying the LDAP server
        config.insert(
            AUTH_TRUST_STORE_PATH.to_string(),
            Some(format!("{}/truststore.p12", store_directory)),
        );
        config.insert(
            AUTH_TRUST_STORE_TYPE.to_string(),
            Some(TLS_STORE_TYPE.to_string()),
        );
        config.insert(
            AUTH_TRUST_STORE_PASSWORD.to_string(),
            Some(TLS_STORE_PASSWORD.to_string()),
        );
    }

    fn add_tls_auth_config_properties(
        config: &mut BTreeMap<String, Option<String>>,
        store_directory: &str,
        store_alias: &str,
    ) {
        config.insert(
            CLIENT_HTTPS_KEY_STORE_PATH.to_string(),
            Some(format!("{store_directory}/keystore.p12")),
        );
        config.insert(
            CLIENT_HTTPS_KEY_STORE_TYPE.to_string(),
            Some(TLS_STORE_TYPE.to_string()),
        );
        config.insert(
            CLIENT_HTTPS_KEY_STORE_PASSWORD.to_string(),
            Some(TLS_STORE_PASSWORD.to_string()),
        );
        // This is required because PKCS12 does not use any key passwords but it will
        // be checked and would lead to an exception:
        // java.security.UnrecoverableKeyException: Get Key failed: null
        // Must be set to the store password or we get a bad padding exception:
        // javax.crypto.BadPaddingException: Given final block not properly padded. Such issues can arise if a bad key is used during decryption.
        config.insert(
            CLIENT_HTTPS_KEY_MANAGER_PASSWORD.to_string(),
            Some(TLS_STORE_PASSWORD.to_string()),
        );
        config.insert(
            CLIENT_HTTPS_CERT_ALIAS.to_string(),
            Some(store_alias.to_string()),
        );
        // FIXME: https://github.com/stackabletech/druid-operator/issues/372
        // This is required because the server will send its pod ip which is not in the SANs of the certificates
        config.insert(
            CLIENT_HTTPS_VALIDATE_HOST_NAMES.to_string(),
            Some("false".to_string()),
        );

        // This will enforce the client to authenticate itself
        config.insert(
            SERVER_HTTPS_REQUIRE_CLIENT_CERTIFICATE.to_string(),
            Some("true".to_string()),
        );

        config.insert(
            SERVER_HTTPS_TRUST_STORE_PATH.to_string(),
            Some(format!("{store_directory}/truststore.p12")),
        );
        config.insert(
            SERVER_HTTPS_TRUST_STORE_TYPE.to_string(),
            Some(TLS_STORE_TYPE.to_string()),
        );
        config.insert(
            SERVER_HTTPS_TRUST_STORE_PASSWORD.to_string(),
            Some(TLS_STORE_PASSWORD.to_string()),
        );
        // This is required because PKCS12 does not use any key passwords but it will
        // be checked and would lead to an exception:
        // java.security.UnrecoverableKeyException: Get Key failed: null
        // Must be set to the store password or we get a bad padding exception:
        // javax.crypto.BadPaddingException: Given final block not properly padded. Such issues can arise if a bad key is used during decryption.
        config.insert(
            SERVER_HTTPS_KEY_MANAGER_PASSWORD.to_string(),
            Some(TLS_STORE_PASSWORD.to_string()),
        );
        // FIXME: https://github.com/stackabletech/druid-operator/issues/372
        // This is required because the client will send its pod ip which is not in the SANs of the certificates
        config.insert(
            SERVER_HTTPS_VALIDATE_HOST_NAMES.to_string(),
            Some("false".to_string()),
        );
    }

    pub fn build_tls_key_stores_cmd(&self) -> Vec<String> {
        if !self.tls_enabled() {
            return vec![];
        }

        vec![
            // FIXME: *Technically* we should only add the system truststore in case any webPki usage is detected,
            // wether that's in S3, LDAP, OIDC, FTE or whatnot.
            format!(
                "cert-tools generate-pkcs12-truststore --pkcs12 '{STACKABLE_MOUNT_TLS_DIR}/truststore.p12:{TLS_STORE_PASSWORD}' --pem /etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem --out {STACKABLE_TLS_DIR}/truststore.p12 --out-password '{TLS_STORE_PASSWORD}'"
            ),
            // We can copy the keystore as is.
            format!("cp {STACKABLE_MOUNT_TLS_DIR}/keystore.p12 {STACKABLE_TLS_DIR}/keystore.p12"),
        ]
    }

    pub fn get_tcp_socket_probe(
        &self,
        initial_delay_seconds: i32,
        period_seconds: i32,
        failure_threshold: i32,
        timeout_seconds: i32,
    ) -> Probe {
        let port = if self.tls_enabled() {
            IntOrString::String(TLS_PORT_NAME.to_string())
        } else {
            IntOrString::String(PLAINTEXT_PORT_NAME.to_string())
        };

        Probe {
            tcp_socket: Some(TCPSocketAction {
                port,
                ..Default::default()
            }),
            initial_delay_seconds: Some(initial_delay_seconds),
            period_seconds: Some(period_seconds),
            failure_threshold: Some(failure_threshold),
            timeout_seconds: Some(timeout_seconds),
            ..Default::default()
        }
    }
}

/// Generate a bash command to add a CA to a truststore
pub fn add_cert_to_trust_store_cmd(
    cert_file: &str,
    destination_directory: &str,
    store_password: &str,
) -> String {
    let truststore = format!("{destination_directory}/truststore.p12");
    format!(
        "cert-tools generate-pkcs12-truststore --pkcs12 {truststore}:{store_password} --pem {cert_file} --out {truststore} --out-password {store_password}"
    )
}

/// Generate a bash command to add a CA to the truststore that is passed to the JVM
pub fn add_cert_to_jvm_trust_store_cmd(cert_file: &str) -> String {
    add_cert_to_trust_store_cmd(cert_file, "/stackable", STACKABLE_TRUST_STORE_PASSWORD)
}