diff --git a/CHANGELOG.md b/CHANGELOG.md index ae18c75a1..0492b5a38 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/modules/kafka/examples/security/02-authorization-opa.yaml b/docs/modules/kafka/examples/security/02-authorization-opa.yaml deleted file mode 100644 index 754bf5525..000000000 --- a/docs/modules/kafka/examples/security/02-authorization-opa.yaml +++ /dev/null @@ -1,63 +0,0 @@ ---- -apiVersion: opa.stackable.tech/v1alpha1 -kind: OpaCluster -metadata: - name: simple-opa -spec: - image: - productVersion: 1.20.2 - servers: - config: - logging: - containers: - opa: - loggers: - decision: - level: INFO - roleGroups: - default: {} ---- -apiVersion: v1 -kind: ConfigMap -metadata: - name: kafka-rego-rules - labels: - opa.stackable.tech/bundle: "true" -data: - kafka.rego: | - package kafka - - is_internal_request if input.requestContext.listenerName == "INTERNAL" - is_external_request if not is_internal_request - - # FQDNs of the brokers as regular expressions - brokers_fqdn := [ - `local`, # cluster domain - `cluster`, # cluster domain - `svc`, - `default`, # namespace - `simple-kafka-broker-default-headless`, # Pod subdomain - `simple-kafka-broker-default-[0-9]+`, # Pod name - ] - - domain_components = [concat("", [`DC=`, dc]) | some dc in brokers_fqdn] - common_name = `CN=generated certificate for pod` - subject_dn_fields = array.concat(domain_components, [common_name]) - brokers_subject_dn_pattern := concat("", ["^", concat(",", subject_dn_fields), "$"]) - - default allow := false - - allow if { - is_internal_request - - regex.match( - brokers_subject_dn_pattern, - input.requestContext.principal.name, - ) - } - - allow if { - is_external_request - - # TODO Add your own rules - } diff --git a/docs/modules/kafka/pages/usage-guide/operations/cluster-operations.adoc b/docs/modules/kafka/pages/usage-guide/operations/cluster-operations.adoc index bddb9b72c..0a89c476d 100644 --- a/docs/modules/kafka/pages/usage-guide/operations/cluster-operations.adoc +++ b/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. diff --git a/docs/modules/kafka/pages/usage-guide/operations/pod-placement.adoc b/docs/modules/kafka/pages/usage-guide/operations/pod-placement.adoc index 0a03be18c..2035fb271 100644 --- a/docs/modules/kafka/pages/usage-guide/operations/pod-placement.adoc +++ b/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] ---- diff --git a/docs/modules/kafka/pages/usage-guide/security.adoc b/docs/modules/kafka/pages/usage-guide/security.adoc index 977803cae..b0957bd7d 100644 --- a/docs/modules/kafka/pages/usage-guide/security.adoc +++ b/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 <>. -[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]. diff --git a/rust/operator-binary/src/controller/build/command.rs b/rust/operator-binary/src/controller/build/command.rs index 04c1f5e07..6b7879081 100644 --- a/rust/operator-binary/src/controller/build/command.rs +++ b/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, diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs index 9a17d5844..59cc807b2 100644 --- a/rust/operator-binary/src/controller/build/mod.rs +++ b/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, ], diff --git a/rust/operator-binary/src/controller/build/properties/broker_properties.rs b/rust/operator-binary/src/controller/build/properties/broker_properties.rs index ee76e0a78..9dee8b335 100644 --- a/rust/operator-binary/src/controller/build/properties/broker_properties.rs +++ b/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()); diff --git a/rust/operator-binary/src/controller/build/properties/controller_properties.rs b/rust/operator-binary/src/controller/build/properties/controller_properties.rs index 6804653f2..c5de74c62 100644 --- a/rust/operator-binary/src/controller/build/properties/controller_properties.rs +++ b/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); diff --git a/rust/operator-binary/src/controller/build/properties/mod.rs b/rust/operator-binary/src/controller/build/properties/mod.rs index 8cd11fc51..29dca78a3 100644 --- a/rust/operator-binary/src/controller/build/properties/mod.rs +++ b/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`. /// @@ -80,31 +83,107 @@ pub(crate) fn kraft_controllers( controllers .map(|desc| { - if kerberos_enabled { - format!( - "{sts}-{replica}.{service}.{namespace}.svc.{cluster_domain}:{client_port}", - sts = desc.role_group_statefulset_name, - replica = desc.replica, - service = desc.role_group_service_name, - namespace = desc.namespace, - cluster_domain = desc.cluster_domain, - client_port = desc.client_port, - ) + let host = if kerberos_enabled { + desc.pod_fqdn() } else { - format!( - "{service}.{namespace}.svc.{cluster_domain}:{client_port}", - service = desc.role_group_service_name, - namespace = desc.namespace, - cluster_domain = desc.cluster_domain, - client_port = desc.client_port, - ) - } + desc.service_fqdn() + }; + format!("{host}:{port}", port = desc.client_port) }) .collect::>() .into_iter() .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 { + 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=,DC=, +/// DC=,DC=svc,DC=`, see the secret-operator. +/// With Kerberos, the node principals are mapped to the short name of the service. +/// +/// # Why the `CN=generated certificate for pod` entry is required +/// +/// The secret-operator always adds this common name to the subject of the pod certificates, +/// regardless of whether domain components are added as well. Kafka's default principal for an +/// mTLS client is the complete subject DN in RFC 2253 notation, which lists the RDNs in reverse +/// order, so the CN is the last component: `DC=local,...,DC=,CN=generated certificate for pod`. +/// +/// Kafka compares each `super.users` entry for exact equality with the principal; there is no +/// support for wildcards or prefixes (the only wildcard Kafka knows is `User:*` in ACLs, which +/// matches all users). An entry without the CN, or with a `*` in place of it, therefore never +/// matches, and the brokers and controllers would be denied access to each other. +/// +/// This could only be avoided by rewriting the principal with `ssl.principal.mapping.rules`, +/// which is not done here. +/// +/// See the Kafka documentation: +/// - [Authorization and ACLs](https://kafka.apache.org/44/security/authorization-and-acls/) +/// - [`ssl.principal.mapping.rules`](https://kafka.apache.org/44/security/authorization-and-acls/#customizing-ssl-user-name) +fn super_users(pod_descriptors: &[KafkaPodDescriptor], kerberos_enabled: bool) -> String { + let mut principals: BTreeSet = 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 pod_fqdn = desc.pod_fqdn(); + let mut rdns: Vec = vec!["CN=generated certificate for pod".to_string()]; + rdns.extend(pod_fqdn.split('.').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::>().join(";") +} + #[cfg(test)] mod tests { use super::*; @@ -150,6 +229,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![ diff --git a/rust/operator-binary/src/controller/build/role_group/controller.rs b/rust/operator-binary/src/controller/build/role_group/controller.rs index 99d24dd99..af391e36f 100644 --- a/rust/operator-binary/src/controller/build/role_group/controller.rs +++ b/rust/operator-binary/src/controller/build/role_group/controller.rs @@ -45,6 +45,7 @@ use crate::{ security::{ self, STACKABLE_TLS_KAFKA_INTERNAL_DIR, STACKABLE_TLS_KAFKA_INTERNAL_VOLUME_NAME, controller_admin_client_properties, mount_controller_tls_volume, + mount_opa_tls_volume, }, }, node_id_hasher::node_id_hash32_offset, @@ -80,7 +81,7 @@ pub enum Error { BuildPodDescriptors { source: PodDescriptorsError }, #[snafu(display("failed to build the TLS volume"))] - BuildTlsVolumes { source: security::Error }, + BuildInternalTlsVolume { source: security::Error }, #[snafu(display("failed to build the Kerberos volume"))] BuildKerberosVolume { source: kerberos::Error }, @@ -90,6 +91,9 @@ pub enum Error { #[snafu(display("failed to construct JVM arguments"))] ConstructJvmArguments { source: jvm::Error }, + + #[snafu(display("failed to build the OPA TLS volume"))] + BuildOpaTlsVolume { source: security::Error }, } // Env vars the operator sets on the controller's `kafka` and `quorum-manager` containers. @@ -338,9 +342,16 @@ fn role_pod( &mut cb_kafka, &config.requested_secret_lifetime, ) - .context(BuildTlsVolumesSnafu)?, + .context(BuildInternalTlsVolumeSnafu)?, ); + // Mount the OPA TLS certificate if any. + if let Some(opa_tls_volume) = + mount_opa_tls_volume(kafka_security, &mut cb_kafka).context(BuildOpaTlsVolumeSnafu)? + { + pod_volumes.push(opa_tls_volume); + } + let containers = vec![ cb_kafka.build(), build_quorum_manager_container(image, kafka_security, quorum_manager_env), diff --git a/rust/operator-binary/src/controller/build/security.rs b/rust/operator-binary/src/controller/build/security.rs index bffc7fc51..dccf549ee 100644 --- a/rust/operator-binary/src/controller/build/security.rs +++ b/rust/operator-binary/src/controller/build/security.rs @@ -40,7 +40,7 @@ const OPA_TLS_MOUNT_PATH: &str = "/stackable/tls-opa"; constant!(pub(crate) OPA_TLS_VOLUME_NAME: VolumeName = "tls-opa"); const SSL_STORE_PASSWORD: &str = ""; const SSL_STORE_TYPE_PKCS12: &str = "PKCS12"; -const SSL_CLIENT_AUTH_REQUIRED: &str = "required"; +pub(super) const SSL_CLIENT_AUTH_REQUIRED: &str = "required"; const SASL_MECHANISM_GSSAPI: &str = "GSSAPI"; const STACKABLE_KCAT_BINARY: &str = "/stackable/kcat"; const PROPERTY_SECURITY_PROTOCOL: &str = "security.protocol"; @@ -76,7 +76,11 @@ pub enum Error { pub fn copy_opa_tls_cert_command(security: &ValidatedKafkaSecurity) -> String { match security.opa_secret_class().is_some() { true => format!( - "keytool -importcert -file {opa_mount_path}/ca.crt -keystore {tls_dir}/{truststore} -storepass '{tls_password}' -alias opa-ca -noprompt", + // The truststore lives on a volume that survives container restarts, so drop any + // previously imported alias first. This prevents keytool -import from failing and + // leaving the pods in a crash loop. + "keytool -delete -alias opa-ca -keystore {tls_dir}/{truststore} -storepass '{tls_password}' 2>/dev/null || true\n\ + keytool -importcert -file {opa_mount_path}/ca.crt -keystore {tls_dir}/{truststore} -storepass '{tls_password}' -alias opa-ca -noprompt", opa_mount_path = OPA_TLS_MOUNT_PATH, tls_dir = STACKABLE_TLS_KAFKA_INTERNAL_DIR, truststore = TRUSTSTORE_P12_FILE_NAME, @@ -256,6 +260,37 @@ pub fn controller_admin_client_properties( properties } +/// Adds the volume mount holding the CA of the OPA TLS `SecretClass`, if configured, and returns +/// the pod volume it mounts. +/// +/// The CA is imported into the internal truststore on startup, see [`copy_opa_tls_cert_command`]. +pub fn mount_opa_tls_volume( + security: &ValidatedKafkaSecurity, + cb_kafka: &mut ContainerBuilder, +) -> Result, Error> { + let Some(secret_class) = security.opa_secret_class() else { + return Ok(None); + }; + + cb_kafka + .add_volume_mount(&*OPA_TLS_VOLUME_NAME, OPA_TLS_MOUNT_PATH) + .expect("The mount paths are statically defined and there should be no duplicates."); + + let volume = VolumeBuilder::new(&*OPA_TLS_VOLUME_NAME) + .ephemeral( + SecretOperatorVolumeSourceBuilder::new( + secret_class, + // Only the truststore is required to connect to OPA. + SecretClassVolumeProvisionParts::Public, + ) + .build() + .context(OpaTlsCertSecretClassVolumeBuildSnafu)?, + ) + .build(); + + Ok(Some(volume)) +} + /// Adds the volume mounts the tls and authentication settings require to the broker's container /// builder and returns the pod volumes they mount, in pod order. /// @@ -308,35 +343,18 @@ pub fn mount_broker_tls_volumes( ) .expect("The mount paths are statically defined and there should be no duplicates."); - if let Some(secret_class) = security.opa_secret_class() { - cb_kafka - .add_volume_mount(&*OPA_TLS_VOLUME_NAME, OPA_TLS_MOUNT_PATH) - .expect("The mount paths are statically defined and there should be no duplicates."); - - volumes.push( - VolumeBuilder::new(&*OPA_TLS_VOLUME_NAME) - .ephemeral( - SecretOperatorVolumeSourceBuilder::new( - secret_class, - // Only the truststore is required to connect to OPA. - SecretClassVolumeProvisionParts::Public, - ) - .build() - .context(OpaTlsCertSecretClassVolumeBuildSnafu)?, - ) - .build(), - ); + if let Some(opa_tls_volume) = mount_opa_tls_volume(security, cb_kafka)? { + volumes.push(opa_tls_volume); } Ok(volumes) } -/// Adds the internal TLS volume mount to the controller's container builder and returns the pod -/// volume it mounts. +/// Adds the internal TLS volume mount to the controller's container builder. /// /// # Panics /// -/// Panics if the volume mounts cannot be added to the container builder. Only call this on a +/// Panics if the volume mount cannot be added to the container builder. Only call this on a /// container builder whose mount paths are still distinct from the ones added here. pub fn mount_controller_tls_volume( security: &ValidatedKafkaSecurity, @@ -437,6 +455,32 @@ fn push_client_ssl_truststore(props: &mut Vec<(String, Option)>, dir: &s )); } +/// Inserts the `opa.authorizer.truststore.*` settings if OPA is configured with TLS. +/// +/// The OPA CA is imported into the internal truststore, see [`copy_opa_tls_cert_command`]. +fn insert_opa_truststore_settings( + config: &mut BTreeMap, + security: &ValidatedKafkaSecurity, +) { + if security.opa_secret_class().is_some() { + config.insert( + "opa.authorizer.truststore.path".to_string(), + format!( + "{}/{}", + STACKABLE_TLS_KAFKA_INTERNAL_DIR, TRUSTSTORE_P12_FILE_NAME + ), + ); + config.insert( + "opa.authorizer.truststore.password".to_string(), + SSL_STORE_PASSWORD.to_string(), + ); + config.insert( + "opa.authorizer.truststore.type".to_string(), + SSL_STORE_TYPE_PKCS12.to_string(), + ); + } +} + /// Returns required Kafka configuration settings for the `broker.properties` file /// depending on the tls and authentication settings. pub fn broker_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap { @@ -507,24 +551,7 @@ pub fn broker_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap BTreeMap BTreeMap { +pub fn controller_config_settings( + security: &ValidatedKafkaSecurity, + opa_authorization: bool, +) -> BTreeMap { let mut config = BTreeMap::new(); insert_listener_ssl_stores( @@ -556,7 +586,13 @@ pub fn controller_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap // We set either client tls with authentication or client tls without authentication // If authentication is explicitly required we do not want to have any other CAs to // be trusted. - if security.tls_client_authentication_class().is_some() { + // + // With OPA, 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. + if security.tls_client_authentication_class().is_some() + || (opa_authorization && !security.has_kerberos_enabled()) + { // client auth required config.insert( KafkaListenerName::Controller.listener_ssl_client_auth(), @@ -564,6 +600,8 @@ pub fn controller_config_settings(security: &ValidatedKafkaSecurity) -> BTreeMap ); } + insert_opa_truststore_settings(&mut config, security); + // Kerberos if security.has_kerberos_enabled() { config.insert( @@ -1140,12 +1178,20 @@ pub(crate) mod tests { // ---- controller_config_settings ---- + #[test] + fn controller_config_opa_adds_authorizer_truststore() { + let config = controller_config_settings(&opa(), false); + assert!(config.contains_key("opa.authorizer.truststore.path")); + assert!(config.contains_key("opa.authorizer.truststore.password")); + assert!(config.contains_key("opa.authorizer.truststore.type")); + } + #[test] fn controller_config_plaintext_has_internal_tls() { // Internal TLS is mandatory, so the controller config is non-empty even for a plaintext // cluster: it carries the controller + internal SSL stores, but no client-cert auth // (there is no client authentication class). - let config = controller_config_settings(&plaintext()); + let config = controller_config_settings(&plaintext(), false); assert!( config.contains_key(&KafkaListenerName::Controller.listener_ssl_keystore_location()) ); @@ -1153,9 +1199,20 @@ pub(crate) mod tests { assert!(!config.contains_key(&KafkaListenerName::Controller.listener_ssl_client_auth())); } + #[test] + fn controller_config_opa_requires_client_auth_unless_kerberos() { + let key = KafkaListenerName::Controller.listener_ssl_client_auth(); + + let config = controller_config_settings(&internal_tls(), true); + assert_eq!(config.get(&key).map(String::as_str), Some("required")); + + assert!(!controller_config_settings(&internal_tls(), false).contains_key(&key)); + assert!(!controller_config_settings(&kerberos(), true).contains_key(&key)); + } + #[test] fn controller_config_internal_tls_covers_controller_and_internal() { - let config = controller_config_settings(&internal_tls()); + let config = controller_config_settings(&internal_tls(), false); assert!(config.contains_key("listener.name.controller.ssl.keystore.location")); assert!(config.contains_key("listener.name.internal.ssl.keystore.location")); } @@ -1171,7 +1228,7 @@ pub(crate) mod tests { #[test] fn controller_config_sets_the_controller_sasl_mechanism_with_kerberos() { - let config = controller_config_settings(&kerberos()); + let config = controller_config_settings(&kerberos(), false); assert_eq!( config.get("sasl.mechanism.controller.protocol"), Some(&"GSSAPI".to_string()) @@ -1185,14 +1242,14 @@ pub(crate) mod tests { .contains_key("sasl.mechanism.controller.protocol") ); assert!( - !controller_config_settings(&internal_tls()) + !controller_config_settings(&internal_tls(), false) .contains_key("sasl.mechanism.controller.protocol") ); } #[test] fn controller_config_kerberos_adds_sasl() { - let config = controller_config_settings(&kerberos()); + let config = controller_config_settings(&kerberos(), false); assert_eq!( config.get("sasl.enabled.mechanisms"), Some(&"GSSAPI".to_string()) diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index 3f0f2322d..634a21b52 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -300,6 +300,7 @@ pub fn validate( ControllerConfig::default_config( &kafka.name_any(), &KafkaRole::Controller.to_string(), + kafka.spec.cluster_config.authorization.opa.as_ref(), ), cluster_id, &vector_aggregator_config_map_name, diff --git a/rust/operator-binary/src/crd/affinity.rs b/rust/operator-binary/src/crd/affinity.rs index e46b71ecc..dda570a16 100644 --- a/rust/operator-binary/src/crd/affinity.rs +++ b/rust/operator-binary/src/crd/affinity.rs @@ -8,7 +8,8 @@ use stackable_operator::{ use crate::crd::APP_NAME; -/// `opa_config` is only passed for roles that send authorization requests to OPA. +/// `opa_config` is only passed if OPA authorization is configured. Brokers and controllers +/// both run the OPA authorizer, so both send authorization requests to OPA. pub fn get_affinity( cluster_name: &str, role: &str, @@ -112,39 +113,32 @@ mod tests { assert_eq!( *affinity, StackableAffinity { - pod_affinity: match role { - // Only brokers are configured with the OPA authorizer. - KafkaRole::Broker => Some(PodAffinity { - preferred_during_scheduling_ignored_during_execution: Some(vec![ - WeightedPodAffinityTerm { - pod_affinity_term: PodAffinityTerm { - label_selector: Some(LabelSelector { - match_expressions: None, - match_labels: Some(BTreeMap::from([ - ( - "app.kubernetes.io/name".to_string(), - "opa".to_string() - ), - ( - "app.kubernetes.io/instance".to_string(), - "simple-opa".to_string(), - ), - ( - "app.kubernetes.io/component".to_string(), - "server".to_string(), - ), - ])), - }), - topology_key: "kubernetes.io/hostname".to_string(), - ..PodAffinityTerm::default() - }, - weight: 50, - } - ]), - required_during_scheduling_ignored_during_execution: None, - }), - KafkaRole::Controller => None, - }, + pod_affinity: Some(PodAffinity { + preferred_during_scheduling_ignored_during_execution: Some(vec![ + WeightedPodAffinityTerm { + pod_affinity_term: PodAffinityTerm { + label_selector: Some(LabelSelector { + match_expressions: None, + match_labels: Some(BTreeMap::from([ + ("app.kubernetes.io/name".to_string(), "opa".to_string()), + ( + "app.kubernetes.io/instance".to_string(), + "simple-opa".to_string(), + ), + ( + "app.kubernetes.io/component".to_string(), + "server".to_string(), + ), + ])), + }), + topology_key: "kubernetes.io/hostname".to_string(), + ..PodAffinityTerm::default() + }, + weight: 50, + } + ]), + required_during_scheduling_ignored_during_execution: None, + }), pod_anti_affinity: Some(PodAntiAffinity { preferred_during_scheduling_ignored_during_execution: Some(vec![ WeightedPodAffinityTerm { diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 77bbfd17c..388ed5452 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -346,6 +346,32 @@ pub struct KafkaPodDescriptor { pub client_port: Port, } +impl KafkaPodDescriptor { + /// The FQDN of the role group's headless `Service`: + /// `..svc.`. + pub(crate) fn service_fqdn(&self) -> String { + format!( + "{service}.{namespace}.svc.{cluster_domain}", + service = self.role_group_service_name, + namespace = self.namespace, + cluster_domain = self.cluster_domain, + ) + } + + /// The FQDN of the `Pod`: `-.`. + /// + /// The statefulset, service and namespace names are DNS labels and contain no dots, so + /// splitting the result on `.` yields the individual labels. + pub(crate) fn pod_fqdn(&self) -> String { + format!( + "{sts}-{replica}.{service_fqdn}", + sts = self.role_group_statefulset_name, + replica = self.replica, + service_fqdn = self.service_fqdn(), + ) + } +} + #[derive(Clone, Default, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] pub struct KafkaClusterStatus { diff --git a/rust/operator-binary/src/crd/role/controller.rs b/rust/operator-binary/src/crd/role/controller.rs index 58540482b..e13a0114d 100644 --- a/rust/operator-binary/src/crd/role/controller.rs +++ b/rust/operator-binary/src/crd/role/controller.rs @@ -2,9 +2,12 @@ use std::str::FromStr; use serde::{Deserialize, Serialize}; use stackable_operator::{ - commons::resources::{ - CpuLimitsFragment, MemoryLimitsFragment, NoRuntimeLimits, NoRuntimeLimitsFragment, - PvcConfigFragment, Resources, ResourcesFragment, + commons::{ + opa::OpaConfig, + resources::{ + CpuLimitsFragment, MemoryLimitsFragment, NoRuntimeLimits, NoRuntimeLimitsFragment, + PvcConfigFragment, Resources, ResourcesFragment, + }, }, config::{fragment::Fragment, merge::Merge}, constant, @@ -78,11 +81,13 @@ pub struct ControllerConfig { } impl ControllerConfig { - pub fn default_config(cluster_name: &str, role: &str) -> ControllerConfigFragment { + pub fn default_config( + cluster_name: &str, + role: &str, + opa_config: Option<&OpaConfig>, + ) -> ControllerConfigFragment { ControllerConfigFragment { - // Only brokers are configured with the OPA authorizer, so controllers get no affinity - // to the OPA Pods. - common_config: CommonConfig::default_config(cluster_name, role, None), + common_config: CommonConfig::default_config(cluster_name, role, opa_config), logging: product_logging::spec::default_logging(), resources: ResourcesFragment { cpu: CpuLimitsFragment { diff --git a/tests/templates/kuttl/configuration/10-assert.yaml.j2 b/tests/templates/kuttl/configuration/10-assert.yaml.j2 index 3d75ac23f..507b9c6d2 100644 --- a/tests/templates/kuttl/configuration/10-assert.yaml.j2 +++ b/tests/templates/kuttl/configuration/10-assert.yaml.j2 @@ -2,6 +2,8 @@ apiVersion: kuttl.dev/v1beta1 kind: TestAssert timeout: 300 +commands: + - script: kubectl -n $NAMESPACE wait --for=condition=available=true kafkaclusters.kafka.stackable.tech/test-kafka --timeout 601s --- apiVersion: apps/v1 kind: StatefulSet diff --git a/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2 b/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2 index af9f97e80..c026bdaae 100644 --- a/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2 +++ b/tests/templates/kuttl/configuration/10-install-kafka.yaml.j2 @@ -17,8 +17,8 @@ spec: {% endif %} pullPolicy: IfNotPresent clusterConfig: + metadataManager: KRaft {% if lookup('env', 'VECTOR_AGGREGATOR') %} - clusterConfig: vectorAggregatorConfigMapName: vector-aggregator-discovery {% endif %} controllers: diff --git a/tests/templates/kuttl/opa-kraft/00-patch-ns.yaml.j2 b/tests/templates/kuttl/opa-kraft/00-patch-ns.yaml.j2 new file mode 100644 index 000000000..67185acfd --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/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 %} diff --git a/tests/templates/kuttl/opa-kraft/00-rbac.yaml.j2 b/tests/templates/kuttl/opa-kraft/00-rbac.yaml.j2 new file mode 100644 index 000000000..7ee61d23e --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/00-rbac.yaml.j2 @@ -0,0 +1,29 @@ +--- +kind: Role +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: test-role +rules: +{% if test_scenario['values']['openshift'] == "true" %} + - apiGroups: ["security.openshift.io"] + resources: ["securitycontextconstraints"] + resourceNames: ["privileged"] + verbs: ["use"] +{% endif %} +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: test-sa +--- +kind: RoleBinding +apiVersion: rbac.authorization.k8s.io/v1 +metadata: + name: test-rb +subjects: + - kind: ServiceAccount + name: test-sa +roleRef: + kind: Role + name: test-role + apiGroup: rbac.authorization.k8s.io diff --git a/tests/templates/kuttl/opa-kraft/01-assert.yaml b/tests/templates/kuttl/opa-kraft/01-assert.yaml new file mode 100644 index 000000000..615b16ab1 --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/01-assert.yaml @@ -0,0 +1,16 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: vector-aggregator-discovery +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: krb5-kdc +status: + readyReplicas: 1 + replicas: 1 diff --git a/tests/templates/kuttl/opa-kraft/01-install-krb5-kdc.yaml.j2 b/tests/templates/kuttl/opa-kraft/01-install-krb5-kdc.yaml.j2 new file mode 100644 index 000000000..b77a5fc20 --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/01-install-krb5-kdc.yaml.j2 @@ -0,0 +1,144 @@ +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: krb5-kdc +spec: + selector: + matchLabels: + app: krb5-kdc + template: + metadata: + labels: + app: krb5-kdc + spec: + serviceAccountName: test-sa + initContainers: + - name: init + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + args: + - sh + - -euo + - pipefail + - -c + - | + test -e /var/kerberos/krb5kdc/principal || kdb5_util create -s -P asdf + kadmin.local get_principal -terse root/admin || kadmin.local add_principal -pw asdf root/admin + # stackable-secret-operator principal must match the keytab specified in the SecretClass + kadmin.local get_principal -terse stackable-secret-operator || kadmin.local add_principal -e aes256-cts-hmac-sha384-192:normal -pw asdf stackable-secret-operator + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + - mountPath: /var/kerberos/krb5kdc + name: data + containers: + - name: kdc + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + args: + - krb5kdc + - -n + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + - mountPath: /var/kerberos/krb5kdc + name: data +# Root permissions required on Openshift to bind to privileged port numbers +{% if test_scenario['values']['openshift'] == "true" %} + securityContext: + runAsUser: 0 +{% endif %} + - name: kadmind + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + args: + - kadmind + - -nofork + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + - mountPath: /var/kerberos/krb5kdc + name: data +# Root permissions required on Openshift to bind to privileged port numbers +{% if test_scenario['values']['openshift'] == "true" %} + securityContext: + runAsUser: 0 +{% endif %} + - name: client + image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev + tty: true + stdin: true + env: + - name: KRB5_CONFIG + value: /stackable/config/krb5.conf + volumeMounts: + - mountPath: /stackable/config + name: config + volumes: + - name: config + configMap: + name: krb5-kdc + volumeClaimTemplates: + - metadata: + name: data + spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 1Gi +--- +apiVersion: v1 +kind: Service +metadata: + name: krb5-kdc +spec: + selector: + app: krb5-kdc + ports: + - name: kadmin + port: 749 + - name: kdc + port: 88 + - name: kdc-udp + port: 88 + protocol: UDP +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: krb5-kdc +data: + krb5.conf: | + [logging] + default = STDERR + kdc = STDERR + admin_server = STDERR + # default = FILE:/var/log/krb5libs.log + # kdc = FILE:/var/log/krb5kdc.log + # admin_server = FILE:/vaggr/log/kadmind.log + [libdefaults] + dns_lookup_realm = false + ticket_lifetime = 24h + renew_lifetime = 7d + forwardable = true + rdns = false + default_realm = CLUSTER.LOCAL + spake_preauth_groups = edwards25519 + [realms] + CLUSTER.LOCAL = { + acl_file = /stackable/config/kadm5.acl + disable_encrypted_timestamp = false + } + [domain_realm] + .cluster.local = CLUSTER.LOCAL + cluster.local = CLUSTER.LOCAL + kadm5.acl: | + root/admin *e + stackable-secret-operator *e diff --git a/tests/templates/kuttl/opa-kraft/01-install-vector-aggregator-discovery-configmap.yaml b/tests/templates/kuttl/opa-kraft/01-install-vector-aggregator-discovery-configmap.yaml new file mode 100644 index 000000000..91752a602 --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/01-install-vector-aggregator-discovery-configmap.yaml @@ -0,0 +1,17 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + - script: >- + helm install vector-aggregator vector + --namespace $NAMESPACE + --version 0.58.0 `# app version 0.58.0` + --repo https://helm.vector.dev + --values vector-aggregator-values.yaml +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: vector-aggregator-discovery +data: + ADDRESS: http://vector-aggregator:6123 diff --git a/tests/templates/kuttl/opa-kraft/02-create-kerberos-secretclass.yaml.j2 b/tests/templates/kuttl/opa-kraft/02-create-kerberos-secretclass.yaml.j2 new file mode 100644 index 000000000..0d443b614 --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/02-create-kerberos-secretclass.yaml.j2 @@ -0,0 +1,37 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + - script: | + kubectl apply -n $NAMESPACE -f - < + envsubst '$NAMESPACE' < 20_opa.yaml | + kubectl apply -n $NAMESPACE -f - diff --git a/tests/templates/kuttl/opa-kraft/20_opa.yaml.j2 b/tests/templates/kuttl/opa-kraft/20_opa.yaml.j2 new file mode 100644 index 000000000..a98e20aac --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/20_opa.yaml.j2 @@ -0,0 +1,77 @@ +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: test-rego-kafka + labels: + opa.stackable.tech/bundle: "true" +data: + kafka.rego: | + package authz + + default allow := false + + # The Kafka nodes are super users and do not hit OPA for their own traffic, + # so only the developer needs a rule. + allow if { + input.requestContext.principal.name == "developer" + resource_is_allowed + } + + # The (idempotent) producer needs this cluster-level permission. + allow if { + input.requestContext.principal.name == "developer" + input.action.resourcePattern.resourceType == "CLUSTER" + input.action.operation == "IDEMPOTENT_WRITE" + } + + # only allow access to a specific topic + resource_is_allowed if { + input.action.resourcePattern.resourceType == "TOPIC" + input.action.resourcePattern.name == "test-topic" + action_is_allowed + } + + resource_is_allowed if { + input.action.resourcePattern.resourceType == "GROUP" + input.action.operation in ["DESCRIBE", "READ"] + } + + action_is_allowed if { + input.action.operation in ["CREATE", "DESCRIBE", "READ", "WRITE"] + } +--- +apiVersion: opa.stackable.tech/v1alpha1 +kind: OpaCluster +metadata: + name: test-opa +spec: + image: +{% if test_scenario['values']['opa-latest'].find(",") > 0 %} + custom: "{{ test_scenario['values']['opa-latest'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['opa-latest'].split(',')[0] }}" +{% else %} + productVersion: "{{ test_scenario['values']['opa-latest'] }}" +{% endif %} + pullPolicy: IfNotPresent + clusterConfig: +{% if test_scenario['values']['use-opa-tls'] == "true" %} + tls: + serverSecretClass: opa-tls-$NAMESPACE +{% endif %} + vectorAggregatorConfigMapName: vector-aggregator-discovery + servers: + config: + logging: + enableVectorAgent: true + containers: + opa: + console: + level: INFO + file: + level: INFO + loggers: + decision: + level: INFO + roleGroups: + default: {} diff --git a/tests/templates/kuttl/opa-kraft/30-assert.yaml.j2 b/tests/templates/kuttl/opa-kraft/30-assert.yaml.j2 new file mode 100644 index 000000000..55d42ab0d --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/30-assert.yaml.j2 @@ -0,0 +1,40 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 300 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-controller-default +status: + readyReplicas: 3 + replicas: 3 +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: test-kafka-broker-default +status: + readyReplicas: 3 + replicas: 3 +--- +apiVersion: v1 +kind: PersistentVolumeClaim +metadata: + name: log-dirs-test-kafka-broker-default-0 +spec: + resources: + requests: + storage: 2Gi +status: + phase: Bound +--- +apiVersion: policy/v1 +kind: PodDisruptionBudget +metadata: + name: test-kafka-broker +status: + expectedPods: 3 + currentHealthy: 3 + disruptionsAllowed: 1 diff --git a/tests/templates/kuttl/opa-kraft/30-install-kafka.yaml.j2 b/tests/templates/kuttl/opa-kraft/30-install-kafka.yaml.j2 new file mode 100644 index 000000000..b897a9216 --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/30-install-kafka.yaml.j2 @@ -0,0 +1,56 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + - script: | + kubectl apply -n $NAMESPACE -f - < 0 %} + custom: "{{ test_scenario['values']['kafka-kraft'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['kafka-kraft'].split(',')[0] }}" +{% else %} + productVersion: "{{ test_scenario['values']['kafka-kraft'] }}" +{% endif %} + pullPolicy: IfNotPresent + clusterConfig: + # KRaft: metadata is managed by the controllers role, no ZooKeeper involved. + metadataManager: KRaft + authentication: + - authenticationClass: kerberos-auth-$NAMESPACE + authorization: + opa: + configMapName: test-opa + package: authz + tls: + serverSecretClass: tls + vectorAggregatorConfigMapName: vector-aggregator-discovery + controllers: + config: + logging: + enableVectorAgent: true + roleGroups: + default: + replicas: 3 + brokers: + config: + logging: + enableVectorAgent: true + roleGroups: + default: + replicas: 3 + EOF diff --git a/tests/templates/kuttl/opa-kraft/40-access-kafka.txt.j2 b/tests/templates/kuttl/opa-kraft/40-access-kafka.txt.j2 new file mode 100644 index 000000000..93bf47fc9 --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/40-access-kafka.txt.j2 @@ -0,0 +1,177 @@ +--- +apiVersion: batch/v1 +kind: Job +metadata: + name: access-kafka +spec: + template: + spec: + serviceAccountName: test-sa + containers: + - name: access-kafka +{% if test_scenario['values']['kafka-kraft'].find(",") > 0 %} + image: {{ test_scenario['values']['kafka-kraft'].split(',')[1] }} +{% else %} + image: oci.stackable.tech/sdp/kafka:{{ test_scenario['values']['kafka-kraft'] }}-stackable0.0.0-dev +{% endif %} + command: + - /bin/bash + - /tmp/script/script.sh + env: + - name: KRB5_CONFIG + value: /stackable/kerberos/krb5.conf + - name: KAFKA_OPTS + value: -Djava.security.krb5.conf=/stackable/kerberos/krb5.conf + - name: KAFKA + valueFrom: + configMapKeyRef: + name: test-kafka + key: KAFKA + volumeMounts: + - name: script + mountPath: /tmp/script + - mountPath: /stackable/tls-ca-cert-mount + name: tls-ca-cert-mount + - name: kerberos + mountPath: /stackable/kerberos + volumes: + - name: script + configMap: + name: access-kafka-script + - name: kerberos + ephemeral: + volumeClaimTemplate: + metadata: + annotations: + secrets.stackable.tech/class: kerberos-$NAMESPACE + secrets.stackable.tech/scope: service=access-kafka + secrets.stackable.tech/kerberos.service.names: developer + spec: + storageClassName: secrets.stackable.tech + accessModes: + - ReadWriteOnce + resources: + requests: + storage: "1" + - name: tls-ca-cert-mount + ephemeral: + volumeClaimTemplate: + metadata: + annotations: + secrets.stackable.tech/class: tls + secrets.stackable.tech/scope: pod + spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: "1" + storageClassName: secrets.stackable.tech + volumeMode: Filesystem + securityContext: + fsGroup: 1000 + restartPolicy: OnFailure +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: access-kafka-script +data: + script.sh: | + set -euxo pipefail + + TOPIC=test-topic + DISALLOWED_TOPIC=disallowed-topic + CONSUMER_GROUP=test-consumer-group + ADMIN_CONFIG=/stackable/admin.properties + + # In KRaft mode, the brokers forward topic management requests to the controllers, which + # have to authorize them with OPA. Test this with the Kafka admin client. + cat > $ADMIN_CONFIG <&1) + exit_code=$? + echo "$output" | tail -5 + if [ $exit_code -eq 0 ] || ! echo "$output" | grep -qi "authorization failed"; then + echo "Test failed: creating $DISALLOWED_TOPIC should have been denied" + exit 1 + fi + + set -e + + echo "Creating the allowed topic must succeed" + /stackable/kafka/bin/kafka-topics.sh --bootstrap-server $KAFKA --command-config $ADMIN_CONFIG --create --if-not-exists --topic $TOPIC + + set +e + + echo "Deleting the allowed topic must fail, as only CREATE, DESCRIBE, READ and WRITE are allowed" + output=$(/stackable/kafka/bin/kafka-topics.sh --bootstrap-server $KAFKA --command-config $ADMIN_CONFIG --delete --topic $TOPIC 2>&1) + exit_code=$? + echo "$output" | tail -5 + if [ $exit_code -eq 0 ] || ! echo "$output" | grep -qi "authorization failed"; then + echo "Test failed: deleting $TOPIC should have been denied" + exit 1 + fi + + set -e + + sent_message="Hello Stackable!" + + # kcat (librdkafka) cannot query offsets on Kafka 4.x, so use the Kafka tools to produce and + # consume. + echo $sent_message | /stackable/kafka/bin/kafka-console-producer.sh \ + --bootstrap-server $KAFKA \ + --producer.config $ADMIN_CONFIG \ + --topic $TOPIC + + echo Sent message: \"$sent_message\" + + # Kafka 3.9 (and newer versions) print a deprecation warning for --consumer.config on stdout, so filter it + received_message=$(/stackable/kafka/bin/kafka-console-consumer.sh \ + --bootstrap-server $KAFKA \ + --consumer.config $ADMIN_CONFIG \ + --group $CONSUMER_GROUP \ + --topic $TOPIC \ + --from-beginning \ + --max-messages 1 \ + --timeout-ms 30000 | grep -v "is deprecated") + + echo Received message: \"$received_message\" + + if [ "$received_message" = "$sent_message" ]; then + echo "Test passed" + else + echo "Test failed" + exit 1 + fi + + # Temporarily disable exit-on-error + set +e + + # This should fail as the rego rule filters topic access + output=$(echo "$sent_message" | /stackable/kafka/bin/kafka-console-producer.sh \ + --bootstrap-server $KAFKA \ + --producer.config $ADMIN_CONFIG \ + --topic $DISALLOWED_TOPIC 2>&1) + exit_code=$? + + set -e + + if [ $exit_code -ne 0 ] || echo "$output" | grep -qi "authorization failed"; then + echo "Test passed" + exit 0 + else + echo "Test failed" + exit 1 + fi diff --git a/tests/templates/kuttl/opa-kraft/40-access-kafka.yaml b/tests/templates/kuttl/opa-kraft/40-access-kafka.yaml new file mode 100644 index 000000000..966e1ebce --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/40-access-kafka.yaml @@ -0,0 +1,6 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + # We need to replace $NAMESPACE (by KUTTL) + - script: envsubst '$NAMESPACE' < 40-access-kafka.txt | kubectl apply -n $NAMESPACE -f - diff --git a/tests/templates/kuttl/opa-kraft/40-assert.yaml b/tests/templates/kuttl/opa-kraft/40-assert.yaml new file mode 100644 index 000000000..edc6c317f --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/40-assert.yaml @@ -0,0 +1,11 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestAssert +timeout: 600 +--- +apiVersion: batch/v1 +kind: Job +metadata: + name: access-kafka +status: + succeeded: 1 diff --git a/tests/templates/kuttl/opa-kraft/vector-aggregator-values.yaml.j2 b/tests/templates/kuttl/opa-kraft/vector-aggregator-values.yaml.j2 new file mode 100644 index 000000000..1eacd03f2 --- /dev/null +++ b/tests/templates/kuttl/opa-kraft/vector-aggregator-values.yaml.j2 @@ -0,0 +1,82 @@ +--- +role: Aggregator +service: + ports: + - name: api + port: 8686 + protocol: TCP + targetPort: 8686 + - name: vector + port: 6123 + protocol: TCP + targetPort: 6000 +customConfig: + api: + address: 0.0.0.0:8686 + enabled: true + sources: + vector: + address: 0.0.0.0:6000 + type: vector + version: "2" + transforms: + validEvents: + type: filter + inputs: [vector] + condition: is_null(.errors) + filteredAutomaticLogConfigServerOpaDecision: + type: filter + inputs: [validEvents] + condition: >- + starts_with(string!(.pod), "test-opa-server-automatic-log-config") && + .container == "opa" && + .logger == "decision" + filteredAutomaticLogConfigServerOpa: + type: filter + inputs: [validEvents] + condition: >- + starts_with(string!(.pod), "test-opa-server-automatic-log-config") && + .container == "opa" && + .logger != "decision" + filteredAutomaticLogConfigServerBundleBuilder: + type: filter + inputs: [validEvents] + condition: >- + starts_with(string!(.pod), "test-opa-server-automatic-log-config") && + .container == "bundle-builder" + filteredAutomaticLogConfigServerVector: + type: filter + inputs: [validEvents] + condition: >- + starts_with(string!(.pod), "test-opa-server-automatic-log-config") && + .container == "vector" + filteredAutomaticLogConfigServerPrepare: + type: filter + inputs: [validEvents] + condition: >- + starts_with(string!(.pod), "test-opa-server-automatic-log-config") && + .container == "prepare" + filteredInvalidEvents: + type: filter + inputs: [vector] + condition: |- + .timestamp == from_unix_timestamp!(0) || + is_null(.level) || + is_null(.logger) || + is_null(.message) + sinks: + test: + inputs: [filtered*] + type: blackhole +{% if lookup('env', 'VECTOR_AGGREGATOR') %} + aggregator: + inputs: [vector] + type: vector + routing: + endpoints: + - {{ lookup('env', 'VECTOR_AGGREGATOR') }} + buffer: + # Avoid back pressure from VECTOR_AGGREGATOR. The test should + # not fail if the aggregator is not available. + when_full: drop_newest +{% endif %} diff --git a/tests/test-definition.yaml b/tests/test-definition.yaml index c4c294ffd..7efe4f22d 100644 --- a/tests/test-definition.yaml +++ b/tests/test-definition.yaml @@ -38,16 +38,16 @@ dimensions: - 4.3.1 - name: use-client-tls values: - - "true" - "false" + - "true" - name: use-client-auth-tls values: - - "true" - "false" + - "true" - name: use-opa-tls values: - - "true" - "false" + - "true" - name: openshift values: - "false" @@ -95,6 +95,15 @@ tests: - use-opa-tls - opa-latest - krb5 + # Tests OPA authorization with KRaft controllers, which have to authorize the requests + # that the brokers forward to them (e.g. creating and deleting topics). + - name: opa-kraft + dimensions: + - kafka-kraft + - openshift + - use-opa-tls + - opa-latest + - krb5 - name: configuration dimensions: - kafka-latest