KAFKA-15513: Test SCRAM authentication for KRaft controllers

tests/kafkatest/sanity_checks/test_verifiable_producer.py

@@ -151,3 +151,31 @@ def test_multiple_kraft_sasl_mechanisms(self, metadata_quorum):
         num_produced = self.producer.num_acked
         assert num_produced == self.num_messages, "num_produced: %d, num_messages: %d" % (num_produced, self.num_messages)
 
+    @cluster(num_nodes=4)
+    @matrix(controller_sasl_mechanism=['SCRAM-SHA-256', 'SCRAM-SHA-512'], metadata_quorum=[quorum.isolated_kraft])
+    def test_isolated_kraft_controller_scram(self, controller_sasl_mechanism, metadata_quorum):
+        """Verify that an isolated KRaft quorum can bootstrap with SCRAM controller listeners."""
+        self.kafka.security_protocol = self.kafka.interbroker_security_protocol = 'PLAINTEXT'
+        self.kafka.client_sasl_mechanism = controller_sasl_mechanism
+
+        controller_quorum = self.kafka.controller_quorum
+        controller_quorum.security_protocol = 'SASL_PLAINTEXT'
+        controller_quorum.client_sasl_mechanism = controller_sasl_mechanism
+        controller_quorum.interbroker_security_protocol = 'SASL_PLAINTEXT'
+        controller_quorum.interbroker_sasl_mechanism = controller_sasl_mechanism
+        controller_quorum.controller_security_protocol = 'SASL_PLAINTEXT'
+        controller_quorum.controller_sasl_mechanism = controller_sasl_mechanism
+        controller_quorum.intercontroller_security_protocol = 'SASL_PLAINTEXT'
+        controller_quorum.intercontroller_sasl_mechanism = controller_sasl_mechanism
+
+        self.kafka.start()
+
+        node = self.producer.nodes[0]
+        node.version = KafkaVersion(str(DEV_BRANCH))
+        self.producer.start()
+        wait_until(lambda: self.producer.num_acked > 5, timeout_sec=15,
+                   err_msg="Producer failed to start in a reasonable amount of time.")
+
+        assert is_version(node, [node.version.vstring], proc_grep_string=VerifiableProducer.__qualname__, logger=self.logger)
+        self.producer.wait()
+        assert self.producer.num_acked == self.num_messages

tests/kafkatest/services/kafka/kafka.py

@@ -863,7 +863,7 @@ def start_node(self, node, timeout_sec=60, **kwargs):
                     self.close_port(controller_listener)
 
         self.security_config.setup_node(node)
-        if self.quorum_info.using_zk or self.quorum_info.has_brokers: # TODO: SCRAM currently unsupported for controller quorum
+        if self.quorum_info.using_zk or self.quorum_info.has_brokers:
             self.maybe_setup_broker_scram_credentials(node)
 
         if self.quorum_info.using_kraft:
@@ -917,6 +917,14 @@ def start_node(self, node, timeout_sec=60, **kwargs):
                 # build (max supported is 4.3-IV0), and Feature.validateVersion() does not enforce
                 # the dependency, so the feature can be bootstrapped on its own.
                 cmd += " --feature share.version=%s" % self.share_version
+            if self.node_quorum_info.has_controller_role:
+                for mechanism in sorted(self.security_config.serves_kraft_sasl):
+                    if self.security_config.is_sasl_scram(mechanism):
+                        cmd += " --add-scram %s=[name=%s,password=%s]" % (
+                            mechanism,
+                            SecurityConfig.SCRAM_BROKER_USER,
+                            SecurityConfig.SCRAM_BROKER_PASSWORD,
+                        )
             self.logger.info("Running log directory format command...\n%s" % cmd)
             node.account.ssh(cmd)
 

tests/kafkatest/services/security/templates/jaas.conf

@@ -66,7 +66,7 @@ KafkaServer {
 	user_client="client-secret"
 	user_kafka="kafka-secret";
 {% endif %}
-{% if "SCRAM-SHA-256" in client_sasl_mechanism or "SCRAM-SHA-512" in client_sasl_mechanism %}
+{% if "SCRAM-SHA-256" in enabled_sasl_mechanisms or "SCRAM-SHA-512" in enabled_sasl_mechanisms %}
 	org.apache.kafka.common.security.scram.ScramLoginModule required
 	username="{{ SecurityConfig.SCRAM_BROKER_USER }}"
 	password="{{ SecurityConfig.SCRAM_BROKER_PASSWORD }}";