Core: Add support for HashiCorp Vault KMS client

build.gradle

@@ -802,6 +802,30 @@ project(':iceberg-gcp') {
   }
 }
 
+project(':iceberg-hashicorp') {
+    apply plugin: 'com.gradleup.shadow'
+
+    build.dependsOn shadowJar
+
+    test {
+        useJUnitPlatform()
+    }
+
+    dependencies {
+        implementation project(path: ':iceberg-bundled-guava', configuration: 'shadow')
+        implementation project(':iceberg-core')
+        implementation libs.httpcomponents.httpclient5
+        implementation libs.jackson.databind
+
+        testImplementation project(path: ':iceberg-api', configuration: 'testArtifacts')
+        testImplementation libs.esotericsoftware.kryo
+        testImplementation libs.jackson.databind
+        testImplementation libs.testcontainers
+        testImplementation libs.testcontainers.junit.jupiter
+        testImplementation libs.testcontainers.vault
+    }
+}
+
 project(':iceberg-hive-metastore') {
   test {
     useJUnitPlatform()

core/src/main/java/org/apache/iceberg/CatalogProperties.java

@@ -174,6 +174,7 @@ private CatalogProperties() {}
   public static final String ENCRYPTION_KMS_TYPE_AWS = "aws";
   public static final String ENCRYPTION_KMS_TYPE_AZURE = "azure";
   public static final String ENCRYPTION_KMS_TYPE_GCP = "gcp";
+  public static final String ENCRYPTION_KMS_TYPE_HASHICORP = "hashicorp";
 
   public static final String ENCRYPTION_KMS_IMPL = "encryption.kms-impl";
   public static final String ENCRYPTION_KMS_IMPL_AWS =
@@ -182,4 +183,6 @@ private CatalogProperties() {}
       "org.apache.iceberg.azure.keymanagement.AzureKeyManagementClient";
   public static final String ENCRYPTION_KMS_IMPL_GCP =
       "org.apache.iceberg.gcp.GcpKeyManagementClient";
+  public static final String ENCRYPTION_KMS_IMPL_HASHICORP =
+      "org.apache.iceberg.hashicorp.VaultKeyManagementClient";
 }

core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java

@@ -63,6 +63,8 @@ public static KeyManagementClient createKmsClient(Map<String, String> catalogPro
                 CatalogProperties.ENCRYPTION_KMS_IMPL_AZURE;
             case CatalogProperties.ENCRYPTION_KMS_TYPE_GCP ->
                 CatalogProperties.ENCRYPTION_KMS_IMPL_GCP;
+            case CatalogProperties.ENCRYPTION_KMS_TYPE_HASHICORP ->
+                CatalogProperties.ENCRYPTION_KMS_IMPL_HASHICORP;
             default -> throw new IllegalStateException("Unsupported KMS type: " + kmsType);
           };
     }

docs/docs/encryption.md

@@ -28,7 +28,7 @@ Currently, encryption is supported in the Hive and REST catalogs for tables with
 
 Two parameters are required to activate encryption of a table:
 
-1. Catalog property that specifies the KMS ("key management service"). It can be either `encryption.kms-type` for pre-defined KMS clients (`aws`, `azure` or `gcp`) or `encryption.kms-impl` with the client class path for custom KMS clients.
+1. Catalog property that specifies the KMS ("key management service"). It can be either `encryption.kms-type` for pre-defined KMS clients (`aws`, `azure`, `gcp` or `hashicorp`) or `encryption.kms-impl` with the client class path for custom KMS clients.
 2. Table property `encryption.key-id`, that specifies the ID of a master key used to encrypt and decrypt the table. Master keys are stored and managed in the KMS.
 
 For more details on table encryption, see the "Appendix: Internals Overview" [subsection](#appendix-internals-overview).

gradle/libs.versions.toml

@@ -229,5 +229,6 @@ sqlite-jdbc = { module = "org.xerial:sqlite-jdbc", version.ref = "sqlite-jdbc" }
 testcontainers = { module = "org.testcontainers:testcontainers", version.ref = "testcontainers" }
 testcontainers-junit-jupiter = { module = "org.testcontainers:testcontainers-junit-jupiter", version.ref = "testcontainers" }
 testcontainers-minio = { module = "org.testcontainers:testcontainers-minio", version.ref = "testcontainers" }
+testcontainers-vault = { module = "org.testcontainers:testcontainers-vault", version.ref = "testcontainers" }
 tez08-dag = { module = "org.apache.tez:tez-dag", version.ref = "tez08" }
 tez08-mapreduce = { module = "org.apache.tez:tez-mapreduce", version.ref = "tez08" }

hashicorp/src/main/java/org/apache/iceberg/hashicorp/VaultClient.java

@@ -0,0 +1,221 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.iceberg.hashicorp;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.OptionalLong;
+import org.apache.hc.client5.http.classic.methods.HttpPost;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.protocol.HttpClientContext;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.ParseException;
+import org.apache.hc.core5.http.io.HttpClientResponseHandler;
+import org.apache.hc.core5.http.io.entity.EntityUtils;
+import org.apache.hc.core5.http.io.entity.StringEntity;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/**
+ * HTTP client for interacting with HashiCorp Vault REST API.
+ *
+ * @see <a href=https://developer.hashicorp.com/vault/api-docs>HashiCorp Vault HTTP API</a>
+ */
+class VaultClient implements Closeable {
+  private static final ObjectMapper MAPPER = new ObjectMapper();
+  private static final String VAULT_TOKEN_HEADER = "X-Vault-Token";
+
+  private final String address;
+  private final String transitMount;
+  private final String appRolePath;
+
+  private transient volatile CloseableHttpClient httpClient;
+
+  VaultClient(String address, String transitMount, String appRolePath) {
+    this.address = address;
+    this.transitMount = transitMount;
+    this.appRolePath = appRolePath;
+  }
+
+  AuthResult authenticate(String roleId, String secretId) {
+    ObjectNode requestBody = MAPPER.createObjectNode();
+    requestBody.put("role_id", roleId);
+    requestBody.put("secret_id", secretId);
+
+    JsonNode response = post("/v1/auth/" + appRolePath + "/login", null, requestBody);
+    JsonNode authNode = response.get("auth");
+    if (authNode == null) {
+      throw new RuntimeException("Failed to authenticate: no auth section in response");
+    }
+
+    String clientToken = authNode.get("client_token").asText();
+    OptionalLong leaseDuration =
+        authNode.has("lease_duration")
+            ? OptionalLong.of(authNode.get("lease_duration").asLong())
+            : OptionalLong.empty();
+    return new AuthResult(clientToken, leaseDuration);
+  }
+
+  String encrypt(String vaultToken, String wrappingKeyId, String plaintext) {
+    ObjectNode requestBody = MAPPER.createObjectNode();
+    requestBody.put("plaintext", plaintext);
+
+    JsonNode response =
+        post("/v1/" + transitMount + "/encrypt/" + wrappingKeyId, vaultToken, requestBody);
+
+    JsonNode dataNode = response.get("data");
+    if (dataNode == null || !dataNode.has("ciphertext")) {
+      throw new RuntimeException("Failed to wrap key: no ciphertext returned");
+    }
+
+    return dataNode.get("ciphertext").asText();
+  }
+
+  String decrypt(String vaultToken, String wrappingKeyId, String ciphertext) {
+    ObjectNode requestBody = MAPPER.createObjectNode();
+    requestBody.put("ciphertext", ciphertext);
+
+    JsonNode response =
+        post("/v1/" + transitMount + "/decrypt/" + wrappingKeyId, vaultToken, requestBody);
+
+    JsonNode dataNode = response.get("data");
+    if (dataNode == null || !dataNode.has("plaintext")) {
+      throw new RuntimeException("Failed to unwrap key: no plaintext returned");
+    }
+
+    return dataNode.get("plaintext").asText();
+  }
+
+  DataKey generateKey(String vaultToken, String wrappingKeyId) {
+    ObjectNode requestBody = MAPPER.createObjectNode();
+
+    JsonNode response =
+        post(
+            "/v1/" + transitMount + "/datakey/plaintext/" + wrappingKeyId, vaultToken, requestBody);
+
+    JsonNode dataNode = response.get("data");
+    if (dataNode == null || !dataNode.has("plaintext") || !dataNode.has("ciphertext")) {
+      throw new RuntimeException("Failed to generate key: missing plaintext or ciphertext");
+    }
+
+    String plaintext = dataNode.get("plaintext").asText();
+    String ciphertext = dataNode.get("ciphertext").asText();
+    return new DataKey(plaintext, ciphertext);
+  }
+
+  private JsonNode post(String path, String token, ObjectNode requestBody) {
+    HttpPost request = new HttpPost(address + path);
+    if (token != null) {
+      request.setHeader(VAULT_TOKEN_HEADER, token);
+    }
+
+    try {
+      request.setEntity(
+          new StringEntity(MAPPER.writeValueAsString(requestBody), ContentType.APPLICATION_JSON));
+    } catch (IOException e) {
+      throw new UncheckedIOException("Failed to serialize request body", e);
+    }
+
+    try {
+      return httpClient().execute(request, HttpClientContext.create(), new VaultResponseHandler());
+    } catch (IOException e) {
+      throw new UncheckedIOException("Failed to execute Vault request to " + path, e);
+    }
+  }
+
+  private static class VaultResponseHandler implements HttpClientResponseHandler<JsonNode> {
+    @Override
+    public JsonNode handleResponse(ClassicHttpResponse response) {
+      int statusCode = response.getCode();
+      Preconditions.checkState(statusCode == 200, "Status must be 200: %d", statusCode);
+
+      try {
+        String responseBody = EntityUtils.toString(response.getEntity());
+        return MAPPER.readTree(responseBody);
+      } catch (ParseException e) {
+        throw new RuntimeException("Failed to parse Vault error response", e);
+      } catch (IOException e) {
+        throw new UncheckedIOException("Failed to read response", e);
+      }
+    }
+  }
+
+  private CloseableHttpClient httpClient() {
+    if (httpClient == null) {
+      synchronized (this) {
+        if (httpClient == null) {
+          httpClient = HttpClients.createDefault();
+        }
+      }
+    }
+    return httpClient;
+  }
+
+  @Override
+  public void close() {
+    if (httpClient != null) {
+      try {
+        httpClient.close();
+      } catch (IOException e) {
+        throw new UncheckedIOException("Failed to close HTTP client", e);
+      }
+    }
+  }
+
+  static class AuthResult {
+    private final String clientToken;
+    private final OptionalLong leaseDuration;
+
+    AuthResult(String clientToken, OptionalLong leaseDuration) {
+      this.clientToken = clientToken;
+      this.leaseDuration = leaseDuration;
+    }
+
+    public String clientToken() {
+      return clientToken;
+    }
+
+    public OptionalLong leaseDuration() {
+      return leaseDuration;
+    }
+  }
+
+  static class DataKey {
+    private final String plaintext;
+    private final String ciphertext;
+
+    DataKey(String plaintext, String ciphertext) {
+      this.plaintext = plaintext;
+      this.ciphertext = ciphertext;
+    }
+
+    public String plaintext() {
+      return plaintext;
+    }
+
+    public String ciphertext() {
+      return ciphertext;
+    }
+  }
+}