feat(data-masking): added new ability to mask the record key and values

application.example.yml

@@ -295,4 +295,13 @@ akhq:
         - username: header-admin
           groups:
             - admin
-
+
+    # Data masking configuration
+    data-masking:
+      filters:
+        - description: "Masks value for secret-key fields"
+          search-regex: '"(secret-key)":".*"'
+          replacement: '"$1":"xxxx"'
+        - description: "Masks last digits of phone numbers"
+          search-regex: '"([\+]?[(]?[0-9]{3}[)]?[-\s\.]?[0-9]{3}[-\s\.]?)[0-9]{4,6}"'
+          replacement: '"$1xxxx"'

docs/docs/configuration/akhq.md

@@ -67,3 +67,18 @@ akhq:
         value: "none"
 ```
 
+## Data Masking
+If you want to hide some data in your records, you can configure this with the following filters.
+These will be applied to all record values and keys.
+```yaml
+akhq:
+  security:
+    data-masking:
+      filters:
+        - description: "Masks value for secret-key fields"
+          search-regex: '"(secret-key)":".*"'
+          replacement: '"$1":"xxxx"'
+        - description: "Masks last digits of phone numbers"
+          search-regex: '"([\+]?[(]?[0-9]{3}[)]?[-\s\.]?[0-9]{3}[-\s\.]?)[0-9]{4,6}"'
+          replacement: '"$1xxxx"'
+```

src/main/java/org/akhq/configs/DataMasking.java

@@ -0,0 +1,13 @@
+package org.akhq.configs;
+
+import io.micronaut.context.annotation.ConfigurationProperties;
+import lombok.Data;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@ConfigurationProperties("akhq.security.data-masking")
+@Data
+public class DataMasking {
+   List<DataMaskingFilter> filters = new ArrayList<>();
+}

src/main/java/org/akhq/configs/DataMaskingFilter.java

@@ -0,0 +1,12 @@
+package org.akhq.configs;
+
+import io.micronaut.context.annotation.EachProperty;
+import lombok.Data;
+
+@EachProperty("filters")
+@Data
+public class DataMaskingFilter {
+    String description;
+    String searchRegex;
+    String replacement;
+}

src/main/java/org/akhq/models/Record.java

@@ -157,6 +157,10 @@ public void setValue(String value) {
         this.value = value;
     }
 
+    public void setKey(String key) {
+        this.key = key;
+    }
+
     public void setTruncated(Boolean truncated) {
         this.truncated = truncated;
     }

src/main/java/org/akhq/repositories/RecordRepository.java

@@ -22,6 +22,7 @@
 import org.akhq.modules.schemaregistry.RecordWithSchemaSerializerFactory;
 import org.akhq.utils.AvroToJsonSerializer;
 import org.akhq.utils.Debug;
+import org.akhq.utils.MaskingUtils;
 import org.apache.kafka.clients.admin.DeletedRecords;
 import org.apache.kafka.clients.admin.RecordsToDelete;
 import org.apache.kafka.clients.consumer.*;
@@ -72,6 +73,9 @@ public class RecordRepository extends AbstractRepository {
     @Inject
     private AvroWireFormatConverter avroWireFormatConverter;
 
+    @Inject
+    private MaskingUtils maskingUtils;
+
     @Value("${akhq.topic-data.poll-timeout:1000}")
     protected int pollTimeout;
 
@@ -443,7 +447,7 @@ private ConsumerRecords<byte[], byte[]> poll(KafkaConsumer<byte[], byte[]> consu
     private Record newRecord(ConsumerRecord<byte[], byte[]> record, String clusterId, Topic topic) {
         SchemaRegistryType schemaRegistryType = this.schemaRegistryRepository.getSchemaRegistryType(clusterId);
         SchemaRegistryClient client = this.kafkaModule.getRegistryClient(clusterId);
-        return new Record(
+        return maskingUtils.maskRecord(new Record(
             client,
             record,
             this.schemaRegistryRepository.getSchemaRegistryType(clusterId),
@@ -456,13 +460,13 @@ private Record newRecord(ConsumerRecord<byte[], byte[]> record, String clusterId
             avroWireFormatConverter.convertValueToWireFormat(record, client,
                     this.schemaRegistryRepository.getSchemaRegistryType(clusterId)),
             topic
-        );
+        ));
     }
 
     private Record newRecord(ConsumerRecord<byte[], byte[]> record, BaseOptions options, Topic topic) {
         SchemaRegistryType schemaRegistryType = this.schemaRegistryRepository.getSchemaRegistryType(options.clusterId);
         SchemaRegistryClient client = this.kafkaModule.getRegistryClient(options.clusterId);
-        return new Record(
+        return maskingUtils.maskRecord(new Record(
             client,
             record,
             schemaRegistryType,
@@ -475,7 +479,7 @@ private Record newRecord(ConsumerRecord<byte[], byte[]> record, BaseOptions opti
             avroWireFormatConverter.convertValueToWireFormat(record, client,
                     this.schemaRegistryRepository.getSchemaRegistryType(options.clusterId)),
             topic
-        );
+        ));
     }
 
     public List<RecordMetadata> produce(

src/main/java/org/akhq/utils/MaskingUtils.java

@@ -0,0 +1,38 @@
+package org.akhq.utils;
+
+import jakarta.inject.Inject;
+import jakarta.inject.Singleton;
+import org.akhq.configs.DataMasking;
+import org.akhq.configs.DataMaskingFilter;
+import org.akhq.models.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@Singleton
+public class MaskingUtils {
+    private static final Logger LOG = LoggerFactory.getLogger(MaskingUtils.class);
+
+    @Inject
+    DataMasking dataMasking;
+
+    public Record maskRecord(Record record) {
+        LOG.trace("masking record");
+
+        String value = record.getValue();
+        String key = record.getKey();
+
+        for (DataMaskingFilter filter : dataMasking.getFilters()) {
+            if (value != null) {
+                value = value.replaceAll(filter.getSearchRegex(), filter.getReplacement());
+            }
+            if (key != null) {
+                key = key.replaceAll(filter.getSearchRegex(), filter.getReplacement());
+            }
+        }
+
+        record.setValue(value);
+        record.setKey(key);
+
+        return record;
+    }
+}

src/test/java/org/akhq/utils/MaskingUtilsTest.java

@@ -0,0 +1,64 @@
+package org.akhq.utils;
+
+import io.micronaut.context.ApplicationContext;
+import org.akhq.models.Record;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class MaskingUtilsTest {
+
+    @Test
+    void shouldMasksRecordValue() {
+        ApplicationContext ctx = ApplicationContext.run("data-masking");
+        MaskingUtils maskingUtils = ctx.getBean(MaskingUtils.class);
+
+        Record record = new Record();
+        record.setValue("{\"secret-key\":\"my-secret-value\"}");
+
+        Record maskedRecord = maskingUtils.maskRecord(record);
+
+        assertEquals(
+            "{\"secret-key\":\"xxxx\"}",
+            maskedRecord.getValue()
+        );
+
+        ctx.close();
+    }
+
+    @Test
+    void shouldMasksRecordKey() {
+        ApplicationContext ctx = ApplicationContext.run("data-masking");
+        MaskingUtils maskingUtils = ctx.getBean(MaskingUtils.class);
+
+        Record record = new Record();
+        record.setKey("{\"secret-key\":\"my-secret-value\"}");
+
+        Record maskedRecord = maskingUtils.maskRecord(record);
+
+        assertEquals(
+            "{\"secret-key\":\"xxxx\"}",
+            maskedRecord.getKey()
+        );
+
+        ctx.close();
+    }
+
+    @Test
+    void shouldReturnGroupsToAllowPartialMasking() {
+        ApplicationContext ctx = ApplicationContext.run("data-masking");
+        MaskingUtils maskingUtils = ctx.getBean(MaskingUtils.class);
+
+        Record record = new Record();
+        record.setValue("{\"some-key\":\"+12092503766\"}");
+
+        Record maskedRecord = maskingUtils.maskRecord(record);
+
+        assertEquals(
+            "{\"some-key\":\"+120925xxxx\"}",
+            maskedRecord.getValue()
+        );
+
+        ctx.close();
+    }
+}

src/test/resources/application-data-masking.yml

@@ -0,0 +1,10 @@
+akhq:
+  security:
+    data-masking:
+      filters:
+        - description: "Masks value for secret-key fields"
+          search-regex: '"(secret-key)":".*"'
+          replacement: '"$1":"xxxx"'
+        - description: "Masks last digits of phone numbers"
+          search-regex: '"([\+]?[(]?[0-9]{3}[)]?[-\s\.]?[0-9]{3}[-\s\.]?)[0-9]{4,6}"'
+          replacement: '"$1xxxx"'