Merge pull request #324 from aiven/ivanyu/multiple-rsa-keypairs

Provide keyring in configuration

core/src/integration-test/java/io/aiven/kafka/tieredstorage/RemoteStorageManagerTest.java

@@ -60,7 +60,6 @@
 import io.aiven.kafka.tieredstorage.security.DataKeyAndAAD;
 import io.aiven.kafka.tieredstorage.security.EncryptedDataKey;
 import io.aiven.kafka.tieredstorage.security.RsaEncryptionProvider;
-import io.aiven.kafka.tieredstorage.security.RsaKeyReader;
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -108,8 +107,6 @@ class RemoteStorageManagerTest extends RsaKeyAwareTest {
     static final String TARGET_MANIFEST_FILE =
         "test/topic-AAAAAAAAAAAAAAAAAAAAAQ/7/00000000000000000023-AAAAAAAAAAAAAAAAAAAAAA.rsm-manifest";
 
-    static final String KEY_ENCRYPTION_KEY_ID = "static-key-id";
-
     private static List<Arguments> provideEndToEnd() {
         final List<Arguments> result = new ArrayList<>();
         for (final int chunkSize : List.of(1024 * 1024 - 1, 1024 * 1024 * 1024 - 1, Integer.MAX_VALUE / 2)) {
@@ -128,9 +125,7 @@ private static List<Arguments> provideEndToEnd() {
     void init() throws IOException {
         rsm = new RemoteStorageManager();
 
-        rsaEncryptionProvider = new RsaEncryptionProvider(
-            KEY_ENCRYPTION_KEY_ID,
-            Map.of(KEY_ENCRYPTION_KEY_ID, RsaKeyReader.read(publicKeyPem, privateKeyPem)));
+        rsaEncryptionProvider = new RsaEncryptionProvider(KEY_ENCRYPTION_KEY_ID, keyRing);
         aesEncryptionProvider = new AesEncryptionProvider();
 
         sourceDir = Path.of(tmpDir.toString(), "source");
@@ -194,8 +189,10 @@ void endToEnd(final int chunkSize,
             "chunk.cache.size", Integer.toString(100 * 1024 * 1024)
         ));
         if (encryption) {
-            config.put("encryption.public.key.file", publicKeyPem.toString());
-            config.put("encryption.private.key.file", privateKeyPem.toString());
+            config.put("encryption.key.pair.id", KEY_ENCRYPTION_KEY_ID);
+            config.put("encryption.key.pairs", KEY_ENCRYPTION_KEY_ID);
+            config.put("encryption.key.pairs." + KEY_ENCRYPTION_KEY_ID + ".public.key.file", publicKeyPem.toString());
+            config.put("encryption.key.pairs." + KEY_ENCRYPTION_KEY_ID + ".private.key.file", privateKeyPem.toString());
         }
         rsm.configure(config);
 

core/src/main/java/io/aiven/kafka/tieredstorage/KeyPairPaths.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2023 Aiven Oy
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.aiven.kafka.tieredstorage;
+
+import java.nio.file.Path;
+import java.util.Objects;
+
+public class KeyPairPaths {
+    public final Path publicKey;
+    public final Path privateKey;
+
+    public KeyPairPaths(final Path publicKey, final Path privateKey) {
+        this.publicKey = Objects.requireNonNull(publicKey, "publicKey cannot be null");
+        this.privateKey = Objects.requireNonNull(privateKey, "privateKey cannot be null");
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final KeyPairPaths that = (KeyPairPaths) o;
+        return Objects.equals(publicKey, that.publicKey) && Objects.equals(privateKey, that.privateKey);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(publicKey, privateKey);
+    }
+
+    @Override
+    public String toString() {
+        return "KeyPairPaths("
+            + "publicKey=" + publicKey
+            + ", privateKey=" + privateKey
+            + ")";
+    }
+}

core/src/main/java/io/aiven/kafka/tieredstorage/RemoteStorageManager.java

@@ -24,6 +24,7 @@
 import java.io.InputStream;
 import java.nio.file.Files;
 import java.security.KeyPair;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
@@ -110,8 +111,6 @@ public class RemoteStorageManager implements org.apache.kafka.server.log.remote.
 
     private SegmentManifestProvider segmentManifestProvider;
 
-    private static final String KEY_ENCRYPTION_KEY_ID = "static-key-id";
-
     public RemoteStorageManager() {
         this(Time.SYSTEM);
     }
@@ -134,11 +133,11 @@ public void configure(final Map<String, ?> configs) {
         objectKey = new ObjectKey(config.keyPrefix());
         encryptionEnabled = config.encryptionEnabled();
         if (encryptionEnabled) {
-            final KeyPair keyPair =
-                RsaKeyReader.read(config.encryptionPublicKeyFile(), config.encryptionPrivateKeyFile());
-            rsaEncryptionProvider = new RsaEncryptionProvider(
-                KEY_ENCRYPTION_KEY_ID,
-                Map.of(KEY_ENCRYPTION_KEY_ID, keyPair));
+            final Map<String, KeyPair> keyRing = new HashMap<>();
+            config.encryptionKeyRing().forEach((keyId, keyPaths) ->
+                keyRing.put(keyId, RsaKeyReader.read(keyPaths.publicKey, keyPaths.privateKey)
+            ));
+            rsaEncryptionProvider = new RsaEncryptionProvider(config.encryptionKeyPairId(), keyRing);
             aesEncryptionProvider = new AesEncryptionProvider();
         }
         final ChunkManagerFactory chunkManagerFactory = new ChunkManagerFactory();

core/src/main/java/io/aiven/kafka/tieredstorage/RemoteStorageManagerConfig.java

@@ -18,6 +18,8 @@
 
 import java.nio.file.Path;
 import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 
@@ -68,10 +70,6 @@ public class RemoteStorageManagerConfig extends AbstractConfig {
 
     private static final String ENCRYPTION_CONFIG = "encryption.enabled";
     private static final String ENCRYPTION_DOC = "Whether to enable encryption";
-    private static final String ENCRYPTION_PUBLIC_KEY_FILE_CONFIG = "encryption.public.key.file";
-    private static final String ENCRYPTION_PUBLIC_KEY_FILE_DOC = "The path to the RSA public key file";
-    private static final String ENCRYPTION_PRIVATE_KEY_FILE_CONFIG = "encryption.private.key.file";
-    private static final String ENCRYPTION_PRIVATE_KEY_FILE_DOC = "The path to the RSA private key file";
     // TODO add possibility to pass keys as strings
 
 
@@ -158,20 +156,6 @@ public class RemoteStorageManagerConfig extends AbstractConfig {
             ConfigDef.Importance.HIGH,
             ENCRYPTION_DOC
         );
-        CONFIG.define(
-            ENCRYPTION_PUBLIC_KEY_FILE_CONFIG,
-            ConfigDef.Type.STRING,
-            null,
-            ConfigDef.Importance.HIGH,
-            ENCRYPTION_PUBLIC_KEY_FILE_DOC
-        );
-        CONFIG.define(
-            ENCRYPTION_PRIVATE_KEY_FILE_CONFIG,
-            ConfigDef.Type.STRING,
-            null,
-            ConfigDef.Importance.HIGH,
-            ENCRYPTION_PRIVATE_KEY_FILE_DOC
-        );
 
         CONFIG
             .define(METRICS_SAMPLE_WINDOW_MS_CONFIG,
@@ -196,14 +180,107 @@ public class RemoteStorageManagerConfig extends AbstractConfig {
                 METRICS_RECORDING_LEVEL_DOC);
     }
 
+    /**
+     * Internal config for encryption.
+     *
+     * <p>It's needed for more convenient dynamic config definition.
+     */
+    private static class EncryptionConfig extends AbstractConfig {
+        private static final String ENCRYPTION_KEY_PAIR_ID_CONFIG = "encryption.key.pair.id";
+        private static final String ENCRYPTION_KEY_PAIR_ID_DOC =
+            "The ID of the key pair to be used for encryption";
+
+        private static final String ENCRYPTION_KEY_PAIRS_CONFIG = "encryption.key.pairs";
+        private static final String ENCRYPTION_KEY_PAIRS_DOC = "The list of encryption key pair IDs";
+
+        private static final String ENCRYPTION_PUBLIC_KEY_FILE_DOC = "The path to the RSA public key file";
+        private static final String ENCRYPTION_PRIVATE_KEY_FILE_DOC = "The path to the RSA private key file";
+
+        private EncryptionConfig(final ConfigDef configDef, final Map<String, ?> props) {
+            super(configDef, props);
+        }
+
+        Path encryptionPublicKeyFile(final String keyPairId) {
+            return Path.of(getString(publicKeyFileConfig(keyPairId)));
+        }
+
+        Path encryptionPrivateKeyFile(final String keyPairId) {
+            return Path.of(getString(privateKeyFileConfig(keyPairId)));
+        }
+
+        public static EncryptionConfig create(final Map<String, ?> props) {
+            final ConfigDef configDef = new ConfigDef();
+            // First, define the active key ID and key ID list fields, they are required always.
+            configDef.define(
+                ENCRYPTION_KEY_PAIR_ID_CONFIG,
+                ConfigDef.Type.STRING,
+                ConfigDef.NO_DEFAULT_VALUE,
+                ConfigDef.Importance.HIGH,
+                ENCRYPTION_KEY_PAIR_ID_DOC
+            );
+            configDef.define(
+                ENCRYPTION_KEY_PAIRS_CONFIG,
+                ConfigDef.Type.LIST,
+                ConfigDef.NO_DEFAULT_VALUE,
+                ConfigDef.Importance.HIGH,
+                ENCRYPTION_KEY_PAIRS_DOC
+            );
+            final EncryptionConfig interimEncryptionConfig = new EncryptionConfig(configDef, props);
+
+            // Check that the active ID is present in the list.
+            if (!interimEncryptionConfig.keyPairIds().contains(interimEncryptionConfig.activeKeyPairId())) {
+                throw new ConfigException(
+                    "Encryption key '" + interimEncryptionConfig.activeKeyPairId() + "' must be provided");
+            }
+
+            // Then, define key fields dynamically based on the key pair IDs provided above.
+            // See e.g. the ConnectorConfig.enrich in the Kafka code.
+            for (final String keyPairId : interimEncryptionConfig.keyPairIds()) {
+                configDef.define(
+                    publicKeyFileConfig(keyPairId),
+                    ConfigDef.Type.STRING,
+                    ConfigDef.NO_DEFAULT_VALUE,
+                    ConfigDef.Importance.HIGH,
+                    ENCRYPTION_PUBLIC_KEY_FILE_DOC
+                );
+                configDef.define(
+                    privateKeyFileConfig(keyPairId),
+                    ConfigDef.Type.STRING,
+                    ConfigDef.NO_DEFAULT_VALUE,
+                    ConfigDef.Importance.HIGH,
+                    ENCRYPTION_PRIVATE_KEY_FILE_DOC
+                );
+            }
+
+            return new EncryptionConfig(configDef, props);
+        }
+
+        String activeKeyPairId() {
+            return getString(ENCRYPTION_KEY_PAIR_ID_CONFIG);
+        }
+
+        List<String> keyPairIds() {
+            return getList(ENCRYPTION_KEY_PAIRS_CONFIG);
+        }
+
+        private static String publicKeyFileConfig(final String keyPairId) {
+            return "encryption.key.pairs." + keyPairId + ".public.key.file";
+        }
+
+        private static String privateKeyFileConfig(final String keyPairId) {
+            return "encryption.key.pairs." + keyPairId + ".private.key.file";
+        }
+    }
+
+    private final EncryptionConfig encryptionConfig;
 
     RemoteStorageManagerConfig(final Map<String, ?> props) {
         super(CONFIG, props);
+        encryptionConfig = encryptionEnabled() ? EncryptionConfig.create(props) : null;
         validate();
     }
 
     private void validate() {
-        validateEncryption();
         validateCompression();
     }
 
@@ -214,17 +291,6 @@ private void validateCompression() {
         }
     }
 
-    private void validateEncryption() {
-        if (getBoolean(ENCRYPTION_CONFIG) && getString(ENCRYPTION_PUBLIC_KEY_FILE_CONFIG) == null) {
-            throw new ConfigException(
-                ENCRYPTION_PUBLIC_KEY_FILE_CONFIG + " must be provided if encryption is enabled");
-        }
-        if (getBoolean(ENCRYPTION_CONFIG) && getString(ENCRYPTION_PRIVATE_KEY_FILE_CONFIG) == null) {
-            throw new ConfigException(
-                ENCRYPTION_PRIVATE_KEY_FILE_CONFIG + " must be provided if encryption is enabled");
-        }
-    }
-
     StorageBackend storage() {
         final Class<?> storageClass = getClass(STORAGE_BACKEND_CLASS_CONFIG);
         final StorageBackend storage = Utils.newInstance(storageClass, StorageBackend.class);
@@ -268,19 +334,25 @@ boolean encryptionEnabled() {
         return getBoolean(ENCRYPTION_CONFIG);
     }
 
-    Path encryptionPublicKeyFile() {
-        final String value = getString(ENCRYPTION_PUBLIC_KEY_FILE_CONFIG);
-        if (value == null) {
+    String encryptionKeyPairId() {
+        if (!encryptionEnabled()) {
             return null;
         }
-        return Path.of(value);
+        return encryptionConfig.activeKeyPairId();
     }
 
-    Path encryptionPrivateKeyFile() {
-        final String value = getString(ENCRYPTION_PRIVATE_KEY_FILE_CONFIG);
-        if (value == null) {
+    Map<String, KeyPairPaths> encryptionKeyRing() {
+        if (!encryptionEnabled()) {
             return null;
         }
-        return Path.of(value);
+
+        final Map<String, KeyPairPaths> result = new HashMap<>();
+        for (final String keyPairId : encryptionConfig.keyPairIds()) {
+            final KeyPairPaths keyPair = new KeyPairPaths(
+                encryptionConfig.encryptionPublicKeyFile(keyPairId),
+                encryptionConfig.encryptionPrivateKeyFile(keyPairId));
+            result.put(keyPairId, keyPair);
+        }
+        return result;
     }
 }