Merge pull request #47 from strimzi/kms-module

Kms module

encmod/pom.xml

@@ -7,15 +7,26 @@
   </parent>
   <modelVersion>4.0.0</modelVersion>
   <artifactId>encmod</artifactId>
-  <name>encryption module</name>
-  <description>desc</description>
+  <name>Encryption module</name>
+  <description>The encryption engine.</description>
 
   <dependencies>
 	<dependency>
 	  <groupId>io.strimzi</groupId>
 	  <artifactId>common</artifactId>
 	  <version>${project.version}</version>
 	</dependency>
+	<dependency>
+	  <groupId>io.strimzi</groupId>
+	  <artifactId>kms</artifactId>
+	  <version>${project.version}</version>
+	</dependency>
+	<dependency>
+	  <groupId>io.strimzi</groupId>
+	  <artifactId>kms-test</artifactId>
+	  <version>${project.version}</version>
+	  <scope>test</scope>
+	</dependency>
   	<dependency>
   		<groupId>org.apache.kafka</groupId>
   		<artifactId>kafka-clients</artifactId>
@@ -43,6 +54,10 @@
         <groupId>junit</groupId>
         <artifactId>junit</artifactId>
         <scope>test</scope>
+    </dependency>
+        <dependency>
+        <groupId>com.github.spotbugs</groupId>
+        <artifactId>spotbugs-annotations</artifactId>
     </dependency>
     <dependency>
         <groupId>com.fasterxml.jackson.core</groupId>

encmod/src/main/java/io/strimzi/kafka/topicenc/policy/JsonPolicyLoader.java

@@ -1,6 +1,6 @@
 /*
- * Copyright Strimzi authors. License: Apache License 2.0 (see the file LICENSE or
- * http://apache.org/licenses/LICENSE-2.0.html).
+ * Copyright Strimzi authors. 
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
  */
 package io.strimzi.kafka.topicenc.policy;
 
@@ -25,7 +25,8 @@
 
 import io.strimzi.kafka.topicenc.kms.KeyMgtSystem;
 import io.strimzi.kafka.topicenc.kms.KmsDefinition;
-import io.strimzi.kafka.topicenc.kms.KmsFactory;
+import io.strimzi.kafka.topicenc.kms.KmsException;
+import io.strimzi.kafka.topicenc.kms.KmsFactoryManager;
 
 /**
  * Utility methods for loading and processing configuration information residing
@@ -77,10 +78,11 @@ public static List<TopicPolicy> loadTopicPolicies(File file,
                 });
 
         // validating each topic policy, assign a KMS instance,
-        // and ensuring unique topic names by creating a map.
+        // ensuring unique topic names by filling a map.
         Map<String, KeyMgtSystem> kmsPool = new HashMap<>();
         policies.stream()
                 .map(policy -> policy.validate())
+                .map(policy -> validateKms(policy, kmsDefs))
                 .map(policy -> assignKms(policy, kmsDefs, kmsPool))
                 .collect(Collectors.toMap(JsonPolicyLoader::key, Function.identity()));
 
@@ -100,34 +102,64 @@ public static Map<String, KmsDefinition> loadKmsDefs(File file) throws IOExcepti
         List<KmsDefinition> kmsDefs = OBJ_MAPPER.readValue(file,
                 new TypeReference<List<KmsDefinition>>() {
                 });
-
         return kmsDefs.stream()
                 .map(kmsDef -> kmsDef.validate())
                 .collect(Collectors.toMap(JsonPolicyLoader::key, Function.identity()));
     }
 
     /**
-     * Assign a KeyMgtSystem instance to a topic policy. This is done by matching
-     * the KMS name in the topic policy to a KeyMgtSystem instance.
+     * Assert that the kms name in the topic policy corresponds to a valid, known
+     * KMS definition. If not valid, an IllegalArgumentException is thrown.
      * 
-     * @param policy  the topic policy to which a KMS instance is assigned
-     * @param kmsDefs A map of KmsDefinitions, indexed by name
-     * @param kmsPool A map of already instantiated KMS instances
-     * @return
+     * @param policy       the policy being verified
+     * @param validKmsDefs a map of known, valid KMS definitions.
+     * @return the policy instance passed as an argument
      */
-    private static TopicPolicy assignKms(TopicPolicy policy, Map<String, KmsDefinition> kmsDefs,
-            Map<String, KeyMgtSystem> kmsPool) {
+    private static TopicPolicy validateKms(TopicPolicy policy,
+            Map<String, KmsDefinition> validKmsDefs) {
 
         String kmsName = createKey(policy.getKmsName(), Locale.getDefault());
-        KmsDefinition kmsDef = kmsDefs.computeIfAbsent(
+
+        // verify the policy refers to a known KMS definition
+        validKmsDefs.computeIfAbsent(
                 kmsName,
                 k -> {
                     throw new IllegalArgumentException(
-                            "Topic " + policy.getTopic() + " refers to unknown KMS");
+                            "Policy for topic, " + policy.getTopic() + ", refers to unknown KMS.");
                 });
-        KeyMgtSystem kms = kmsPool.computeIfAbsent(
-                kmsName, k -> KmsFactory.createKms(kmsDef));
+        // if this far, the topic policy contains a known KMS def, as it should.
+        return policy;
+    }
+
+    /**
+     * Assign a KeyMgtSystem instance to a topic policy by matching the policy's KMS
+     * name to a KeyMgtSystem instance.
+     * 
+     * @param policy       the topic policy to which a KMS instance is assigned
+     * @param validKmsDefs A map of valid KmsDefinitions, indexed by name
+     * @param kmsPool      A pool of instantiated KMS instances
+     * @return
+     */
+    private static TopicPolicy assignKms(TopicPolicy policy,
+            Map<String, KmsDefinition> validKmsDefs,
+            Map<String, KeyMgtSystem> kmsPool) {
+
+        String kmsName = createKey(policy.getKmsName(), Locale.getDefault());
+
+        // Check the pool (aka cache) of KMS instances first
+        KeyMgtSystem kms = kmsPool.get(kmsName);
+        if (kms == null) {
+            // kms not instantiated yet. Create and add to pool.
+            KmsDefinition kmsDef = validKmsDefs.get(kmsName);
+            try {
+                kms = KmsFactoryManager.getInstance().createKms(kmsDef);
+            } catch (KmsException e) {
+                throw new RuntimeException(e);
+            }
+            kmsPool.put(kmsName, kms);
+        }
         policy.setKms(kms);
+        // return policy for method chaining
         return policy;
     }
 
@@ -158,4 +190,3 @@ private static void logUnassignedKmsDefs(Set<String> kmsPool, Set<String> kmsDef
         }
     }
 }
-

encmod/src/main/java/io/strimzi/kafka/topicenc/policy/TestPolicyRepository.java

@@ -1,13 +1,14 @@
 /*
- * Copyright Strimzi authors. License: Apache License 2.0 (see the file LICENSE or
- * http://apache.org/licenses/LICENSE-2.0.html).
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
  */
 package io.strimzi.kafka.topicenc.policy;
 
 import io.strimzi.kafka.topicenc.kms.KeyMgtSystem;
 import io.strimzi.kafka.topicenc.kms.KmsDefinition;
 import io.strimzi.kafka.topicenc.kms.KmsException;
 import io.strimzi.kafka.topicenc.kms.KmsFactory;
+import io.strimzi.kafka.topicenc.kms.KmsFactoryManager;
 
 /**
  * An trivial implementation of a policy repository used only for testing. All
@@ -32,7 +33,7 @@ public TestPolicyRepository() throws KmsException {
                 .setName("test")
                 .setType("test");
 
-        KeyMgtSystem kms = KmsFactory.createKms(kmsDef);
+        KeyMgtSystem kms = KmsFactoryManager.getInstance().createKms(kmsDef);
 
         // create the single test policy for all topics:
         policy = new TopicPolicy()
@@ -47,4 +48,3 @@ public TopicPolicy getTopicPolicy(String topicName) {
         return policy;
     }
 }
-

encmod/src/test/java/io/strimzi/kafka/topicenc/enc/AesGcmEncTests.java

@@ -17,20 +17,28 @@
 import org.junit.Test;
 
 import io.strimzi.kafka.topicenc.common.EncUtils;
-import io.strimzi.kafka.topicenc.kms.TestKms;
+import io.strimzi.kafka.topicenc.kms.KeyMgtSystem;
+import io.strimzi.kafka.topicenc.kms.KmsDefinition;
+import io.strimzi.kafka.topicenc.kms.KmsException;
+import io.strimzi.kafka.topicenc.kms.KmsFactoryManager;
 import io.strimzi.kafka.topicenc.ser.AesGcmV1SerDer;
 import io.strimzi.kafka.topicenc.ser.EncSerDerException;
 
 public class AesGcmEncTests {
 
     private static final String TEST_MSG = "abcdefghijklmnopqrstuvwxyz0123456789!@#$%^&*()_-=+[]{}";
 
-    TestKms kms;
+    KeyMgtSystem kms;
     AesGcmEncrypter enc;
 
     @Before
-    public void testsInit() throws NoSuchAlgorithmException {
-        kms = new TestKms();
+    public void testsInit() throws NoSuchAlgorithmException, KmsException {
+
+        KmsDefinition kmsDef = new KmsDefinition()
+                .setName("test")
+                .setType("test");
+
+        kms = KmsFactoryManager.getInstance().createKms(kmsDef);
         SecretKey key = kms.getKey("test");
         enc = new AesGcmEncrypter(key);
     }