diff --git a/ci/scripts/generate_public_headers.py b/ci/scripts/generate_public_headers.py index 56f196d87..9629236a9 100755 --- a/ci/scripts/generate_public_headers.py +++ b/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}"] diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt index c283b8af9..313b91606 100644 --- a/src/iceberg/CMakeLists.txt +++ b/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) diff --git a/src/iceberg/cpp20_compatibility_internal.h b/src/iceberg/cpp20_compatibility_internal.h index 359e5ddff..9eac44f9e 100644 --- a/src/iceberg/cpp20_compatibility_internal.h +++ b/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" diff --git a/src/iceberg/encryption/aes_gcm_internal.cc b/src/iceberg/encryption/aes_gcm_internal.cc new file mode 100644 index 000000000..63fe63036 --- /dev/null +++ b/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 + +#include +#include + +#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(ctx); } + +} // namespace + +Status SecureRandomBytes(std::span out) { + if (out.size() > static_cast(std::numeric_limits::max()) || + RAND_bytes(out.data(), static_cast(out.size())) != 1) { + return IOError("Failed to generate secure random bytes"); + } + return {}; +} + +Result> AesGcmCipher::Make(std::span 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( + new AesGcmCipher(std::vector(key.begin(), key.end()), ctx)); +} + +AesGcmCipher::AesGcmCipher(std::vector key, void* ctx) + : key_(std::move(key)), ctx_(ctx) {} + +AesGcmCipher::~AesGcmCipher() { EVP_CIPHER_CTX_free(Ctx(ctx_)); } + +Status AesGcmCipher::Encrypt(std::span plaintext, + std::span aad, std::span out) { + if (out.size() != plaintext.size() + kNonceLength + kGcmTagLength) { + return InvalidArgument("Invalid ciphertext buffer size: {}", out.size()); + } + if (plaintext.size() > static_cast(std::numeric_limits::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(aad.size())) != 1) { + return IOError("Failed to set AES GCM AAD"); + } + if (!plaintext.empty() && + EVP_EncryptUpdate(ctx, ciphertext.data(), &len, plaintext.data(), + static_cast(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 ciphertext, + std::span aad, std::span out) { + if (ciphertext.size() < static_cast(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(aad.size())) != 1) { + return IOError("Failed to set AES GCM AAD"); + } + if (!data.empty() && EVP_DecryptUpdate(ctx, out.data(), &len, data.data(), + static_cast(data.size())) != 1) { + return IOError("Failed to decrypt"); + } + if (EVP_CIPHER_CTX_ctrl(ctx, EVP_CTRL_GCM_SET_TAG, kGcmTagLength, + const_cast(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> AesGcmCipher::Encrypt(std::span plaintext, + std::span aad) { + std::vector out(plaintext.size() + kNonceLength + kGcmTagLength); + ICEBERG_RETURN_UNEXPECTED(Encrypt(plaintext, aad, out)); + return out; +} + +Result> AesGcmCipher::Decrypt(std::span ciphertext, + std::span aad) { + if (ciphertext.size() < static_cast(kNonceLength + kGcmTagLength)) { + return Invalid("Cannot decrypt cipher text of length {}", ciphertext.size()); + } + std::vector out(ciphertext.size() - kNonceLength - kGcmTagLength); + ICEBERG_RETURN_UNEXPECTED(Decrypt(ciphertext, aad, out)); + return out; +} + +} // namespace iceberg::encryption diff --git a/src/iceberg/encryption/aes_gcm_internal.h b/src/iceberg/encryption/aes_gcm_internal.h new file mode 100644 index 000000000..e09a7802e --- /dev/null +++ b/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 +#include +#include +#include + +#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 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> Make(std::span key); + + ~AesGcmCipher(); + + /// \brief Encrypt `plaintext` into `out`, which must hold plaintext.size() + 28 bytes. + Status Encrypt(std::span plaintext, std::span aad, + std::span out); + + /// \brief Decrypt and verify `ciphertext` into `out`, which must hold + /// ciphertext.size() - 28 bytes. + Status Decrypt(std::span ciphertext, std::span aad, + std::span out); + + Result> Encrypt(std::span plaintext, + std::span aad); + Result> Decrypt(std::span ciphertext, + std::span aad); + + private: + AesGcmCipher(std::vector key, void* ctx); + + std::vector key_; + void* ctx_; // EVP_CIPHER_CTX +}; + +} // namespace iceberg::encryption diff --git a/src/iceberg/encryption/aes_gcm_stream.cc b/src/iceberg/encryption/aes_gcm_stream.cc new file mode 100644 index 000000000..7fb417c11 --- /dev/null +++ b/src/iceberg/encryption/aes_gcm_stream.cc @@ -0,0 +1,344 @@ +/* + * 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_stream.h" + +#include +#include +#include +#include +#include + +#include "iceberg/encryption/aes_gcm_internal.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +namespace { + +using encryption::AesGcmCipher; + +constexpr int64_t kPlainBlock = AesGcmStream::kPlainBlockSize; +constexpr int64_t kCipherBlock = AesGcmStream::kCipherBlockSize; +constexpr int64_t kBlockOverhead = + AesGcmStream::kNonceLength + AesGcmStream::kGcmTagLength; + +/// \brief Block AAD: file AAD prefix followed by the block index as a 4-byte LE int. +std::vector BlockAad(const std::vector& aad_prefix, + int32_t block_index) { + std::vector aad(aad_prefix); + for (int i = 0; i < 4; ++i) { + aad.push_back( + static_cast((static_cast(block_index) >> (8 * i)) & 0xFF)); + } + return aad; +} + +std::array HeaderBytes() { + std::array header{}; + std::memcpy(header.data(), AesGcmStream::kMagic.data(), AesGcmStream::kMagic.size()); + for (int i = 0; i < 4; ++i) { + header[4 + i] = static_cast( + (static_cast(AesGcmStream::kPlainBlockSize) >> (8 * i)) & 0xFF); + } + return header; +} + +class AesGcmInputStream : public SeekableInputStream { + public: + AesGcmInputStream(std::unique_ptr source, + std::unique_ptr cipher, std::vector aad_prefix, + int64_t encrypted_length) + : source_(std::move(source)), + cipher_(std::move(cipher)), + aad_prefix_(std::move(aad_prefix)), + cipher_block_(kCipherBlock), + plain_block_(kPlainBlock) { + int64_t stream_length = encrypted_length - AesGcmStream::kHeaderLength; + int64_t num_full_blocks = stream_length / kCipherBlock; + int64_t last_bytes = stream_length - num_full_blocks * kCipherBlock; + bool full_blocks_only = last_bytes == 0; + num_blocks_ = full_blocks_only ? num_full_blocks : num_full_blocks + 1; + last_cipher_block_size_ = full_blocks_only ? kCipherBlock : last_bytes; + plain_size_ = AesGcmStream::PlaintextLength(encrypted_length); + } + + Status ValidateHeader() { + std::array header{}; + ICEBERG_RETURN_UNEXPECTED(source_->ReadFully(0, header)); + if (std::memcmp(header.data(), AesGcmStream::kMagic.data(), + AesGcmStream::kMagic.size()) != 0) { + return Invalid("Invalid GCM stream: magic does not match AGS1"); + } + uint32_t plain_block_size = 0; + for (int i = 0; i < 4; ++i) { + plain_block_size |= static_cast(header[4 + i]) << (8 * i); + } + if (plain_block_size != static_cast(AesGcmStream::kPlainBlockSize)) { + return Invalid("Invalid GCM stream: block size {} != {}", plain_block_size, + AesGcmStream::kPlainBlockSize); + } + if (last_cipher_block_size_ < kBlockOverhead) { + return Invalid("Invalid GCM stream: truncated last block"); + } + return {}; + } + + Result Position() const override { return position_; } + + Status Seek(int64_t position) override { + if (position < 0 || position > plain_size_) { + return IOError("Invalid position {} for stream of length {}", position, + plain_size_); + } + position_ = position; + return {}; + } + + Result Read(std::span out) override { + int64_t total = 0; + while (std::cmp_less(total, out.size()) && position_ < plain_size_) { + int64_t block_index = position_ / kPlainBlock; + ICEBERG_RETURN_UNEXPECTED(DecryptBlock(block_index)); + int64_t offset = position_ % kPlainBlock; + int64_t n = std::min(current_block_size_ - offset, + static_cast(out.size()) - total); + std::memcpy(out.data() + total, plain_block_.data() + offset, n); + total += n; + position_ += n; + } + return total; + } + + Status ReadFully(int64_t position, std::span out) override { + if (position < 0 || position + static_cast(out.size()) > plain_size_) { + return IOError("Cannot read {} bytes at {} from stream of length {}", out.size(), + position, plain_size_); + } + ICEBERG_RETURN_UNEXPECTED(Seek(position)); + ICEBERG_ASSIGN_OR_RAISE(auto n, Read(out)); + if (std::cmp_not_equal(n, out.size())) { + return IOError("Short read from GCM stream"); + } + return {}; + } + + Status Close() override { return source_->Close(); } + + private: + Status DecryptBlock(int64_t block_index) { + if (block_index == current_block_index_) { + return {}; + } + if (block_index > std::numeric_limits::max()) { + return Invalid("GCM stream block index overflow"); + } + int64_t size = + block_index == num_blocks_ - 1 ? last_cipher_block_size_ : kCipherBlock; + auto cipher_block = std::span(cipher_block_).first(size); + ICEBERG_RETURN_UNEXPECTED( + source_->ReadFully(AesGcmStream::kHeaderLength + block_index * kCipherBlock, + std::as_writable_bytes(cipher_block))); + auto aad = BlockAad(aad_prefix_, static_cast(block_index)); + int64_t plain_size = size - kBlockOverhead; + // Decryption writes plaintext before the tag is checked, so the cached block is + // invalid until the whole block authenticates. + current_block_index_ = -1; + current_block_size_ = 0; + ICEBERG_RETURN_UNEXPECTED( + cipher_->Decrypt(cipher_block, aad, std::span(plain_block_).first(plain_size))); + current_block_size_ = plain_size; + current_block_index_ = block_index; + return {}; + } + + std::unique_ptr source_; + std::unique_ptr cipher_; + std::vector aad_prefix_; + std::vector cipher_block_; + std::vector plain_block_; + int64_t num_blocks_ = 0; + int64_t last_cipher_block_size_ = 0; + int64_t plain_size_ = 0; + int64_t position_ = 0; + int64_t current_block_index_ = -1; + int64_t current_block_size_ = 0; +}; + +class AesGcmOutputStream : public PositionOutputStream { + public: + AesGcmOutputStream(std::unique_ptr target, + std::unique_ptr cipher, + std::vector aad_prefix) + : target_(std::move(target)), + cipher_(std::move(cipher)), + aad_prefix_(std::move(aad_prefix)), + plain_block_(kPlainBlock), + cipher_block_(kCipherBlock) {} + + Result Position() const override { + // Closing flushes the last block, which resets the in-block position, so the + // plaintext length is captured before that happens. + if (closed_) { + return final_position_; + } + return block_index_ * kPlainBlock + position_in_block_; + } + + Result StoredLength() const override { return target_->StoredLength(); } + + Status Write(std::span data) override { + if (closed_) { + return IOError("Writing to closed stream"); + } + ICEBERG_RETURN_UNEXPECTED(WriteHeaderIfNeeded()); + while (!data.empty()) { + int64_t n = std::min(kPlainBlock - position_in_block_, data.size()); + std::memcpy(plain_block_.data() + position_in_block_, data.data(), n); + position_in_block_ += n; + data = data.subspan(n); + if (position_in_block_ == kPlainBlock) { + ICEBERG_RETURN_UNEXPECTED(EncryptAndWriteBlock()); + } + } + return {}; + } + + Status Flush() override { return target_->Flush(); } + + Status Close() override { + if (closed_) { + return {}; + } + ICEBERG_RETURN_UNEXPECTED(WriteHeaderIfNeeded()); + ICEBERG_ASSIGN_OR_RAISE(final_position_, Position()); + // Like Java, an empty stream gets one empty block; otherwise only a non-empty + // partial block is written. + if (position_in_block_ > 0 || block_index_ == 0) { + ICEBERG_RETURN_UNEXPECTED(EncryptAndWriteBlock()); + } + closed_ = true; + return target_->Close(); + } + + private: + Status WriteHeaderIfNeeded() { + if (!header_written_) { + auto header = HeaderBytes(); + ICEBERG_RETURN_UNEXPECTED(target_->Write(header)); + header_written_ = true; + } + return {}; + } + + Status EncryptAndWriteBlock() { + if (last_block_written_) { + return Invalid("Cannot encrypt block: a partial block has already been written"); + } + if (block_index_ == std::numeric_limits::max()) { + return IOError("Cannot write block: exceeded Integer.MAX_VALUE blocks"); + } + if (position_in_block_ != kPlainBlock) { + last_block_written_ = true; + } + auto aad = BlockAad(aad_prefix_, static_cast(block_index_)); + auto out = std::span(cipher_block_).first(position_in_block_ + kBlockOverhead); + ICEBERG_RETURN_UNEXPECTED( + cipher_->Encrypt(std::span(plain_block_).first(position_in_block_), aad, out)); + ICEBERG_RETURN_UNEXPECTED(target_->Write(std::as_bytes(out))); + position_in_block_ = 0; + ++block_index_; + return {}; + } + + std::unique_ptr target_; + std::unique_ptr cipher_; + std::vector aad_prefix_; + std::vector plain_block_; + std::vector cipher_block_; + int64_t position_in_block_ = 0; + int64_t block_index_ = 0; + int64_t final_position_ = 0; + bool header_written_ = false; + bool last_block_written_ = false; + bool closed_ = false; +}; + +} // namespace + +int64_t AesGcmStream::PlaintextLength(int64_t encrypted_length) { + int64_t stream_length = encrypted_length - kHeaderLength; + if (stream_length <= 0) { + return 0; + } + int64_t num_full_blocks = stream_length / kCipherBlockSize; + int64_t last_bytes = stream_length - num_full_blocks * kCipherBlockSize; + int64_t last_plain = last_bytes == 0 ? 0 : last_bytes - kNonceLength - kGcmTagLength; + return num_full_blocks * kPlainBlockSize + std::max(last_plain, 0); +} + +AesGcmInputFile::AesGcmInputFile(std::unique_ptr source, + std::vector key, + std::vector aad_prefix, + int64_t encrypted_length) + : source_(std::move(source)), + key_(std::move(key)), + aad_prefix_(std::move(aad_prefix)), + encrypted_length_(encrypted_length) {} + +Result AesGcmInputFile::Size() const { + return AesGcmStream::PlaintextLength(encrypted_length_); +} + +Result> AesGcmInputFile::Open() { + if (encrypted_length_ < AesGcmStream::kMinStreamLength) { + return Invalid( + "Invalid encrypted stream: {} is shorter than the minimum possible stream length", + encrypted_length_); + } + ICEBERG_ASSIGN_OR_RAISE(auto cipher, AesGcmCipher::Make(key_)); + ICEBERG_ASSIGN_OR_RAISE(auto source, source_->Open()); + auto stream = std::make_unique(std::move(source), std::move(cipher), + aad_prefix_, encrypted_length_); + ICEBERG_RETURN_UNEXPECTED(stream->ValidateHeader()); + return stream; +} + +AesGcmOutputFile::AesGcmOutputFile(std::unique_ptr target, + std::vector key, + std::vector aad_prefix) + : target_(std::move(target)), + key_(std::move(key)), + aad_prefix_(std::move(aad_prefix)) {} + +Result> AesGcmOutputFile::Create() { + ICEBERG_ASSIGN_OR_RAISE(auto cipher, encryption::AesGcmCipher::Make(key_)); + ICEBERG_ASSIGN_OR_RAISE(auto target, target_->Create()); + return std::make_unique(std::move(target), std::move(cipher), + aad_prefix_); +} + +Result> AesGcmOutputFile::CreateOrOverwrite() { + ICEBERG_ASSIGN_OR_RAISE(auto cipher, encryption::AesGcmCipher::Make(key_)); + ICEBERG_ASSIGN_OR_RAISE(auto target, target_->CreateOrOverwrite()); + return std::make_unique(std::move(target), std::move(cipher), + aad_prefix_); +} + +} // namespace iceberg diff --git a/src/iceberg/encryption/aes_gcm_stream.h b/src/iceberg/encryption/aes_gcm_stream.h new file mode 100644 index 000000000..99404fa48 --- /dev/null +++ b/src/iceberg/encryption/aes_gcm_stream.h @@ -0,0 +1,89 @@ +/* + * 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_stream.h +/// AES GCM Stream (AGS1) files, see format/gcm-stream-spec.md. + +#include +#include +#include +#include +#include + +#include "iceberg/file_io.h" +#include "iceberg/iceberg_bundle_export.h" +#include "iceberg/result.h" + +namespace iceberg { + +/// \brief AES GCM Stream format constants. +struct AesGcmStream { + static constexpr int32_t kPlainBlockSize = 1024 * 1024; + static constexpr int32_t kNonceLength = 12; + static constexpr int32_t kGcmTagLength = 16; + static constexpr int32_t kCipherBlockSize = + kPlainBlockSize + kNonceLength + kGcmTagLength; + static constexpr std::array kMagic = {'A', 'G', 'S', '1'}; + static constexpr int32_t kHeaderLength = 8; // magic + plain block size (LE int32) + static constexpr int64_t kMinStreamLength = + kHeaderLength + kNonceLength + kGcmTagLength; + + /// \brief Plaintext length of a stream with the given encrypted length. + static int64_t PlaintextLength(int64_t encrypted_length); +}; + +/// \brief An InputFile that decrypts an AGS1 file. +/// +/// The encrypted length must come from a trusted source (the key metadata), so that +/// truncation of the file is detected. +class ICEBERG_BUNDLE_EXPORT AesGcmInputFile : public InputFile { + public: + AesGcmInputFile(std::unique_ptr source, std::vector key, + std::vector aad_prefix, int64_t encrypted_length); + + std::string_view location() const override { return source_->location(); } + Result Size() const override; + Result> Open() override; + + private: + std::unique_ptr source_; + std::vector key_; + std::vector aad_prefix_; + int64_t encrypted_length_; +}; + +/// \brief An OutputFile that writes an AGS1 file. +class ICEBERG_BUNDLE_EXPORT AesGcmOutputFile : public OutputFile { + public: + AesGcmOutputFile(std::unique_ptr target, std::vector key, + std::vector aad_prefix); + + std::string_view location() const override { return target_->location(); } + Result> Create() override; + Result> CreateOrOverwrite() override; + + private: + std::unique_ptr target_; + std::vector key_; + std::vector aad_prefix_; +}; + +} // namespace iceberg diff --git a/src/iceberg/encryption/in_memory_kms.cc b/src/iceberg/encryption/in_memory_kms.cc new file mode 100644 index 000000000..da891ff75 --- /dev/null +++ b/src/iceberg/encryption/in_memory_kms.cc @@ -0,0 +1,70 @@ +/* + * 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/in_memory_kms.h" + +#include "iceberg/encryption/aes_gcm_internal.h" +#include "iceberg/util/base64.h" +#include "iceberg/util/macros.h" + +namespace iceberg { + +void InMemoryKms::Register() { + KmsRegistry::Register(std::string(kName), + []() -> Result> { + return std::make_unique(); + }); +} + +Status InMemoryKms::Initialize( + const std::unordered_map& properties) { + for (const auto& [name, value] : properties) { + if (!name.starts_with(kKeyPropertyPrefix)) { + continue; + } + ICEBERG_ASSIGN_OR_RAISE(auto key, Base64::Decode(value)); + master_keys_.insert_or_assign(name.substr(kKeyPropertyPrefix.size()), + std::vector(key.begin(), key.end())); + } + return {}; +} + +Result*> InMemoryKms::MasterKey(std::string_view id) const { + auto it = master_keys_.find(std::string(id)); + if (it == master_keys_.end()) { + return NotFound("Master key {} is not found", id); + } + return &it->second; +} + +Result> InMemoryKms::WrapKey(std::span key, + std::string_view wrapping_key_id) { + ICEBERG_ASSIGN_OR_RAISE(auto master_key, MasterKey(wrapping_key_id)); + ICEBERG_ASSIGN_OR_RAISE(auto cipher, encryption::AesGcmCipher::Make(*master_key)); + return cipher->Encrypt(key, {}); +} + +Result> InMemoryKms::UnwrapKey(std::span wrapped_key, + std::string_view wrapping_key_id) { + ICEBERG_ASSIGN_OR_RAISE(auto master_key, MasterKey(wrapping_key_id)); + ICEBERG_ASSIGN_OR_RAISE(auto cipher, encryption::AesGcmCipher::Make(*master_key)); + return cipher->Decrypt(wrapped_key, {}); +} + +} // namespace iceberg diff --git a/src/iceberg/encryption/in_memory_kms.h b/src/iceberg/encryption/in_memory_kms.h new file mode 100644 index 000000000..b97025cac --- /dev/null +++ b/src/iceberg/encryption/in_memory_kms.h @@ -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. + */ + +#pragma once + +/// \file iceberg/encryption/in_memory_kms.h +/// KMS client with master keys held in memory, for tests and local development. + +#include +#include +#include +#include +#include +#include + +#include "iceberg/encryption/key_management_client.h" +#include "iceberg/iceberg_bundle_export.h" + +namespace iceberg { + +/// \brief KMS client with in-memory master keys. Not for production use. +/// +/// Wire compatible with Java's MemoryMockKMS (used by Java tests through UnitestKMS): +/// keys are wrapped with AES-GCM under the master key, as nonce | ciphertext | tag, +/// without AAD. +class ICEBERG_BUNDLE_EXPORT InMemoryKms : public KeyManagementClient { + public: + /// \brief The name used with `encryption.kms-impl` after RegisterInMemoryKms(). + static constexpr std::string_view kName = "in-memory"; + /// \brief Catalog property prefix for master keys: `encryption.in-memory-kms.key.` + /// = Base64 of the key bytes. + static constexpr std::string_view kKeyPropertyPrefix = "encryption.in-memory-kms.key."; + + InMemoryKms() = default; + explicit InMemoryKms(std::unordered_map> master_keys) + : master_keys_(std::move(master_keys)) {} + + /// \brief Register this client in KmsRegistry under kName. + static void Register(); + + /// \brief Add master keys from `encryption.in-memory-kms.key.` properties. + Status Initialize( + const std::unordered_map& properties) override; + + Result> WrapKey(std::span key, + std::string_view wrapping_key_id) override; + Result> UnwrapKey(std::span wrapped_key, + std::string_view wrapping_key_id) override; + + private: + Result*> MasterKey(std::string_view id) const; + + std::unordered_map> master_keys_; +}; + +} // namespace iceberg diff --git a/src/iceberg/encryption/key_management_client.cc b/src/iceberg/encryption/key_management_client.cc new file mode 100644 index 000000000..470717cc3 --- /dev/null +++ b/src/iceberg/encryption/key_management_client.cc @@ -0,0 +1,97 @@ +/* + * 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/key_management_client.h" + +#include + +#include "iceberg/util/macros.h" +#include "iceberg/util/string_util.h" + +namespace iceberg { + +namespace { + +struct RegistryState { + std::mutex mutex; + std::unordered_map factories; +}; + +RegistryState& State() { + static RegistryState state; + return state; +} + +} // namespace + +void KmsRegistry::Register(std::string name, Factory factory) { + auto& state = State(); + std::lock_guard lock(state.mutex); + state.factories.insert_or_assign(std::move(name), std::move(factory)); +} + +Result> KmsRegistry::Load( + std::string_view name, const Properties& properties) { + Factory factory; + { + auto& state = State(); + std::lock_guard lock(state.mutex); + auto it = state.factories.find(std::string(name)); + if (it == state.factories.end()) { + return NotFound( + "KMS client not found: {}. Built-in clients must be enabled at build time " + "(ICEBERG_KMS_AWS, ICEBERG_KMS_GCP, ICEBERG_KMS_AZURE); custom clients must be " + "registered with KmsRegistry::Register", + name); + } + factory = it->second; + } + ICEBERG_ASSIGN_OR_RAISE(auto client, factory()); + ICEBERG_PRECHECK(client != nullptr, "KMS client factory for {} returned null", name); + ICEBERG_RETURN_UNEXPECTED(client->Initialize(properties)); + return std::shared_ptr(std::move(client)); +} + +Result> KmsRegistry::FromCatalogProperties( + const Properties& properties) { + auto type_it = properties.find(std::string(kKmsType)); + auto impl_it = properties.find(std::string(kKmsImpl)); + // Like Java, a property that is present but empty counts as set, so that a + // misconfigured catalog fails instead of silently going unencrypted. + bool has_type = type_it != properties.end(); + bool has_impl = impl_it != properties.end(); + + if (has_type && has_impl) { + return InvalidArgument("Cannot set both KMS type ({}) and KMS impl ({})", + type_it->second, impl_it->second); + } + if (has_impl) { + return Load(impl_it->second, properties); + } + if (has_type) { + auto type = StringUtils::ToLower(type_it->second); + if (type != kKmsTypeAws && type != kKmsTypeGcp && type != kKmsTypeAzure) { + return InvalidArgument("Unsupported KMS type: {}", type_it->second); + } + return Load(type, properties); + } + return std::shared_ptr{}; +} + +} // namespace iceberg diff --git a/src/iceberg/encryption/key_management_client.h b/src/iceberg/encryption/key_management_client.h new file mode 100644 index 000000000..1f7aac81b --- /dev/null +++ b/src/iceberg/encryption/key_management_client.h @@ -0,0 +1,118 @@ +/* + * 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/key_management_client.h +/// Client of a key management service (KMS), and the registry that creates one from +/// catalog properties. + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "iceberg/iceberg_export.h" +#include "iceberg/result.h" + +namespace iceberg { + +/// \brief Wraps and unwraps secret keys with master keys kept in a KMS. +/// +/// Mirrors Java's org.apache.iceberg.encryption.KeyManagementClient. The table master +/// key (table property `encryption.key-id`) never leaves the KMS; it only wraps and +/// unwraps the table's key encryption keys. +class ICEBERG_EXPORT KeyManagementClient { + public: + /// \brief A generated key and its wrapped form. + struct GeneratedKey { + std::vector key; + std::vector wrapped_key; + }; + + virtual ~KeyManagementClient() = default; + + /// \brief Initialize the client with the catalog properties. + virtual Status Initialize( + const std::unordered_map& properties) { + return {}; + } + + /// \brief Wrap (encrypt) a secret key with the master key `wrapping_key_id`. + /// + /// The result may include KMS-specific metadata needed to unwrap it. + virtual Result> WrapKey(std::span key, + std::string_view wrapping_key_id) = 0; + + /// \brief Unwrap a key returned by WrapKey. + virtual Result> UnwrapKey(std::span wrapped_key, + std::string_view wrapping_key_id) = 0; + + /// \brief Whether the KMS can generate keys (GenerateKey). + virtual bool SupportsKeyGeneration() const { return false; } + + /// \brief Generate a new key in the KMS, returned in plain and wrapped form. + virtual Result GenerateKey(std::string_view wrapping_key_id) { + return NotSupported("Key generation is not supported by this KMS client"); + } +}; + +/// \brief Creates KMS clients by name. +/// +/// Catalogs create their KMS client from the catalog properties, with the same +/// property names as Java: +/// - `encryption.kms-type`: a built-in client, `aws`, `gcp` or `azure`; +/// - `encryption.kms-impl`: the name a custom client was registered under. +/// +/// Java also accepts the type `aliyun`, which has no client here; register one with +/// Register and select it through `encryption.kms-impl`. +/// +/// Built-in clients register themselves when their library is linked and initialized; +/// applications register custom clients before creating catalogs. +class ICEBERG_EXPORT KmsRegistry { + public: + static constexpr std::string_view kKmsType = "encryption.kms-type"; + static constexpr std::string_view kKmsImpl = "encryption.kms-impl"; + static constexpr std::string_view kKmsTypeAws = "aws"; + static constexpr std::string_view kKmsTypeGcp = "gcp"; + static constexpr std::string_view kKmsTypeAzure = "azure"; + + using Properties = std::unordered_map; + using Factory = std::function>()>; + + /// \brief Register a client factory under `name`, replacing any previous one. + static void Register(std::string name, Factory factory); + + /// \brief Create and initialize the client registered under `name`. + static Result> Load(std::string_view name, + const Properties& properties); + + /// \brief Create the KMS client configured by catalog properties. + /// + /// \return the initialized client, or nullptr if neither `encryption.kms-type` nor + /// `encryption.kms-impl` is set. + static Result> FromCatalogProperties( + const Properties& properties); +}; + +} // namespace iceberg diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt index a86bc9087..a1b8b2c1c 100644 --- a/src/iceberg/test/CMakeLists.txt +++ b/src/iceberg/test/CMakeLists.txt @@ -116,7 +116,7 @@ add_iceberg_test(expression_test residual_evaluator_test.cc strict_metrics_evaluator_test.cc) -add_iceberg_test(encryption_test SOURCES standard_key_metadata_test.cc) +add_iceberg_test(encryption_test SOURCES kms_test.cc standard_key_metadata_test.cc) add_iceberg_test(json_serde_test SOURCES @@ -228,6 +228,10 @@ if(ICEBERG_BUILD_BUNDLE) target_sources(parquet_test PRIVATE parquet_encryption_test.cc) endif() + if(ICEBERG_ENCRYPTION) + add_iceberg_test(crypto_test USE_BUNDLE SOURCES aes_gcm_test.cc) + endif() + add_iceberg_test(scan_test USE_BUNDLE SOURCES @@ -408,6 +412,10 @@ if(ICEBERG_CXX20_COMPAT) ) target_compile_definitions(cpp20_compatibility_test PRIVATE ICEBERG_PUBLIC_HEADERS_BUNDLE) + if(ICEBERG_ENCRYPTION) + target_compile_definitions(cpp20_compatibility_test + PRIVATE ICEBERG_PUBLIC_HEADERS_ENCRYPTION) + endif() endif() if(ICEBERG_BUILD_REST) target_link_libraries(cpp20_compatibility_test diff --git a/src/iceberg/test/aes_gcm_test.cc b/src/iceberg/test/aes_gcm_test.cc new file mode 100644 index 000000000..8c5a9af5e --- /dev/null +++ b/src/iceberg/test/aes_gcm_test.cc @@ -0,0 +1,320 @@ +/* + * 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 +#include +#include +#include +#include +#include +#include + +#include + +#include "iceberg/arrow/arrow_io_util.h" +#include "iceberg/encryption/aes_gcm_internal.h" +#include "iceberg/encryption/aes_gcm_stream.h" +#include "iceberg/encryption/in_memory_kms.h" +#include "iceberg/encryption/standard_key_metadata.h" +#include "iceberg/test/matchers.h" +#include "iceberg/test/temp_file_test_base.h" + +namespace iceberg { + +namespace { + +using encryption::AesGcmCipher; + +/// \brief Deterministic test bytes: byte i is (i * 31 + seed). +std::vector TestBytes(size_t length, int seed) { + std::vector bytes(length); + for (size_t i = 0; i < length; ++i) { + bytes[i] = static_cast(static_cast(i) * 31 + seed); + } + return bytes; +} + +std::span AsBytes(std::string_view s) { + return {reinterpret_cast(s.data()), s.size()}; +} + +// The fixtures below were produced by Iceberg Java. It used TestBytes(16, 1) as the +// data key, TestBytes(16, 2) as the file AAD prefix and TestBytes(16, 3) as the key +// encryption key. +const std::vector kDataKey = TestBytes(16, 1); +const std::vector kAadPrefix = TestBytes(16, 2); +const std::vector kKek = TestBytes(16, 3); + +// StandardEncryptionManager encrypts manifest list key metadata under the key +// encryption key, with the KEK timestamp as AAD. +constexpr std::string_view kKekTimestamp = "1791455219054"; + +// Manifest list key metadata encrypted by Java under kKek with kKekTimestamp +// as AAD: encryption key TestBytes(16, 1), AAD prefix TestBytes(16, 2), +// file length 4242. +const std::vector kJavaManifestListKeyMetadata = { + 0xf2, 0x80, 0x57, 0xac, 0x14, 0xdc, 0x8d, 0xa7, 0xb2, 0x25, 0x58, 0x24, 0xa3, 0xe1, + 0x88, 0xb0, 0xde, 0x2e, 0x94, 0x09, 0x9b, 0x7d, 0xe4, 0x7d, 0x0e, 0xd6, 0xb0, 0x7a, + 0xb1, 0x86, 0x7b, 0xca, 0x61, 0x11, 0x2a, 0x11, 0x4e, 0xc1, 0xa8, 0x23, 0xd8, 0xa5, + 0xca, 0x43, 0x9c, 0x33, 0x6a, 0xe9, 0x6b, 0xf2, 0xc6, 0xdb, 0xa8, 0xe0, 0x2e, 0x7f, + 0xc0, 0xcc, 0x0e, 0xf0, 0xb7, 0xec, 0xb9, 0x1f, 0x06, 0x3a, 0xe7}; +// AES GCM stream of TestBytes(0, 5) written by Java's AesGcmOutputStream +// with kDataKey and kAadPrefix. +const std::vector kJavaAgs1Empty = { + 0x41, 0x47, 0x53, 0x31, 0x00, 0x00, 0x10, 0x00, 0x92, 0xd6, 0x53, 0x8c, + 0x2b, 0x71, 0x45, 0x37, 0x37, 0x86, 0x76, 0x8f, 0xef, 0xd4, 0xdc, 0x98, + 0x15, 0xf1, 0x93, 0x9d, 0x6c, 0x56, 0x4b, 0xae, 0x24, 0xff, 0xdc, 0xb4}; +// AES GCM stream of TestBytes(1, 5) written by Java's AesGcmOutputStream +// with kDataKey and kAadPrefix. +const std::vector kJavaAgs1OneByte = { + 0x41, 0x47, 0x53, 0x31, 0x00, 0x00, 0x10, 0x00, 0xf8, 0xde, 0xe0, 0x44, 0x00, + 0x8d, 0xbb, 0x2b, 0x6b, 0x29, 0x58, 0xb2, 0x70, 0x62, 0x46, 0xbb, 0xb1, 0xf8, + 0x76, 0x29, 0x69, 0x86, 0x14, 0x6c, 0x08, 0x98, 0x84, 0xdc, 0x3b}; +// AES GCM stream of TestBytes(100, 5) written by Java's AesGcmOutputStream +// with kDataKey and kAadPrefix. +const std::vector kJavaAgs1Hundred = { + 0x41, 0x47, 0x53, 0x31, 0x00, 0x00, 0x10, 0x00, 0x02, 0xb0, 0x1c, 0x1f, 0x7e, 0x37, + 0xeb, 0x85, 0xd3, 0x4b, 0x14, 0x75, 0x91, 0x83, 0xcf, 0x68, 0xde, 0xdd, 0xe0, 0xff, + 0xd3, 0xe0, 0xeb, 0x31, 0x10, 0x4b, 0x95, 0x86, 0x5c, 0x67, 0x1f, 0x68, 0xdf, 0x24, + 0xbd, 0x0a, 0x07, 0xf1, 0x06, 0xc7, 0x13, 0x1e, 0x42, 0xd4, 0x7a, 0xff, 0x87, 0x32, + 0xc3, 0xe5, 0xea, 0x75, 0x6a, 0x53, 0xaa, 0xab, 0xab, 0xe0, 0x81, 0x2c, 0x71, 0x4c, + 0xb4, 0x89, 0x32, 0x89, 0xc6, 0xbb, 0x84, 0x5d, 0x58, 0x93, 0x6b, 0x80, 0xbd, 0xa3, + 0xe6, 0xc3, 0xef, 0x53, 0x83, 0xa4, 0x00, 0x9c, 0x30, 0x02, 0xd4, 0xca, 0x7e, 0x0e, + 0xb3, 0xa3, 0xff, 0x4c, 0x96, 0xe3, 0x31, 0xa1, 0x6a, 0xc1, 0xf2, 0xb7, 0xa0, 0x40, + 0xc4, 0x0b, 0x5c, 0x45, 0x09, 0xb7, 0x35, 0xda, 0x6c, 0x03, 0xce, 0xd5, 0x59, 0xd0, + 0x47, 0xf9, 0x0c, 0x9d, 0x1f, 0x89, 0x2f, 0x1b, 0xcb, 0x44}; + +Result> ReadAll(InputFile& file) { + ICEBERG_ASSIGN_OR_RAISE(auto size, file.Size()); + ICEBERG_ASSIGN_OR_RAISE(auto stream, file.Open()); + std::vector out(size); + if (size > 0) { + ICEBERG_RETURN_UNEXPECTED( + stream->ReadFully(0, std::as_writable_bytes(std::span(out)))); + } + return out; +} + +} // namespace + +TEST(AesGcmCipherTest, RoundTrip) { + for (size_t key_length : {16, 24, 32}) { + ICEBERG_UNWRAP_OR_FAIL(auto cipher, AesGcmCipher::Make(TestBytes(key_length, 1))); + auto plaintext = TestBytes(100, 2); + ICEBERG_UNWRAP_OR_FAIL(auto ciphertext, cipher->Encrypt(plaintext, AsBytes("aad"))); + EXPECT_EQ(ciphertext.size(), plaintext.size() + 28); + ICEBERG_UNWRAP_OR_FAIL(auto decrypted, cipher->Decrypt(ciphertext, AsBytes("aad"))); + EXPECT_EQ(decrypted, plaintext); + + // Random nonces: encrypting twice gives different ciphertexts + ICEBERG_UNWRAP_OR_FAIL(auto again, cipher->Encrypt(plaintext, AsBytes("aad"))); + EXPECT_NE(again, ciphertext); + + // Wrong AAD, tampered ciphertext and truncated ciphertext fail authentication + EXPECT_THAT(cipher->Decrypt(ciphertext, AsBytes("other")), + IsError(ErrorKind::kInvalid)); + ciphertext[20] ^= 1; + EXPECT_THAT(cipher->Decrypt(ciphertext, AsBytes("aad")), + IsError(ErrorKind::kInvalid)); + EXPECT_THAT(cipher->Decrypt(std::span(ciphertext).first(27), AsBytes("aad")), + IsError(ErrorKind::kInvalid)); + } + EXPECT_THAT(AesGcmCipher::Make(TestBytes(15, 1)), IsError(ErrorKind::kInvalidArgument)); +} + +TEST(AesGcmCipherTest, DecryptsJavaManifestListKeyMetadata) { + ICEBERG_UNWRAP_OR_FAIL(auto cipher, AesGcmCipher::Make(kKek)); + ICEBERG_UNWRAP_OR_FAIL(auto decrypted, cipher->Decrypt(kJavaManifestListKeyMetadata, + AsBytes(kKekTimestamp))); + ICEBERG_UNWRAP_OR_FAIL(auto metadata, StandardKeyMetadata::Parse(decrypted)); + EXPECT_EQ(metadata.encryption_key, TestBytes(16, 1)); + EXPECT_EQ(metadata.aad_prefix, TestBytes(16, 2)); + EXPECT_EQ(metadata.file_length, 4242); +} + +class AesGcmStreamTest : public TempFileTestBase { + protected: + void SetUp() override { + TempFileTestBase::SetUp(); + io_ = std::shared_ptr(arrow::MakeLocalFileIO()); + } + + // Write `plain` as an AGS1 stream and return the stored length. + int64_t Write(const std::string& path, std::span plain, + const std::vector& key, const std::vector& aad, + size_t chunk = 1 << 20) { + auto out = io_->NewOutputFile(path); + EXPECT_THAT(out, IsOk()); + AesGcmOutputFile file(std::move(out.value()), key, aad); + auto stream = file.CreateOrOverwrite(); + EXPECT_THAT(stream, IsOk()); + for (size_t pos = 0; pos < plain.size(); pos += chunk) { + auto part = plain.subspan(pos, std::min(chunk, plain.size() - pos)); + EXPECT_THAT((*stream)->Write(std::as_bytes(part)), IsOk()); + } + EXPECT_EQ((*stream)->Position().value(), static_cast(plain.size())); + EXPECT_THAT((*stream)->Close(), IsOk()); + // Java's AesGcmOutputStream keeps reporting the plaintext length once closed + EXPECT_EQ((*stream)->Position().value(), static_cast(plain.size())); + auto stored = (*stream)->StoredLength().value(); + EXPECT_EQ(stored, static_cast(std::filesystem::file_size(path))); + return stored; + } + + std::unique_ptr Open(const std::string& path, + const std::vector& key, + const std::vector& aad, int64_t length) { + auto in = io_->NewInputFile(path); + EXPECT_THAT(in, IsOk()); + return std::make_unique(std::move(in.value()), key, aad, length); + } + + // Materialize an inlined Java fixture as a file the reader can open. + std::string WriteRaw(const std::vector& bytes) { + auto path = CreateNewTempFilePath(); + std::ofstream out(path, std::ios::binary); + out.write(reinterpret_cast(bytes.data()), + static_cast(bytes.size())); + out.close(); + return path; + } + + std::shared_ptr io_; +}; + +TEST_F(AesGcmStreamTest, ReadsJavaVectors) { + // Streams of TestBytes(0, 5), TestBytes(1, 5) and TestBytes(100, 5) written by Java + const std::vector*>> vectors = { + {0, &kJavaAgs1Empty}, {1, &kJavaAgs1OneByte}, {100, &kJavaAgs1Hundred}}; + for (const auto& [length, encrypted] : vectors) { + auto path = WriteRaw(*encrypted); + auto file = Open(path, kDataKey, kAadPrefix, static_cast(encrypted->size())); + ICEBERG_UNWRAP_OR_FAIL(auto plain, ReadAll(*file)); + EXPECT_EQ(plain, TestBytes(length, 5)) << "plaintext length " << length; + } +} + +TEST_F(AesGcmStreamTest, WritesJavaCompatibleLayout) { + // Same plaintext, key and AAD prefix as Java's 100-byte vector: same header and + // length (the ciphertext differs only by the random nonce). + auto path = CreateNewTempFilePath(); + auto stored = Write(path, TestBytes(100, 5), kDataKey, kAadPrefix); + ASSERT_EQ(stored, static_cast(kJavaAgs1Hundred.size())); + std::ifstream in(path, std::ios::binary); + std::vector header(8); + in.read(reinterpret_cast(header.data()), 8); + EXPECT_EQ(header, + std::vector(kJavaAgs1Hundred.begin(), kJavaAgs1Hundred.begin() + 8)); +} + +TEST_F(AesGcmStreamTest, RoundTripAcrossBlocks) { + auto key = TestBytes(32, 7); + auto aad = TestBytes(16, 9); + constexpr size_t kBlock = AesGcmStream::kPlainBlockSize; + for (size_t size : + {size_t{0}, size_t{1}, kBlock - 1, kBlock, kBlock + 1, 2 * kBlock + 17}) { + auto path = CreateNewTempFilePath(); + auto plain = TestBytes(size, 5); + // Odd write sizes exercise writes that straddle blocks + int64_t stored = Write(path, plain, key, aad, /*chunk=*/333331); + size_t blocks = size == 0 ? 1 : (size + kBlock - 1) / kBlock; + EXPECT_EQ(stored, static_cast(8 + size + blocks * 28)); + EXPECT_EQ(AesGcmStream::PlaintextLength(stored), static_cast(size)); + + auto file = Open(path, key, aad, stored); + ICEBERG_UNWRAP_OR_FAIL(auto read, ReadAll(*file)); + EXPECT_EQ(read, plain); + + if (size > kBlock + 10) { + // Random access across a block boundary + ICEBERG_UNWRAP_OR_FAIL(auto stream, file->Open()); + std::vector window(10); + ASSERT_THAT(stream->ReadFully(kBlock - 5, window), IsOk()); + EXPECT_EQ(std::memcmp(window.data(), plain.data() + kBlock - 5, 10), 0); + } + } +} + +TEST_F(AesGcmStreamTest, DetectsTampering) { + auto key = TestBytes(16, 7); + auto aad = TestBytes(16, 9); + auto path = CreateNewTempFilePath(); + auto plain = TestBytes(AesGcmStream::kPlainBlockSize + 100, 5); + int64_t stored = Write(path, plain, key, aad); + + // Wrong key or AAD prefix (a file swapped in from another manifest entry) + EXPECT_THAT(ReadAll(*Open(path, TestBytes(16, 8), aad, stored)), + IsError(ErrorKind::kInvalid)); + EXPECT_THAT(ReadAll(*Open(path, key, TestBytes(16, 10), stored)), + IsError(ErrorKind::kInvalid)); + + // Truncation: the trusted length from the key metadata no longer matches the file + std::filesystem::resize_file(path, 8 + AesGcmStream::kCipherBlockSize); + EXPECT_THAT(ReadAll(*Open(path, key, aad, stored)), IsError(ErrorKind::kIOError)); + // AGS1 blocks do not mark the end of the stream: a truncated file read with its own + // (untrusted) length still decrypts. This is why readers must take the length from + // the key metadata. + auto truncated = Open(path, key, aad, 8 + AesGcmStream::kCipherBlockSize); + ICEBERG_UNWRAP_OR_FAIL(auto first_block, ReadAll(*truncated)); + EXPECT_EQ(first_block.size(), AesGcmStream::kPlainBlockSize); + + // Bad magic + std::filesystem::resize_file(path, 0); + std::ofstream(path, std::ios::binary) << "XXXX0000" << std::string(28, '\0'); + EXPECT_THAT(Open(path, key, aad, 36)->Open(), IsError(ErrorKind::kInvalid)); + // Shorter than an empty stream + EXPECT_THAT(Open(path, key, aad, 35)->Open(), IsError(ErrorKind::kInvalid)); +} + +// The key encryption key of the Java fixture table, wrapped by Java's UnitestKMS under +// its keyA master key. +const std::vector kJavaKekWrappedByKeyA = { + 0x04, 0x0d, 0xdf, 0x90, 0xe8, 0x41, 0xa3, 0xff, 0xf7, 0xec, 0x77, + 0xf1, 0x04, 0x0d, 0x65, 0x50, 0x46, 0xd2, 0xae, 0x0a, 0xef, 0xde, + 0x4d, 0x3a, 0xa2, 0xf5, 0x06, 0xca, 0x2e, 0xf4, 0x00, 0x83, 0x4b, + 0xbe, 0xf0, 0x94, 0xf6, 0x20, 0x94, 0x76, 0x1c, 0x11, 0xf6, 0x53}; + +// Java's UnitestKMS.MASTER_KEY1, the master key it registers under "keyA". +constexpr std::string_view kJavaMasterKeyA = "0123456789012345"; + +TEST(InMemoryKmsTest, UnwrapsJavaWrappedKey) { + auto master_key = AsBytes(kJavaMasterKeyA); + InMemoryKms kms({{"keyA", std::vector(master_key.begin(), master_key.end())}}); + ICEBERG_UNWRAP_OR_FAIL(auto kek, kms.UnwrapKey(kJavaKekWrappedByKeyA, "keyA")); + EXPECT_EQ(kek.size(), 16); + + ICEBERG_UNWRAP_OR_FAIL(auto wrapped, kms.WrapKey(kek, "keyA")); + ICEBERG_UNWRAP_OR_FAIL(auto unwrapped, kms.UnwrapKey(wrapped, "keyA")); + EXPECT_EQ(unwrapped, kek); + EXPECT_THAT(kms.UnwrapKey(wrapped, "keyB"), IsError(ErrorKind::kNotFound)); +} + +TEST(InMemoryKmsTest, RegistryAndProperties) { + InMemoryKms::Register(); + ICEBERG_UNWRAP_OR_FAIL( + auto kms, KmsRegistry::FromCatalogProperties( + {{"encryption.kms-impl", "in-memory"}, + // Base64 of "0123456789012345", UnitestKMS's keyA + {"encryption.in-memory-kms.key.keyA", "MDEyMzQ1Njc4OTAxMjM0NQ=="}})); + ASSERT_NE(kms, nullptr); + ICEBERG_UNWRAP_OR_FAIL(auto kek, kms->UnwrapKey(kJavaKekWrappedByKeyA, "keyA")); + EXPECT_EQ(kek.size(), 16); +} + +} // namespace iceberg diff --git a/src/iceberg/test/kms_test.cc b/src/iceberg/test/kms_test.cc new file mode 100644 index 000000000..f32cb742b --- /dev/null +++ b/src/iceberg/test/kms_test.cc @@ -0,0 +1,104 @@ +/* + * 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 + +#include "iceberg/encryption/key_management_client.h" +#include "iceberg/test/matchers.h" + +namespace iceberg { + +namespace { + +/// Records the properties it was initialized with; "wraps" by reversing the bytes. +class FakeKms : public KeyManagementClient { + public: + Status Initialize(const std::unordered_map& props) override { + properties = props; + return {}; + } + Result> WrapKey(std::span key, + std::string_view) override { + return std::vector(key.rbegin(), key.rend()); + } + Result> UnwrapKey(std::span wrapped, + std::string_view) override { + return std::vector(wrapped.rbegin(), wrapped.rend()); + } + + std::unordered_map properties; +}; + +void RegisterFake(const std::string& name) { + KmsRegistry::Register(name, []() -> Result> { + return std::make_unique(); + }); +} + +} // namespace + +TEST(KmsRegistryTest, NotConfigured) { + ICEBERG_UNWRAP_OR_FAIL(auto client, KmsRegistry::FromCatalogProperties({})); + EXPECT_EQ(client, nullptr); +} + +TEST(KmsRegistryTest, LoadsCustomImplementation) { + RegisterFake("fake-kms"); + KmsRegistry::Properties properties{{"encryption.kms-impl", "fake-kms"}, + {"fake.endpoint", "localhost"}}; + ICEBERG_UNWRAP_OR_FAIL(auto client, KmsRegistry::FromCatalogProperties(properties)); + ASSERT_NE(client, nullptr); + auto* fake = dynamic_cast(client.get()); + ASSERT_NE(fake, nullptr); + // Initialized with the catalog properties + EXPECT_EQ(fake->properties.at("fake.endpoint"), "localhost"); + + std::vector key{1, 2, 3}; + ICEBERG_UNWRAP_OR_FAIL(auto wrapped, client->WrapKey(key, "master")); + ICEBERG_UNWRAP_OR_FAIL(auto unwrapped, client->UnwrapKey(wrapped, "master")); + EXPECT_EQ(unwrapped, key); + EXPECT_FALSE(client->SupportsKeyGeneration()); + EXPECT_THAT(client->GenerateKey("master"), IsError(ErrorKind::kNotSupported)); +} + +TEST(KmsRegistryTest, LoadsBuiltInType) { + // Built-in types resolve to the client registered under the type name. The registry + // is process-wide, so this shadows a real "aws" client for the rest of the binary. + RegisterFake("aws"); + ICEBERG_UNWRAP_OR_FAIL( + auto client, KmsRegistry::FromCatalogProperties({{"encryption.kms-type", "AWS"}})); + EXPECT_NE(dynamic_cast(client.get()), nullptr); +} + +TEST(KmsRegistryTest, InvalidConfiguration) { + EXPECT_THAT(KmsRegistry::FromCatalogProperties( + {{"encryption.kms-type", "aws"}, {"encryption.kms-impl", "fake-kms"}}), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(KmsRegistry::FromCatalogProperties({{"encryption.kms-type", "vault"}}), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(KmsRegistry::FromCatalogProperties({{"encryption.kms-impl", "missing"}}), + IsError(ErrorKind::kNotFound)); + // Set but empty is a misconfiguration, not "no KMS", as in Java + EXPECT_THAT(KmsRegistry::FromCatalogProperties({{"encryption.kms-type", ""}}), + IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(KmsRegistry::FromCatalogProperties({{"encryption.kms-impl", ""}}), + IsError(ErrorKind::kNotFound)); +} + +} // namespace iceberg