feat: OPA TLS support (#812)

* feat: OPA TLS support

* chore: changelog

* recfactor: OPA tls mount helper method

Author: dervoeti

CHANGELOG.md

@@ -16,6 +16,7 @@ All notable changes to this project will be documented in this file.
 - Add support for Trino 477 ([#801]).
 - Add support for Hive 4.1.0 ([#805]).
 - Add `prometheus.io/path|port|scheme` annotations to metrics service ([#807]).
+- Add support for OPA with TLS enabled ([#812]).
 
 ### Changed
 
@@ -45,6 +46,7 @@ All notable changes to this project will be documented in this file.
 [#805]: https://github.com/stackabletech/trino-operator/pull/805
 [#807]: https://github.com/stackabletech/trino-operator/pull/807
 [#810]: https://github.com/stackabletech/trino-operator/pull/810
+[#812]: https://github.com/stackabletech/trino-operator/pull/812
 
 ## [25.7.0] - 2025-07-23
 

rust/operator-binary/src/authorization/opa.rs

@@ -3,10 +3,14 @@ use std::collections::BTreeMap;
 use stackable_operator::{
     client::Client,
     commons::opa::{OpaApiVersion, OpaConfig},
+    k8s_openapi::api::core::v1::ConfigMap,
+    kube::ResourceExt,
 };
 
 use crate::crd::v1alpha1::TrinoCluster;
 
+pub const OPA_TLS_VOLUME_NAME: &str = "opa-tls";
+
 pub struct TrinoOpaConfig {
     /// URI for OPA policies, e.g.
     /// `http://localhost:8081/v1/data/trino/allow`
@@ -28,6 +32,10 @@ pub struct TrinoOpaConfig {
     /// such operations, they will be bulk allowed or denied depending
     /// on this setting
     pub(crate) allow_permission_management_operations: bool,
+    /// Optional TLS secret class for OPA communication.
+    /// If set, the CA certificate from this secret class will be added
+    /// to Trino's truststore to make it trust OPA's TLS certificate.
+    pub(crate) tls_secret_class: Option<String>,
 }
 
 impl TrinoOpaConfig {
@@ -66,12 +74,24 @@ impl TrinoOpaConfig {
                 OpaApiVersion::V1,
             )
             .await?;
+
+        let tls_secret_class = client
+            .get::<ConfigMap>(
+                &opa_config.config_map_name,
+                trino.namespace().as_deref().unwrap_or("default"),
+            )
+            .await
+            .ok()
+            .and_then(|cm| cm.data)
+            .and_then(|mut data| data.remove("OPA_SECRET_CLASS"));
+
         Ok(TrinoOpaConfig {
             non_batched_connection_string,
             batched_connection_string,
             row_filters_connection_string: Some(row_filters_connection_string),
             column_masking_connection_string: Some(column_masking_connection_string),
             allow_permission_management_operations: true,
+            tls_secret_class,
         })
     }
 
@@ -107,4 +127,10 @@ impl TrinoOpaConfig {
         }
         config
     }
+
+    pub fn tls_mount_path(&self) -> Option<String> {
+        self.tls_secret_class
+            .as_ref()
+            .map(|_| format!("/stackable/secrets/{OPA_TLS_VOLUME_NAME}"))
+    }
 }

rust/operator-binary/src/controller.rs

@@ -76,10 +76,10 @@ use strum::{EnumDiscriminants, IntoStaticStr};
 
 use crate::{
     authentication::{TrinoAuthenticationConfig, TrinoAuthenticationTypes},
-    authorization::opa::TrinoOpaConfig,
+    authorization::opa::{OPA_TLS_VOLUME_NAME, TrinoOpaConfig},
     catalog::{FromTrinoCatalogError, config::CatalogConfig},
-    command, config,
-    config::{client_protocol, fault_tolerant_execution},
+    command,
+    config::{self, client_protocol, fault_tolerant_execution},
     crd::{
         ACCESS_CONTROL_PROPERTIES, APP_NAME, CONFIG_DIR_NAME, CONFIG_PROPERTIES, Container,
         DISCOVERY_URI, ENV_INTERNAL_SECRET, ENV_SPOOLING_SECRET, EXCHANGE_MANAGER_PROPERTIES,
@@ -630,6 +630,7 @@ pub async fn reconcile_trino(
                 &rbac_sa.name_any(),
                 &resolved_fte_config,
                 &resolved_client_protocol_config,
+                &trino_opa_config,
             )?;
 
             cluster_resources
@@ -1037,6 +1038,7 @@ fn build_rolegroup_statefulset(
     sa_name: &str,
     resolved_fte_config: &Option<fault_tolerant_execution::ResolvedFaultTolerantExecutionConfig>,
     resolved_spooling_config: &Option<client_protocol::ResolvedClientProtocolConfig>,
+    trino_opa_config: &Option<TrinoOpaConfig>,
 ) -> Result<StatefulSet> {
     let role = trino
         .role(trino_role)
@@ -1140,6 +1142,7 @@ fn build_rolegroup_statefulset(
         &requested_secret_lifetime,
         resolved_fte_config,
         resolved_spooling_config,
+        trino_opa_config,
     )?;
 
     let mut prepare_args = vec![];
@@ -1165,6 +1168,17 @@ fn build_rolegroup_statefulset(
     prepare_args
         .extend(trino_authentication_config.commands(&TrinoRole::Coordinator, &Container::Prepare));
 
+    // Add OPA TLS certificate to truststore if configured
+    if let Some(tls_mount_path) = trino_opa_config
+        .as_ref()
+        .and_then(|opa_config| opa_config.tls_mount_path())
+    {
+        prepare_args.extend(command::add_cert_to_truststore(
+            format!("{}/ca.crt", tls_mount_path).as_str(),
+            STACKABLE_CLIENT_TLS_DIR,
+        ));
+    }
+
     let container_prepare = cb_prepare
         .image_from_product_image(resolved_product_image)
         .command(vec![
@@ -1710,6 +1724,7 @@ fn tls_volume_mounts(
     requested_secret_lifetime: &Duration,
     resolved_fte_config: &Option<fault_tolerant_execution::ResolvedFaultTolerantExecutionConfig>,
     resolved_spooling_config: &Option<client_protocol::ResolvedClientProtocolConfig>,
+    trino_opa_config: &Option<TrinoOpaConfig>,
 ) -> Result<()> {
     if let Some(server_tls) = trino.get_server_tls() {
         cb_prepare
@@ -1789,6 +1804,32 @@ fn tls_volume_mounts(
             .context(AddVolumeSnafu)?;
     }
 
+    // Add OPA TLS certs if configured
+    if let Some((tls_secret_class, tls_mount_path)) =
+        trino_opa_config.as_ref().and_then(|opa_config| {
+            opa_config
+                .tls_secret_class
+                .as_ref()
+                .zip(opa_config.tls_mount_path())
+        })
+    {
+        cb_prepare
+            .add_volume_mount(OPA_TLS_VOLUME_NAME, &tls_mount_path)
+            .context(AddVolumeMountSnafu)?;
+
+        let opa_tls_volume = VolumeBuilder::new(OPA_TLS_VOLUME_NAME)
+            .ephemeral(
+                SecretOperatorVolumeSourceBuilder::new(tls_secret_class)
+                    .build()
+                    .context(TlsCertSecretClassVolumeBuildSnafu)?,
+            )
+            .build();
+
+        pod_builder
+            .add_volume(opa_tls_volume)
+            .context(AddVolumeSnafu)?;
+    }
+
     // fault tolerant execution S3 credentials and other resources
     if let Some(resolved_fte) = resolved_fte_config {
         cb_prepare
@@ -2028,6 +2069,7 @@ mod tests {
                     .to_string(),
             ),
             allow_permission_management_operations: true,
+            tls_secret_class: None,
         });
         let resolved_fte_config = match &trino.spec.cluster_config.fault_tolerant_execution {
             Some(fault_tolerant_execution) => Some(

tests/templates/kuttl/opa-authorization/10-install-opa.yaml.j2

@@ -5,6 +5,19 @@ commands:
   - script: |
       kubectl apply -n $NAMESPACE -f - <<EOF
       ---
+      apiVersion: secrets.stackable.tech/v1alpha1
+      kind: SecretClass
+      metadata:
+        name: opa-tls-$NAMESPACE
+      spec:
+        backend:
+          autoTls:
+            ca:
+              autoGenerate: true
+              secret:
+                name: opa-tls-ca-$NAMESPACE
+                namespace: $NAMESPACE
+      ---
       apiVersion: opa.stackable.tech/v1alpha1
       kind: OpaCluster
       metadata:
@@ -19,6 +32,8 @@ commands:
 {% endif %}
           pullPolicy: IfNotPresent
         clusterConfig:
+          tls:
+            serverSecretClass: opa-tls-$NAMESPACE
           userInfo:
             backend:
               keycloak: