feat(encryption) [3/N] Support encryption: KMS (#2339)

## Which issue does this PR close?

Part of #2034 

## What changes are included in this PR?

Adds the `KeyManagementClient` trait and an in-memory implementation for
testing.

- `KeyManagementClient` trait mirrors Java's `KeyManagementClient`
[interface](https://github.com/apache/iceberg/blob/main/core/src/main/java/org/apache/iceberg/encryption/KeyManagementClient.java).
- `InMemoryKeyManagementClient` for testing-only KMS that stores master
keys in memory to wrap/unwrap. Supports configurable key sizes and
explicit key bytes for cross-language interop tests.


## Are these changes tested?

Tests covered for `InMemoryKeyManagementClient`

Author: xanderbailey

crates/iceberg/src/encryption/crypto.rs

@@ -43,7 +43,7 @@ use crate::{Error, ErrorKind, Result};
 /// containing `SensitiveBytes` can safely derive or implement `Debug`
 /// without risk of leaking key material.
 #[derive(Clone, PartialEq, Eq)]
-struct SensitiveBytes(Zeroizing<Box<[u8]>>);
+pub struct SensitiveBytes(Zeroizing<Box<[u8]>>);
 
 impl SensitiveBytes {
     /// Wraps the given bytes as sensitive material.
@@ -57,13 +57,11 @@ impl SensitiveBytes {
     }
 
     /// Returns the number of bytes.
-    #[allow(dead_code)] // Encryption work is ongoing so currently unused
     pub fn len(&self) -> usize {
         self.0.len()
     }
 
     /// Returns `true` if the byte slice is empty.
-    #[allow(dead_code)] // Encryption work is ongoing so currently unused
     pub fn is_empty(&self) -> bool {
         self.0.is_empty()
     }
@@ -85,9 +83,10 @@ impl fmt::Display for SensitiveBytes {
 ///
 /// The Iceberg spec supports 128, 192, and 256-bit keys for AES-GCM.
 /// See: <https://iceberg.apache.org/gcm-stream-spec/#goals>
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 pub enum AesKeySize {
-    /// 128-bit AES key (16 bytes)
+    /// 128-bit AES key (16 bytes). Default per the Iceberg spec.
+    #[default]
     Bits128 = 128,
     /// 192-bit AES key (24 bytes)
     Bits192 = 192,

crates/iceberg/src/encryption/kms/client.rs

@@ -0,0 +1,98 @@
+// 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.
+
+//! Key management client trait for encryption key operations.
+//!
+//! Mirrors the Java `KeyManagementClient` interface from the Apache Iceberg spec.
+
+use async_trait::async_trait;
+
+use crate::Result;
+use crate::encryption::SensitiveBytes;
+
+/// Result of a server-side key generation operation.
+///
+/// Returned by [`KeyManagementClient::generate_key`] when the KMS supports
+/// atomic key generation and wrapping.
+pub struct GeneratedKey {
+    key: SensitiveBytes,
+    wrapped_key: Vec<u8>,
+}
+
+impl GeneratedKey {
+    /// Creates a new `GeneratedKey` from plaintext key bytes and wrapped key bytes.
+    pub fn new(key: SensitiveBytes, wrapped_key: Vec<u8>) -> Self {
+        Self { key, wrapped_key }
+    }
+
+    /// Returns the plaintext key bytes. Zeroized on drop, redacted in Debug.
+    pub fn key(&self) -> &SensitiveBytes {
+        &self.key
+    }
+
+    /// Returns the wrapped (encrypted) key bytes.
+    pub fn wrapped_key(&self) -> &[u8] {
+        &self.wrapped_key
+    }
+}
+
+/// Pluggable interface for key management systems (AWS KMS, Azure Key Vault, etc.).
+#[async_trait]
+pub trait KeyManagementClient: Send + Sync + std::fmt::Debug {
+    /// Wrap (encrypt) a key using a wrapping key managed by the KMS.
+    async fn wrap_key(&self, key: &[u8], wrapping_key_id: &str) -> Result<Vec<u8>>;
+
+    /// Unwrap (decrypt) a previously wrapped key.
+    async fn unwrap_key(&self, wrapped_key: &[u8], wrapping_key_id: &str)
+    -> Result<SensitiveBytes>;
+
+    /// Whether this KMS supports server-side key generation.
+    ///
+    /// If `true`, callers can use [`generate_key`](Self::generate_key) for atomic
+    /// key generation and wrapping, which is more secure than generating a key
+    /// locally and then wrapping it.
+    fn supports_key_generation(&self) -> bool;
+
+    /// Generate a new key and wrap it atomically on the server side.
+    ///
+    /// This is only supported when [`supports_key_generation`](Self::supports_key_generation)
+    /// returns `true`.
+    async fn generate_key(&self, wrapping_key_id: &str) -> Result<GeneratedKey>;
+}
+
+#[async_trait]
+impl<T: AsRef<dyn KeyManagementClient> + Send + Sync + std::fmt::Debug> KeyManagementClient for T {
+    async fn wrap_key(&self, key: &[u8], wrapping_key_id: &str) -> Result<Vec<u8>> {
+        self.as_ref().wrap_key(key, wrapping_key_id).await
+    }
+
+    async fn unwrap_key(
+        &self,
+        wrapped_key: &[u8],
+        wrapping_key_id: &str,
+    ) -> Result<SensitiveBytes> {
+        self.as_ref().unwrap_key(wrapped_key, wrapping_key_id).await
+    }
+
+    fn supports_key_generation(&self) -> bool {
+        self.as_ref().supports_key_generation()
+    }
+
+    async fn generate_key(&self, wrapping_key_id: &str) -> Result<GeneratedKey> {
+        self.as_ref().generate_key(wrapping_key_id).await
+    }
+}