Add SASL based authentication (#1)

* Add SASL based authentication

* Fix default tests

defaults/main.yml

@@ -47,6 +47,11 @@ kafka_socket_request_max_bytes: 104857600
 # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
 #kafka_listener_security_protocol_map: "PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL"
 
+# Additional optional security and sasl related settings
+#security.inter.broker.protocol=PLAINTEXT
+#sasl.mechanism.inter.broker.protocol=PLAIN
+#sasl.enabled.mechanisms=PLAIN
+
 # The default number of log partitions per topic. More partitions allow greater
 # parallelism for consumption, but this will also result in more files across
 # the brokers.
@@ -120,3 +125,9 @@ kafka_group_initial_rebalance_delay_ms: 0
 
 # Environment file that will be created by the role and used in the Kafka systemd service definition if kafka_environment_variables is defined
 kafka_environment_file: /etc/kafka/kafka.env
+
+kafka_server_username: broker
+kafka_server_password: broker
+kafka_client_users:
+  - username: guest
+    password: guest

molecule/default/verify.yml

@@ -17,7 +17,7 @@
       register: stat_log_result
       failed_when: not stat_log_result.stat.exists
     - name: Add a topic to the kafka server
-      command: /usr/local/kafka/bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test_default
+      command: /usr/local/kafka/bin/kafka-topics.sh --create --bootstrap-server {{ inventory_hostname }}:9092 --replication-factor 1 --partitions 1 --topic test_default
       register: create_topic_result
     - name: Verify topic creation return code
       assert:
@@ -28,7 +28,7 @@
         that:
           - "'Created topic test_default' in create_topic_result.stdout"
     - name: List topics from the kafka server
-      command: /usr/local/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092
+      command: /usr/local/kafka/bin/kafka-topics.sh --list --bootstrap-server {{ inventory_hostname }}:9092
       register: list_topics_result
     - name: Verify topic listing return code
       assert:

tasks/config.yml

@@ -1,13 +1,32 @@
 ---
+- name: Create kafka group
+  group:
+    name: '{{ kafka_group }}'
+    state: present
+
+- name: Create kafka user
+  user:
+    name: '{{ kafka_user }}'
+    group: '{{ kafka_group }}'
+    state: present
+    createhome: no
+
 - name: Create kafka config dir
   file:
     path: "{{ kafka_conf_dir }}"
     state: directory
+    owner: '{{ kafka_user }}'
+    group: '{{ kafka_group }}'
     mode: '0755'
 
+- import_tasks: sasl-auth.yml
+  tags: [kafka, kafka_config, config, sasl, auth]
+
 - name: Create kafka server properties file
   template:
     src: server.properties.j2
     dest: "{{ kafka_conf_dir }}/server.properties"
+    owner: '{{ kafka_user }}'
+    group: '{{ kafka_group }}'
     mode: '0644'
   notify: Reload kafka

tasks/install.yml

@@ -1,16 +1,4 @@
 ---
-- name: Create kafka group
-  group:
-    name: '{{ kafka_group }}'
-    state: present
-
-- name: Create kafka user
-  user:
-    name: '{{ kafka_user }}'
-    group: '{{ kafka_group }}'
-    state: present
-    createhome: no
-
 - name: Set kafka_archive_name variable
   set_fact:
     kafka_archive_name: "kafka_{{ kafka_scala_version }}-{{ kafka_version }}"

tasks/main.yml

@@ -1,5 +1,5 @@
 ---
-  - import_tasks: install.yml
-    tags: [kafka, kafka_install, install]
   - import_tasks: config.yml
     tags: [kafka, kafka_config, config]
+  - import_tasks: install.yml
+    tags: [kafka, kafka_install, install]

tasks/sasl-auth.yml

@@ -0,0 +1,40 @@
+---
+  - name: Create JAAS configuration for Kafka broker
+    template:
+      src: kafka_server_jaas.conf.j2
+      dest: "{{ kafka_conf_dir }}/kafka_server_jaas.conf"
+      owner: '{{ kafka_user }}'
+      group: '{{ kafka_group }}'
+      mode: '0644'
+    notify: Reload kafka
+
+  - name: Add to kafka_environment_variables variable if it already exists
+    set_fact:
+      kafka_environment_variables: "{{ kafka_environment_variables | combine({'KAFKA_OPTS': '-Djava.security.auth.login.config=' + kafka_conf_dir + '/kafka_server_jaas.conf'}) }}"
+    when: kafka_environment_variables is defined
+
+  - name: Set fresh kafka_environment_variables variable if it does not exist
+    set_fact:
+      kafka_environment_variables:
+        KAFKA_OPTS: "-Djava.security.auth.login.config={{ kafka_conf_dir }}/kafka_server_jaas.conf"
+    when: kafka_environment_variables is not defined
+
+  - name: Set Kafka listeners variable
+    set_fact:
+      kafka_listeners: "PLAINTEXT://{{ inventory_hostname }}:9092"
+    when: kafka_listeners is not defined
+
+  - name: Set Kafka inter-broker communication protocol
+    set_fact:
+      kafka_security_inter_broker_protocol: "PLAINTEXT"
+    when: kafka_security_inter_broker_protocol is not defined
+
+  - name: Set SASL mechanism for Kafka inter-broker communication protocol
+    set_fact:
+      kafka_sasl_mechanism_inter_broker_protocol: "PLAIN"
+    when: kafka_sasl_mechanism_inter_broker_protocol is not defined
+
+  - name: Set enabled SASL mechanisms for Kafka
+    set_fact:
+      kafka_sasl_enabled_mechanisms: "PLAIN"
+    when: kafka_sasl_enabled_mechanisms is not defined

templates/kafka_server_jaas.conf.j2

@@ -0,0 +1,9 @@
+KafkaServer {
+    org.apache.kafka.common.security.plain.PlainLoginModule required
+    username="{{ kafka_server_username }}"
+    password="{{ kafka_server_password }}"
+    user_{{ kafka_server_username }}="{{ kafka_server_password }}"
+{% for client in kafka_client_users %}
+    user_{{ client.username }}="{{ client.password }}"
+{% endfor %};
+};

templates/server.properties.j2

@@ -47,6 +47,19 @@ advertised.listeners={{ kafka_advertised_listeners }}
 listener.security.protocol.map={{ kafka_listener_security_protocol_map }}
 {% endif %}
 
+# Additional security and sasl related settings
+{% if kafka_security_inter_broker_protocol is defined %}
+security.inter.broker.protocol={{ kafka_security_inter_broker_protocol }}
+{% endif %}
+
+{% if kafka_sasl_mechanism_inter_broker_protocol is defined %}
+sasl.mechanism.inter.broker.protocol={{ kafka_sasl_mechanism_inter_broker_protocol }}
+{% endif %}
+
+{% if kafka_sasl_enabled_mechanisms is defined %}
+sasl.enabled.mechanisms={{ kafka_sasl_enabled_mechanisms }}
+{% endif %}
+
 # The number of threads that the server uses for receiving requests from the network and sending responses to the network
 num.network.threads={{ kafka_num_network_threads }}