refactor: remove client_auth Kafka listener

Author: razvan

rust/operator-binary/src/crd/listener.rs

@@ -38,8 +38,6 @@ pub enum KafkaListenerProtocol {
 pub enum KafkaListenerName {
     #[strum(serialize = "CLIENT")]
     Client,
-    #[strum(serialize = "CLIENT_AUTH")]
-    ClientAuth,
     #[strum(serialize = "INTERNAL")]
     Internal,
     #[strum(serialize = "BOOTSTRAP")]
@@ -179,28 +177,7 @@ pub fn get_kafka_listener_config(
         BTreeMap::new();
 
     // CLIENT
-    if kafka_security.tls_client_authentication_class().is_some() {
-        // 1) If client authentication required, we expose only CLIENT_AUTH connection with SSL
-        listeners.push(KafkaListener {
-            name: KafkaListenerName::ClientAuth,
-            host: LISTENER_LOCAL_ADDRESS.to_string(),
-            port: kafka_security.client_port().to_string(),
-        });
-        advertised_listeners.push(KafkaListener {
-            name: KafkaListenerName::ClientAuth,
-            host: node_address_cmd(STACKABLE_LISTENER_BROKER_DIR),
-            port: node_port_cmd(
-                STACKABLE_LISTENER_BROKER_DIR,
-                kafka_security.client_port_name(),
-            ),
-        });
-        listener_security_protocol_map
-            .insert(KafkaListenerName::ClientAuth, KafkaListenerProtocol::Ssl);
-        listener_security_protocol_map.insert(
-            KafkaListenerName::ControllerAuth,
-            KafkaListenerProtocol::Ssl,
-        );
-    } else if kafka_security.has_kerberos_enabled() {
+    if kafka_security.has_kerberos_enabled() {
         // 2) Kerberos and TLS authentication classes are mutually exclusive
         listeners.push(KafkaListener {
             name: KafkaListenerName::Client,
@@ -217,12 +194,10 @@ pub fn get_kafka_listener_config(
         });
         listener_security_protocol_map
             .insert(KafkaListenerName::Client, KafkaListenerProtocol::SaslSsl);
-        listener_security_protocol_map.insert(
-            KafkaListenerName::Controller,
-            KafkaListenerProtocol::SaslSsl,
-        );
-    } else if kafka_security.tls_server_secret_class().is_some() {
-        // 3) If no client authentication but tls is required we expose CLIENT with SSL
+    } else if kafka_security.tls_client_authentication_class().is_some()
+        || kafka_security.tls_server_secret_class().is_some()
+    {
+        // 1) Client listener uses TLS (possibly with authentication)
         listeners.push(KafkaListener {
             name: KafkaListenerName::Client,
             host: LISTENER_LOCAL_ADDRESS.to_string(),
@@ -239,7 +214,8 @@ pub fn get_kafka_listener_config(
         listener_security_protocol_map
             .insert(KafkaListenerName::Client, KafkaListenerProtocol::Ssl);
     } else {
-        // 4) If no client auth or tls is required we expose CLIENT with PLAINTEXT
+        // 3) If no client auth or tls is required we expose CLIENT with PLAINTEXT
+        // This is actually never the case because
         listeners.push(KafkaListener {
             name: KafkaListenerName::Client,
             host: LISTENER_LOCAL_ADDRESS.to_string(),
@@ -414,7 +390,7 @@ mod tests {
             config.listeners(),
             format!(
                 "{name}://{host}:{port},{internal_name}://{internal_host}:{internal_port}",
-                name = KafkaListenerName::ClientAuth,
+                name = KafkaListenerName::Client,
                 host = LISTENER_LOCAL_ADDRESS,
                 port = kafka_security.client_port(),
                 internal_name = KafkaListenerName::Internal,
@@ -427,7 +403,7 @@ mod tests {
             config.advertised_listeners(),
             format!(
                 "{name}://{host}:{port},{internal_name}://{internal_host}:{internal_port}",
-                name = KafkaListenerName::ClientAuth,
+                name = KafkaListenerName::Client,
                 host = node_address_cmd(STACKABLE_LISTENER_BROKER_DIR),
                 port = node_port_cmd(
                     STACKABLE_LISTENER_BROKER_DIR,
@@ -447,15 +423,13 @@ mod tests {
         assert_eq!(
             config.listener_security_protocol_map(),
             format!(
-                "{name}:{protocol},{internal_name}:{internal_protocol},{controller_name}:{controller_protocol},{controller_auth_name}:{controller_auth_protocol}",
-                name = KafkaListenerName::ClientAuth,
+                "{name}:{protocol},{internal_name}:{internal_protocol},{controller_name}:{controller_protocol}",
+                name = KafkaListenerName::Client,
                 protocol = KafkaListenerProtocol::Ssl,
                 internal_name = KafkaListenerName::Internal,
                 internal_protocol = KafkaListenerProtocol::Ssl,
                 controller_name = KafkaListenerName::Controller,
                 controller_protocol = KafkaListenerProtocol::Ssl,
-                controller_auth_name = KafkaListenerName::ControllerAuth,
-                controller_auth_protocol = KafkaListenerProtocol::Ssl,
             )
         );
 

rust/operator-binary/src/crd/security.rs

@@ -531,37 +531,9 @@ impl KafkaTlsSecurity {
         // 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 self.tls_client_authentication_class().is_some() {
-            config.insert(
-                KafkaListenerName::ClientAuth.listener_ssl_keystore_location(),
-                format!("{}/keystore.p12", Self::STACKABLE_TLS_KAFKA_SERVER_DIR),
-            );
-            config.insert(
-                KafkaListenerName::ClientAuth.listener_ssl_keystore_password(),
-                Self::SSL_STORE_PASSWORD.to_string(),
-            );
-            config.insert(
-                KafkaListenerName::ClientAuth.listener_ssl_keystore_type(),
-                "PKCS12".to_string(),
-            );
-            config.insert(
-                KafkaListenerName::ClientAuth.listener_ssl_truststore_location(),
-                format!("{}/truststore.p12", Self::STACKABLE_TLS_KAFKA_SERVER_DIR),
-            );
-            config.insert(
-                KafkaListenerName::ClientAuth.listener_ssl_truststore_password(),
-                Self::SSL_STORE_PASSWORD.to_string(),
-            );
-            config.insert(
-                KafkaListenerName::ClientAuth.listener_ssl_truststore_type(),
-                "PKCS12".to_string(),
-            );
-            // client auth required
-            config.insert(
-                KafkaListenerName::ClientAuth.listener_ssl_client_auth(),
-                "required".to_string(),
-            );
-        } else if self.tls_server_secret_class().is_some() {
+        if self.tls_client_authentication_class().is_some()
+            || self.tls_server_secret_class().is_some()
+        {
             config.insert(
                 KafkaListenerName::Client.listener_ssl_keystore_location(),
                 format!("{}/keystore.p12", Self::STACKABLE_TLS_KAFKA_SERVER_DIR),
@@ -586,6 +558,13 @@ impl KafkaTlsSecurity {
                 KafkaListenerName::Client.listener_ssl_truststore_type(),
                 "PKCS12".to_string(),
             );
+            if self.tls_client_authentication_class().is_some() {
+                // client auth required
+                config.insert(
+                    KafkaListenerName::Client.listener_ssl_client_auth(),
+                    "required".to_string(),
+                );
+            }
         }
 
         if self.has_kerberos_enabled() {