Fix/custom s3 certificates (#247)

# Description

This PR is supposed to enable TLS for spark-k8s for S3 buckets. 
Fixes #194 
TODO: add s3 certificate for history server access.



Co-authored-by: Andrew Kenworthy <[email protected]>
Co-authored-by: Razvan-Daniel Mihai <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -8,6 +8,8 @@ All notable changes to this project will be documented in this file.
 
 - Generate OLM bundle for Release 23.4.0 ([#238]).
 - Add support for Spark 3.4.0 ([#243]).
+- Add support for using custom certificates when accessing S3 with TLS ([#247]).
+- Use bitnami charts for testing S3 access with TLS ([#247]).
 
 ### Changed
 
@@ -25,6 +27,7 @@ All notable changes to this project will be documented in this file.
 [#238]: https://github.com/stackabletech/spark-k8s-operator/pull/238
 [#241]: https://github.com/stackabletech/spark-k8s-operator/pull/241
 [#243]: https://github.com/stackabletech/spark-k8s-operator/pull/243
+[#247]: https://github.com/stackabletech/spark-k8s-operator/pull/247
 
 ## [23.4.0] - 2023-04-17
 

docs/modules/spark-k8s/pages/usage-guide/s3.adoc

@@ -2,6 +2,8 @@
 
 You can specify S3 connection details directly inside the `SparkApplication` specification or by referring to an external `S3Bucket` custom resource.
 
+== S3 access using credentials
+
 To specify S3 connection details directly as part of the `SparkApplication` resource you add an inline connection configuration as shown below.
 
 [source,yaml]
@@ -17,7 +19,7 @@ s3connection: # <1>
 <1> Entry point for the S3 connection configuration.
 <2> Connection host.
 <3> Optional connection port.
-<4> Name of the `Secret` object expected to contain the following keys: `ACCESS_KEY_ID` and `SECRET_ACCESS_KEY`
+<4> Name of the `Secret` object expected to contain the following keys: `accessKey` and `secretKey`
 
 It is also possible to configure the  connection details as a separate Kubernetes resource and only refer to that object from the `SparkApplication` like this:
 
@@ -46,3 +48,44 @@ spec:
 ----
 
 This has the advantage that one connection configuration can be shared across `SparkApplications` and reduces the cost of updating these details.
+
+== S3 access with TLS
+
+A custom certificate can also be used for S3 access. In the example below, a Secret containing a custom certificate is referenced, which will used a to create a custom truststore which is used by Spark for S3-bucket access:
+
+[source,yaml]
+----
+---
+apiVersion: s3.stackable.tech/v1alpha1
+kind: S3Connection
+metadata:
+  name: s3-connection-resource
+spec:
+  host: test-minio
+  port: 9000
+  accessStyle: Path
+  credentials:
+    secretClass: minio-credentials-class  # <1>
+  tls:
+    verification:
+      server:
+        caCert:
+          secretClass: minio-tls-certificates  # <2>
+----
+<1> Name of the `Secret` object expected to contain the following keys: `accessKey` and `secretKey` (as in the previous example).
+<2> Name of the `Secret` object containing the custom certificate. The certificate should comprise the 3 files named as shown below:
+
+[source,yaml]
+----
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: minio-tls-certificates
+  labels:
+    secrets.stackable.tech/class: minio-tls-certificates
+data:
+  ca.crt: ...
+  tls.crt: ...
+  tls.key: ...
+----

rust/crd/src/constants.rs

@@ -28,6 +28,13 @@ pub const LOG4J2_CONFIG_FILE: &str = "log4j2.properties";
 pub const ACCESS_KEY_ID: &str = "accessKey";
 pub const SECRET_ACCESS_KEY: &str = "secretKey";
 pub const S3_SECRET_DIR_NAME: &str = "/stackable/secrets";
+pub const SYSTEM_TRUST_STORE: &str = "/etc/pki/java/cacerts";
+pub const STACKABLE_TRUST_STORE: &str = "/stackable/truststore";
+pub const STACKABLE_TRUST_STORE_NAME: &str = "stackable-truststore";
+pub const STACKABLE_TLS_STORE_PASSWORD: &str = "changeit";
+pub const SYSTEM_TRUST_STORE_PASSWORD: &str = "changeit";
+pub const STACKABLE_MOUNT_PATH_TLS: &str = "/stackable/mount_server_tls";
+pub const STACKABLE_MOUNT_NAME_TLS: &str = "servertls";
 
 pub const MIN_MEMORY_OVERHEAD: u32 = 384;
 pub const JVM_OVERHEAD_FACTOR: f32 = 0.1;

rust/crd/src/lib.rs

@@ -4,6 +4,7 @@ pub mod affinity;
 pub mod constants;
 pub mod history;
 pub mod s3logdir;
+pub mod tlscerts;
 
 use std::{
     cmp::max,
@@ -18,7 +19,7 @@ use s3logdir::S3LogDir;
 use serde::{Deserialize, Serialize};
 use snafu::{OptionExt, ResultExt, Snafu};
 use stackable_operator::{
-    builder::VolumeBuilder,
+    builder::{SecretOperatorVolumeSourceBuilder, VolumeBuilder},
     commons::{
         affinity::{StackableAffinity, StackableAffinityFragment},
         resources::{
@@ -332,6 +333,21 @@ impl SparkApplication {
                 .build(),
         );
 
+        if let Some(cert_secrets) = tlscerts::tls_secret_names(s3conn, s3logdir) {
+            result.push(
+                VolumeBuilder::new(STACKABLE_TRUST_STORE_NAME)
+                    .with_empty_dir(None::<String>, Some(Quantity("5Mi".to_string())))
+                    .build(),
+            );
+            for cert_secret in cert_secrets {
+                result.push(
+                    VolumeBuilder::new(cert_secret)
+                        .ephemeral(SecretOperatorVolumeSourceBuilder::new(cert_secret).build())
+                        .build(),
+                );
+            }
+        }
+
         result
     }
 
@@ -427,6 +443,22 @@ impl SparkApplication {
             ..VolumeMount::default()
         });
 
+        if let Some(cert_secrets) = tlscerts::tls_secret_names(s3conn, s3logdir) {
+            mounts.push(VolumeMount {
+                name: STACKABLE_TRUST_STORE_NAME.into(),
+                mount_path: STACKABLE_TRUST_STORE.into(),
+                ..VolumeMount::default()
+            });
+            for cert_secret in cert_secrets {
+                let secret_dir = format!("{STACKABLE_MOUNT_PATH_TLS}/{cert_secret}");
+                mounts.push(VolumeMount {
+                    name: cert_secret.to_string(),
+                    mount_path: secret_dir,
+                    ..VolumeMount::default()
+                });
+            }
+        }
+
         mounts
     }
 
@@ -508,6 +540,12 @@ impl SparkApplication {
             }
         }
 
+        // s3 with TLS
+        if tlscerts::tls_secret_names(s3conn, s3_log_dir).is_some() {
+            submit_cmd.push(format!("--conf spark.driver.extraJavaOptions=\"-Djavax.net.ssl.trustStore={STACKABLE_TRUST_STORE}/truststore.p12 -Djavax.net.ssl.trustStorePassword={STACKABLE_TLS_STORE_PASSWORD} -Djavax.net.ssl.trustStoreType=pkcs12 -Djavax.net.debug=ssl,handshake\""));
+            submit_cmd.push(format!("--conf spark.executor.extraJavaOptions=\"-Djavax.net.ssl.trustStore={STACKABLE_TRUST_STORE}/truststore.p12 -Djavax.net.ssl.trustStorePassword={STACKABLE_TLS_STORE_PASSWORD}  -Djavax.net.ssl.trustStoreType=pkcs12 -Djavax.net.debug=ssl,handshake\""));
+        }
+
         // repositories and packages arguments
         if let Some(deps) = self.spec.deps.clone() {
             submit_cmd.extend(
@@ -675,7 +713,11 @@ impl SparkApplication {
         Ok(format!("{}m", original_memory - deduction))
     }
 
-    pub fn env(&self) -> Vec<EnvVar> {
+    pub fn env(
+        &self,
+        s3conn: &Option<S3ConnectionSpec>,
+        s3logdir: &Option<S3LogDir>,
+    ) -> Vec<EnvVar> {
         let tmp = self.spec.env.as_ref();
         let mut e: Vec<EnvVar> = tmp.iter().flat_map(|e| e.iter()).cloned().collect();
         if self.requirements().is_some() {
@@ -687,6 +729,25 @@ impl SparkApplication {
                 value_from: None,
             });
         }
+        if tlscerts::tls_secret_names(s3conn, s3logdir).is_some() {
+            e.push(EnvVar {
+                name: "STACKABLE_TLS_STORE_PASSWORD".to_string(),
+                value: Some(STACKABLE_TLS_STORE_PASSWORD.to_string()),
+                value_from: None,
+            });
+        }
+        if let Some(s3logdir) = s3logdir {
+            if tlscerts::tls_secret_name(&s3logdir.bucket.connection).is_some() {
+                e.push(EnvVar {
+                    name: "SPARK_DAEMON_JAVA_OPTS".to_string(),
+                    value: Some(format!(
+                        "-Djavax.net.ssl.trustStore={STACKABLE_TRUST_STORE}/truststore.p12 -Djavax.net.ssl.trustStorePassword={STACKABLE_TLS_STORE_PASSWORD} -Djavax.net.ssl.trustStoreType=pkcs12"
+                    )),
+                    value_from: None,
+                });
+            }
+        }
+
         e
     }
 
@@ -811,6 +872,7 @@ pub enum SparkContainer {
     Requirements,
     Spark,
     Vector,
+    Tls,
 }
 
 #[derive(Clone, Debug, Default, Fragment, JsonSchema, PartialEq)]

rust/crd/src/s3logdir.rs

@@ -4,8 +4,10 @@ use crate::{
         LogFileDirectorySpec::{self, S3},
         S3LogFileDirectorySpec,
     },
+    tlscerts,
 };
 use stackable_operator::{
+    builder::{SecretOperatorVolumeSourceBuilder, VolumeBuilder},
     commons::{
         s3::{InlinedS3BucketSpec, S3AccessStyle},
         secret_class::SecretClassVolume,
@@ -79,9 +81,7 @@ impl S3LogDir {
                     TlsVerification::Server(server_verification) => {
                         match &server_verification.ca_cert {
                             CaCert::WebPki {} => {}
-                            CaCert::SecretClass(_) => {
-                                return S3TlsCaVerificationNotSupportedSnafu.fail()
-                            }
+                            CaCert::SecretClass(_) => {}
                         }
                     }
                 }
@@ -120,6 +120,7 @@ impl S3LogDir {
                 );
             }
         }
+
         result
     }
 
@@ -177,6 +178,35 @@ impl S3LogDir {
         )
     }
 
+    pub fn volumes(&self) -> Vec<Volume> {
+        let mut volumes: Vec<Volume> = self.credentials_volume().into_iter().collect();
+
+        if let Some(secret_name) = tlscerts::tls_secret_name(&self.bucket.connection) {
+            volumes.push(
+                VolumeBuilder::new(secret_name)
+                    .ephemeral(SecretOperatorVolumeSourceBuilder::new(secret_name).build())
+                    .build(),
+            );
+        }
+        volumes
+    }
+
+    pub fn volume_mounts(&self) -> Vec<VolumeMount> {
+        let mut volume_mounts: Vec<VolumeMount> =
+            self.credentials_volume_mount().into_iter().collect();
+
+        if let Some(secret_name) = tlscerts::tls_secret_name(&self.bucket.connection) {
+            let secret_dir = format!("{STACKABLE_MOUNT_PATH_TLS}/{secret_name}");
+
+            volume_mounts.push(VolumeMount {
+                name: secret_name.to_string(),
+                mount_path: secret_dir,
+                ..VolumeMount::default()
+            });
+        }
+        volume_mounts
+    }
+
     pub fn credentials_volume(&self) -> Option<Volume> {
         self.credentials()
             .map(|credentials| credentials.to_volume(credentials.secret_class.as_ref()))

rust/crd/src/tlscerts.rs

@@ -0,0 +1,62 @@
+use stackable_operator::commons::{
+    s3::S3ConnectionSpec,
+    tls::{CaCert, TlsVerification},
+};
+
+use crate::{
+    constants::{
+        STACKABLE_MOUNT_PATH_TLS, STACKABLE_TLS_STORE_PASSWORD, STACKABLE_TRUST_STORE,
+        SYSTEM_TRUST_STORE, SYSTEM_TRUST_STORE_PASSWORD,
+    },
+    s3logdir::S3LogDir,
+};
+
+pub fn tls_secret_name(s3conn: &Option<S3ConnectionSpec>) -> Option<&str> {
+    if let Some(conn) = s3conn.as_ref() {
+        if let Some(tls) = &conn.tls {
+            if let TlsVerification::Server(verification) = &tls.verification {
+                if let CaCert::SecretClass(secret_name) = &verification.ca_cert {
+                    return Some(secret_name);
+                }
+            }
+        }
+    }
+    None
+}
+
+pub fn tls_secret_names<'a>(
+    s3conn: &'a Option<S3ConnectionSpec>,
+    s3logdir: &'a Option<S3LogDir>,
+) -> Option<Vec<&'a str>> {
+    let mut names = Vec::new();
+
+    if let Some(secret_name) = tls_secret_name(s3conn) {
+        names.push(secret_name);
+    }
+
+    if let Some(logdir) = s3logdir {
+        if let Some(secret_name) = tls_secret_name(&logdir.bucket.connection) {
+            names.push(secret_name);
+        }
+    }
+    if names.is_empty() {
+        None
+    } else {
+        Some(names)
+    }
+}
+
+pub fn create_key_and_trust_store() -> Vec<String> {
+    vec![
+        format!("keytool -importkeystore -srckeystore {SYSTEM_TRUST_STORE} -srcstoretype jks -srcstorepass {SYSTEM_TRUST_STORE_PASSWORD} -destkeystore {STACKABLE_TRUST_STORE}/truststore.p12 -deststoretype pkcs12 -deststorepass {STACKABLE_TLS_STORE_PASSWORD} -noprompt"),
+    ]
+}
+
+pub fn add_cert_to_stackable_truststore(secret_name: &str) -> Vec<String> {
+    vec![
+        format!("echo [{STACKABLE_MOUNT_PATH_TLS}/{secret_name}/ca.crt] Adding cert..."),
+        format!("keytool -importcert -file {STACKABLE_MOUNT_PATH_TLS}/{secret_name}/ca.crt -alias stackable-{secret_name} -keystore {STACKABLE_TRUST_STORE}/truststore.p12 -storepass {STACKABLE_TLS_STORE_PASSWORD} -noprompt"),
+        format!("echo [{STACKABLE_MOUNT_PATH_TLS}/{secret_name}/ca.crt] Checking for cert..."),
+        format!("keytool -list -keystore {STACKABLE_TRUST_STORE}/truststore.p12 -storepass {STACKABLE_TLS_STORE_PASSWORD} -noprompt | grep stackable"),
+    ]
+}