docs: add section about in transit encryption

docs/modules/spark-k8s/examples/example-sparkapp-encryption-crypto.yaml

@@ -0,0 +1,54 @@
+---
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
+metadata:
+  name: spark-encryption-crypto
+spec:
+  sparkImage:
+    productVersion: 4.1.2
+  mode: cluster
+  mainApplicationFile: local:///stackable/spark/jobs/my-job.py
+  sparkConf:
+    spark.authenticate: "true" # <1>
+    spark.network.crypto.enabled: "true" # <2>
+    spark.io.encryption.enabled: "true" # <3>
+    spark.io.encryption.keySizeBits: "256"
+    spark.ssl.ui.enabled: "true" # <4>
+    spark.ssl.ui.keyStore: /stackable/tls/keystore.p12
+    spark.ssl.ui.keyStorePassword: ""
+    spark.ssl.ui.keyStoreType: PKCS12
+    spark.ssl.ui.protocol: TLSv1.3
+  job:
+    config:
+      volumeMounts:
+        - name: jobs
+          mountPath: /stackable/spark/jobs
+  driver:
+    config:
+      volumeMounts: &mounts
+        - name: jobs
+          mountPath: /stackable/spark/jobs
+        - name: tls
+          mountPath: /stackable/tls
+  executor:
+    replicas: 2
+    config:
+      volumeMounts: *mounts
+  volumes:
+    - name: jobs
+      configMap:
+        name: my-job
+    - name: tls # <5>
+      ephemeral:
+        volumeClaimTemplate:
+          metadata:
+            annotations:
+              secrets.stackable.tech/class: tls
+              secrets.stackable.tech/scope: pod
+              secrets.stackable.tech/format: tls-pkcs12
+          spec:
+            accessModes: ["ReadWriteOnce"]
+            resources:
+              requests:
+                storage: "1"
+            storageClassName: secrets.stackable.tech

docs/modules/spark-k8s/pages/usage-guide/security/encryption-in-transit.adoc

@@ -0,0 +1,128 @@
+= Encryption in transit
+:description: Encrypt Apache Spark RPC and block transfer, shuffle and spill data, and the Web UI.
+
+Spark protects traffic and on-disk data with three unrelated mechanisms, each configured by its own set of properties.
+For the full property reference see the https://spark.apache.org/docs/latest/security.html[Spark security documentation]; this page covers what is specific to running Spark on the Stackable Data Platform (SDP).
+
+[cols="1,2,2"]
+|===
+| What you want to protect | Property namespace | Notes
+
+| RPC (control plane) and block transfer (shuffle data over the network)
+| `spark.network.crypto.\*` or `spark.ssl.rpc.*`
+| Two alternatives, described below. Pick one, never both.
+
+| Web UI and history server
+| `spark.ssl.ui.\*`, `spark.ssl.historyServer.*`
+| Independent of the RPC choice.
+
+| Shuffle files, shuffle spills, on-disk cached and broadcast blocks
+| `spark.io.encryption.*`
+| At-rest encryption of Spark's own temporary files, not a transport setting.
+|===
+
+WARNING: `spark.ssl.enabled=true` does not enable RPC encryption; `spark.ssl.rpc.enabled` must be set explicitly.
+
+== Authentication
+
+Both transport-encryption options build on Spark's shared-secret authentication, so set `spark.authenticate: "true"` first.
+On Kubernetes Spark generates and propagates the secret itself, so no `spark.authenticate.secret` is needed.
+The driver and the executors log `authentication enabled` once this is active.
+
+== Encrypting RPC with `spark.network.crypto` (recommended)
+
+Encrypts RPC *and* block transfer with AES, keyed off the authentication secret above.
+Upstream documents this as RPC encryption only; block transfer runs over the same transport and is covered as well.
+There are no keystores or passwords to deliver:
+
+[source,yaml]
+----
+sparkConf:
+  spark.authenticate: "true"
+  spark.network.crypto.enabled: "true"
+----
+
+== Encrypting RPC with TLS (`spark.ssl.rpc`)
+
+Use this when a policy explicitly demands TLS on internal traffic.
+Two requirements apply:
+
+1. The store passwords must reach the driver and the executors as environment variables.
+   Spark strips every `spark.ssl.*Password` property from the executor startup configuration and expects the password to arrive in `_SPARK_SSL_RPC_KEY_STORE_PASSWORD`, `_SPARK_SSL_RPC_KEY_PASSWORD` and `_SPARK_SSL_RPC_TRUST_STORE_PASSWORD` instead.
+   Use `spec.env`, which the operator propagates to the job pod and to both the driver and the executor pods.
+2. The PKCS#12 passphrase must not be empty.
+   The Stackable Secret Operator generates stores with an empty passphrase by default.
+   That is fine for the Web UI, but the RPC keystore holds a private key and cannot be loaded without a real password.
+
+Only the parts that differ from the <<complete-example,complete example>> are shown below; the volume mounts are the same.
+
+[source,yaml]
+----
+# SparkApplication, showing only what differs from the complete example
+spec:
+  sparkConf:
+    spark.authenticate: "true"
+    spark.ssl.rpc.enabled: "true" # <1>
+    spark.ssl.rpc.keyStore: /stackable/tls/keystore.p12
+    spark.ssl.rpc.keyStoreType: PKCS12
+    spark.ssl.rpc.trustStore: /stackable/tls/truststore.p12
+    spark.ssl.rpc.trustStoreType: PKCS12
+    spark.ssl.rpc.protocol: TLSv1.3 # <2>
+  env: # <3>
+    - name: _SPARK_SSL_RPC_KEY_STORE_PASSWORD
+      value: "changeit" # <4>
+    - name: _SPARK_SSL_RPC_KEY_PASSWORD
+      value: "changeit" # <4>
+    - name: _SPARK_SSL_RPC_TRUST_STORE_PASSWORD
+      value: "changeit" # <4>
+  volumes:
+    - name: tls
+      ephemeral:
+        volumeClaimTemplate:
+          metadata:
+            annotations:
+              secrets.stackable.tech/format.compatibility.tls-pkcs12.password: "changeit" # <4>
+----
+<1> Covers RPC and block transfer. Do not combine with `spark.network.crypto.enabled`.
+<2> `spark.ssl.rpc.protocol` has no default and must be set explicitly.
+<3> The store passwords, delivered to the job pod and to both the driver and the executor pods.
+<4> The PKCS#12 passphrase. Must be non-empty and identical in all four places.
+    It is hardcoded on purpose: the volume annotation only accepts a literal value, so the passphrase is visible in the `SparkApplication` either way.
+    Reading only the environment variables from a `Secret` would suggest a confidentiality that is not there.
+    If you also enable the Web UI over TLS, `spark.ssl.ui.keyStorePassword` must carry this same passphrase.
+
+=== Verifying that it took effect
+
+The driver and every executor log `RPC SSL enabled` when the mode is active, and `RPC SSL disabled` when RPC and block transfer are in plaintext.
+The two requirements above fail with distinct symptoms:
+
+* Missing password environment variables -- the driver fails during startup and no executor is created; look for `SSLFactory creation failed`.
+* Empty PKCS#12 passphrase -- both sides report `RPC SSL enabled` but no executor registers, and the driver logs a TLS `handshake_failure`.
+
+[IMPORTANT]
+====
+Collect these logs while the application is running.
+When an application reaches a terminal phase the operator deletes the driver pod -- regardless of `spark.kubernetes.driver.deleteOnTermination` -- and the executor pods are garbage-collected along with it, because they are owned by the driver pod.
+====
+
+== Web UI over TLS
+
+Unlike the RPC keystore, the UI keystore works with the Stackable Secret Operator's default empty passphrase; see the <<complete-example,complete example>>.
+
+The TLS listener binds to the UI port plus 400 (`4440` for the default `4040`).
+The plain port stays bound and answers with a `302` redirect to the TLS port; it serves no content of its own.
+
+[#complete-example]
+== Complete example
+
+The recommended combination -- authenticated, AES-encrypted transport, encrypted spill data and a TLS Web UI:
+
+[source,yaml]
+----
+include::example$example-sparkapp-encryption-crypto.yaml[]
+----
+<1> Shared-secret authentication; the secret is generated and propagated automatically.
+<2> RPC and block transfer encrypted with AES.
+<3> Shuffle files, spills, and on-disk cached and broadcast blocks encrypted at rest.
+<4> Web UI over TLS.
+<5> Provides the `keystore.p12` used above, from the `tls` SecretClass.

docs/modules/spark-k8s/pages/usage-guide/security.adoc → docs/modules/spark-k8s/pages/usage-guide/security/index.adoc

@@ -1,5 +1,6 @@
 = Security
 :description: Learn how to configure Apache Spark applications with Kerberos authentication using Stackable Secret Operator for secure data access in HDFS.
+:page-aliases: usage-guide/security.adoc
 
 == Authentication
 
@@ -9,6 +10,8 @@ Kerberos is a network authentication protocol that works on the basis of "ticket
 
 In this guide we show how to configure Spark applications to use Kerberos while accessing data in an HDFS cluster. The Stackable Secret Operator is used to generate the keytab files. In production environments, users might have different means to provision the keytab files.
 
+This page covers authentication against external services only. To encrypt traffic between the Spark components themselves, see xref:usage-guide/security/encryption-in-transit.adoc[].
+
 
 == Prerequisites
 

docs/modules/spark-k8s/partials/nav.adoc

@@ -7,7 +7,8 @@
 ** xref:spark-k8s:usage-guide/resources.adoc[]
 ** xref:spark-k8s:usage-guide/s3.adoc[]
 ** xref:spark-k8s:usage-guide/app_templates.adoc[]
-** xref:spark-k8s:usage-guide/security.adoc[]
+** xref:spark-k8s:usage-guide/security/index.adoc[]
+*** xref:spark-k8s:usage-guide/security/encryption-in-transit.adoc[]
 ** xref:spark-k8s:usage-guide/logging.adoc[]
 ** xref:spark-k8s:usage-guide/history-server.adoc[]
 ** xref:spark-k8s:usage-guide/spark-connect.adoc[]

tests/templates/kuttl/transit-encryption/00-patch-ns.yaml.j2

@@ -0,0 +1,9 @@
+{% if test_scenario['values']['openshift'] == 'true' %}
+# see https://github.com/stackabletech/issues/issues/566
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}'
+    timeout: 120
+{% endif %}

tests/templates/kuttl/transit-encryption/01-assert.yaml.j2

@@ -0,0 +1,10 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+{% endif %}

tests/templates/kuttl/transit-encryption/01-install-vector-aggregator-discovery-configmap.yaml.j2

@@ -0,0 +1,9 @@
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+data:
+  ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }}
+{% endif %}

tests/templates/kuttl/transit-encryption/10-assert.yaml

@@ -0,0 +1,11 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 900
+---
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
+metadata:
+  name: spark-encryption-crypto
+status:
+  phase: Succeeded

tests/templates/kuttl/transit-encryption/10-deploy-spark-app.yaml.j2

@@ -0,0 +1,143 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: my-job
+data:
+  my-job.py: |
+    """Verify that authentication, encrypted transport and a TLS Web UI are active.
+
+    Configuration errors in any of the three mechanisms make the driver fail
+    during startup, so reaching the end of this script already covers a lot.
+    On top of that the settings are read back from the running SparkContext,
+    a shuffle is forced across executors, and both Web UI ports are probed.
+
+    A failing assertion here surfaces as phase "Failed" only because the image
+    entrypoint propagates the driver exit code, see the signal-propagation test.
+    """
+    import ssl
+    import urllib.error
+    import urllib.request
+
+    from pyspark.sql import SparkSession
+
+    EXPECTED_CONF = {
+        "spark.authenticate": "true",
+        "spark.network.crypto.enabled": "true",
+        "spark.io.encryption.enabled": "true",
+        "spark.io.encryption.keySizeBits": "256",
+        "spark.ssl.ui.enabled": "true",
+    }
+
+    spark = SparkSession.builder.appName("spark-encryption-crypto").getOrCreate()
+    conf = spark.sparkContext.getConf()
+
+    for key, expected in EXPECTED_CONF.items():
+        actual = conf.get(key, None)
+        assert actual == expected, f"{key}: expected {expected!r}, got {actual!r}"
+
+    # Force a shuffle, so that encrypted block transfer between the executors and
+    # the encrypted shuffle files are exercised and not just configured.
+    counts = (
+        spark.sparkContext.parallelize(range(7000), 8)
+        .map(lambda i: (i % 7, 1))
+        .reduceByKey(lambda a, b: a + b)
+        .collectAsMap()
+    )
+    assert counts == {i: 1000 for i in range(7)}, f"unexpected shuffle result: {counts}"
+
+    # The TLS listener binds to the UI port plus 400, the plain port redirects to it.
+    ui_port = int(conf.get("spark.ui.port", "4040"))
+    tls_port = ui_port + 400
+
+    # The certificate is issued by the Secret Operator CA, which is not in the
+    # image's trust store, so only the TLS handshake itself is checked here.
+    with urllib.request.urlopen(
+        f"https://localhost:{tls_port}/", context=ssl._create_unverified_context(), timeout=30
+    ) as response:
+        assert response.status == 200, f"TLS UI returned {response.status}"
+
+    class NoRedirects(urllib.request.HTTPRedirectHandler):
+        def redirect_request(self, *args, **kwargs):
+            return None
+
+    try:
+        urllib.request.build_opener(NoRedirects).open(f"http://localhost:{ui_port}/", timeout=30)
+        raise AssertionError(f"expected a redirect from the plain UI port {ui_port}")
+    except urllib.error.HTTPError as error:
+        assert error.code == 302, f"plain UI port returned {error.code}, expected 302"
+        location = error.headers["Location"]
+        assert location.startswith("https://") and f":{tls_port}" in location, (
+            f"plain UI port redirects to {location}, expected the TLS port {tls_port}"
+        )
+
+    spark.stop()
+---
+# Mirrors docs/modules/spark-k8s/examples/example-sparkapp-encryption-crypto.yaml.
+# Keep both in sync: the docs page claims this combination works.
+apiVersion: spark.stackable.tech/v1alpha1
+kind: SparkApplication
+metadata:
+  name: spark-encryption-crypto
+spec:
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+  vectorAggregatorConfigMapName: vector-aggregator-discovery
+{% endif %}
+  sparkImage:
+{% if test_scenario['values']['spark'].find(",") > 0 %}
+    custom: "{{ test_scenario['values']['spark'].split(',')[1] }}"
+    productVersion: "{{ test_scenario['values']['spark'].split(',')[0] }}"
+{% else %}
+    productVersion: "{{ test_scenario['values']['spark'] }}"
+{% endif %}
+    pullPolicy: IfNotPresent
+  mode: cluster
+  mainApplicationFile: local:///stackable/spark/jobs/my-job.py
+  sparkConf:
+    spark.authenticate: "true"
+    spark.network.crypto.enabled: "true"
+    spark.io.encryption.enabled: "true"
+    spark.io.encryption.keySizeBits: "256"
+    spark.ssl.ui.enabled: "true"
+    spark.ssl.ui.keyStore: /stackable/tls/keystore.p12
+    spark.ssl.ui.keyStorePassword: ""
+    spark.ssl.ui.keyStoreType: PKCS12
+    spark.ssl.ui.protocol: TLSv1.3
+  job:
+    config:
+      volumeMounts:
+        - name: jobs
+          mountPath: /stackable/spark/jobs
+  driver:
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+      volumeMounts: &mounts
+        - name: jobs
+          mountPath: /stackable/spark/jobs
+        - name: tls
+          mountPath: /stackable/tls
+  executor:
+    replicas: 2
+    config:
+      logging:
+        enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }}
+      volumeMounts: *mounts
+  volumes:
+    - name: jobs
+      configMap:
+        name: my-job
+    - name: tls
+      ephemeral:
+        volumeClaimTemplate:
+          metadata:
+            annotations:
+              secrets.stackable.tech/class: tls
+              secrets.stackable.tech/scope: pod
+              secrets.stackable.tech/format: tls-pkcs12
+          spec:
+            accessModes: ["ReadWriteOnce"]
+            resources:
+              requests:
+                storage: "1"
+            storageClassName: secrets.stackable.tech