Merge pull request #51 from strimzi/kms-vault

Kms vault

kms-vault/pom.xml

@@ -0,0 +1,65 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>io.strimzi</groupId>
+    <artifactId>topic-encryption</artifactId>
+    <version>0.0.1-SNAPSHOT</version>
+  </parent>
+  <artifactId>kms-vault</artifactId>
+  <name>KMS - Vault</name>
+  <description>Implementation of the KeyManagementSystem with Vault as the underlying KMS.</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>org.apache.logging.log4j</groupId>
+        <artifactId>log4j-api</artifactId>
+        <scope>test</scope>
+    </dependency>
+    <dependency>
+        <groupId>org.apache.logging.log4j</groupId>
+        <artifactId>log4j-core</artifactId>
+        <scope>test</scope>
+    </dependency>
+    <dependency>
+        <groupId>org.apache.logging.log4j</groupId>
+        <artifactId>log4j-slf4j-impl</artifactId>
+        <scope>test</scope>
+    </dependency>   
+    <dependency>
+        <groupId>org.slf4j</groupId>
+        <artifactId>slf4j-api</artifactId>
+    </dependency>
+    <dependency>
+        <groupId>com.github.spotbugs</groupId>
+        <artifactId>spotbugs-annotations</artifactId>
+    </dependency>
+    <dependency>
+        <groupId>com.fasterxml.jackson.core</groupId>
+        <artifactId>jackson-core</artifactId>
+    </dependency>
+    <dependency>
+        <groupId>com.fasterxml.jackson.core</groupId>
+        <artifactId>jackson-annotations</artifactId>
+    </dependency>
+    <dependency>
+        <groupId>com.fasterxml.jackson.core</groupId>
+        <artifactId>jackson-databind</artifactId>
+    </dependency>
+    <dependency>
+        <groupId>junit</groupId>
+        <artifactId>junit</artifactId>
+        <scope>test</scope>
+    </dependency>
+  </dependencies>
+
+</project>

kms-vault/src/main/java/io/strimzi/kafka/topicenc/kms/vault/VaultKms.java

@@ -0,0 +1,152 @@
+/*
+ * 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.kms.vault;
+
+import static java.util.Objects.isNull;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.net.http.HttpResponse.BodyHandlers;
+import java.nio.charset.StandardCharsets;
+
+import javax.crypto.SecretKey;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import io.strimzi.kafka.topicenc.common.EncUtils;
+import io.strimzi.kafka.topicenc.kms.KeyMgtSystem;
+import io.strimzi.kafka.topicenc.kms.KmsDefinition;
+import io.strimzi.kafka.topicenc.kms.KmsException;
+
+/**
+ * Key Management System interface implemented with Vault.
+ */
+public class VaultKms implements KeyMgtSystem {
+
+    public static final String VAULT_TOKEN_HEADER = "X-Vault-Token";
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(VaultKms.class);
+    private static final ObjectMapper OBJ_MAPPER = new ObjectMapper();
+    private static final String KEY_PATH = "/data/data/%s";
+
+    private HttpClient client;
+    private KmsDefinition config;
+
+    public VaultKms() {
+    }
+
+    /**
+     * Constructor
+     * 
+     * @param config a valid KmsDefiniton instance containing the info to interact
+     *               with Vault.
+     */
+    public VaultKms(KmsDefinition config) {
+        if (isNull(config)) {
+            throw new IllegalArgumentException("Null KmsConfig argument.");
+        }
+        if (isNull(config.getUri())) {
+            throw new IllegalArgumentException("Required argument 'baseUri' is missing.");
+        }
+        if (isNull(config.getCredential())) {
+            throw new IllegalArgumentException("Required argument 'token' is missing.");
+        }
+        this.config = config;
+        this.client = HttpClient.newBuilder().build();
+        LOGGER.debug("Vault KMS created");
+    }
+
+    /**
+     * Return the key corresponding to the requested key reference.
+     */
+    @Override
+    public SecretKey getKey(String keyReference) throws KmsException {
+
+        URI uri = createKeyUri(config.getUri(), keyReference);
+
+        // create request
+        HttpRequest request = HttpRequest.newBuilder()
+                .uri(uri)
+                .header(VAULT_TOKEN_HEADER, config.getCredential())
+                .GET()
+                .build();
+
+        // send request
+        HttpResponse<String> rsp;
+        try {
+            rsp = client.send(request, BodyHandlers.ofString());
+        } catch (IOException | InterruptedException e) {
+            throw new KmsException("Error requesting key.", e);
+        }
+
+        // check HTTP status code
+        if (rsp.statusCode() != 200) {
+            LOGGER.error("Error requesting key from Vault: HTTP {}", rsp.statusCode());
+            throw new KmsException("Error accessing Vault instance: HTTP " + rsp.statusCode());
+        }
+
+        // parse out key, decode and return:
+        String key = getKey(rsp.body(), keyReference);
+        LOGGER.debug("Vault KMS returned key");
+        return EncUtils.base64Decode(key);
+    }
+
+    public static URI createKeyUri(URI baseUri, String keyRef)
+            throws KmsException {
+        String uriStr;
+        try {
+            uriStr = String.format("%s/%s",
+                    baseUri.toString(),
+                    URLEncoder.encode(keyRef, StandardCharsets.UTF_8.toString()));
+        } catch (UnsupportedEncodingException e) {
+            throw new KmsException("Error encoding URL", e);
+        }
+        try {
+            return new URI(uriStr);
+        } catch (URISyntaxException e) {
+            throw new KmsException("Error creating Vault URL", e);
+        }
+    }
+
+    /**
+     * Utility to process the JSON response to a key request and return the key
+     * payload.
+     * 
+     * @param jsonStr      a JSON response document
+     * @param keyReference the reference of the key to return.
+     * @return a String in base64 encoding
+     * @throws KmsException if any error occurs or the requested key is not
+     *                      returned.
+     */
+    private String getKey(String jsonStr, String keyReference) throws KmsException {
+        JsonNode jsonObj;
+        try {
+            jsonObj = OBJ_MAPPER.readTree(jsonStr);
+        } catch (JsonProcessingException e) {
+            throw new KmsException("Error processing KMS response", e);
+        }
+        String path = String.format(KEY_PATH, keyReference);
+        JsonNode keyNode = jsonObj.at(path);
+        if (keyNode != null) {
+            String key = keyNode.asText();
+            if (key != null) {
+                return key;
+            }
+        }
+        // key not found:
+        throw new KmsException("Key " + keyReference + " not found");
+    }
+}

kms-vault/src/main/java/io/strimzi/kafka/topicenc/kms/vault/VaultKmsFactory.java

@@ -0,0 +1,27 @@
+/*
+ * 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.kms.vault;
+
+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;
+
+/**
+ * A KmsFactory, made available over the Java SPI, for creating instance of
+ * VaultKms.
+ */
+public class VaultKmsFactory implements KmsFactory {
+
+    @Override
+    public KeyMgtSystem createKms(KmsDefinition kmsDef) throws KmsException {
+        return new VaultKms(kmsDef);
+    }
+
+    @Override
+    public String getName() {
+        return VaultKmsFactory.class.getName();
+    }
+}

kms-vault/src/main/resources/META-INF/services/io.strimzi.kafka.topicenc.kms.KmsFactory

@@ -0,0 +1 @@
+io.strimzi.kafka.topicenc.kms.vault.VaultKmsFactory

kms-vault/src/test/java/io/strimzi/kafka/topicenc/kms/vault/VaultKmsTests.java

@@ -9,7 +9,6 @@
 
 import java.io.IOException;
 import java.net.URI;
-import java.net.URISyntaxException;
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
@@ -25,14 +24,19 @@
 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.VaultKms;
+import io.strimzi.kafka.topicenc.kms.KmsFactoryManager;
 
 /**
- * Testing of the Vault KMS.
+ * Testing of the Vault KMS. This is not complete because: 1) it assumes a vault
+ * instance is running on localhost 2) vault token currently must be entered as
+ * a string below.
+ * 
+ * This will be addressed in a separate PR involving the use of
+ * testcontainers.org
  */
 public class VaultKmsTests {
 
-    private static final String BASE_URI = "http://127.0.0.1:8200/v1/secret/data";
+    private static final String BASE_URI = "https://127.0.0.1:8200/v1/secret/data";
     private static final String STORE_KEY_DATA = "{\"data\":{\"%s\":\"%s\"}}";
 
     KeyMgtSystem vaultKms;
@@ -44,11 +48,10 @@ public void setUp() {
         try {
             config = new KmsDefinition()
                     .setUri(new URI(BASE_URI))
-                    .setType("vault")
-                    //.setKmsClassname(VaultKms.class.getName())
+                    .setType(VaultKmsFactory.class.getName())
                     .setCredential("<vault token>");
 
-            vaultKms = new VaultKms(config);
+            vaultKms = KmsFactoryManager.getInstance().createKms(config);
 
         } catch (Exception e) {
             fail("Error creating vault kms: " + e.toString());
@@ -70,7 +73,7 @@ public void basicVaultTests() {
         // store test key
         try {
             storeKey("test", testKey);
-        } catch (URISyntaxException | IOException | InterruptedException e) {
+        } catch (KmsException | IOException | InterruptedException e) {
             fail("Error storing test key " + e.toString());
             return;
         }
@@ -93,7 +96,7 @@ public void basicVaultTests() {
     }
 
     private void storeKey(String keyRef, String key)
-            throws URISyntaxException, IOException, InterruptedException {
+            throws KmsException, IOException, InterruptedException {
 
         URI uri = VaultKms.createKeyUri(config.getUri(), keyRef);
         String data = String.format(STORE_KEY_DATA, keyRef, key);
@@ -122,4 +125,3 @@ private String createTestKey() throws NoSuchAlgorithmException {
         return EncUtils.base64Encode(key);
     }
 }
-

pom.xml

@@ -86,6 +86,7 @@
         <module>common</module>
         <module>kms</module>
         <module>kms-test</module>
+        <module>kms-vault</module>
     </modules>
 
     <dependencyManagement>