Go for Secured Consumer and dev-broker

Author: mmohamed

README.md

@@ -78,7 +78,7 @@ So, we need to add a DIGEST authentication layer to Zookeeper (doesn’t support
 
 Follow the [security section documentation](SECURITY.md)
 
-## 5- Sourcing
+## 6- Sourcing
 * Zookeeper Docker image : we use the [kubernetes-zookeeper @kow3ns](https://github.com/kow3ns/kubernetes-zookeeper) as base image with 2 modifications:
     * Add JVM flags to be injected in Java environment file [@see start-zookeeper.sh](zookeeper/docker/scripts/start-zookeeper)
     ```bash
@@ -104,7 +104,7 @@ Follow the [security section documentation](SECURITY.md)
     fi
     ```
 
-## 6- Tips
+## 7- Tips
 * For debugging, you can bypass the Kafka broker for topics management (kafka and ZooKeeper helpers script) :
 ```bash
 kubectl exec -ti kafka-0 -- kafka-topics.sh --create --topic k8s --zookeeper zk-cs.kafka.svc.cluster.local:2181

SECURITY.md

@@ -125,15 +125,23 @@ kubectl exec -ti kafka-1 -- kafka-console-producer.sh --bootstrap-server kafka-0
 >> Hello with secured connection
 ```
 ```bash
-kubectl exec -ti kafka-1 -- kafka-console-consumer.sh --bootstrap-server kafka-0.kafka-broker.kafka.svc.cluster.local:9093 --topic k8s --consumer.config /opt/kafka/config/client.properties --from-beginning
+kubectl exec -ti kafka-1 -- kafka-console-consumer.sh --bootstrap-server kafka-0.kafka-broker.kafka.svc.cluster.local:9093 --topic k8s -consumer.config /opt/kafka/config/client.properties --from-beginning
 << Hello with secured connection
 ```
 
+Prepare Secret for client:
+```bash
+kubectl create secret generic client-ssl --from-file=ca-certs.pem=ssl/ca-cert --from-file=cert.pem=ssl/cert-signed-client
+kubectl apply -f consumer.secured.yaml
+kubectl logs consumer-secured
+```
+
+
 6. Sources & Links:
-- https://access.redhat.com/documentation/en-us/red_hat_amq/7.2/html/using_amq_streams_on_red_hat_enterprise_linux_rhel/configuring_kafka
-- https://cwiki.apache.org/confluence/display/ZOOKEEPER/Server-Server+mutual+authentication
-- https://kafka.apache.org/documentation/#security_overview
-- https://github.com/bitnami/charts/issues/1279
-- https://stackoverflow.com/questions/54903381/kafka-failed-authentication-due-to-ssl-handshake-failed
-- https://www.vertica.com/docs/9.2.x/HTML/Content/Authoring/KafkaIntegrationGuide/TLS-SSL/KafkaTLS-SSLExamplePart3ConfigureKafka.htm?tocpath=Integrating%20with%20Apache%20Kafka%7CUsing%20TLS%2FSSL%20Encryption%20with%20Kafka%7C_____7
-- https://gist.github.com/anoopl/85d869f7a85a70c6c60542922fc314a8
+- [Redhat-Kafka](https://access.redhat.com/documentation/en-us/red_hat_amq/7.2/html/using_amq_streams_on_red_hat_enterprise_linux_rhel/configuring_kafka)
+- [Confluence-Zookeeper](https://cwiki.apache.org/confluence/display/ZOOKEEPER/Server-Server+mutual+authentication)
+- [Apache-Kafka](https://kafka.apache.org/documentation/#security_overview)
+- [Bitnami-Kafka](https://github.com/bitnami/charts/issues/1279)
+- [Kafka-SSL](https://stackoverflow.com/questions/54903381/kafka-failed-authentication-due-to-ssl-handshake-failed)
+- [Vertica-Kafka-SSL](https://www.vertica.com/docs/9.2.x/HTML/Content/Authoring/KafkaIntegrationGuide/TLS-SSL/KafkaTLS-SSLExamplePart3ConfigureKafka.htm?tocpath=Integrating%20with%20Apache%20Kafka%7CUsing%20TLS%2FSSL%20Encryption%20with%20Kafka%7C_____7)
+- [Kafka-SSL](https://gist.github.com/anoopl/85d869f7a85a70c6c60542922fc314a8)

kafka/consumer.secured.yaml

@@ -0,0 +1,51 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: consumer-secured
+data:
+  client: |
+    from kafka import KafkaConsumer
+    import logging
+    import sys
+
+    logger = logging.getLogger('kafka')
+    logger.addHandler(logging.StreamHandler(sys.stdout))
+    logger.setLevel(logging.INFO)
+
+    print('Init consumer...')
+
+    consumer = KafkaConsumer(
+      'k8s',
+      security_protocol='SSL',
+      ssl_cafile='/var/ssl/ca-certs.pem',
+      ssl_certfile='/var/ssl/cert.pem',
+      ssl_check_hostname= False,
+      ssl_password='passcode',
+      bootstrap_servers=['kafka-service.kafka.svc.cluster.local:9093'])
+    
+    print('Waiting for message')
+    for message in consumer:
+      print ("%s:%d:%d: key=%s value=%s" % (message.topic, message.partition, message.offset, message.key, message.value))
+---
+
+apiVersion: v1
+kind: Pod
+metadata:
+  name: consumer-secured
+spec:
+  containers:
+  - name: client
+    image: python:3.8.5
+    command: ['sh', '-c', 'pip install kafka-python && python /var/static/client']
+    volumeMounts:
+      - name: static-volume
+        mountPath: /var/static
+      - name: secret-volume
+        mountPath: /var/ssl
+  volumes:
+    - name: static-volume
+      configMap:
+        name: consumer-secured
+    - name: secret-volume
+      secret:
+        secretName: client-ssl

kafka/dev-brocker.secured.yaml

@@ -0,0 +1,41 @@
+apiVersion: v1
+kind: Pod
+metadata:
+  name: dev-secured-brocker
+spec:
+  affinity:
+    nodeAffinity:
+      requiredDuringSchedulingIgnoredDuringExecution:
+        nodeSelectorTerms:
+        - matchExpressions:
+          - key: kubernetes.io/arch
+            operator: In
+            values:
+            - amd64
+            - arm64
+  containers:
+  - name: single-secured-broker
+    image: medinvention/kafka:2.13-2.7.0
+    env:
+    - name: HOSTNAME_VALUE
+      value: "127.0.0.1"
+    - name: KAFKA_ZOOKEEPER_CONNECT
+      value: "zk-cs.kafka.svc.cluster.local:2181"
+    - name: KAFKA_LOG4J_OPTS
+      value: "-Dlog4j.configuration=file:/opt/kafka/config/log4j.properties -Djava.security.auth.login.config=/var/lib/kafka/jass/jaas.conf"
+    - name: KAFAK_AUTO_CREATE_TOPICS_ENABLE
+      value: "false"
+    ports:
+    - containerPort: 9092
+    volumeMounts:
+    - name: data
+      mountPath: /var/lib/kafka/data
+    - name: jaas
+      mountPath: /var/lib/kafka/jass
+      readOnly: true
+  volumes:
+  - name: data
+    emptyDir: {}
+  - name: jaas
+    secret:
+      secretName: kafka-jaas