WIP:sasl-client--broker-angepasst--so-funzts

* Für das Setup werden noch weitere Schalter benötigt!
** Das Warten auf den Cluster schlägt fehl, da die
   Zugriffsrechte fehlen!
** Außerdem: Die `client.properties` enthalten hier erst
   mal die Broker-Rechte! Umbenennen! Und/Oder einen
   expliziten Nutzer admin einühren...
** Außerdem: Die Listener-Interfaces `BROKER` und `DOCKER`
   sind so eigentlich doppelt
** ENTWEDER: BROKER ohne Authentifizierung mit `User:ANONYMOUS`,
   wie bei den alten Vorführ-Setups, ODER: Nur ein Interface.
* Außerdem: Das Skript enthält noch Fehler

diff --git a/README.sh b/README.sh
new file mode 100755 (executable)
index 0000000..f1d4061
--- /dev/null
+++ b/README.sh
@@ -0,0 +1,29 @@
+#!/bin/bash
+
+docker compose up -d zookeeper
+sleep 10
+docker compose exec zookeeper zookeeper-shell zookeeper:2181 create /cluster
+docker compose exec zookeeper zookeeper-shell zookeeper:2181 create /cluster/id
+docker compose exec zookeeper zookeeper-shell zookeeper:2181 set /cluster/id '{"version":"1","id":"8e-KqCzwQPqK7EibrzJE5g"}'
+docker compose exec zookeeper zookeeper-shell zookeeper:2181 get /cluster/id
+
+docker compose up -d producer consumer
+docker compose up setup
+
+# client
+docker compose exec cli kafka-acls \
+  --command-config /mnt/admin.properties \
+  --bootstrap-server kafka:9092 \
+  --add \
+  --allow-principal User:client \
+  --topic test \
+  --consumer --group=my-group \
+  --producer
+
+docker compose exec cli kafka-acls \
+  --command-config /mnt/admin.properties \
+  --bootstrap-server kafka:9092 \
+  --list \
+  --user-principal User:client \
+  --group=my-group \
+  --topic test

diff --git a/admin.properties b/admin.properties
new file mode 100644 (file)
index 0000000..b040250
--- /dev/null
+++ b/admin.properties
@@ -0,0 +1,5 @@
+sasl.mechanism=PLAIN
+security.protocol=SASL_PLAINTEXT
+sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
+  username="broker" \
+  password="geheim";

diff --git a/docker-compose.yml b/docker-compose.yml
index 93bd7ad..b42639f 100644 (file)
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -14,14 +14,40 @@ services:
     environment:
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
       KAFKA_LISTENERS: BROKER://:9091, DOCKER://:9092, LOCALHOST://:9081
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: BROKER:PLAINTEXT, DOCKER:PLAINTEXT, LOCALHOST:PLAINTEXT
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: BROKER:SASL_PLAINTEXT, DOCKER:SASL_PLAINTEXT, LOCALHOST:SASL_PLAINTEXT
       KAFKA_ADVERTISED_LISTENERS: BROKER://kafka-1:9091, DOCKER://kafka-1:9092, LOCALHOST://localhost:9081
       KAFKA_BROKER_ID: 1
       KAFKA_INTER_BROKER_LISTENER_NAME: BROKER
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
-      KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: 1000
-      KAFKA_LOG4J_LOGGERS: "org.apache.kafka.metadata.migration=TRACE"
+      KAFKA_SUPER_USERS: User:broker
+      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
+      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "false"
+      KAFKA_LISTENER_NAME_BROKER_SASL_ENABLED_MECHANISMS: PLAIN
+      KAFKA_LISTENER_NAME_BROKER_PLAIN_SASL_JAAS_CONFIG: |
+        org.apache.kafka.common.security.plain.PlainLoginModule required \
+        user_broker="geheim" \
+        user_client="geheim" \
+        username="broker" \
+        password="geheim"; \
+      KAFKA_LISTENER_NAME_DOCKER_SASL_ENABLED_MECHANISMS: PLAIN
+      KAFKA_LISTENER_NAME_DOCKER_PLAIN_SASL_JAAS_CONFIG: |
+        org.apache.kafka.common.security.plain.PlainLoginModule required \
+        user_broker="geheim" \
+        user_client="geheim" \
+        username="broker" \
+        password="geheim"; \
+      KAFKA_LISTENER_NAME_LOCALHOST_SASL_ENABLED_MECHANISMS: PLAIN
+      KAFKA_LISTENER_NAME_LOCALHOST_PLAIN_SASL_JAAS_CONFIG: |
+        org.apache.kafka.common.security.plain.PlainLoginModule required \
+        user_broker="geheim" \
+        user_client="geheim" \
+        username="broker" \
+        password="geheim"; \
+      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
+      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
+      KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: 10000
+      KAFKA_LOG4J_LOGGERS: org.apache.kafka.metadata.migration=TRACE, kafka.authorizer.logger=INFO
     volumes:
       - kafka-1-data:/var/lib/kafka/data
     ports:
@@ -35,14 +61,40 @@ services:
     environment:
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
       KAFKA_LISTENERS: BROKER://:9091, DOCKER://:9092, LOCALHOST://:9082
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: BROKER:PLAINTEXT, DOCKER:PLAINTEXT, LOCALHOST:PLAINTEXT
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: BROKER:SASL_PLAINTEXT, DOCKER:SASL_PLAINTEXT, LOCALHOST:SASL_PLAINTEXT
       KAFKA_ADVERTISED_LISTENERS: BROKER://kafka-2:9091, DOCKER://kafka-2:9092, LOCALHOST://localhost:9082
       KAFKA_BROKER_ID: 2
       KAFKA_INTER_BROKER_LISTENER_NAME: BROKER
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
+      KAFKA_SUPER_USERS: User:broker
+      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
+      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "false"
+      KAFKA_LISTENER_NAME_BROKER_SASL_ENABLED_MECHANISMS: PLAIN
+      KAFKA_LISTENER_NAME_BROKER_PLAIN_SASL_JAAS_CONFIG: |
+        org.apache.kafka.common.security.plain.PlainLoginModule required \
+        user_broker="geheim" \
+        user_client="geheim" \
+        username="broker" \
+        password="geheim"; \
+      KAFKA_LISTENER_NAME_DOCKER_SASL_ENABLED_MECHANISMS: PLAIN
+      KAFKA_LISTENER_NAME_DOCKER_PLAIN_SASL_JAAS_CONFIG: |
+        org.apache.kafka.common.security.plain.PlainLoginModule required \
+        user_broker="geheim" \
+        user_client="geheim" \
+        username="broker" \
+        password="geheim"; \
+      KAFKA_LISTENER_NAME_LOCALHOST_SASL_ENABLED_MECHANISMS: PLAIN
+      KAFKA_LISTENER_NAME_LOCALHOST_PLAIN_SASL_JAAS_CONFIG: |
+        org.apache.kafka.common.security.plain.PlainLoginModule required \
+        user_broker="geheim" \
+        user_client="geheim" \
+        username="broker" \
+        password="geheim"; \
+      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
+      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
       KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: 10000
-      KAFKA_LOG4J_LOGGERS: "org.apache.kafka.metadata.migration=TRACE"
+      KAFKA_LOG4J_LOGGERS: org.apache.kafka.metadata.migration=TRACE, kafka.authorizer.logger=INFO
     volumes:
       - kafka-2-data:/var/lib/kafka/data
     ports:
@@ -61,14 +113,40 @@ services:
     environment:
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
       KAFKA_LISTENERS: BROKER://:9091, DOCKER://:9092, LOCALHOST://:9083
-      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: BROKER:PLAINTEXT, DOCKER:PLAINTEXT, LOCALHOST:PLAINTEXT
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: BROKER:SASL_PLAINTEXT, DOCKER:SASL_PLAINTEXT, LOCALHOST:SASL_PLAINTEXT
       KAFKA_ADVERTISED_LISTENERS: BROKER://kafka-3:9091, DOCKER://kafka-3:9092, LOCALHOST://localhost:9083
       KAFKA_BROKER_ID: 3
       KAFKA_INTER_BROKER_LISTENER_NAME: BROKER
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false"
+      KAFKA_SUPER_USERS: User:broker
+      KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
+      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: "false"
+      KAFKA_LISTENER_NAME_BROKER_SASL_ENABLED_MECHANISMS: PLAIN
+      KAFKA_LISTENER_NAME_BROKER_PLAIN_SASL_JAAS_CONFIG: |
+        org.apache.kafka.common.security.plain.PlainLoginModule required \
+        user_broker="geheim" \
+        user_client="geheim" \
+        username="broker" \
+        password="geheim"; \
+      KAFKA_LISTENER_NAME_DOCKER_SASL_ENABLED_MECHANISMS: PLAIN
+      KAFKA_LISTENER_NAME_DOCKER_PLAIN_SASL_JAAS_CONFIG: |
+        org.apache.kafka.common.security.plain.PlainLoginModule required \
+        user_broker="geheim" \
+        user_client="geheim" \
+        username="broker" \
+        password="geheim"; \
+      KAFKA_LISTENER_NAME_LOCALHOST_SASL_ENABLED_MECHANISMS: PLAIN
+      KAFKA_LISTENER_NAME_LOCALHOST_PLAIN_SASL_JAAS_CONFIG: |
+        org.apache.kafka.common.security.plain.PlainLoginModule required \
+        user_broker="geheim" \
+        user_client="geheim" \
+        username="broker" \
+        password="geheim"; \
+      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
+      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
       KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: 10000
-      KAFKA_LOG4J_LOGGERS: "org.apache.kafka.metadata.migration=TRACE"
+      KAFKA_LOG4J_LOGGERS: org.apache.kafka.metadata.migration=TRACE, kafka.authorizer.logger=INFO
     volumes:
       - kafka-3-data:/var/lib/kafka/data
     ports:
@@ -80,6 +158,8 @@ services:
   cli:
     image: juplo/toolbox
     command: sleep infinity
+    volumes:
+      - ./admin.properties:/mnt/admin.properties:ro
     stop_grace_period: 0s
     depends_on:
       - kafka-1
@@ -88,29 +168,33 @@ services:
 
   setup:
     image: juplo/toolbox
+    volumes:
+      - ./admin.properties:/mnt/admin.properties:ro
     command:
       - bash
       - -c
       - |
-        cub kafka-ready -b kafka-1:9092,kafka-2:9092,kafka-3:9092 3 60 > /dev/null 2>&1 || exit 1
+        cub kafka-ready -c admin.properties -b kafka-1:9092,kafka-2:9092,kafka-3:9092 3 60
         if [ -e INITIALIZED ]
         then
           echo -n Bereits konfiguriert: 
           cat INITIALIZED
-          kafka-topics --bootstrap-server kafka:9092 --describe --topic test
+          kafka-topics --bootstrap-server kafka:9092 --command-config /mnt/admin.properties --describe --topic test
         else
           kafka-topics --bootstrap-server kafka:9092 \
+                       --command-config /mnt/admin.properties \
                        --delete \
                        --if-exists \
                        --topic test
           kafka-topics --bootstrap-server kafka:9092 \
+                       --command-config /mnt/admin.properties \
                        --create \
                        --topic test \
                        --partitions 2 \
                        --replication-factor 3 \
                        --config min.insync.replicas=2 \
           && echo Das Topic \'test\' wurde erfolgreich angelegt: \
-          && kafka-topics --bootstrap-server kafka:9092 --describe --topic test \
+          && kafka-topics --bootstrap-server kafka:9092 --command-config /mnt/admin.properties --describe --topic test \
           && date > INITIALIZED
         fi
     stop_grace_period: 0s
@@ -143,6 +227,10 @@ services:
     environment:
       spring.kafka.bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
       spring.kafka.client-id: producer
+      spring.kafka.jaas.enabled: "true"
+      spring.kafka.properties.security.protocol: SASL_PLAINTEXT
+      spring.kafka.properties.sasl.mechanism: PLAIN
+      spring.kafka.properties.sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="client" password="geheim";
       juplo.producer.topic: test
 
   consumer:
@@ -150,6 +238,10 @@ services:
     environment:
       spring.kafka.bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
       spring.kafka.client-id: consumer
+      spring.kafka.jaas.enabled: "true"
+      spring.kafka.properties.security.protocol: SASL_PLAINTEXT
+      spring.kafka.properties.sasl.mechanism: PLAIN
+      spring.kafka.properties.sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="client" password="geheim";
       spring.kafka.consumer.auto-offset-reset: earliest
       logging.level.org.apache.kafka.clients.consumer: INFO
       juplo.consumer.topic: test
@@ -159,6 +251,10 @@ services:
     environment:
       spring.kafka.bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
       spring.kafka.client-id: peter
+      spring.kafka.jaas.enabled: "true"
+      spring.kafka.properties.security.protocol: SASL_PLAINTEXT
+      spring.kafka.properties.sasl.mechanism: PLAIN
+      spring.kafka.properties.sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="client" password="geheim";
       spring.kafka.consumer.auto-offset-reset: earliest
       logging.level.org.apache.kafka.clients.consumer: INFO
       juplo.consumer.topic: test
@@ -168,6 +264,10 @@ services:
     environment:
       spring.kafka.bootstrap-servers: kafka-1:9092,kafka-2:9092,kafka-3:9092
       spring.kafka.client-id: ute
+      spring.kafka.jaas.enabled: "true"
+      spring.kafka.properties.security.protocol: SASL_PLAINTEXT
+      spring.kafka.properties.sasl.mechanism: PLAIN
+      spring.kafka.properties.sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="client" password="geheim";
       spring.kafka.consumer.auto-offset-reset: earliest
       logging.level.org.apache.kafka.clients.consumer: INFO
       juplo.consumer.topic: test