Fixes for mTLS

kafka-trainings · Jun 24, 2024 · f7d9450 · f7d9450
1 parent afb2d23
commit f7d9450
Show file tree

Hide file tree

Showing 16 changed files with 708 additions and 31 deletions.
diff --git a/csp1_consumer/gradle/wrapper/gradle-wrapper.properties b/csp1_consumer/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.4-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.8-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/csp1_consumer/src/main/java/charging/Main.java b/csp1_consumer/src/main/java/charging/Main.java
@@ -7,16 +7,30 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.util.Collections;
+import java.util.Locale;
+import java.util.Map;
 import java.util.Properties;
+import java.util.stream.Collectors;
 
 public class Main {
     public static void main(final String[] args) throws IOException, InterruptedException {
         final Properties props = new Properties();
-        String configFile = "csp1_consumer.properties";
         if (args.length == 1) {
-            configFile = args[0];
+            props.load(new FileReader(args[0]));
         }
-        props.load(new FileReader(configFile));
+        // Load from environment variables
+        props.putAll(System.getenv()
+                .entrySet()
+                .stream()
+                .filter(mapEntry -> mapEntry.getKey().startsWith("KAFKA_"))
+                .collect(Collectors.toMap(
+                        mapEntry -> {
+                            String envVar = mapEntry.getKey();
+                            return envVar.substring(envVar.indexOf("_") + 1).toLowerCase(Locale.ENGLISH).replace("_", ".");
+                        },
+                        Map.Entry::getValue)));
+
+        System.out.println("Properties: " + props);
         // What do you need to configure here?
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CSP1TransactionDeserializer.class);

diff --git a/csp1_producer/gradle/wrapper/gradle-wrapper.properties b/csp1_producer/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.4-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.8-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/csp1_producer/src/main/java/charging/Main.java b/csp1_producer/src/main/java/charging/Main.java
@@ -8,17 +8,31 @@
 
 import java.io.FileReader;
 import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
 import java.util.Properties;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 public class Main {
     public static void main(final String[] args) throws IOException {
         final Properties props = new Properties();
-        String configFile = "csp1_producer.properties";
         if (args.length == 1) {
-            configFile = args[0];
+            props.load(new FileReader(args[0]));
         }
-        props.load(new FileReader(configFile));
+        // Load from environment variables
+        props.putAll(System.getenv()
+                .entrySet()
+                .stream()
+                .filter(mapEntry -> mapEntry.getKey().startsWith("KAFKA_"))
+                .collect(Collectors.toMap(
+                        mapEntry -> {
+                            String envVar = mapEntry.getKey();
+                            return envVar.substring(envVar.indexOf("_") + 1).toLowerCase(Locale.ENGLISH).replace("_", ".");
+                        },
+                        Map.Entry::getValue)));
+
+        System.out.println("Properties: " + props);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CSP1TransactionSerializer.class);
 

diff --git a/csp1_transformer/src/main/java/charging/Main.java b/csp1_transformer/src/main/java/charging/Main.java
@@ -12,21 +12,37 @@
 import java.io.FileReader;
 import java.io.IOException;
 import java.time.Duration;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Properties;
+import java.util.*;
+import java.util.stream.Collectors;
 
 public class Main {
     public static void main(final String[] args) throws IOException, InterruptedException {
 
-        final Properties consumerProps = new Properties();
-        final Properties producerProps = new Properties();
-        String configFile = "csp1_transformer.properties";
+        final Properties props = new Properties();
         if (args.length == 1) {
-            configFile = args[0];
+            props.load(new FileReader(args[0]));
         }
-        consumerProps.load(new FileReader(configFile));
-        producerProps.load(new FileReader(configFile));
+        // Load from environment variables
+        props.putAll(System.getenv()
+                .entrySet()
+                .stream()
+                .filter(mapEntry -> mapEntry.getKey().startsWith("KAFKA_"))
+                .collect(Collectors.toMap(
+                        mapEntry -> {
+                            String envVar = mapEntry.getKey();
+                            return envVar.substring(envVar.indexOf("_") + 1).toLowerCase(Locale.ENGLISH).replace("_", ".");
+                        },
+                        Map.Entry::getValue)));
+
+        System.out.println("Properties: " + props);
+
+        // copy from Props
+        final Properties consumerProps = new Properties();
+        consumerProps.putAll(props);
+        final Properties producerProps = new Properties();
+        producerProps.putAll(props);
+
+
         consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
         consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, CSP1TransactionDeserializer.class);
         consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

diff --git a/csp2_producer/gradle/wrapper/gradle-wrapper.properties b/csp2_producer/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.4-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.8-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/csp2_producer/src/main/java/charging/Main.java b/csp2_producer/src/main/java/charging/Main.java
@@ -8,17 +8,31 @@
 
 import java.io.FileReader;
 import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
 import java.util.Properties;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 public class Main {
     public static void main(final String[] args) throws IOException {
         final Properties props = new Properties();
-        String configFile = "csp2_producer.properties";
         if (args.length == 1) {
-            configFile = args[0];
+            props.load(new FileReader(args[0]));
         }
-        props.load(new FileReader(configFile));
+        // Load from environment variables
+        props.putAll(System.getenv()
+                .entrySet()
+                .stream()
+                .filter(mapEntry -> mapEntry.getKey().startsWith("KAFKA_"))
+                .collect(Collectors.toMap(
+                        mapEntry -> {
+                            String envVar = mapEntry.getKey();
+                            return envVar.substring(envVar.indexOf("_") + 1).toLowerCase(Locale.ENGLISH).replace("_", ".");
+                        },
+                        Map.Entry::getValue)));
+
+        System.out.println("Properties: " + props);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, CSP2TransactionSerializer.class);
 

diff --git a/csp2_transformer/src/main/java/charging/Main.java b/csp2_transformer/src/main/java/charging/Main.java
@@ -11,18 +11,31 @@
 
 import java.io.FileReader;
 import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
+import java.util.stream.Collectors;
 
 public class Main {
     public static void main(final String[] args) throws IOException {
-
         final Properties props = new Properties();
-        String configFile = "csp2_transformer.properties";
         if (args.length == 1) {
-            configFile = args[0];
+            props.load(new FileReader(args[0]));
         }
-        props.load(new FileReader(configFile));
+        // Load from environment variables
+        props.putAll(System.getenv()
+                .entrySet()
+                .stream()
+                .filter(mapEntry -> mapEntry.getKey().startsWith("KAFKA_"))
+                .collect(Collectors.toMap(
+                        mapEntry -> {
+                            String envVar = mapEntry.getKey();
+                            return envVar.substring(envVar.indexOf("_") + 1).toLowerCase(Locale.ENGLISH).replace("_", ".");
+                        },
+                        Map.Entry::getValue)));
+
+        System.out.println("Properties: " + props);
         props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
         props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
 

diff --git a/dashboard/gradle/wrapper/gradle-wrapper.properties b/dashboard/gradle/wrapper/gradle-wrapper.properties
@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.4-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.8-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
diff --git a/dashboard/src/main/java/charging/Main.java b/dashboard/src/main/java/charging/Main.java
@@ -11,20 +11,33 @@
 
 import java.io.FileReader;
 import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
+import java.util.stream.Collectors;
 
 import static charging.Server.*;
 
 public class Main {
     public static void main(final String[] args) throws IOException {
-
         final Properties props = new Properties();
-        String configFile = "dashboard.properties";
-        if(args.length == 1) {
-            configFile = args[0];
+        if (args.length == 1) {
+            props.load(new FileReader(args[0]));
         }
-        props.load(new FileReader(configFile));
+        // Load from environment variables
+        props.putAll(System.getenv()
+                .entrySet()
+                .stream()
+                .filter(mapEntry -> mapEntry.getKey().startsWith("KAFKA_"))
+                .collect(Collectors.toMap(
+                        mapEntry -> {
+                            String envVar = mapEntry.getKey();
+                            return envVar.substring(envVar.indexOf("_") + 1).toLowerCase(Locale.ENGLISH).replace("_", ".");
+                        },
+                        Map.Entry::getValue)));
+
+        System.out.println("Properties: " + props);
         props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
         props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
         int port = Integer.parseInt(props.getProperty("web.port"));

diff --git a/kubernetes/acls/csp1_consumer.yaml b/kubernetes/acls/csp1_consumer.yaml
@@ -0,0 +1,92 @@
+apiVersion: kafka.strimzi.io/v1beta2
+kind: KafkaUser
+metadata:
+  name: csp1_consumer
+  labels:
+    strimzi.io/cluster: kafka
+spec:
+  authentication:
+    type: tls
+  authorization:
+    type: simple
+    acls:
+      - resource:
+          type: group
+          name: csp1_consumer
+          patternType: literal
+        operation: Read
+        host: "*"
+      - resource:
+          type: topic
+          name: csp1.transactions
+          patternType: literal
+        operations:
+          - Read
+        host: "*"
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: csp1-consumer
+  labels:
+    app: csp1-consumer
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: csp1-consumer
+  template:
+    metadata:
+      labels:
+        app: csp1-consumer
+    spec:
+      containers:
+        - name: csp1-consumer
+          image: ghcr.io/kafka-trainings/example-project/csp1_consumer:latest
+          imagePullPolicy: Always
+          ports:
+            - name: jmx
+              containerPort: 3012
+          volumeMounts:
+            - name: user-certs
+              mountPath: /tls/user.p12
+              subPath: user.p12
+            - name: cluster-certs
+              mountPath: /tls/ca.p12
+              subPath: ca.p12
+          env:
+            # Replace ConfigMap by Environment
+            - name: KAFKA_BOOTSTRAP_SERVERS
+              value: kafka-kafka-bootstrap:9093
+            - name: KAFKA_GROUP_ID
+              value: csp1_consumer
+            - name: KAFKA_TOPIC
+              value: csp1.transactions
+            - name: KAFKA_CLIENT_ID
+              value: csp1_consumer
+            - name: KAFKA_APP_LOG_INFOS
+              value: "true"
+            # SSL Configs
+            - name: KAFKA_SECURITY_PROTOCOL
+              value: SSL
+            - name: KAFKA_SSL_KEYSTORE_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: csp1_consumer
+                  key: user.password
+            - name: KAFKA_SSL_KEYSTORE_LOCATION
+              value: /tls/user.p12
+            - name: KAFKA_SSL_TRUSTSTORE_LOCATION
+              value: /tls/ca.p12
+            - name: KAFKA_SSL_TRUSTSTORE_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: kafka-cluster-ca-cert
+                  key: ca.password
+      volumes:
+        - name: user-certs
+          secret:
+            secretName: csp1_consumer
+        - name: cluster-certs
+          secret:
+            secretName: kafka-cluster-ca-cert