feat: Add OPA support to KRaft controllers

CHANGELOG.md

@@ -13,6 +13,10 @@ 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 now configured as super users, so that traffic between them does not hit OPA.
+  Controllers now get the same affinity to OPA as brokers ([#1051]).
 
 ### Changed
 
@@ -100,6 +104,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

docs/modules/kafka/pages/usage-guide/operations/cluster-operations.adoc

@@ -2,3 +2,12 @@
 = Cluster operation
 
 The managed Kafka instances can be configured with different cluster operations like pausing reconciliation or stopping the cluster. See xref:concepts:operations/cluster_operations.adoc[cluster operations] for more details.
+
+== [[scaling-restarts]]Scaling operations restart the whole cluster
+
+Changing the number of replicas of any Kafka role (brokers or controllers) restarts all Pods of the Kafka cluster, not only the Pods of the scaled role.
+
+Scaling a role changes the ConfigMap with the node properties.
+The restart controller detects this change and triggers a restart, which results in all Pods of the cluster being restarted.
+
+Plan scaling operations accordingly, for example by performing them during a maintenance window.

docs/modules/kafka/pages/usage-guide/operations/pod-placement.adoc

@@ -1,6 +1,6 @@
 = Pod placement
 
-You can configure Pod placement for Kafka brokers as described in xref:concepts:pod_placement.adoc[].
+You can configure Pod placement for Kafka brokers and controllers as described in xref:concepts:pod_placement.adoc[].
 
 By default, the operator configures the following Pod placement constraints:
 
@@ -21,8 +21,8 @@ affinity:
 
 In the example above `cluster-name` is the name of the Kafka custom resource that owns this Pod.
 
-If OPA authorization is configured, the broker Pods additionally prefer to run on the same Kubernetes nodes as the OPA Pods (weight 50).
-Only brokers are configured with the OPA authorizer, as they are the only role issuing OPA requests.
+If OPA authorization is configured, the broker and controller Pods additionally prefer to run on the same Kubernetes nodes as the OPA Pods (weight 50).
+Both roles run the OPA authorizer, so both issue OPA requests.
 
 [source,yaml]
 ----

docs/modules/kafka/pages/usage-guide/security.adoc

@@ -231,12 +231,31 @@ include::example$security/03b-authorization-kafkacluster-custom-properties.yaml[
 
 A full list of settings and their respective defaults can be found https://github.com/anderseknert/opa-kafka-plugin[here].
 
-The OPA Rego rules can authorize both external and internal communication.
-For internal communication, the principal name is the subject DN of the broker's certificate.
-The subject DN contains the FQDN of the pod and a generic common name, e.g. `DC=local,DC=cluster,DC=svc,DC=default,DC=simple-kafka-broker-default-headless,DC=simple-kafka-broker-default-0,CN=generated certificate for pod`.
-An OpaCluster and the Rego rules that verify the subject DNs could look as follows:
+The OPA Rego rules authorize requests from external clients.
+Communication between the Kafka nodes is not authorized by OPA, see <<authorization-internal-traffic>>.
 
-[source,yaml]
-----
-include::example$security/02-authorization-opa.yaml[]
-----
+=== [[authorization-internal-traffic]]Internal traffic
+
+The Kafka nodes (brokers and controllers) are configured as `super.users`, so that the traffic between them is allowed without asking OPA.
+You do not need to write Rego rules for it.
+Requests that a broker forwards to a controller on behalf of a client, such as creating a topic, are still authorized against the principal of the original client.
+
+The operator sets `super.users` to the principals of all nodes of the cluster:
+
+* The nodes authenticate each other with the TLS certificates that the secret operator provides for the `internalSecretClass`.
+  The principal is the subject DN of the certificate, which contains the FQDN of the pod and a generic common name, e.g. `DC=local,DC=cluster,DC=svc,DC=default,DC=simple-kafka-broker-default-headless,DC=simple-kafka-broker-default-0,CN=generated certificate for pod`.
+* If Kerberos is enabled, the short names of the Kerberos services of the brokers and the controllers are added as well.
+
+Kafka compares the principal of a request with each `super.users` entry for exact equality; wildcards are not supported.
+The common name `CN=generated certificate for pod` is therefore part of every certificate-based entry.
+The secret operator always sets it in the subject of the pod certificates.
+
+NOTE: Everything that presents a certificate with the subject DN of a node, issued by the CA of the `internalSecretClass`, is treated as a Kafka node.
+Restrict who can obtain such certificates accordingly.
+
+If you previously set `super.users` for the Kafka nodes yourself, for example with `configOverrides`, you can remove that override.
+The operator now sets the principals of all nodes automatically.
+Keep an override only if you need additional super users, such as specific client principals, and then include the entries for the nodes as well, because the override replaces the value set by the operator.
+
+NOTE: Scaling a role changes the nodes of the cluster and with them the node properties.
+This restarts all Pods of the cluster, see xref:usage-guide/operations/cluster-operations.adoc#scaling-restarts[Scaling operations restart the whole cluster].

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,6 +1,6 @@
 use std::collections::BTreeMap;
 
-use super::kraft_controllers;
+use super::{kraft_controllers, opa_authorizer_properties};
 use crate::{
     controller::{
         ValidatedClusterConfig,
@@ -73,7 +73,11 @@ pub fn build(
         );
     }
 
-    result.extend(controller_config_settings(&cluster_config.kafka_security));
+    result.extend(opa_authorizer_properties(cluster_config, pod_descriptors));
+    result.extend(controller_config_settings(
+        &cluster_config.kafka_security,
+        cluster_config.opa_connect().is_some(),
+    ));
     result.extend(graceful_shutdown_config_properties());
     result.extend(overrides);