feat: Add OPA support to KRaft controllers

CHANGELOG.md

@@ -13,6 +13,9 @@ All notable changes to this project will be documented in this file.
 - Support for Kafka 4.3.1 ([#1022]).
 - Add `/ready` endpoint to the operator Deployment, which reports the CRD installation status ([#1029]).
 - Brokers now have a default affinity to the OPA Pods when OPA authorization is configured ([#1035]).
+- The OPA authorizer is now also configured on KRaft controllers. Previously, requests that brokers
+  forward to the controllers (e.g. creating or deleting topics) were not authorized at all.
+  The Kafka nodes are super users, so that traffic between them does not hit OPA ([#1051]).
 
 ### Changed
 
@@ -100,6 +103,7 @@ All notable changes to this project will be documented in this file.
 [#1046]: https://github.com/stackabletech/kafka-operator/pull/1046
 [#1047]: https://github.com/stackabletech/kafka-operator/pull/1047
 [#1049]: https://github.com/stackabletech/kafka-operator/pull/1049
+[#1051]: https://github.com/stackabletech/kafka-operator/pull/1051
 [#1052]: https://github.com/stackabletech/kafka-operator/pull/1052
 
 ## [26.7.0] - 2026-07-21

rust/operator-binary/src/controller/build/command.rs

@@ -157,6 +157,8 @@ pub fn controller_kafka_container_command(
         containerdebug --output={STACKABLE_LOG_DIR}/containerdebug-state.json --loop &
         {set_realm_env}
 
+        {import_opa_tls_cert}
+
         {derive_pod_index}
         {export_replica_id}
 
@@ -180,6 +182,7 @@ pub fn controller_kafka_container_command(
         remove_vector_shutdown_file_command = remove_vector_shutdown_file_command(STACKABLE_LOG_DIR),
         // Mirrors `broker_kafka_container_commands`: empty when Kerberos is disabled.
         set_realm_env = export_kerberos_realm_command(kafka_security).unwrap_or_default(),
+        import_opa_tls_cert = copy_opa_tls_cert_command(kafka_security),
         derive_pod_index = DERIVE_POD_INDEX,
         export_replica_id = EXPORT_REPLICA_ID,
         config_dir = STACKABLE_CONFIG_DIR,

rust/operator-binary/src/controller/build/mod.rs

@@ -715,6 +715,7 @@ mod tests {
                     &*KERBEROS_VOLUME_NAME,
                     &*STACKABLE_LOG_CONFIG_DIR_NAME,
                     &*STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME,
+                    &*OPA_TLS_VOLUME_NAME,
                     &*STACKABLE_CONFIG_DIR_NAME,
                     &*STACKABLE_LOG_DIR_NAME,
                 ],

rust/operator-binary/src/controller/build/properties/broker_properties.rs

@@ -1,6 +1,6 @@
 use std::collections::BTreeMap;
 
-use super::kraft_controllers;
+use super::{kraft_controllers, opa_authorizer_properties};
 use crate::{
     controller::{
         ValidatedClusterConfig,
@@ -95,23 +95,7 @@ pub fn build(
         }
     }
 
-    // Enable OPA authorization
-    if let Some(opa_connect_string) = cluster_config.opa_connect() {
-        result.extend([
-            (
-                "authorizer.class.name".to_string(),
-                "org.openpolicyagent.kafka.OpaAuthorizer".to_string(),
-            ),
-            (
-                "opa.authorizer.metrics.enabled".to_string(),
-                "true".to_string(),
-            ),
-            (
-                "opa.authorizer.url".to_string(),
-                opa_connect_string.to_string(),
-            ),
-        ]);
-    }
+    result.extend(opa_authorizer_properties(cluster_config, pod_descriptors));
 
     result.extend(broker_config_settings(&cluster_config.kafka_security));
     result.extend(graceful_shutdown_config_properties());

rust/operator-binary/src/controller/build/properties/controller_properties.rs

@@ -1,12 +1,12 @@
 use std::collections::BTreeMap;
 
-use super::kraft_controllers;
+use super::{kraft_controllers, opa_authorizer_properties};
 use crate::{
     controller::{
         ValidatedClusterConfig,
         build::{
             graceful_shutdown::graceful_shutdown_config_properties,
-            security::controller_config_settings,
+            security::{SSL_CLIENT_AUTH_REQUIRED, controller_config_settings},
         },
     },
     crd::{
@@ -73,6 +73,18 @@ pub fn build(
         );
     }
 
+    result.extend(opa_authorizer_properties(cluster_config, pod_descriptors));
+    if cluster_config.opa_connect().is_some()
+        && !cluster_config.kafka_security.has_kerberos_enabled()
+    {
+        // The nodes are identified as super users by the subject of their internal certificate.
+        // Without client authentication, the principal on the CONTROLLER listener is ANONYMOUS.
+        // With Kerberos, the principal is derived from the SASL authentication instead.
+        result.insert(
+            KafkaListenerName::Controller.listener_ssl_client_auth(),
+            SSL_CLIENT_AUTH_REQUIRED.to_string(),
+        );
+    }
     result.extend(controller_config_settings(&cluster_config.kafka_security));
     result.extend(graceful_shutdown_config_properties());
     result.extend(overrides);

rust/operator-binary/src/controller/build/properties/mod.rs

@@ -6,9 +6,12 @@ pub mod listener;
 pub mod product_logging;
 pub mod security_properties;
 
-use std::collections::BTreeSet;
+use std::collections::{BTreeMap, BTreeSet};
 
-use crate::crd::{KafkaPodDescriptor, role::KafkaRole};
+use crate::{
+    controller::ValidatedClusterConfig,
+    crd::{KafkaPodDescriptor, role::KafkaRole},
+};
 
 /// The names of the config files assembled into the rolegroup `ConfigMap`.
 ///
@@ -105,6 +108,84 @@ pub(crate) fn kraft_controllers(
         .collect()
 }
 
+/// The OPA authorizer properties shared by brokers and controllers.
+///
+/// In KRaft mode, brokers forward requests such as `CreateTopics` to the controllers without
+/// authorizing them, so the controllers must run the authorizer too. Empty if OPA is not
+/// configured.
+///
+/// The Kafka nodes themselves are super users, so that traffic between them does not
+/// hit OPA. Forwarded requests are still authorized against the original client principal.
+pub(crate) fn opa_authorizer_properties(
+    cluster_config: &ValidatedClusterConfig,
+    pod_descriptors: &[KafkaPodDescriptor],
+) -> BTreeMap<String, String> {
+    let Some(opa_connect_string) = cluster_config.opa_connect() else {
+        return BTreeMap::new();
+    };
+
+    BTreeMap::from([
+        (
+            "authorizer.class.name".to_string(),
+            "org.openpolicyagent.kafka.OpaAuthorizer".to_string(),
+        ),
+        (
+            "opa.authorizer.metrics.enabled".to_string(),
+            "true".to_string(),
+        ),
+        (
+            "opa.authorizer.url".to_string(),
+            opa_connect_string.to_string(),
+        ),
+        (
+            "super.users".to_string(),
+            super_users(
+                pod_descriptors,
+                cluster_config.kafka_security.has_kerberos_enabled(),
+            ),
+        ),
+    ])
+}
+
+/// The `super.users` value: the principals of all Kafka nodes.
+///
+/// Internal traffic is authenticated with the pod certificates of the internal `SecretClass`
+/// (mTLS), whose subject DN is `CN=generated certificate for pod,DC=<pod>,DC=<service>,
+/// DC=<namespace>,DC=svc,DC=<cluster domain parts>`, see the secret-operator.
+/// With Kerberos, the node principals are mapped to the short name of the service.
+fn super_users(pod_descriptors: &[KafkaPodDescriptor], kerberos_enabled: bool) -> String {
+    let mut principals: BTreeSet<String> = pod_descriptors
+        .iter()
+        .map(|desc| {
+            // The certificate encodes `CN`, then the `DC`s from the pod up to the cluster domain
+            // suffix. Java prints the subject DN in RFC 2253 order, which is reversed.
+            let cluster_domain = desc.cluster_domain.to_string();
+            let domain_components = [
+                format!("{}-{}", desc.role_group_statefulset_name, desc.replica),
+                desc.role_group_service_name.to_string(),
+                desc.namespace.to_string(),
+                "svc".to_string(),
+            ]
+            .into_iter()
+            .chain(cluster_domain.split('.').map(str::to_string));
+            let mut rdns: Vec<String> = vec!["CN=generated certificate for pod".to_string()];
+            rdns.extend(domain_components.map(|dc| format!("DC={dc}")));
+            rdns.reverse();
+            format!("User:{}", rdns.join(","))
+        })
+        .collect();
+
+    if kerberos_enabled {
+        principals.extend(
+            [KafkaRole::Broker, KafkaRole::Controller]
+                .iter()
+                .map(|role| format!("User:{}", role.kerberos_service_name())),
+        );
+    }
+
+    principals.into_iter().collect::<Vec<_>>().join(";")
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -150,6 +231,28 @@ mod tests {
         }
     }
 
+    #[test]
+    fn super_users_contains_the_pod_certificate_subject_of_every_node() {
+        let pod_descriptors = vec![
+            pod_descriptor(KafkaRole::Controller, 0, 9093),
+            pod_descriptor(KafkaRole::Broker, 1, 9092),
+        ];
+
+        assert_eq!(
+            super_users(&pod_descriptors, false),
+            "User:DC=local,DC=cluster,DC=svc,DC=default,DC=kafka-controller-default-headless,DC=kafka-controller-default-0,CN=generated certificate for pod;\
+             User:DC=local,DC=cluster,DC=svc,DC=default,DC=kafka-controller-default-headless,DC=kafka-controller-default-1,CN=generated certificate for pod"
+        );
+    }
+
+    #[test]
+    fn super_users_contains_the_service_name_with_kerberos() {
+        let pod_descriptors = vec![pod_descriptor(KafkaRole::Controller, 0, 9093)];
+
+        assert!(super_users(&pod_descriptors, true).ends_with(";User:kafka"));
+        assert!(!super_users(&pod_descriptors, false).contains("User:kafka"));
+    }
+
     #[test]
     fn kraft_controllers_lists_individual_pod_fqdns_under_kerberos() {
         let pod_descriptors = vec![

rust/operator-binary/src/controller/build/role_group/controller.rs

@@ -332,7 +332,7 @@ fn role_pod(
     pod_volumes.push(log_config_volume(logging, &resource_names));
 
     // The TLS certificate is requested with the role group's secret lifetime.
-    pod_volumes.push(
+    pod_volumes.extend(
         mount_controller_tls_volume(
             kafka_security,
             &mut cb_kafka,