This is an automated email from the ASF dual-hosted git repository.
singhpk234 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 0f3131f02f Aliyun: Add AliyunKeyManagementClient for client-side
encryption (#18210)
0f3131f02f is described below
commit 0f3131f02fc2560bd122f3b323d13ea935892eb3
Author: mengna-lin <[email protected]>
AuthorDate: Fri Oct 9 17:21:13 2026 -0700
Aliyun: Add AliyunKeyManagementClient for client-side encryption (#18210)
* Aliyun: Add AliyunKeyManagementClient for client-side encryption
Add an Alibaba Cloud KMS implementation of the KeyManagementClient SPI so
tables using Iceberg envelope encryption can wrap/unwrap data keys with
Aliyun KMS, at parity with AwsKeyManagementClient. The aliyun module
previously supported only OSS.
- AliyunKeyManagementClient: wrapKey, unwrapKey, and server-side
generateKey; supportsKeyGeneration() returns true.
- AliyunClientFactory#newKmsClient: build the KMS client from
client.region, resolving credentials from access key/secret/token and
falling back to the default chain (env vars, RRSA/OIDC, ECS RAM role).
- AliyunProperties: add client.region and kms.data-key-spec (default
AES_256).
---
.../iceberg/aliyun/AliyunClientFactories.java | 65 +++++-
.../apache/iceberg/aliyun/AliyunClientFactory.java | 11 +
.../iceberg/aliyun/AliyunKeyManagementClient.java | 149 +++++++++++++
.../apache/iceberg/aliyun/AliyunProperties.java | 64 ++++++
.../iceberg/aliyun/TestAliyunClientFactories.java | 28 +++
.../aliyun/TestAliyunKeyManagementClient.java | 231 +++++++++++++++++++++
build.gradle | 3 +
.../java/org/apache/iceberg/CatalogProperties.java | 3 +
.../apache/iceberg/encryption/EncryptionUtil.java | 2 +
docs/docs/encryption.md | 2 +-
gradle/libs.versions.toml | 4 +-
11 files changed, 558 insertions(+), 4 deletions(-)
diff --git
a/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunClientFactories.java
b/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunClientFactories.java
index 8bb88db83e..31a34aabf6 100644
--- a/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunClientFactories.java
+++ b/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunClientFactories.java
@@ -20,11 +20,13 @@ package org.apache.iceberg.aliyun;
import com.aliyun.credentials.models.CredentialModel;
import com.aliyun.credentials.provider.OIDCRoleArnCredentialProvider;
+import com.aliyun.kms20160120.Client;
import com.aliyun.oss.OSS;
import com.aliyun.oss.OSSClientBuilder;
import com.aliyun.oss.common.auth.BasicCredentials;
import com.aliyun.oss.common.auth.Credentials;
import com.aliyun.oss.common.auth.CredentialsProvider;
+import com.aliyun.teaopenapi.models.Config;
import java.util.Map;
import org.apache.iceberg.common.DynConstructors;
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -111,6 +113,11 @@ public class AliyunClientFactories {
&& !Strings.isNullOrEmpty(oidcTokenFile);
}
+ /** Build an RRSA (OIDC role) credential provider with built-in caching
and auto-refresh. */
+ private OIDCRoleArnCredentialProvider rrsaCredentialProvider() {
+ return OIDCRoleArnCredentialProvider.builder().build();
+ }
+
@Override
public OSS newOSSClient() {
Preconditions.checkNotNull(
@@ -127,8 +134,7 @@ public class AliyunClientFactories {
endpoint);
// Use OIDCRoleArnCredentialProvider directly with built-in caching
and auto-refresh
- final OIDCRoleArnCredentialProvider oidcProvider =
- OIDCRoleArnCredentialProvider.builder().build();
+ final OIDCRoleArnCredentialProvider oidcProvider =
rrsaCredentialProvider();
CredentialsProvider ossCredProvider =
new CredentialsProvider() {
@@ -185,6 +191,61 @@ public class AliyunClientFactories {
this.aliyunProperties = new AliyunProperties(properties);
}
+ /**
+ * Creates the Aliyun KMS client. Credentials are resolved in the same
order as {@link
+ * #newOSSClient()}:
+ *
+ * <ol>
+ * <li>RRSA/OIDC when its environment variables are present; this takes
precedence over the
+ * static access-key properties.
+ * <li>otherwise the static access key / secret (+ optional security
token).
+ * </ol>
+ */
+ @Override
+ public Client newKmsClient() {
+ Preconditions.checkNotNull(
+ aliyunProperties,
+ "Cannot create aliyun kms client before initializing the
AliyunClientFactory.");
+ Preconditions.checkArgument(
+ !Strings.isNullOrEmpty(aliyunProperties.region()),
+ "Cannot create aliyun kms client, %s is not set.",
+ AliyunProperties.CLIENT_REGION);
+
+ // the KMS client resolves kms.<region>.aliyuncs.com from the region id
+ Config config = new Config().setRegionId(aliyunProperties.region());
+
+ // Optional endpoint override, e.g. a KMS Instance (DKMS) endpoint.
+ if (!Strings.isNullOrEmpty(aliyunProperties.kmsEndpoint())) {
+ config.setEndpoint(aliyunProperties.kmsEndpoint());
+ }
+
+ // Check if RRSA environment is available (same credential resolution
order as newOSSClient)
+ if (isRrsaEnvironmentAvailable()) {
+ try {
+ LOG.info(
+ "Detected RRSA environment variables, creating KMS client with
RRSA credentials");
+ // Use OIDCRoleArnCredentialProvider directly with built-in caching
and auto-refresh
+ OIDCRoleArnCredentialProvider oidcProvider =
rrsaCredentialProvider();
+ config.setCredential(new
com.aliyun.credentials.Client(oidcProvider));
+ return new Client(config);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to create RRSA KMS client", e);
+ }
+ }
+
+ try {
+ config
+ .setAccessKeyId(aliyunProperties.accessKeyId())
+ .setAccessKeySecret(aliyunProperties.accessKeySecret());
+ if (!Strings.isNullOrEmpty(aliyunProperties.securityToken())) {
+ config.setSecurityToken(aliyunProperties.securityToken());
+ }
+ return new Client(config);
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to create aliyun kms client", e);
+ }
+ }
+
@Override
public AliyunProperties aliyunProperties() {
return aliyunProperties;
diff --git
a/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunClientFactory.java
b/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunClientFactory.java
index 67490db578..b8c8e49871 100644
--- a/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunClientFactory.java
+++ b/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunClientFactory.java
@@ -18,6 +18,7 @@
*/
package org.apache.iceberg.aliyun;
+import com.aliyun.kms20160120.Client;
import com.aliyun.oss.OSS;
import java.io.Serializable;
import java.util.Map;
@@ -31,6 +32,16 @@ public interface AliyunClientFactory extends Serializable {
*/
OSS newOSSClient();
+ /**
+ * Create an aliyun KMS client.
+ *
+ * @return kms client.
+ */
+ default Client newKmsClient() {
+ throw new UnsupportedOperationException(
+ this.getClass().getName() + " does not support creating a KMS client");
+ }
+
/**
* Initialize Aliyun client factory from catalog properties.
*
diff --git
a/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunKeyManagementClient.java
b/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunKeyManagementClient.java
new file mode 100644
index 0000000000..3bebc63af6
--- /dev/null
+++
b/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunKeyManagementClient.java
@@ -0,0 +1,149 @@
+/*
+ * 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.aliyun;
+
+import com.aliyun.kms20160120.Client;
+import com.aliyun.kms20160120.models.DecryptRequest;
+import com.aliyun.kms20160120.models.DecryptResponse;
+import com.aliyun.kms20160120.models.EncryptRequest;
+import com.aliyun.kms20160120.models.EncryptResponse;
+import com.aliyun.teautil.models.RuntimeOptions;
+import java.nio.ByteBuffer;
+import java.util.Base64;
+import java.util.Map;
+import org.apache.iceberg.encryption.KeyManagementClient;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.util.ByteBuffers;
+import org.apache.iceberg.util.SerializableMap;
+
+/**
+ * Key management client implementation that uses Alibaba Cloud KMS. Wraps
(encrypts) and unwraps
+ * (decrypts) data keys with a KMS-managed master key referenced by its key id.
+ */
+public class AliyunKeyManagementClient implements KeyManagementClient {
+
+ private static final int BACKOFF_PERIOD_MS = 100;
+ private static final String ALIAS_PREFIX = "alias/";
+ private static final String ARN_KEY_SEPARATOR = "key/";
+
+ private Map<String, String> allProperties;
+ private AliyunProperties aliyunProperties;
+
+ private transient volatile ClientState state;
+
+ @Override
+ public void initialize(Map<String, String> properties) {
+ this.allProperties = SerializableMap.copyOf(properties);
+ this.aliyunProperties = new AliyunProperties(properties);
+ }
+
+ @Override
+ public ByteBuffer wrapKey(ByteBuffer key, String wrappingKeyId) {
+ EncryptRequest request =
+ new EncryptRequest()
+ .setKeyId(wrappingKeyId)
+
.setPlaintext(Base64.getEncoder().encodeToString(ByteBuffers.toByteArray(key)));
+ try {
+ EncryptResponse response = client().encryptWithOptions(request,
runtimeOptions());
+ return base64ToBuffer(response.getBody().getCiphertextBlob());
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to wrap key with Aliyun KMS", e);
+ }
+ }
+
+ @Override
+ public ByteBuffer unwrapKey(ByteBuffer wrappedKey, String wrappingKeyId) {
+ DecryptResponse response;
+ try {
+ DecryptRequest request =
+ new DecryptRequest()
+ .setCiphertextBlob(
+
Base64.getEncoder().encodeToString(ByteBuffers.toByteArray(wrappedKey)));
+ response = client().decryptWithOptions(request, runtimeOptions());
+ } catch (Exception e) {
+ throw new RuntimeException("Failed to unwrap key with Aliyun KMS", e);
+ }
+
+ // Verify the key that wrapped the data key. Skip aliases: an alias is a
mutable pointer whose
+ // target may differ from the key that wrapped older data. Only bare ids
and ARNs are checked.
+ String actualKeyId = response.getBody().getKeyId();
+ if (!isAlias(wrappingKeyId)) {
+ String expectedKeyId = keyIdFromRef(wrappingKeyId);
+ Preconditions.checkState(
+ expectedKeyId.equals(actualKeyId),
+ "Data key was wrapped by KMS key %s, but unwrap expected key %s",
+ actualKeyId,
+ expectedKeyId);
+ }
+ return base64ToBuffer(response.getBody().getPlaintext());
+ }
+
+ private static ByteBuffer base64ToBuffer(String base64) {
+ return ByteBuffer.wrap(Base64.getDecoder().decode(base64));
+ }
+
+ // bare alias "alias/name" or alias ARN "acs:kms:...:alias/name"
+ private static boolean isAlias(String keyRef) {
+ return keyRef.contains(ALIAS_PREFIX);
+ }
+
+ // key ARN -> id after "key/"; bare id -> unchanged
+ private static String keyIdFromRef(String keyRef) {
+ int idx = keyRef.lastIndexOf(ARN_KEY_SEPARATOR);
+ return idx >= 0 ? keyRef.substring(idx + ARN_KEY_SEPARATOR.length()) :
keyRef;
+ }
+
+ private Client client() {
+ return state().client;
+ }
+
+ private RuntimeOptions runtimeOptions() {
+ return state().runtimeOptions;
+ }
+
+ private ClientState state() {
+ if (state == null) {
+ synchronized (this) {
+ if (state == null) {
+ Client kmsClient =
AliyunClientFactories.from(allProperties).newKmsClient();
+ RuntimeOptions options =
+ new RuntimeOptions()
+ .setAutoretry(true)
+ .setMaxAttempts(aliyunProperties.kmsClientMaxAttempts())
+ .setBackoffPolicy("fixed")
+ .setBackoffPeriod(BACKOFF_PERIOD_MS)
+
.setConnectTimeout(aliyunProperties.kmsClientConnectTimeoutMs())
+ .setReadTimeout(aliyunProperties.kmsClientReadTimeoutMs());
+ state = new ClientState(kmsClient, options);
+ }
+ }
+ }
+ return state;
+ }
+
+ private static class ClientState {
+ private final Client client;
+ private final RuntimeOptions runtimeOptions;
+
+ ClientState(Client client, RuntimeOptions runtimeOptions) {
+ this.client = client;
+ this.runtimeOptions = runtimeOptions;
+ }
+ }
+}
diff --git
a/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunProperties.java
b/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunProperties.java
index 4de784f40c..0faf387fcd 100644
--- a/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunProperties.java
+++ b/aliyun/src/main/java/org/apache/iceberg/aliyun/AliyunProperties.java
@@ -76,11 +76,43 @@ public class AliyunProperties implements Serializable {
*/
public static final String OSS_STAGING_DIRECTORY = "oss.staging-dir";
+ /**
+ * The region id used to derive a service endpoint, for example {@code
cn-hangzhou}. Used by the
+ * Aliyun KMS client to resolve {@code kms.<region>.aliyuncs.com}.
+ */
+ public static final String CLIENT_REGION = "client.region";
+
+ /**
+ * Overrides the KMS endpoint. Defaults to the region-derived {@code
kms.<region>.aliyuncs.com}.
+ * Set to a KMS Instance (DKMS) endpoint for keys that live in a KMS
Instance.
+ */
+ public static final String CLIENT_KMS_ENDPOINT = "client.kms-endpoint";
+
+ /** Maximum number of attempts (including the first) for each KMS call. */
+ public static final String KMS_CLIENT_MAX_ATTEMPTS =
"kms.client.aliyun.max.attempts";
+
+ public static final int KMS_CLIENT_MAX_ATTEMPTS_DEFAULT = 3;
+
+ /** Connect timeout in milliseconds for KMS calls. */
+ public static final String KMS_CLIENT_CONNECT_TIMEOUT_MS =
"kms.client.aliyun.connect.timeout.ms";
+
+ public static final int KMS_CLIENT_CONNECT_TIMEOUT_MS_DEFAULT = 2_000;
+
+ /** Read timeout in milliseconds for KMS calls. */
+ public static final String KMS_CLIENT_READ_TIMEOUT_MS =
"kms.client.aliyun.read.timeout.ms";
+
+ public static final int KMS_CLIENT_READ_TIMEOUT_MS_DEFAULT = 30_000;
+
private final String ossEndpoint;
private final String accessKeyId;
private final String accessKeySecret;
private final String securityToken;
private final String ossStagingDirectory;
+ private final String region;
+ private final String kmsEndpoint;
+ private final int kmsClientMaxAttempts;
+ private final int kmsClientConnectTimeoutMs;
+ private final int kmsClientReadTimeoutMs;
public AliyunProperties() {
this(ImmutableMap.of());
@@ -96,6 +128,18 @@ public class AliyunProperties implements Serializable {
this.ossStagingDirectory =
PropertyUtil.propertyAsString(
properties, OSS_STAGING_DIRECTORY,
System.getProperty("java.io.tmpdir"));
+
+ this.region = properties.get(CLIENT_REGION);
+ this.kmsEndpoint = properties.get(CLIENT_KMS_ENDPOINT);
+ this.kmsClientMaxAttempts =
+ PropertyUtil.propertyAsInt(
+ properties, KMS_CLIENT_MAX_ATTEMPTS,
KMS_CLIENT_MAX_ATTEMPTS_DEFAULT);
+ this.kmsClientConnectTimeoutMs =
+ PropertyUtil.propertyAsInt(
+ properties, KMS_CLIENT_CONNECT_TIMEOUT_MS,
KMS_CLIENT_CONNECT_TIMEOUT_MS_DEFAULT);
+ this.kmsClientReadTimeoutMs =
+ PropertyUtil.propertyAsInt(
+ properties, KMS_CLIENT_READ_TIMEOUT_MS,
KMS_CLIENT_READ_TIMEOUT_MS_DEFAULT);
}
public String ossEndpoint() {
@@ -117,4 +161,24 @@ public class AliyunProperties implements Serializable {
public String ossStagingDirectory() {
return ossStagingDirectory;
}
+
+ public String region() {
+ return region;
+ }
+
+ public String kmsEndpoint() {
+ return kmsEndpoint;
+ }
+
+ public int kmsClientMaxAttempts() {
+ return kmsClientMaxAttempts;
+ }
+
+ public int kmsClientConnectTimeoutMs() {
+ return kmsClientConnectTimeoutMs;
+ }
+
+ public int kmsClientReadTimeoutMs() {
+ return kmsClientReadTimeoutMs;
+ }
}
diff --git
a/aliyun/src/test/java/org/apache/iceberg/aliyun/TestAliyunClientFactories.java
b/aliyun/src/test/java/org/apache/iceberg/aliyun/TestAliyunClientFactories.java
index 3fd4cfe87a..f0d1cd36bc 100644
---
a/aliyun/src/test/java/org/apache/iceberg/aliyun/TestAliyunClientFactories.java
+++
b/aliyun/src/test/java/org/apache/iceberg/aliyun/TestAliyunClientFactories.java
@@ -19,7 +19,9 @@
package org.apache.iceberg.aliyun;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import com.aliyun.kms20160120.Client;
import com.aliyun.oss.OSS;
import java.util.Map;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
@@ -135,6 +137,27 @@ public class TestAliyunClientFactories {
.isFalse();
}
+ @Test
+ public void testNewKmsClientResolvesEndpointFromRegion() {
+ Map<String, String> properties = Maps.newHashMap();
+ properties.put(AliyunProperties.CLIENT_REGION, "cn-hangzhou");
+ properties.put(AliyunProperties.CLIENT_ACCESS_KEY_ID, "key");
+ properties.put(AliyunProperties.CLIENT_ACCESS_KEY_SECRET, "secret");
+
+ Client kmsClient = AliyunClientFactories.from(properties).newKmsClient();
+ assertThat(kmsClient)
+ .as("KMS client should be created with an endpoint resolved from the
region")
+ .isNotNull();
+ }
+
+ @Test
+ public void testNewKmsClientRequiresRegion() {
+ AliyunClientFactory factory =
AliyunClientFactories.from(Maps.newHashMap());
+ assertThatThrownBy(factory::newKmsClient)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining(AliyunProperties.CLIENT_REGION);
+ }
+
public static class CustomFactory implements AliyunClientFactory {
AliyunProperties aliyunProperties;
@@ -146,6 +169,11 @@ public class TestAliyunClientFactories {
return null;
}
+ @Override
+ public Client newKmsClient() {
+ return null;
+ }
+
@Override
public void initialize(Map<String, String> properties) {
this.aliyunProperties = new AliyunProperties(properties);
diff --git
a/aliyun/src/test/java/org/apache/iceberg/aliyun/TestAliyunKeyManagementClient.java
b/aliyun/src/test/java/org/apache/iceberg/aliyun/TestAliyunKeyManagementClient.java
new file mode 100644
index 0000000000..ba352b5f36
--- /dev/null
+++
b/aliyun/src/test/java/org/apache/iceberg/aliyun/TestAliyunKeyManagementClient.java
@@ -0,0 +1,231 @@
+/*
+ * 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.aliyun;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.aliyun.kms20160120.Client;
+import com.aliyun.kms20160120.models.DecryptRequest;
+import com.aliyun.kms20160120.models.DecryptResponse;
+import com.aliyun.kms20160120.models.DecryptResponseBody;
+import com.aliyun.kms20160120.models.EncryptRequest;
+import com.aliyun.kms20160120.models.EncryptResponse;
+import com.aliyun.kms20160120.models.EncryptResponseBody;
+import com.aliyun.oss.OSS;
+import com.aliyun.teautil.models.RuntimeOptions;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.Map;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.TestHelpers;
+import org.apache.iceberg.encryption.EncryptionUtil;
+import org.apache.iceberg.encryption.KeyManagementClient;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
+import org.mockito.ArgumentCaptor;
+
+public class TestAliyunKeyManagementClient {
+
+ private static final String WRAPPING_KEY_ID = "test-wrapping-key";
+ private static final String SAMPLE_KEY_ID =
"202b9877-5a25-46e3-a763-e20791b5abcd";
+ private static final byte[] RAW_KEY =
+ "0123456789abcdef0123456789abcdef".getBytes(StandardCharsets.UTF_8);
+ private static final String RAW_KEY_B64 =
Base64.getEncoder().encodeToString(RAW_KEY);
+ // Aliyun KMS returns the ciphertext blob as a Base64 string; the wrapped
key carries the raw
+ // (decoded) ciphertext bytes.
+ private static final String CIPHERTEXT_BLOB =
+
Base64.getEncoder().encodeToString("wrapped-key-material".getBytes(StandardCharsets.UTF_8));
+ private static final ByteBuffer WRAPPED_KEY =
+ ByteBuffer.wrap(Base64.getDecoder().decode(CIPHERTEXT_BLOB));
+
+ // shared with the reflectively-loaded MockClientFactory below; ThreadLocal
keeps it
+ // parallel-safe.
+ private static final ThreadLocal<Client> MOCK_KMS = new ThreadLocal<>();
+
+ @BeforeEach
+ public void before() {
+ MOCK_KMS.set(mock(Client.class));
+ }
+
+ @AfterEach
+ public void after() {
+ MOCK_KMS.remove();
+ }
+
+ private static Client mockKms() {
+ return MOCK_KMS.get();
+ }
+
+ private static void stubDecrypt(String responseKeyId) throws Exception {
+ DecryptResponse response =
+ new DecryptResponse()
+ .setBody(new
DecryptResponseBody().setPlaintext(RAW_KEY_B64).setKeyId(responseKeyId));
+ when(mockKms().decryptWithOptions(any(DecryptRequest.class),
any(RuntimeOptions.class)))
+ .thenReturn(response);
+ }
+
+ @Test
+ public void testCreateFromKmsType() {
+ KeyManagementClient client =
+ EncryptionUtil.createKmsClient(
+ ImmutableMap.of(
+ CatalogProperties.ENCRYPTION_KMS_TYPE,
+ CatalogProperties.ENCRYPTION_KMS_TYPE_ALIYUN));
+ assertThat(client).isInstanceOf(AliyunKeyManagementClient.class);
+ }
+
+ @Test
+ public void testWrapKey() throws Exception {
+ EncryptResponse response =
+ new EncryptResponse().setBody(new
EncryptResponseBody().setCiphertextBlob(CIPHERTEXT_BLOB));
+ when(mockKms().encryptWithOptions(any(EncryptRequest.class),
any(RuntimeOptions.class)))
+ .thenReturn(response);
+
+ KeyManagementClient client = kmsClient(ImmutableMap.of());
+
+ ByteBuffer wrapped = client.wrapKey(ByteBuffer.wrap(RAW_KEY),
WRAPPING_KEY_ID);
+ assertThat(wrapped).isEqualTo(WRAPPED_KEY);
+
+ ArgumentCaptor<EncryptRequest> captor =
ArgumentCaptor.forClass(EncryptRequest.class);
+ verify(mockKms()).encryptWithOptions(captor.capture(),
any(RuntimeOptions.class));
+ assertThat(captor.getValue().getKeyId()).isEqualTo(WRAPPING_KEY_ID);
+ assertThat(captor.getValue().getPlaintext()).isEqualTo(RAW_KEY_B64);
+ }
+
+ @Test
+ public void testUnwrapKey() throws Exception {
+ stubDecrypt(WRAPPING_KEY_ID);
+
+ ByteBuffer unwrapped =
+ kmsClient(ImmutableMap.of()).unwrapKey(WRAPPED_KEY.duplicate(),
WRAPPING_KEY_ID);
+ assertThat(unwrapped).isEqualTo(ByteBuffer.wrap(RAW_KEY));
+
+ ArgumentCaptor<DecryptRequest> captor =
ArgumentCaptor.forClass(DecryptRequest.class);
+ verify(mockKms()).decryptWithOptions(captor.capture(),
any(RuntimeOptions.class));
+
assertThat(captor.getValue().getCiphertextBlob()).isEqualTo(CIPHERTEXT_BLOB);
+ }
+
+ @Test
+ public void testUnwrapKeyRejectsMismatchedKeyId() throws Exception {
+ stubDecrypt("some-other-key");
+
+ KeyManagementClient client = kmsClient(ImmutableMap.of());
+ assertThatThrownBy(() -> client.unwrapKey(WRAPPED_KEY.duplicate(),
WRAPPING_KEY_ID))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("some-other-key")
+ .hasMessageContaining(WRAPPING_KEY_ID);
+ }
+
+ @Test
+ public void testUnwrapKeyWithArnKeyId() throws Exception {
+ String arn = "acs:kms:cn-hangzhou:1234567890123456:key/" + SAMPLE_KEY_ID;
+ stubDecrypt(SAMPLE_KEY_ID);
+
+ // an ARN is normalized to its bare key id locally and matches the response
+ assertThat(kmsClient(ImmutableMap.of()).unwrapKey(WRAPPED_KEY.duplicate(),
arn))
+ .isEqualTo(ByteBuffer.wrap(RAW_KEY));
+ }
+
+ @Test
+ public void testUnwrapKeySkipsVerificationForAlias() throws Exception {
+ // an alias skips key-id verification, so unwrap succeeds even when the
response keyId differs
+ stubDecrypt(SAMPLE_KEY_ID);
+
+ assertThat(
+ kmsClient(ImmutableMap.of()).unwrapKey(WRAPPED_KEY.duplicate(),
"alias/maps-data-key"))
+ .isEqualTo(ByteBuffer.wrap(RAW_KEY));
+ }
+
+ @ParameterizedTest
+ @MethodSource("org.apache.iceberg.TestHelpers#serializers")
+ public void testKmsClientSerialization(
+ TestHelpers.RoundTripSerializer<KeyManagementClient>
roundTripSerializer) throws Exception {
+ when(mockKms().encryptWithOptions(any(EncryptRequest.class),
any(RuntimeOptions.class)))
+ .thenReturn(
+ new EncryptResponse()
+ .setBody(new
EncryptResponseBody().setCiphertextBlob(CIPHERTEXT_BLOB)));
+ stubDecrypt(WRAPPING_KEY_ID);
+
+ Map<String, String> baseProps =
+ ImmutableMap.of(
+ AliyunProperties.CLIENT_REGION, "cn-hangzhou",
+ AliyunProperties.KMS_CLIENT_MAX_ATTEMPTS, "5");
+
+ KeyManagementClient client = kmsClient(baseProps);
+ KeyManagementClient roundTripped = roundTripSerializer.apply(client);
+
+ // wrap + unwrap survive serialization
+ ByteBuffer wrapped = roundTripped.wrapKey(ByteBuffer.wrap(RAW_KEY),
WRAPPING_KEY_ID);
+ assertThat(wrapped).isEqualTo(WRAPPED_KEY);
+ assertThat(client.unwrapKey(wrapped.duplicate(), WRAPPING_KEY_ID))
+ .isEqualTo(ByteBuffer.wrap(RAW_KEY));
+ assertThat(roundTripped.unwrapKey(wrapped.duplicate(), WRAPPING_KEY_ID))
+ .isEqualTo(ByteBuffer.wrap(RAW_KEY));
+ }
+
+ private KeyManagementClient kmsClient(Map<String, String> extraProps) {
+ Map<String, String> properties =
+ ImmutableMap.<String, String>builder()
+ .put(AliyunProperties.CLIENT_FACTORY,
MockClientFactory.class.getName())
+ .putAll(extraProps)
+ .build();
+ KeyManagementClient client = new AliyunKeyManagementClient();
+ client.initialize(properties);
+ return client;
+ }
+
+ /**
+ * Reflectively loaded by {@link AliyunKeyManagementClient} to hand back the
mocked KMS client.
+ */
+ public static class MockClientFactory implements AliyunClientFactory {
+ private AliyunProperties aliyunProperties;
+
+ public MockClientFactory() {}
+
+ @Override
+ public OSS newOSSClient() {
+ return null;
+ }
+
+ @Override
+ public Client newKmsClient() {
+ return mockKms();
+ }
+
+ @Override
+ public void initialize(Map<String, String> properties) {
+ this.aliyunProperties = new AliyunProperties(properties);
+ }
+
+ @Override
+ public AliyunProperties aliyunProperties() {
+ return aliyunProperties;
+ }
+ }
+}
diff --git a/build.gradle b/build.gradle
index 66ce4a3ee0..df44af7e9e 100644
--- a/build.gradle
+++ b/build.gradle
@@ -501,6 +501,7 @@ project(':iceberg-aliyun') {
implementation project(':iceberg-common')
compileOnly libs.aliyun.sdk.oss
+ compileOnly libs.aliyun.sdk.kms
compileOnly libs.aliyun.credentials.java
compileOnly libs.aliyun.tea
compileOnly libs.jaxb.api
@@ -516,6 +517,8 @@ project(':iceberg-aliyun') {
testImplementation platform(libs.jackson.bom)
testImplementation
"com.fasterxml.jackson.dataformat:jackson-dataformat-xml"
testImplementation project(path: ':iceberg-api', configuration:
'testArtifacts')
+ testImplementation libs.esotericsoftware.kryo
+ testImplementation libs.mockito.core
testImplementation libs.junit.pioneer
}
}
diff --git a/core/src/main/java/org/apache/iceberg/CatalogProperties.java
b/core/src/main/java/org/apache/iceberg/CatalogProperties.java
index 6b85ccbc87..8ad32bed03 100644
--- a/core/src/main/java/org/apache/iceberg/CatalogProperties.java
+++ b/core/src/main/java/org/apache/iceberg/CatalogProperties.java
@@ -174,6 +174,7 @@ public class 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_ALIYUN = "aliyun";
public static final String ENCRYPTION_KMS_IMPL = "encryption.kms-impl";
public static final String ENCRYPTION_KMS_IMPL_AWS =
@@ -182,4 +183,6 @@ public class 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_ALIYUN =
+ "org.apache.iceberg.aliyun.AliyunKeyManagementClient";
}
diff --git
a/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java
b/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java
index 382d244883..cab839a7a8 100644
--- a/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java
+++ b/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java
@@ -63,6 +63,8 @@ public class EncryptionUtil {
CatalogProperties.ENCRYPTION_KMS_IMPL_AZURE;
case CatalogProperties.ENCRYPTION_KMS_TYPE_GCP ->
CatalogProperties.ENCRYPTION_KMS_IMPL_GCP;
+ case CatalogProperties.ENCRYPTION_KMS_TYPE_ALIYUN ->
+ CatalogProperties.ENCRYPTION_KMS_IMPL_ALIYUN;
default -> throw new IllegalStateException("Unsupported KMS type:
" + kmsType);
};
}
diff --git a/docs/docs/encryption.md b/docs/docs/encryption.md
index cbdce85e76..f8dcfcbe87 100644
--- a/docs/docs/encryption.md
+++ b/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 `aliyun`) 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).
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index 481da364d4..a7294dca4c 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -21,7 +21,8 @@
[versions]
activation = "1.1.1"
-aliyun-credentials-java = "0.3.12"
+aliyun-credentials-java = "1.0.3"
+aliyun-sdk-kms = "2.3.0"
aliyun-sdk-oss = "3.18.5"
aliyun-tea = "1.4.2"
analyticsaccelerator = "1.3.1"
@@ -98,6 +99,7 @@ tez08 = { strictly = "0.8.4"} # see rich version usage
explanation above
activation = { module = "javax.activation:activation", version.ref =
"activation" }
aircompressor = { module = "io.airlift:aircompressor", version.ref =
"aircompressor" }
aliyun-credentials-java = { module = "com.aliyun:credentials-java",
version.ref = "aliyun-credentials-java" }
+aliyun-sdk-kms = { module = "com.aliyun:kms20160120", version.ref =
"aliyun-sdk-kms" }
aliyun-sdk-oss = { module = "com.aliyun.oss:aliyun-sdk-oss", version.ref =
"aliyun-sdk-oss" }
aliyun-tea = { module = "com.aliyun:tea", version.ref = "aliyun-tea" }
analyticsaccelerator-s3 = { module =
"software.amazon.s3.analyticsaccelerator:analyticsaccelerator-s3", version.ref
= "analyticsaccelerator" }
