feat: add KeyManagementClient, KmsRegistry and InMemoryKms

ci/scripts/generate_public_headers.py

@@ -51,6 +51,14 @@
     ("iceberg/parquet", "ICEBERG_PUBLIC_HEADERS_BUNDLE"),
 ]
 
+# Individual headers of an optional component that lives in a directory which is
+# otherwise always built. iceberg/encryption holds both the metadata types, which are
+# part of core, and the crypto implementation, which needs ICEBERG_ENCRYPTION.
+OPTIONAL_HEADERS = {
+    "iceberg/encryption/aes_gcm_stream.h": "ICEBERG_PUBLIC_HEADERS_ENCRYPTION",
+    "iceberg/encryption/in_memory_kms.h": "ICEBERG_PUBLIC_HEADERS_ENCRYPTION",
+}
+
 INSTALL_CALL = re.compile(r"iceberg_install_all_headers\(\s*([^\s)]+)")
 
 LICENSE = """\
@@ -75,7 +83,9 @@
 """
 
 
-def component_of(install_path: str) -> str | None:
+def component_of(install_path: str, header_path: str | None = None) -> str | None:
+    if header_path is not None and header_path in OPTIONAL_HEADERS:
+        return OPTIONAL_HEADERS[header_path]
     for prefix, macro in OPTIONAL_COMPONENTS:
         if install_path == prefix or install_path.startswith(prefix + "/"):
             return macro
@@ -89,8 +99,9 @@ def collect_public_headers() -> dict[str | None, list[str]]:
             for header in cmake_file.parent.iterdir():
                 if header.suffix != ".h" or "internal" in header.name:
                     continue
-                headers.setdefault(component_of(install_path), []).append(
-                    f"{install_path}/{header.name}"
+                header_path = f"{install_path}/{header.name}"
+                headers.setdefault(component_of(install_path, header_path), []).append(
+                    header_path
                 )
     return headers
 
@@ -107,7 +118,8 @@ def render(headers: dict[str | None, list[str]]) -> str:
         "",
     ]
     lines += [f'#include "{h}"' for h in sorted(headers.get(None, []))]
-    for macro in sorted({m for _, m in OPTIONAL_COMPONENTS}):
+    optional_macros = {m for _, m in OPTIONAL_COMPONENTS} | set(OPTIONAL_HEADERS.values())
+    for macro in sorted(optional_macros):
         if macro not in headers:
             continue
         lines += ["", f"#ifdef {macro}"]

src/iceberg/CMakeLists.txt

@@ -33,6 +33,7 @@ set(ICEBERG_SOURCES
     deletes/position_delete_index.cc
     deletes/position_delete_range_consumer.cc
     deletes/roaring_position_bitmap.cc
+    encryption/key_management_client.cc
     encryption/standard_key_metadata.cc
     expression/aggregate.cc
     expression/binder.cc
@@ -326,6 +327,14 @@ if(ICEBERG_BUILD_BUNDLE)
       parquet/parquet_writer.cc
       parquet/row_number_reader.cc)
 
+  if(ICEBERG_ENCRYPTION)
+    list(APPEND
+         ICEBERG_BUNDLE_SOURCES
+         encryption/aes_gcm_internal.cc
+         encryption/aes_gcm_stream.cc
+         encryption/in_memory_kms.cc)
+  endif()
+
   # Libraries to link with exported libiceberg_bundle.{so,a}.
   set(ICEBERG_BUNDLE_STATIC_BUILD_INTERFACE_LIBS)
   set(ICEBERG_BUNDLE_SHARED_BUILD_INTERFACE_LIBS)

src/iceberg/cpp20_compatibility_internal.h

@@ -49,6 +49,7 @@
 #include "iceberg/deletes/position_delete_range_consumer.h"
 #include "iceberg/deletes/roaring_position_bitmap.h"
 #include "iceberg/encryption/encrypted_key.h"
+#include "iceberg/encryption/key_management_client.h"
 #include "iceberg/encryption/standard_key_metadata.h"
 #include "iceberg/exception.h"
 #include "iceberg/expression/aggregate.h"
@@ -216,6 +217,11 @@
 #  include "iceberg/parquet/parquet_writer.h"
 #endif  // ICEBERG_PUBLIC_HEADERS_BUNDLE
 
+#ifdef ICEBERG_PUBLIC_HEADERS_ENCRYPTION
+#  include "iceberg/encryption/aes_gcm_stream.h"
+#  include "iceberg/encryption/in_memory_kms.h"
+#endif  // ICEBERG_PUBLIC_HEADERS_ENCRYPTION
+
 #ifdef ICEBERG_PUBLIC_HEADERS_HIVE
 #  include "iceberg/catalog/hive/hive_catalog.h"
 #  include "iceberg/catalog/hive/hive_catalog_properties.h"

src/iceberg/encryption/aes_gcm_internal.cc

@@ -0,0 +1,174 @@
+/*
+ * 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.
+ */
+
+#include "iceberg/encryption/aes_gcm_internal.h"
+
+#include <limits>
+
+#include <openssl/evp.h>
+#include <openssl/rand.h>
+
+#include "iceberg/util/macros.h"
+
+namespace iceberg::encryption {
+
+namespace {
+
+const EVP_CIPHER* CipherForKey(size_t key_length) {
+  switch (key_length) {
+    case 16:
+      return EVP_aes_128_gcm();
+    case 24:
+      return EVP_aes_192_gcm();
+    case 32:
+      return EVP_aes_256_gcm();
+    default:
+      return nullptr;
+  }
+}
+
+EVP_CIPHER_CTX* Ctx(void* ctx) { return static_cast<EVP_CIPHER_CTX*>(ctx); }
+
+}  // namespace
+
+Status SecureRandomBytes(std::span<uint8_t> out) {
+  if (out.size() > static_cast<size_t>(std::numeric_limits<int>::max()) ||
+      RAND_bytes(out.data(), static_cast<int>(out.size())) != 1) {
+    return IOError("Failed to generate secure random bytes");
+  }
+  return {};
+}
+
+Result<std::unique_ptr<AesGcmCipher>> AesGcmCipher::Make(std::span<const uint8_t> key) {
+  if (CipherForKey(key.size()) == nullptr) {
+    return InvalidArgument("Invalid key length: {} (must be 16, 24, or 32 bytes)",
+                           key.size());
+  }
+  EVP_CIPHER_CTX* ctx = EVP_CIPHER_CTX_new();
+  if (ctx == nullptr) {
+    return IOError("Failed to create cipher context");
+  }
+  return std::unique_ptr<AesGcmCipher>(
+      new AesGcmCipher(std::vector<uint8_t>(key.begin(), key.end()), ctx));
+}
+
+AesGcmCipher::AesGcmCipher(std::vector<uint8_t> key, void* ctx)
+    : key_(std::move(key)), ctx_(ctx) {}
+
+AesGcmCipher::~AesGcmCipher() { EVP_CIPHER_CTX_free(Ctx(ctx_)); }
+
+Status AesGcmCipher::Encrypt(std::span<const uint8_t> plaintext,
+                             std::span<const uint8_t> aad, std::span<uint8_t> out) {
+  if (out.size() != plaintext.size() + kNonceLength + kGcmTagLength) {
+    return InvalidArgument("Invalid ciphertext buffer size: {}", out.size());
+  }
+  if (plaintext.size() > static_cast<size_t>(std::numeric_limits<int>::max())) {
+    return InvalidArgument("Plaintext too large: {}", plaintext.size());
+  }
+
+  auto nonce = out.first(kNonceLength);
+  auto ciphertext = out.subspan(kNonceLength, plaintext.size());
+  auto tag = out.last(kGcmTagLength);
+  ICEBERG_RETURN_UNEXPECTED(SecureRandomBytes(nonce));
+
+  auto* ctx = Ctx(ctx_);
+  int len = 0;
+  if (EVP_EncryptInit_ex(ctx, CipherForKey(key_.size()), nullptr, nullptr, nullptr) !=
+          1 ||
+      EVP_CIPHER_CTX_ctrl(ctx, EVP_CTRL_GCM_SET_IVLEN, kNonceLength, nullptr) != 1 ||
+      EVP_EncryptInit_ex(ctx, nullptr, nullptr, key_.data(), nonce.data()) != 1) {
+    return IOError("Failed to initialize AES GCM encryption");
+  }
+  if (!aad.empty() && EVP_EncryptUpdate(ctx, nullptr, &len, aad.data(),
+                                        static_cast<int>(aad.size())) != 1) {
+    return IOError("Failed to set AES GCM AAD");
+  }
+  if (!plaintext.empty() &&
+      EVP_EncryptUpdate(ctx, ciphertext.data(), &len, plaintext.data(),
+                        static_cast<int>(plaintext.size())) != 1) {
+    return IOError("Failed to encrypt");
+  }
+  if (EVP_EncryptFinal_ex(ctx, ciphertext.data() + plaintext.size(), &len) != 1 ||
+      EVP_CIPHER_CTX_ctrl(ctx, EVP_CTRL_GCM_GET_TAG, kGcmTagLength, tag.data()) != 1) {
+    return IOError("Failed to finalize AES GCM encryption");
+  }
+  return {};
+}
+
+Status AesGcmCipher::Decrypt(std::span<const uint8_t> ciphertext,
+                             std::span<const uint8_t> aad, std::span<uint8_t> out) {
+  if (ciphertext.size() < static_cast<size_t>(kNonceLength + kGcmTagLength)) {
+    return Invalid(
+        "Cannot decrypt cipher text of length {}: shorter than nonce and tag, text may "
+        "not be encrypted with AES GCM",
+        ciphertext.size());
+  }
+  size_t plaintext_length = ciphertext.size() - kNonceLength - kGcmTagLength;
+  if (out.size() != plaintext_length) {
+    return InvalidArgument("Invalid plaintext buffer size: {}", out.size());
+  }
+
+  auto nonce = ciphertext.first(kNonceLength);
+  auto data = ciphertext.subspan(kNonceLength, plaintext_length);
+  auto tag = ciphertext.last(kGcmTagLength);
+
+  auto* ctx = Ctx(ctx_);
+  int len = 0;
+  if (EVP_DecryptInit_ex(ctx, CipherForKey(key_.size()), nullptr, nullptr, nullptr) !=
+          1 ||
+      EVP_CIPHER_CTX_ctrl(ctx, EVP_CTRL_GCM_SET_IVLEN, kNonceLength, nullptr) != 1 ||
+      EVP_DecryptInit_ex(ctx, nullptr, nullptr, key_.data(), nonce.data()) != 1) {
+    return IOError("Failed to initialize AES GCM decryption");
+  }
+  if (!aad.empty() && EVP_DecryptUpdate(ctx, nullptr, &len, aad.data(),
+                                        static_cast<int>(aad.size())) != 1) {
+    return IOError("Failed to set AES GCM AAD");
+  }
+  if (!data.empty() && EVP_DecryptUpdate(ctx, out.data(), &len, data.data(),
+                                         static_cast<int>(data.size())) != 1) {
+    return IOError("Failed to decrypt");
+  }
+  if (EVP_CIPHER_CTX_ctrl(ctx, EVP_CTRL_GCM_SET_TAG, kGcmTagLength,
+                          const_cast<uint8_t*>(tag.data())) != 1 ||
+      EVP_DecryptFinal_ex(ctx, out.data() + plaintext_length, &len) != 1) {
+    return Invalid(
+        "GCM tag check failed. Possible reasons: wrong decryption key; or "
+        "corrupt/tampered data. AES GCM doesn't differentiate between these two.");
+  }
+  return {};
+}
+
+Result<std::vector<uint8_t>> AesGcmCipher::Encrypt(std::span<const uint8_t> plaintext,
+                                                   std::span<const uint8_t> aad) {
+  std::vector<uint8_t> out(plaintext.size() + kNonceLength + kGcmTagLength);
+  ICEBERG_RETURN_UNEXPECTED(Encrypt(plaintext, aad, out));
+  return out;
+}
+
+Result<std::vector<uint8_t>> AesGcmCipher::Decrypt(std::span<const uint8_t> ciphertext,
+                                                   std::span<const uint8_t> aad) {
+  if (ciphertext.size() < static_cast<size_t>(kNonceLength + kGcmTagLength)) {
+    return Invalid("Cannot decrypt cipher text of length {}", ciphertext.size());
+  }
+  std::vector<uint8_t> out(ciphertext.size() - kNonceLength - kGcmTagLength);
+  ICEBERG_RETURN_UNEXPECTED(Decrypt(ciphertext, aad, out));
+  return out;
+}
+
+}  // namespace iceberg::encryption

src/iceberg/encryption/aes_gcm_internal.h

@@ -0,0 +1,71 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+/// \file iceberg/encryption/aes_gcm_internal.h
+/// AES GCM cipher compatible with Java's org.apache.iceberg.encryption.Ciphers.
+
+#include <cstdint>
+#include <memory>
+#include <span>
+#include <vector>
+
+#include "iceberg/result.h"
+
+namespace iceberg::encryption {
+
+inline constexpr int32_t kNonceLength = 12;
+inline constexpr int32_t kGcmTagLength = 16;
+
+/// \brief Fill `out` with cryptographically secure random bytes.
+Status SecureRandomBytes(std::span<uint8_t> out);
+
+/// \brief AES GCM with a random 12-byte nonce.
+///
+/// Ciphertext layout: nonce (12) | encrypted data | tag (16).
+class AesGcmCipher {
+ public:
+  /// \brief Create a cipher for a 16, 24 or 32 byte key.
+  static Result<std::unique_ptr<AesGcmCipher>> Make(std::span<const uint8_t> key);
+
+  ~AesGcmCipher();
+
+  /// \brief Encrypt `plaintext` into `out`, which must hold plaintext.size() + 28 bytes.
+  Status Encrypt(std::span<const uint8_t> plaintext, std::span<const uint8_t> aad,
+                 std::span<uint8_t> out);
+
+  /// \brief Decrypt and verify `ciphertext` into `out`, which must hold
+  /// ciphertext.size() - 28 bytes.
+  Status Decrypt(std::span<const uint8_t> ciphertext, std::span<const uint8_t> aad,
+                 std::span<uint8_t> out);
+
+  Result<std::vector<uint8_t>> Encrypt(std::span<const uint8_t> plaintext,
+                                       std::span<const uint8_t> aad);
+  Result<std::vector<uint8_t>> Decrypt(std::span<const uint8_t> ciphertext,
+                                       std::span<const uint8_t> aad);
+
+ private:
+  AesGcmCipher(std::vector<uint8_t> key, void* ctx);
+
+  std::vector<uint8_t> key_;
+  void* ctx_;  // EVP_CIPHER_CTX
+};
+
+}  // namespace iceberg::encryption