Repository: kroxylicious/kroxylicious (java)

Issue:
Expose record encryption parameters to configuration
              piggybacking a small feature request onto this issue - `RecordEncryption` currently [hardcodes 5 million encryptions per DEK](https://github.com/kroxylicious/kroxylicious/blob/11c1a14ebecf04604df34f5c535783145247ad82/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/RecordEncryption.java#L106) and it would be nice if that value were configurable - or, failing that, if it were set low enough not to encrypt more data than would weaken the GCM key given a suitably pessimistic average message size. 

After reading that last half sentence, I think it'd be better for it to be configurable. I don't personally have a use-case for having different DEK rotation requirements for different levels of sensitive data, so a single value is enough for me.

_Originally posted by @ddaviesbrackett-dwp in https://github.com/kroxylicious/kroxylicious/issues/1042#issuecomment-2141697948_
            

Pull request description:
Fix #1289: expose maxEncryptionsPerDek for configuration
### Type of change

_Select the type of your PR_

- Enhancement / new feature

### Description

#1289: expose `maxEncryptionsPerDek` for configuration.   This is done as another experimental parameter.

### Additional Context

_Why are you making this pull request?_

### Checklist

_Please go through this checklist and make sure all applicable tasks have been done_

- [x] Write tests
- [ ] Make sure all tests pass
- [ ] Review performance test results. Ensure that any degradations to performance numbers are understood and justified.
- [x] Make sure all Sonarcloud warnings are addressed or are justifiably ignored.
- [ ] Update documentation
- [x] Reference relevant issue(s) and close them after merging
- [x] For user facing changes, update CHANGELOG.md (remember to include changes affecting the API of the test artefacts too).


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 74b2e0a8..d3dce1de 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,7 @@ Format `<github issue/pr number>: <short description>`.
 
 ## SNAPSHOT
 
+* [#1289](https://github.com/kroxylicious/kroxylicious/issues/1289): Record Encryption - expose maxEncryptionsPerDek for configuration
 * [#1356](https://github.com/kroxylicious/kroxylicious/pull/1356): Changes for Kafka 3.8.0 #1356
 * [#1354](https://github.com/kroxylicious/kroxylicious/pull/1354): Make EDEK cache refresh and expiry durations configurable
 * [#1360](https://github.com/kroxylicious/kroxylicious/pull/1360): Bump kafka.version from 3.7.0 to 3.7.1
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/RecordEncryption.java b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/RecordEncryption.java
index 333e85f8..10bfad71 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/RecordEncryption.java
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/RecordEncryption.java
@@ -103,7 +103,9 @@ public class RecordEncryption<K, E> implements FilterFactory<RecordEncryptionCon
         checkCipherSuite();
         Kms<K, E> kms = buildKms(context, configuration);
 
-        DekManager<K, E> dekManager = new DekManager<>(ignored -> kms, null, 5_000_000);
+        var dekConfig = configuration.dekManager();
+        DekManager<K, E> dekManager = new DekManager<>(ignored -> kms, null, dekConfig.maxEncryptionsPerDek());
+
         KmsCacheConfig cacheConfig = configuration.kmsCache();
         EncryptionDekCache<K, E> encryptionDekCache = new EncryptionDekCache<>(dekManager, null, EncryptionDekCache.NO_MAX_CACHE_SIZE,
                 cacheConfig.encryptionDekCacheRefreshAfterWriteDuration(), cacheConfig.encryptionDekCacheExpireAfterWriteDuration());
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/config/DekManagerConfig.java b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/config/DekManagerConfig.java
new file mode 100644
index 00000000..768a9394
--- /dev/null
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/config/DekManagerConfig.java
@@ -0,0 +1,20 @@
+/*
+ * Copyright Kroxylicious Authors.
+ *
+ * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+package io.kroxylicious.filter.encryption.config;
+
+import static java.util.Objects.requireNonNullElse;
+
+/**
+ * Configuration for the {@link io.kroxylicious.filter.encryption.dek.DekManager}.
+ *
+ * @param maxEncryptionsPerDek maximum number of encryptions that will be performed by a DEK
+ */
+public record DekManagerConfig(Long maxEncryptionsPerDek) {
+    public DekManagerConfig {
+        maxEncryptionsPerDek = requireNonNullElse(maxEncryptionsPerDek, 5_000_000L);
+    }
+}
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/config/RecordEncryptionConfig.java b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/config/RecordEncryptionConfig.java
index 946720a1..639bfcd0 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/config/RecordEncryptionConfig.java
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/main/java/io/kroxylicious/filter/encryption/config/RecordEncryptionConfig.java
@@ -40,6 +40,12 @@ public record RecordEncryptionConfig(@JsonProperty(required = true) @PluginImplN
                 resolvedAliasRefreshAfterWriteSeconds, notFoundAliasExpireAfterWriteSeconds, encryptionDekRefreshAfterWriteSeconds, encryptionDekExpireAfterWriteSeconds);
     }
 
+    public DekManagerConfig dekManager() {
+        Long maxEncryptionsPerDek = getExperimentalLong("maxEncryptionsPerDek");
+        return new DekManagerConfig(maxEncryptionsPerDek);
+
+    }
+
     @Nullable
     private Integer getExperimentalInt(String property) {
         return Optional.ofNullable(experimental.get(property)).map(value -> {


Test patch:
diff --git a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/RecordEncryptionTest.java b/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/RecordEncryptionTest.java
index eeab708c..886c8e2f 100644
--- a/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/RecordEncryptionTest.java
+++ b/kroxylicious-filters/kroxylicious-record-encryption/src/test/java/io/kroxylicious/filter/encryption/RecordEncryptionTest.java
@@ -8,6 +8,7 @@ package io.kroxylicious.filter.encryption;
 
 import java.time.Duration;
 import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.Future;
 
 import javax.crypto.Cipher;
@@ -16,6 +17,7 @@ import org.assertj.core.api.AbstractThrowableAssert;
 import org.junit.jupiter.api.Test;
 
 import io.kroxylicious.filter.encryption.config.CipherSpec;
+import io.kroxylicious.filter.encryption.config.DekManagerConfig;
 import io.kroxylicious.filter.encryption.config.EncryptionConfigurationException;
 import io.kroxylicious.filter.encryption.config.KekSelectorService;
 import io.kroxylicious.filter.encryption.config.KmsCacheConfig;
@@ -89,9 +91,15 @@ class RecordEncryptionTest {
         assertThat(config.encryptionDekCacheExpireAfterWriteDuration()).isEqualTo(Duration.ofHours(2));
     }
 
+    @Test
+    void testDekManagerConfigDefaults() {
+        var config = new RecordEncryptionConfig("vault", 1L, "selector", 1L, null).dekManager();
+        assertThat(config.maxEncryptionsPerDek()).isEqualTo(5_000_000L);
+    }
+
     @Test
     void testKmsCacheConfigDefaultsWhenPropertiesNull() {
-        HashMap<String, Object> experimental = new HashMap<>();
+        Map<String, Object> experimental = new HashMap<>();
         experimental.put("decryptedDekCacheSize", null);
         experimental.put("decryptedDekExpireAfterAccessSeconds", null);
         experimental.put("resolvedAliasCacheSize", null);
@@ -112,6 +120,15 @@ class RecordEncryptionTest {
         assertThat(config.encryptionDekCacheExpireAfterWriteDuration()).isEqualTo(Duration.ofHours(2));
     }
 
+    @Test
+    void testDekManagerConfigDefaultsWhenPropertiesNull() {
+        Map<String, Object> experimental = new HashMap<>();
+        experimental.put("maxEncryptionsPerDek", null);
+
+        var config = new RecordEncryptionConfig("vault", 1L, "selector", 1L, null).dekManager();
+        assertThat(config.maxEncryptionsPerDek()).isEqualTo(5_000_000L);
+    }
+
     @Test
     void testKmsCacheConfigOverrides() {
         KmsCacheConfig kmsCacheConfig = new KmsCacheConfig(
@@ -124,7 +141,7 @@ class RecordEncryptionTest {
                 Duration.ofSeconds(7L),
                 Duration.ofSeconds(8L));
 
-        HashMap<String, Object> experimental = new HashMap<>();
+        Map<String, Object> experimental = new HashMap<>();
         experimental.put("decryptedDekCacheSize", 1);
         experimental.put("decryptedDekExpireAfterAccessSeconds", 2);
         experimental.put("resolvedAliasCacheSize", 3);
@@ -137,6 +154,18 @@ class RecordEncryptionTest {
         assertThat(config).isEqualTo(kmsCacheConfig);
     }
 
+    @Test
+    void testDekManagerConfigOverrides() {
+        var dekManagerCacheConfig = new DekManagerConfig(
+                1_000L);
+
+        Map<String, Object> experimental = new HashMap<>();
+        experimental.put("maxEncryptionsPerDek", 1_000L);
+
+        var config = new RecordEncryptionConfig("vault", 1L, "selector", 1L, experimental).dekManager();
+        assertThat(config).isEqualTo(dekManagerCacheConfig);
+    }
+
     @Test
     void testRetryPool() {
         Future<Thread> thread = RecordEncryption.RETRY_POOL.submit(Thread::currentThread);