rgw/kafka/tests: add SASL mechanism tests

Signed-off-by: Yuval Lifshitz <ylifshit@redhat.com>

2 files changed, 99 insertions, 29 deletions

diff --git a/src/test/rgw/bucket_notification/README.rst b/src/test/rgw/bucket_notification/README.rst
index 9686bef71b3..20eee8463cf 100644
--- a/src/test/rgw/bucket_notification/README.rst
+++ b/src/test/rgw/bucket_notification/README.rst
@@ -5,12 +5,10 @@
 You will need to use the sample configuration file named ``bntests.conf.SAMPLE``
 that has been provided at ``/path/to/ceph/src/test/rgw/bucket_notification/``. You can also copy this file to the directory where you are
 running the tests and modify it if needed. This file can be used to run the bucket notification tests on a Ceph cluster started
-with vstart.
+with the `vstart.sh` script.
 For the tests covering Kafka and RabbitMQ security, the RGW will need to accept use/password without TLS connection between the client and the RGW.
 So, the cluster will have to be started with the following ``rgw_allow_notification_secrets_in_cleartext`` parameter set to ``true``.
-For example::
 
-  MON=1 OSD=1 MDS=0 MGR=1 RGW=1 ../src/vstart.sh -n -d -o "rgw_allow_notification_secrets_in_cleartext=true"
 
 ===========
 Kafka Tests
@@ -18,22 +16,10 @@ Kafka Tests
 
 You also need to install Kafka which can be downloaded from: https://kafka.apache.org/downloads
 
-To test Kafka security, you should first run the ``kafka-security.sh`` script inside the Kafka directory.
-
 Then edit the Kafka server properties file (``/path/to/kafka/config/server.properties``)
-to have the following lines::
-
-  listeners=PLAINTEXT://localhost:9092,SSL://localhost:9093,SASL_SSL://localhost:9094
-  ssl.keystore.location=/home/ylifshit/kafka-3.3.1-src/server.keystore.jks 
-  ssl.keystore.password=mypassword 
-  ssl.key.password=mypassword 
-  ssl.truststore.location=/home/ylifshit/kafka-3.3.1-src/server.truststore.jks 
-  ssl.truststore.password=mypassword 
-  sasl.enabled.mechanisms=PLAIN
-  listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
-     username="alice" \
-     password="alice-secret" \
-     user_alice="alice-secret";
+to have the following line::
+
+        listeners=PLAINTEXT://localhost:9092
 
 After following the above steps, start the Zookeeper and Kafka services.
 For starting Zookeeper service run::
@@ -52,13 +38,72 @@ and::
 
         bin/kafka-server-start.sh -daemon config/server.properties
 
-After running vstart, Zookeeper, and Kafka services you're ready to run the Kafka tests::
+After running `vstart.sh`, Zookeeper, and Kafka services you're ready to run the Kafka tests::
 
         BNTESTS_CONF=bntests.conf python -m nose -s /path/to/ceph/src/test/rgw/bucket_notification/test_bn.py -v -a 'kafka_test'
 
+--------------------
+Kafka Security Tests
+--------------------
+
+First, make sure that vstart was initiated with the following ``rgw_allow_notification_secrets_in_cleartext`` parameter set to ``true``::
+
+        MON=1 OSD=1 MDS=0 MGR=1 RGW=1 ../src/vstart.sh -n -d -o "rgw_allow_notification_secrets_in_cleartext=true"
+
+Then you should run the ``kafka-security.sh`` script inside the Kafka directory::
+
+        cd /path/to/kafka/
+        /path/to/ceph/src/test/rgw/bucket_notification/kafka-security.sh
+
+Then make sure the Kafka server properties file (``/path/to/kafka/config/server.properties``) has the following lines::
+
+
+        # all listeners
+        listeners=PLAINTEXT://localhost:9092,SSL://localhost:9093,SASL_SSL://localhost:9094,SASL_PLAINTEXT://localhost:9095
+
+        # SSL configuration matching the kafka-security.sh script
+        ssl.keystore.location=./server.keystore.jks
+        ssl.keystore.password=mypassword
+        ssl.key.password=mypassword
+        ssl.truststore.location=./server.truststore.jks
+        ssl.truststore.password=mypassword
+
+        # SASL mechanisms
+        sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-256
+
+        # SASL over SSL with SCRAM-SHA-256 mechanism
+        listener.name.sasl_ssl.scram-sha-256.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
+          username="alice" \
+          password="alice-secret" \
+          user_alice="alice-secret";
+
+        # SASL over SSL with PLAIN mechanism
+        listener.name.sasl_ssl.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
+          username="alice" \
+          password="alice-secret" \
+          user_alice="alice-secret";
+
+        # PLAINTEXT SASL with SCRAM-SHA-256 mechanism
+        listener.name.sasl_plaintext.scram-sha-256.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \
+          username="alice" \
+          password="alice-secret" \
+          user_alice="alice-secret";
+
+        # PLAINTEXT SASL with PLAIN mechanism
+        listener.name.sasl_plaintext.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
+          username="alice" \
+          password="alice-secret" \
+          user_alice="alice-secret";
+
+
+And restart the Kafka server. Once both Zookeeper and Kafka are up, run the following command (for the SASL SCRAM test) from the Kafka directory::
+
+        bin/kafka-configs.sh --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[iterations=8192,password=alice-secret],SCRAM-SHA-512=[password=alice-secret]' --entity-type users --entity-name alice
+
+
 To run the Kafka security test, you also need to provide the test with the location of the Kafka directory::
 
-        KAFKA_DIR=/path/to/kafkaBNTESTS_CONF=bntests.conf python -m nose -s /path/to/ceph/src/test/rgw/bucket_notification/test_bn.py -v -a 'kafka_ssl_test'
+        KAFKA_DIR=/path/to/kafka BNTESTS_CONF=bntests.conf python -m nose -s /path/to/ceph/src/test/rgw/bucket_notification/test_bn.py -v -a 'kafka_security_test'
 
 ==============
 RabbitMQ Tests
@@ -80,7 +125,7 @@ To confirm that the RabbitMQ server is running you can run the following command
 
         sudo /sbin/service rabbitmq-server status
 
-After running vstart and RabbitMQ server you're ready to run the AMQP tests::
+After running `vstart.sh` and RabbitMQ server you're ready to run the AMQP tests::
 
         BNTESTS_CONF=bntests.conf python -m nose -s /path/to/ceph/src/test/rgw/bucket_notification/test_bn.py -v -a 'amqp_test'
 
@@ -93,4 +138,6 @@ To run the RabbitMQ SSL security tests use the following::
         BNTESTS_CONF=bntests.conf python -m nose -s /path/to/ceph/src/test/rgw/bucket_notification/test_bn.py -v -a 'amqp_ssl_test'
 
 During these tests, the test script will restart the RabbitMQ server with the correct security configuration (``sudo`` privileges will be needed).
+For that reason it is not recommended to run the `amqp_ssl_test` tests, that assumes a manually configured rabbirmq server, in the same run as `amqp_test` tests, 
+that assume the rabbitmq daemon running on the host as a service.
 
diff --git a/src/test/rgw/bucket_notification/test_bn.py b/src/test/rgw/bucket_notification/test_bn.py
index d0f267ef2b7..d3d77ba1a2f 100644
--- a/src/test/rgw/bucket_notification/test_bn.py
+++ b/src/test/rgw/bucket_notification/test_bn.py
@@ -463,7 +463,7 @@ def create_kafka_receiver_thread(topic, security_type='PLAINTEXT'):
     return task, receiver
 
 def stop_kafka_receiver(receiver, task):
-    """stop the receiver thread and wait for it to finis"""
+    """stop the receiver thread and wait for it to finish"""
     receiver.stop = True
     task.join(1)
     try:
@@ -3871,7 +3871,7 @@ def test_ps_s3_multiple_topics_notification():
     http_server.close()
 
 
-def kafka_security(security_type):
+def kafka_security(security_type, mechanism='PLAIN'):
     """ test pushing kafka s3 notification securly to master """
     conn = connection()
     zonegroup = 'default'
@@ -3881,15 +3881,23 @@ def kafka_security(security_type):
     # name is constant for manual testing
     topic_name = bucket_name+'_topic'
     # create s3 topic
-    if security_type == 'SSL_SASL':
+    if security_type == 'SASL_SSL':
         endpoint_address = 'kafka://alice:alice-secret@' + kafka_server + ':9094'
     elif security_type == 'SSL':
         endpoint_address = 'kafka://' + kafka_server + ':9093'
+    elif security_type == 'SASL_PLAINTEXT':
+        endpoint_address = 'kafka://alice:alice-secret@' + kafka_server + ':9095'
     else:
         assert False, 'unknown security method '+security_type
 
-    KAFKA_DIR = os.environ['KAFKA_DIR']
-    endpoint_args = 'push-endpoint='+endpoint_address+'&kafka-ack-level=broker&use-ssl=true&ca-location='+KAFKA_DIR+"/y-ca.crt"
+    if security_type == 'SASL_PLAINTEXT':
+        endpoint_args = 'push-endpoint='+endpoint_address+'&kafka-ack-level=broker&use-ssl=false&mechanism='+mechanism
+    elif security_type == 'SASL_SSL':
+        KAFKA_DIR = os.environ['KAFKA_DIR']
+        endpoint_args = 'push-endpoint='+endpoint_address+'&kafka-ack-level=broker&use-ssl=true&ca-location='+KAFKA_DIR+'/y-ca.crt&mechanism='+mechanism
+    else:
+        KAFKA_DIR = os.environ['KAFKA_DIR']
+        endpoint_args = 'push-endpoint='+endpoint_address+'&kafka-ack-level=broker&use-ssl=true&ca-location='+KAFKA_DIR+'/y-ca.crt'
 
     topic_conf = PSTopicS3(conn, topic_name, zonegroup, endpoint_args=endpoint_args)
     
@@ -3949,12 +3957,27 @@ def kafka_security(security_type):
         stop_kafka_receiver(receiver, task)
 
 
-@attr('kafka_ssl_test')
+@attr('kafka_security_test')
 def test_ps_s3_notification_push_kafka_security_ssl():
     kafka_security('SSL')
 
 
-@attr('kafka_ssl_test')
+@attr('kafka_security_test')
 def test_ps_s3_notification_push_kafka_security_ssl_sasl():
-    kafka_security('SSL_SASL')
+    kafka_security('SASL_SSL')
+
+
+@attr('kafka_security_test')
+def test_ps_s3_notification_push_kafka_security_sasl():
+    kafka_security('SASL_PLAINTEXT')
+
+
+@attr('kafka_security_test')
+def test_ps_s3_notification_push_kafka_security_ssl_sasl_scram():
+    kafka_security('SASL_SSL', mechanism='SCRAM-SHA-256')
+
+
+@attr('kafka_security_test')
+def test_ps_s3_notification_push_kafka_security_sasl_scram():
+    kafka_security('SASL_PLAINTEXT', mechanism='SCRAM-SHA-256')