feat(encryption): add EncryptionManager and StandardEncryptionManager

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/encryption_manager.cc
     encryption/key_management_client.cc
     encryption/standard_key_metadata.cc
     expression/aggregate.cc
@@ -321,6 +322,7 @@ if(ICEBERG_BUILD_BUNDLE)
       avro/avro_register.cc
       avro/avro_schema_util.cc
       avro/avro_stream_internal.cc
+      encryption/encryption_register.cc
       parquet/parquet_data_util.cc
       parquet/parquet_metrics.cc
       parquet/parquet_metrics_row_group_filter.cc
@@ -335,7 +337,8 @@ if(ICEBERG_BUILD_BUNDLE)
          ICEBERG_BUNDLE_SOURCES
          encryption/aes_gcm.cc
          encryption/aes_gcm_stream.cc
-         encryption/in_memory_kms.cc)
+         encryption/in_memory_kms.cc
+         encryption/standard_encryption_manager.cc)
   endif()
 
   # Libraries to link with exported libiceberg_bundle.{so,a}.
@@ -410,6 +413,11 @@ if(ICEBERG_BUILD_BUNDLE)
 
   foreach(target iceberg_bundle_static iceberg_bundle_shared)
     if(TARGET ${target})
+      if(ICEBERG_ENCRYPTION)
+        target_compile_definitions(${target} PRIVATE ICEBERG_ENCRYPTION_ENABLED=1)
+      else()
+        target_compile_definitions(${target} PRIVATE ICEBERG_ENCRYPTION_ENABLED=0)
+      endif()
       if(ICEBERG_S3)
         target_compile_definitions(${target}
                                    PUBLIC "$<BUILD_INTERFACE:ICEBERG_S3_ENABLED=1>")

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/encryption_manager.h"
 #include "iceberg/encryption/key_management_client.h"
 #include "iceberg/encryption/standard_key_metadata.h"
 #include "iceberg/exception.h"
@@ -213,6 +214,7 @@
 #  include "iceberg/avro/avro_reader.h"
 #  include "iceberg/avro/avro_register.h"
 #  include "iceberg/avro/avro_writer.h"
+#  include "iceberg/encryption/encryption_register.h"
 #  include "iceberg/parquet/parquet_reader.h"
 #  include "iceberg/parquet/parquet_register.h"
 #  include "iceberg/parquet/parquet_writer.h"
@@ -221,6 +223,7 @@
 #ifdef ICEBERG_PUBLIC_HEADERS_ENCRYPTION
 #  include "iceberg/encryption/aes_gcm_stream.h"
 #  include "iceberg/encryption/in_memory_kms.h"
+#  include "iceberg/encryption/standard_encryption_manager.h"
 #endif  // ICEBERG_PUBLIC_HEADERS_ENCRYPTION
 
 #ifdef ICEBERG_PUBLIC_HEADERS_HIVE

src/iceberg/encryption/CMakeLists.txt

@@ -18,13 +18,22 @@
 iceberg_install_all_headers(iceberg/encryption
                             PATTERN
                             encrypted_key.h
+                            encryption_manager.h
                             key_management_client.h
                             standard_key_metadata.h)
-if(ICEBERG_BUILD_BUNDLE AND ICEBERG_ENCRYPTION)
+if(ICEBERG_BUILD_BUNDLE)
   iceberg_install_all_headers(iceberg/encryption
                               PATTERN
-                              aes_gcm_stream.h
-                              in_memory_kms.h
+                              encryption_register.h
                               HEADER_GROUP
-                              ICEBERG_PUBLIC_HEADERS_ENCRYPTION)
+                              ICEBERG_PUBLIC_HEADERS_BUNDLE)
+  if(ICEBERG_ENCRYPTION)
+    iceberg_install_all_headers(iceberg/encryption
+                                PATTERN
+                                aes_gcm_stream.h
+                                in_memory_kms.h
+                                standard_encryption_manager.h
+                                HEADER_GROUP
+                                ICEBERG_PUBLIC_HEADERS_ENCRYPTION)
+  endif()
 endif()

src/iceberg/encryption/encryption_manager.cc

@@ -0,0 +1,72 @@
+/*
+ * 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/encryption_manager.h"
+
+#include <mutex>
+
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+namespace {
+
+struct RegistryState {
+  std::mutex mutex;
+  EncryptionManagerRegistry::Factory factory;
+};
+
+RegistryState& State() {
+  static RegistryState state;
+  return state;
+}
+
+}  // namespace
+
+void EncryptionManagerRegistry::Register(Factory factory) {
+  auto& state = State();
+  std::lock_guard lock(state.mutex);
+  state.factory = std::move(factory);
+}
+
+Result<std::shared_ptr<EncryptionManager>> EncryptionManagerRegistry::Make(
+    Options options) {
+  ICEBERG_PRECHECK(!options.table_key_id.empty(), "Invalid encryption key ID: empty");
+  ICEBERG_PRECHECK(options.data_key_length == 16 || options.data_key_length == 24 ||
+                       options.data_key_length == 32,
+                   "Invalid data key length: {} (must be 16, 24, or 32)",
+                   options.data_key_length);
+  ICEBERG_PRECHECK(options.kms != nullptr,
+                   "Cannot create an encryption manager without a KMS client; set the "
+                   "encryption.kms-type or encryption.kms-impl catalog property");
+  Factory factory;
+  {
+    auto& state = State();
+    std::lock_guard lock(state.mutex);
+    factory = state.factory;
+  }
+  if (!factory) {
+    return NotSupported(
+        "Table encryption is not available: link iceberg_bundle built with "
+        "ICEBERG_ENCRYPTION=ON and call iceberg::encryption::RegisterAll()");
+  }
+  return factory(std::move(options));
+}
+
+}  // namespace iceberg

src/iceberg/encryption/encryption_manager.h

@@ -0,0 +1,120 @@
+/*
+ * 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/encryption_manager.h
+/// Table encryption manager interface and the registry that creates one.
+
+#include <cstdint>
+#include <functional>
+#include <memory>
+#include <span>
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include "iceberg/encryption/encrypted_key.h"
+#include "iceberg/encryption/key_management_client.h"
+#include "iceberg/encryption/standard_key_metadata.h"
+#include "iceberg/file_io.h"
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+
+namespace iceberg {
+
+/// \brief Keys created when the key metadata of a new manifest list is registered.
+///
+/// Both must be added to the table metadata `encryption-keys` in the same commit as
+/// the snapshot whose `key-id` is `file_key.key_id`.
+struct ICEBERG_EXPORT ManifestListEncryptionKeys {
+  /// The key encryption key, wrapped by the table master key in the KMS.
+  EncryptedKey key_encryption_key;
+  /// The manifest list key metadata, encrypted by the key encryption key.
+  EncryptedKey file_key;
+};
+
+/// \brief Encrypts and decrypts the files of one table.
+///
+/// The standard implementation (StandardEncryptionManager) needs a crypto provider
+/// and lives in the bundle library; the core library reaches it through
+/// EncryptionManagerRegistry.
+class ICEBERG_EXPORT EncryptionManager {
+ public:
+  virtual ~EncryptionManager() = default;
+
+  /// \brief Generate key metadata (random data key and AAD prefix) for a new file.
+  virtual Result<StandardKeyMetadata> NewKeyMetadata() = 0;
+
+  /// \brief Wrap an AES GCM Stream (AGS1) file in a decrypting InputFile.
+  ///
+  /// \param encrypted the encrypted file
+  /// \param key_metadata serialized key metadata, including the file length
+  virtual Result<std::unique_ptr<InputFile>> NewDecryptingInputFile(
+      std::unique_ptr<InputFile> encrypted, std::span<const uint8_t> key_metadata) = 0;
+
+  /// \brief Wrap a file in an AES GCM Stream (AGS1) encrypting OutputFile.
+  virtual Result<std::unique_ptr<OutputFile>> NewEncryptingOutputFile(
+      std::unique_ptr<OutputFile> plain, const StandardKeyMetadata& key_metadata) = 0;
+
+  /// \brief Decrypt the key metadata of a manifest list.
+  ///
+  /// \param manifest_list_key_id the snapshot `key-id`
+  /// \return the serialized StandardKeyMetadata of the manifest list file
+  virtual Result<std::vector<uint8_t>> DecryptManifestListKeyMetadata(
+      std::string_view manifest_list_key_id) = 0;
+
+  /// \brief Encrypt the key metadata of a new manifest list with a key encryption key.
+  ///
+  /// Creates a key encryption key (wrapped by the table master key) if there is no
+  /// unexpired one.
+  virtual Result<ManifestListEncryptionKeys> RegisterManifestListKeyMetadata(
+      const StandardKeyMetadata& key_metadata) = 0;
+};
+
+/// \brief Creates the encryption manager of encrypted tables.
+///
+/// The bundle library registers the standard implementation (see
+/// iceberg::encryption::RegisterAll()). Without it, encrypted tables cannot be read or
+/// written.
+class ICEBERG_EXPORT EncryptionManagerRegistry {
+ public:
+  struct Options {
+    /// The table master key ID in the KMS (table property `encryption.key-id`), taken
+    /// from a trusted source.
+    std::string table_key_id;
+    /// Length of generated data keys: 16, 24 or 32 bytes.
+    int32_t data_key_length = 16;
+    /// The table metadata `encryption-keys`.
+    std::vector<EncryptedKey> encryption_keys;
+    /// The catalog's KMS client.
+    std::shared_ptr<KeyManagementClient> kms;
+  };
+
+  using Factory =
+      std::function<Result<std::shared_ptr<EncryptionManager>>(Options options)>;
+
+  /// \brief Register the encryption manager implementation.
+  static void Register(Factory factory);
+
+  /// \brief Create an encryption manager.
+  static Result<std::shared_ptr<EncryptionManager>> Make(Options options);
+};
+
+}  // namespace iceberg

src/iceberg/encryption/encryption_register.cc

@@ -0,0 +1,36 @@
+/*
+ * 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/encryption_register.h"
+
+#if ICEBERG_ENCRYPTION_ENABLED
+#  include "iceberg/encryption/standard_encryption_manager.h"
+#endif
+
+namespace iceberg::encryption {
+
+void RegisterAll() {
+#if ICEBERG_ENCRYPTION_ENABLED
+  RegisterStandardEncryptionManager();
+#endif
+}
+
+bool IsEncryptionAvailable() { return ICEBERG_ENCRYPTION_ENABLED; }
+
+}  // namespace iceberg::encryption

src/iceberg/encryption/encryption_register.h

@@ -0,0 +1,36 @@
+/*
+ * 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/encryption_register.h
+/// Registration of the table encryption implementation of the bundle library.
+
+#include "iceberg/iceberg_bundle_export.h"
+
+namespace iceberg::encryption {
+
+/// \brief Register the standard encryption manager, if the bundle was built with
+/// ICEBERG_ENCRYPTION=ON. Otherwise a no-op, and encrypted tables are not supported.
+ICEBERG_BUNDLE_EXPORT void RegisterAll();
+
+/// \brief Whether the bundle was built with table encryption support.
+ICEBERG_BUNDLE_EXPORT bool IsEncryptionAvailable();
+
+}  // namespace iceberg::encryption