wgtmac commented on code in PR #616: URL: https://github.com/apache/iceberg-cpp/pull/616#discussion_r3279131049
########## src/iceberg/catalog/rest/auth/sigv4_auth_manager.cc: ########## @@ -0,0 +1,407 @@ +/* + * 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/catalog/rest/auth/auth_manager_internal.h" +#include "iceberg/catalog/rest/auth/aws_sdk.h" +#include "iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h" + +#ifdef ICEBERG_SIGV4 + +# include <atomic> +# include <mutex> +# include <sstream> + +# include <aws/core/Aws.h> +# include <aws/core/auth/AWSAuthSigner.h> +# include <aws/core/auth/AWSCredentialsProvider.h> +# include <aws/core/auth/AWSCredentialsProviderChain.h> +# include <aws/core/client/ClientConfiguration.h> +# include <aws/core/http/standard/StandardHttpRequest.h> +# include <aws/core/utils/HashingUtils.h> + +# include "iceberg/catalog/rest/auth/auth_managers.h" +# include "iceberg/catalog/rest/auth/auth_properties.h" +# include "iceberg/util/macros.h" +# include "iceberg/util/string_util.h" + +namespace iceberg::rest::auth { + +namespace { + +class AwsSdkLifecycle { + public: + static AwsSdkLifecycle& Instance() { + static AwsSdkLifecycle instance; + return instance; + } + + Status Initialize() { + std::lock_guard<std::mutex> lock(mutex_); + auto s = state_.load(); + if (s == State::kInitialized) return {}; + if (s == State::kFinalized) { + return InvalidArgument("AWS SDK has already been finalized; cannot reinitialize"); + } + Aws::InitAPI(options_); + state_.store(State::kInitialized); + return {}; + } + + Status Finalize() { + std::lock_guard<std::mutex> lock(mutex_); + if (state_.load() != State::kInitialized) return {}; + auto live = active_session_count_.load(); + if (live != 0) { + return Invalid( + "Cannot finalize AWS SDK while {} SigV4 auth session(s) are still alive", live); + } + Aws::ShutdownAPI(options_); + state_.store(State::kFinalized); + return {}; + } + + Status EnsureInitialized() { + if (state_.load() == State::kInitialized) return {}; + return Initialize(); + } + + bool IsInitialized() const { return state_.load() == State::kInitialized; } + bool IsFinalized() const { return state_.load() == State::kFinalized; } + + void IncrementSessionCount() { + active_session_count_.fetch_add(1, std::memory_order_relaxed); + } + void DecrementSessionCount() { + active_session_count_.fetch_sub(1, std::memory_order_relaxed); + } + + private: + enum class State : uint8_t { kUninitialized, kInitialized, kFinalized }; + + AwsSdkLifecycle() = default; + + std::atomic<State> state_{State::kUninitialized}; + std::mutex mutex_; + Aws::SDKOptions options_; + std::atomic<size_t> active_session_count_{0}; +}; + +Aws::Http::HttpMethod ToAwsMethod(HttpMethod method) { + switch (method) { + case HttpMethod::kGet: + return Aws::Http::HttpMethod::HTTP_GET; + case HttpMethod::kPost: + return Aws::Http::HttpMethod::HTTP_POST; + case HttpMethod::kPut: + return Aws::Http::HttpMethod::HTTP_PUT; + case HttpMethod::kDelete: + return Aws::Http::HttpMethod::HTTP_DELETE; + case HttpMethod::kHead: + return Aws::Http::HttpMethod::HTTP_HEAD; + } + return Aws::Http::HttpMethod::HTTP_GET; +} + +std::unordered_map<std::string, std::string> MergeProperties( + const std::unordered_map<std::string, std::string>& base, + const std::unordered_map<std::string, std::string>& overrides) { + auto merged = base; + for (const auto& [key, value] : overrides) { + merged.insert_or_assign(key, value); + } + return merged; +} + +/// Matches Java RESTSigV4AuthSession: canonical headers carry +/// Base64(SHA256(body)), canonical request trailer uses hex. +class RestSigV4Signer : public Aws::Client::AWSAuthV4Signer { + public: + RestSigV4Signer(const std::shared_ptr<Aws::Auth::AWSCredentialsProvider>& creds, + const char* service_name, const Aws::String& region) + : Aws::Client::AWSAuthV4Signer(creds, service_name, region, + PayloadSigningPolicy::Always, + /*urlEscapePath=*/false) { + // Skip the signer's hex overwrite of x-amz-content-sha256 so canonical + // headers see the caller's Base64; ComputePayloadHash still feeds hex + // into the canonical request trailer. + m_includeSha256HashHeader = false; + } +}; + +} // namespace + +// ---- SigV4AuthSession ---- + +SigV4AuthSession::SigV4AuthSession( + std::shared_ptr<AuthSession> delegate, std::string signing_region, + std::string signing_name, + std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider) + : delegate_(std::move(delegate)), + signing_region_(std::move(signing_region)), + signing_name_(std::move(signing_name)), + credentials_provider_(std::move(credentials_provider)), + signer_(std::make_unique<RestSigV4Signer>( + credentials_provider_, signing_name_.c_str(), signing_region_.c_str())) { + AwsSdkLifecycle::Instance().IncrementSessionCount(); +} + +SigV4AuthSession::~SigV4AuthSession() { + AwsSdkLifecycle::Instance().DecrementSessionCount(); +} + +Result<HttpRequest> SigV4AuthSession::Authenticate(const HttpRequest& request) { + ICEBERG_ASSIGN_OR_RAISE(auto delegate_request, delegate_->Authenticate(request)); + const auto& original_headers = delegate_request.headers; + + std::unordered_map<std::string, std::string> signing_headers; + for (const auto& [name, value] : original_headers) { + if (StringUtils::EqualsIgnoreCase(name, "Authorization")) { + signing_headers[std::string(kRelocatedHeaderPrefix) + name] = value; + } else { + signing_headers[name] = value; + } + } + + Aws::Http::URI aws_uri(delegate_request.url.c_str()); + auto aws_request = std::make_shared<Aws::Http::Standard::StandardHttpRequest>( + aws_uri, ToAwsMethod(delegate_request.method)); + for (const auto& [name, value] : signing_headers) { + aws_request->SetHeaderValue(Aws::String(name.c_str()), Aws::String(value.c_str())); + } + + // Empty body: hex EMPTY_BODY_SHA256 (Java parity workaround for the signer + // computing an invalid checksum on empty bodies). Non-empty: Base64. + if (delegate_request.body.empty()) { + aws_request->SetHeaderValue("x-amz-content-sha256", Aws::String(kEmptyBodySha256)); + } else { + auto body_stream = + Aws::MakeShared<std::stringstream>("SigV4Body", delegate_request.body); + aws_request->AddContentBody(body_stream); + auto sha256 = Aws::Utils::HashingUtils::CalculateSHA256( + Aws::String(delegate_request.body.data(), delegate_request.body.size())); + aws_request->SetHeaderValue("x-amz-content-sha256", + Aws::Utils::HashingUtils::Base64Encode(sha256)); + } + + if (!signer_->SignRequest(*aws_request)) { + return std::unexpected<Error>(Error{.kind = ErrorKind::kAuthenticationFailed, + .message = "SigV4 signing failed"}); + } + + HttpRequest signed_request{.method = delegate_request.method, + .url = std::move(delegate_request.url), + .headers = {}, + .body = std::move(delegate_request.body)}; + for (const auto& [aws_name, aws_value] : aws_request->GetHeaders()) { + std::string name(aws_name.c_str(), aws_name.size()); + std::string value(aws_value.c_str(), aws_value.size()); + for (const auto& [orig_name, orig_value] : original_headers) { + if (StringUtils::EqualsIgnoreCase(orig_name, name) && orig_value != value) { + signed_request.headers[std::string(kRelocatedHeaderPrefix) + orig_name] = + orig_value; + break; + } + } + signed_request.headers[std::move(name)] = std::move(value); + } + + return signed_request; +} + +Status SigV4AuthSession::Close() { return delegate_->Close(); } + +// ---- SigV4AuthManager ---- + +SigV4AuthManager::SigV4AuthManager(std::unique_ptr<AuthManager> delegate) + : delegate_(std::move(delegate)) {} + +SigV4AuthManager::~SigV4AuthManager() = default; + +Result<std::shared_ptr<AuthSession>> SigV4AuthManager::InitSession( + HttpClient& init_client, + const std::unordered_map<std::string, std::string>& properties) { + ICEBERG_RETURN_UNEXPECTED(AwsSdkLifecycle::Instance().EnsureInitialized()); + ICEBERG_ASSIGN_OR_RAISE(auto delegate_session, + delegate_->InitSession(init_client, properties)); + return WrapSession(std::move(delegate_session), properties); +} + +Result<std::shared_ptr<AuthSession>> SigV4AuthManager::CatalogSession( + HttpClient& shared_client, + const std::unordered_map<std::string, std::string>& properties) { + ICEBERG_RETURN_UNEXPECTED(AwsSdkLifecycle::Instance().EnsureInitialized()); + catalog_properties_ = properties; + ICEBERG_ASSIGN_OR_RAISE(auto delegate_session, + delegate_->CatalogSession(shared_client, properties)); + return WrapSession(std::move(delegate_session), properties); +} + +// Contextual and table sessions both merge against the stored catalog +// properties, matching Java's RESTSigV4AuthManager. Contextual overrides do +// not propagate into child table sessions; the two derivations are +// independent dimensions on top of the catalog baseline. + +Result<std::shared_ptr<AuthSession>> SigV4AuthManager::ContextualSession( + const std::unordered_map<std::string, std::string>& context, + std::shared_ptr<AuthSession> parent) { + auto sigv4_parent = std::dynamic_pointer_cast<SigV4AuthSession>(std::move(parent)); + ICEBERG_PRECHECK(sigv4_parent != nullptr, + "SigV4AuthManager parent must be a SigV4AuthSession"); + + ICEBERG_ASSIGN_OR_RAISE(auto delegate_session, delegate_->ContextualSession( + context, sigv4_parent->delegate())); + + auto merged = MergeProperties(catalog_properties_, context); + return WrapSession(std::move(delegate_session), merged); +} + +Result<std::shared_ptr<AuthSession>> SigV4AuthManager::TableSession( + const TableIdentifier& table, + const std::unordered_map<std::string, std::string>& properties, + std::shared_ptr<AuthSession> parent) { + auto sigv4_parent = std::dynamic_pointer_cast<SigV4AuthSession>(std::move(parent)); + ICEBERG_PRECHECK(sigv4_parent != nullptr, + "SigV4AuthManager parent must be a SigV4AuthSession"); + + ICEBERG_ASSIGN_OR_RAISE( + auto delegate_session, + delegate_->TableSession(table, properties, sigv4_parent->delegate())); + + auto merged = MergeProperties(catalog_properties_, properties); + return WrapSession(std::move(delegate_session), merged); Review Comment: **Performance & Design Flaw**: In the Java implementation, `TableSession` and `ContextualSession` directly reuse the `Aws4Signer` and AWS properties from the parent session. Here, `WrapSession` is called every time to recreate the `RestSigV4Signer` and `MakeCredentialsProvider`. If `DefaultAWSCredentialsProviderChain` is used, this could trigger expensive EC2 IMDS network requests or file reads on every table session creation, leading to severe performance degradation. Consider reusing the signer from the parent session like Java does. ########## src/iceberg/catalog/rest/auth/sigv4_auth_manager.cc: ########## @@ -0,0 +1,407 @@ +/* + * 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/catalog/rest/auth/auth_manager_internal.h" +#include "iceberg/catalog/rest/auth/aws_sdk.h" +#include "iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h" + +#ifdef ICEBERG_SIGV4 + +# include <atomic> +# include <mutex> +# include <sstream> + +# include <aws/core/Aws.h> +# include <aws/core/auth/AWSAuthSigner.h> +# include <aws/core/auth/AWSCredentialsProvider.h> +# include <aws/core/auth/AWSCredentialsProviderChain.h> +# include <aws/core/client/ClientConfiguration.h> +# include <aws/core/http/standard/StandardHttpRequest.h> +# include <aws/core/utils/HashingUtils.h> + +# include "iceberg/catalog/rest/auth/auth_managers.h" +# include "iceberg/catalog/rest/auth/auth_properties.h" +# include "iceberg/util/macros.h" +# include "iceberg/util/string_util.h" + +namespace iceberg::rest::auth { + +namespace { + +class AwsSdkLifecycle { + public: + static AwsSdkLifecycle& Instance() { + static AwsSdkLifecycle instance; + return instance; + } + + Status Initialize() { + std::lock_guard<std::mutex> lock(mutex_); + auto s = state_.load(); + if (s == State::kInitialized) return {}; + if (s == State::kFinalized) { + return InvalidArgument("AWS SDK has already been finalized; cannot reinitialize"); + } + Aws::InitAPI(options_); + state_.store(State::kInitialized); + return {}; + } + + Status Finalize() { + std::lock_guard<std::mutex> lock(mutex_); + if (state_.load() != State::kInitialized) return {}; + auto live = active_session_count_.load(); + if (live != 0) { + return Invalid( + "Cannot finalize AWS SDK while {} SigV4 auth session(s) are still alive", live); + } + Aws::ShutdownAPI(options_); + state_.store(State::kFinalized); + return {}; + } + + Status EnsureInitialized() { + if (state_.load() == State::kInitialized) return {}; + return Initialize(); + } + + bool IsInitialized() const { return state_.load() == State::kInitialized; } + bool IsFinalized() const { return state_.load() == State::kFinalized; } + + void IncrementSessionCount() { + active_session_count_.fetch_add(1, std::memory_order_relaxed); + } + void DecrementSessionCount() { + active_session_count_.fetch_sub(1, std::memory_order_relaxed); + } + + private: + enum class State : uint8_t { kUninitialized, kInitialized, kFinalized }; + + AwsSdkLifecycle() = default; + + std::atomic<State> state_{State::kUninitialized}; + std::mutex mutex_; + Aws::SDKOptions options_; + std::atomic<size_t> active_session_count_{0}; +}; + +Aws::Http::HttpMethod ToAwsMethod(HttpMethod method) { + switch (method) { + case HttpMethod::kGet: + return Aws::Http::HttpMethod::HTTP_GET; + case HttpMethod::kPost: + return Aws::Http::HttpMethod::HTTP_POST; + case HttpMethod::kPut: + return Aws::Http::HttpMethod::HTTP_PUT; + case HttpMethod::kDelete: + return Aws::Http::HttpMethod::HTTP_DELETE; + case HttpMethod::kHead: + return Aws::Http::HttpMethod::HTTP_HEAD; + } + return Aws::Http::HttpMethod::HTTP_GET; +} + +std::unordered_map<std::string, std::string> MergeProperties( + const std::unordered_map<std::string, std::string>& base, + const std::unordered_map<std::string, std::string>& overrides) { + auto merged = base; + for (const auto& [key, value] : overrides) { + merged.insert_or_assign(key, value); + } + return merged; +} + +/// Matches Java RESTSigV4AuthSession: canonical headers carry +/// Base64(SHA256(body)), canonical request trailer uses hex. +class RestSigV4Signer : public Aws::Client::AWSAuthV4Signer { + public: + RestSigV4Signer(const std::shared_ptr<Aws::Auth::AWSCredentialsProvider>& creds, + const char* service_name, const Aws::String& region) + : Aws::Client::AWSAuthV4Signer(creds, service_name, region, + PayloadSigningPolicy::Always, + /*urlEscapePath=*/false) { + // Skip the signer's hex overwrite of x-amz-content-sha256 so canonical + // headers see the caller's Base64; ComputePayloadHash still feeds hex + // into the canonical request trailer. + m_includeSha256HashHeader = false; + } +}; + +} // namespace + +// ---- SigV4AuthSession ---- + +SigV4AuthSession::SigV4AuthSession( + std::shared_ptr<AuthSession> delegate, std::string signing_region, + std::string signing_name, + std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider) + : delegate_(std::move(delegate)), + signing_region_(std::move(signing_region)), + signing_name_(std::move(signing_name)), + credentials_provider_(std::move(credentials_provider)), + signer_(std::make_unique<RestSigV4Signer>( + credentials_provider_, signing_name_.c_str(), signing_region_.c_str())) { + AwsSdkLifecycle::Instance().IncrementSessionCount(); +} + +SigV4AuthSession::~SigV4AuthSession() { + AwsSdkLifecycle::Instance().DecrementSessionCount(); +} + +Result<HttpRequest> SigV4AuthSession::Authenticate(const HttpRequest& request) { + ICEBERG_ASSIGN_OR_RAISE(auto delegate_request, delegate_->Authenticate(request)); + const auto& original_headers = delegate_request.headers; + + std::unordered_map<std::string, std::string> signing_headers; + for (const auto& [name, value] : original_headers) { + if (StringUtils::EqualsIgnoreCase(name, "Authorization")) { + signing_headers[std::string(kRelocatedHeaderPrefix) + name] = value; + } else { + signing_headers[name] = value; + } + } + + Aws::Http::URI aws_uri(delegate_request.url.c_str()); + auto aws_request = std::make_shared<Aws::Http::Standard::StandardHttpRequest>( + aws_uri, ToAwsMethod(delegate_request.method)); + for (const auto& [name, value] : signing_headers) { + aws_request->SetHeaderValue(Aws::String(name.c_str()), Aws::String(value.c_str())); + } + + // Empty body: hex EMPTY_BODY_SHA256 (Java parity workaround for the signer + // computing an invalid checksum on empty bodies). Non-empty: Base64. + if (delegate_request.body.empty()) { + aws_request->SetHeaderValue("x-amz-content-sha256", Aws::String(kEmptyBodySha256)); + } else { + auto body_stream = + Aws::MakeShared<std::stringstream>("SigV4Body", delegate_request.body); + aws_request->AddContentBody(body_stream); + auto sha256 = Aws::Utils::HashingUtils::CalculateSHA256( + Aws::String(delegate_request.body.data(), delegate_request.body.size())); + aws_request->SetHeaderValue("x-amz-content-sha256", + Aws::Utils::HashingUtils::Base64Encode(sha256)); + } + + if (!signer_->SignRequest(*aws_request)) { + return std::unexpected<Error>(Error{.kind = ErrorKind::kAuthenticationFailed, + .message = "SigV4 signing failed"}); + } + + HttpRequest signed_request{.method = delegate_request.method, + .url = std::move(delegate_request.url), + .headers = {}, + .body = std::move(delegate_request.body)}; + for (const auto& [aws_name, aws_value] : aws_request->GetHeaders()) { + std::string name(aws_name.c_str(), aws_name.size()); + std::string value(aws_value.c_str(), aws_value.size()); + for (const auto& [orig_name, orig_value] : original_headers) { + if (StringUtils::EqualsIgnoreCase(orig_name, name) && orig_value != value) { + signed_request.headers[std::string(kRelocatedHeaderPrefix) + orig_name] = + orig_value; + break; + } + } + signed_request.headers[std::move(name)] = std::move(value); + } + + return signed_request; +} + +Status SigV4AuthSession::Close() { return delegate_->Close(); } + +// ---- SigV4AuthManager ---- + +SigV4AuthManager::SigV4AuthManager(std::unique_ptr<AuthManager> delegate) + : delegate_(std::move(delegate)) {} + +SigV4AuthManager::~SigV4AuthManager() = default; + +Result<std::shared_ptr<AuthSession>> SigV4AuthManager::InitSession( + HttpClient& init_client, + const std::unordered_map<std::string, std::string>& properties) { + ICEBERG_RETURN_UNEXPECTED(AwsSdkLifecycle::Instance().EnsureInitialized()); Review Comment: **Race Condition**: There is no lock protecting the gap between `EnsureInitialized()` and `WrapSession` (which creates the `SigV4AuthSession` and increments `active_session_count_`). If another thread calls `FinalizeAwsSdk()` in this brief window, it could successfully shut down the AWS SDK (since count is still 0), and then this thread would proceed to create a session and use the closed SDK, causing a crash. Consider strengthening the lock protection around the session lifecycle and counter increment. ########## src/iceberg/catalog/rest/auth/sigv4_auth_manager.cc: ########## @@ -0,0 +1,407 @@ +/* + * 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/catalog/rest/auth/auth_manager_internal.h" +#include "iceberg/catalog/rest/auth/aws_sdk.h" +#include "iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h" + +#ifdef ICEBERG_SIGV4 + +# include <atomic> +# include <mutex> +# include <sstream> + +# include <aws/core/Aws.h> +# include <aws/core/auth/AWSAuthSigner.h> +# include <aws/core/auth/AWSCredentialsProvider.h> +# include <aws/core/auth/AWSCredentialsProviderChain.h> +# include <aws/core/client/ClientConfiguration.h> +# include <aws/core/http/standard/StandardHttpRequest.h> +# include <aws/core/utils/HashingUtils.h> + +# include "iceberg/catalog/rest/auth/auth_managers.h" +# include "iceberg/catalog/rest/auth/auth_properties.h" +# include "iceberg/util/macros.h" +# include "iceberg/util/string_util.h" + +namespace iceberg::rest::auth { + +namespace { + +class AwsSdkLifecycle { + public: + static AwsSdkLifecycle& Instance() { + static AwsSdkLifecycle instance; + return instance; + } + + Status Initialize() { + std::lock_guard<std::mutex> lock(mutex_); + auto s = state_.load(); + if (s == State::kInitialized) return {}; + if (s == State::kFinalized) { + return InvalidArgument("AWS SDK has already been finalized; cannot reinitialize"); + } + Aws::InitAPI(options_); + state_.store(State::kInitialized); + return {}; + } + + Status Finalize() { + std::lock_guard<std::mutex> lock(mutex_); + if (state_.load() != State::kInitialized) return {}; + auto live = active_session_count_.load(); + if (live != 0) { + return Invalid( + "Cannot finalize AWS SDK while {} SigV4 auth session(s) are still alive", live); + } + Aws::ShutdownAPI(options_); + state_.store(State::kFinalized); + return {}; + } + + Status EnsureInitialized() { + if (state_.load() == State::kInitialized) return {}; + return Initialize(); + } + + bool IsInitialized() const { return state_.load() == State::kInitialized; } + bool IsFinalized() const { return state_.load() == State::kFinalized; } + + void IncrementSessionCount() { + active_session_count_.fetch_add(1, std::memory_order_relaxed); + } + void DecrementSessionCount() { + active_session_count_.fetch_sub(1, std::memory_order_relaxed); + } + + private: + enum class State : uint8_t { kUninitialized, kInitialized, kFinalized }; + + AwsSdkLifecycle() = default; + + std::atomic<State> state_{State::kUninitialized}; + std::mutex mutex_; + Aws::SDKOptions options_; + std::atomic<size_t> active_session_count_{0}; +}; + +Aws::Http::HttpMethod ToAwsMethod(HttpMethod method) { + switch (method) { + case HttpMethod::kGet: + return Aws::Http::HttpMethod::HTTP_GET; + case HttpMethod::kPost: + return Aws::Http::HttpMethod::HTTP_POST; + case HttpMethod::kPut: + return Aws::Http::HttpMethod::HTTP_PUT; + case HttpMethod::kDelete: + return Aws::Http::HttpMethod::HTTP_DELETE; + case HttpMethod::kHead: + return Aws::Http::HttpMethod::HTTP_HEAD; + } + return Aws::Http::HttpMethod::HTTP_GET; +} + +std::unordered_map<std::string, std::string> MergeProperties( + const std::unordered_map<std::string, std::string>& base, + const std::unordered_map<std::string, std::string>& overrides) { + auto merged = base; + for (const auto& [key, value] : overrides) { + merged.insert_or_assign(key, value); + } + return merged; +} + +/// Matches Java RESTSigV4AuthSession: canonical headers carry +/// Base64(SHA256(body)), canonical request trailer uses hex. +class RestSigV4Signer : public Aws::Client::AWSAuthV4Signer { + public: + RestSigV4Signer(const std::shared_ptr<Aws::Auth::AWSCredentialsProvider>& creds, + const char* service_name, const Aws::String& region) + : Aws::Client::AWSAuthV4Signer(creds, service_name, region, + PayloadSigningPolicy::Always, + /*urlEscapePath=*/false) { + // Skip the signer's hex overwrite of x-amz-content-sha256 so canonical + // headers see the caller's Base64; ComputePayloadHash still feeds hex + // into the canonical request trailer. + m_includeSha256HashHeader = false; + } +}; + +} // namespace + +// ---- SigV4AuthSession ---- + +SigV4AuthSession::SigV4AuthSession( + std::shared_ptr<AuthSession> delegate, std::string signing_region, + std::string signing_name, + std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider) + : delegate_(std::move(delegate)), + signing_region_(std::move(signing_region)), + signing_name_(std::move(signing_name)), + credentials_provider_(std::move(credentials_provider)), + signer_(std::make_unique<RestSigV4Signer>( + credentials_provider_, signing_name_.c_str(), signing_region_.c_str())) { + AwsSdkLifecycle::Instance().IncrementSessionCount(); +} + +SigV4AuthSession::~SigV4AuthSession() { + AwsSdkLifecycle::Instance().DecrementSessionCount(); +} + +Result<HttpRequest> SigV4AuthSession::Authenticate(const HttpRequest& request) { + ICEBERG_ASSIGN_OR_RAISE(auto delegate_request, delegate_->Authenticate(request)); + const auto& original_headers = delegate_request.headers; + + std::unordered_map<std::string, std::string> signing_headers; + for (const auto& [name, value] : original_headers) { + if (StringUtils::EqualsIgnoreCase(name, "Authorization")) { + signing_headers[std::string(kRelocatedHeaderPrefix) + name] = value; + } else { + signing_headers[name] = value; + } + } + + Aws::Http::URI aws_uri(delegate_request.url.c_str()); + auto aws_request = std::make_shared<Aws::Http::Standard::StandardHttpRequest>( + aws_uri, ToAwsMethod(delegate_request.method)); + for (const auto& [name, value] : signing_headers) { + aws_request->SetHeaderValue(Aws::String(name.c_str()), Aws::String(value.c_str())); + } + + // Empty body: hex EMPTY_BODY_SHA256 (Java parity workaround for the signer + // computing an invalid checksum on empty bodies). Non-empty: Base64. + if (delegate_request.body.empty()) { + aws_request->SetHeaderValue("x-amz-content-sha256", Aws::String(kEmptyBodySha256)); + } else { + auto body_stream = + Aws::MakeShared<std::stringstream>("SigV4Body", delegate_request.body); + aws_request->AddContentBody(body_stream); + auto sha256 = Aws::Utils::HashingUtils::CalculateSHA256( + Aws::String(delegate_request.body.data(), delegate_request.body.size())); + aws_request->SetHeaderValue("x-amz-content-sha256", + Aws::Utils::HashingUtils::Base64Encode(sha256)); Review Comment: **Spec Violation**: AWS SigV4 specification strictly requires `x-amz-content-sha256` to be a lowercase hexadecimal string, not Base64. Although the comment mentions this aligns with the Java implementation (which might incorrectly output Base64 due to misuse of `SignerChecksumParams`), blindly matching this bug is problematic. If the backend is a standard AWS service (like API Gateway), the Base64 header will cause signature validation to fail (`SignatureDoesNotMatch`). This should be fixed to use Hex encoding. ########## src/iceberg/catalog/rest/auth/sigv4_auth_manager.cc: ########## @@ -0,0 +1,407 @@ +/* + * 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/catalog/rest/auth/auth_manager_internal.h" +#include "iceberg/catalog/rest/auth/aws_sdk.h" +#include "iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h" + +#ifdef ICEBERG_SIGV4 + +# include <atomic> +# include <mutex> +# include <sstream> + +# include <aws/core/Aws.h> +# include <aws/core/auth/AWSAuthSigner.h> +# include <aws/core/auth/AWSCredentialsProvider.h> +# include <aws/core/auth/AWSCredentialsProviderChain.h> +# include <aws/core/client/ClientConfiguration.h> +# include <aws/core/http/standard/StandardHttpRequest.h> +# include <aws/core/utils/HashingUtils.h> + +# include "iceberg/catalog/rest/auth/auth_managers.h" +# include "iceberg/catalog/rest/auth/auth_properties.h" +# include "iceberg/util/macros.h" +# include "iceberg/util/string_util.h" + +namespace iceberg::rest::auth { + +namespace { + +class AwsSdkLifecycle { + public: + static AwsSdkLifecycle& Instance() { + static AwsSdkLifecycle instance; + return instance; + } + + Status Initialize() { + std::lock_guard<std::mutex> lock(mutex_); + auto s = state_.load(); + if (s == State::kInitialized) return {}; + if (s == State::kFinalized) { + return InvalidArgument("AWS SDK has already been finalized; cannot reinitialize"); + } + Aws::InitAPI(options_); + state_.store(State::kInitialized); + return {}; + } + + Status Finalize() { + std::lock_guard<std::mutex> lock(mutex_); + if (state_.load() != State::kInitialized) return {}; + auto live = active_session_count_.load(); + if (live != 0) { + return Invalid( + "Cannot finalize AWS SDK while {} SigV4 auth session(s) are still alive", live); + } + Aws::ShutdownAPI(options_); + state_.store(State::kFinalized); + return {}; + } + + Status EnsureInitialized() { + if (state_.load() == State::kInitialized) return {}; + return Initialize(); + } + + bool IsInitialized() const { return state_.load() == State::kInitialized; } + bool IsFinalized() const { return state_.load() == State::kFinalized; } + + void IncrementSessionCount() { + active_session_count_.fetch_add(1, std::memory_order_relaxed); Review Comment: **Concurrency Flaw**: Using `std::memory_order_relaxed` for the atomic counter is unsafe here. In a multi-core environment, `relaxed` does not provide memory visibility guarantees. The thread executing `FinalizeAwsSdk()` might read a stale value (e.g., 0) from the CPU cache and prematurely close the SDK while sessions are still active. Consider using `std::memory_order_acquire` and `std::memory_order_release` at a minimum. ########## src/iceberg/catalog/rest/auth/sigv4_auth_manager.cc: ########## @@ -0,0 +1,407 @@ +/* + * 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/catalog/rest/auth/auth_manager_internal.h" +#include "iceberg/catalog/rest/auth/aws_sdk.h" +#include "iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h" + +#ifdef ICEBERG_SIGV4 + +# include <atomic> +# include <mutex> +# include <sstream> + +# include <aws/core/Aws.h> +# include <aws/core/auth/AWSAuthSigner.h> +# include <aws/core/auth/AWSCredentialsProvider.h> +# include <aws/core/auth/AWSCredentialsProviderChain.h> +# include <aws/core/client/ClientConfiguration.h> +# include <aws/core/http/standard/StandardHttpRequest.h> +# include <aws/core/utils/HashingUtils.h> + +# include "iceberg/catalog/rest/auth/auth_managers.h" +# include "iceberg/catalog/rest/auth/auth_properties.h" +# include "iceberg/util/macros.h" +# include "iceberg/util/string_util.h" + +namespace iceberg::rest::auth { + +namespace { + +class AwsSdkLifecycle { + public: + static AwsSdkLifecycle& Instance() { + static AwsSdkLifecycle instance; + return instance; + } + + Status Initialize() { + std::lock_guard<std::mutex> lock(mutex_); + auto s = state_.load(); + if (s == State::kInitialized) return {}; + if (s == State::kFinalized) { + return InvalidArgument("AWS SDK has already been finalized; cannot reinitialize"); + } + Aws::InitAPI(options_); + state_.store(State::kInitialized); + return {}; + } + + Status Finalize() { + std::lock_guard<std::mutex> lock(mutex_); + if (state_.load() != State::kInitialized) return {}; + auto live = active_session_count_.load(); + if (live != 0) { + return Invalid( + "Cannot finalize AWS SDK while {} SigV4 auth session(s) are still alive", live); + } + Aws::ShutdownAPI(options_); + state_.store(State::kFinalized); + return {}; + } + + Status EnsureInitialized() { + if (state_.load() == State::kInitialized) return {}; + return Initialize(); + } + + bool IsInitialized() const { return state_.load() == State::kInitialized; } + bool IsFinalized() const { return state_.load() == State::kFinalized; } + + void IncrementSessionCount() { + active_session_count_.fetch_add(1, std::memory_order_relaxed); + } + void DecrementSessionCount() { + active_session_count_.fetch_sub(1, std::memory_order_relaxed); + } + + private: + enum class State : uint8_t { kUninitialized, kInitialized, kFinalized }; + + AwsSdkLifecycle() = default; + + std::atomic<State> state_{State::kUninitialized}; + std::mutex mutex_; + Aws::SDKOptions options_; + std::atomic<size_t> active_session_count_{0}; +}; + +Aws::Http::HttpMethod ToAwsMethod(HttpMethod method) { + switch (method) { + case HttpMethod::kGet: + return Aws::Http::HttpMethod::HTTP_GET; + case HttpMethod::kPost: + return Aws::Http::HttpMethod::HTTP_POST; + case HttpMethod::kPut: + return Aws::Http::HttpMethod::HTTP_PUT; + case HttpMethod::kDelete: + return Aws::Http::HttpMethod::HTTP_DELETE; + case HttpMethod::kHead: + return Aws::Http::HttpMethod::HTTP_HEAD; + } + return Aws::Http::HttpMethod::HTTP_GET; +} + +std::unordered_map<std::string, std::string> MergeProperties( + const std::unordered_map<std::string, std::string>& base, + const std::unordered_map<std::string, std::string>& overrides) { + auto merged = base; + for (const auto& [key, value] : overrides) { + merged.insert_or_assign(key, value); + } + return merged; +} + +/// Matches Java RESTSigV4AuthSession: canonical headers carry +/// Base64(SHA256(body)), canonical request trailer uses hex. +class RestSigV4Signer : public Aws::Client::AWSAuthV4Signer { + public: + RestSigV4Signer(const std::shared_ptr<Aws::Auth::AWSCredentialsProvider>& creds, + const char* service_name, const Aws::String& region) + : Aws::Client::AWSAuthV4Signer(creds, service_name, region, + PayloadSigningPolicy::Always, + /*urlEscapePath=*/false) { + // Skip the signer's hex overwrite of x-amz-content-sha256 so canonical + // headers see the caller's Base64; ComputePayloadHash still feeds hex + // into the canonical request trailer. + m_includeSha256HashHeader = false; + } +}; + +} // namespace + +// ---- SigV4AuthSession ---- + +SigV4AuthSession::SigV4AuthSession( + std::shared_ptr<AuthSession> delegate, std::string signing_region, + std::string signing_name, + std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider) + : delegate_(std::move(delegate)), + signing_region_(std::move(signing_region)), + signing_name_(std::move(signing_name)), + credentials_provider_(std::move(credentials_provider)), + signer_(std::make_unique<RestSigV4Signer>( + credentials_provider_, signing_name_.c_str(), signing_region_.c_str())) { + AwsSdkLifecycle::Instance().IncrementSessionCount(); +} + +SigV4AuthSession::~SigV4AuthSession() { + AwsSdkLifecycle::Instance().DecrementSessionCount(); +} + +Result<HttpRequest> SigV4AuthSession::Authenticate(const HttpRequest& request) { + ICEBERG_ASSIGN_OR_RAISE(auto delegate_request, delegate_->Authenticate(request)); + const auto& original_headers = delegate_request.headers; + + std::unordered_map<std::string, std::string> signing_headers; + for (const auto& [name, value] : original_headers) { + if (StringUtils::EqualsIgnoreCase(name, "Authorization")) { + signing_headers[std::string(kRelocatedHeaderPrefix) + name] = value; + } else { + signing_headers[name] = value; + } + } + + Aws::Http::URI aws_uri(delegate_request.url.c_str()); + auto aws_request = std::make_shared<Aws::Http::Standard::StandardHttpRequest>( + aws_uri, ToAwsMethod(delegate_request.method)); + for (const auto& [name, value] : signing_headers) { + aws_request->SetHeaderValue(Aws::String(name.c_str()), Aws::String(value.c_str())); + } + + // Empty body: hex EMPTY_BODY_SHA256 (Java parity workaround for the signer + // computing an invalid checksum on empty bodies). Non-empty: Base64. + if (delegate_request.body.empty()) { + aws_request->SetHeaderValue("x-amz-content-sha256", Aws::String(kEmptyBodySha256)); + } else { + auto body_stream = + Aws::MakeShared<std::stringstream>("SigV4Body", delegate_request.body); + aws_request->AddContentBody(body_stream); + auto sha256 = Aws::Utils::HashingUtils::CalculateSHA256( + Aws::String(delegate_request.body.data(), delegate_request.body.size())); + aws_request->SetHeaderValue("x-amz-content-sha256", + Aws::Utils::HashingUtils::Base64Encode(sha256)); + } + + if (!signer_->SignRequest(*aws_request)) { + return std::unexpected<Error>(Error{.kind = ErrorKind::kAuthenticationFailed, + .message = "SigV4 signing failed"}); + } + + HttpRequest signed_request{.method = delegate_request.method, + .url = std::move(delegate_request.url), + .headers = {}, + .body = std::move(delegate_request.body)}; + for (const auto& [aws_name, aws_value] : aws_request->GetHeaders()) { + std::string name(aws_name.c_str(), aws_name.size()); + std::string value(aws_value.c_str(), aws_value.size()); + for (const auto& [orig_name, orig_value] : original_headers) { + if (StringUtils::EqualsIgnoreCase(orig_name, name) && orig_value != value) { + signed_request.headers[std::string(kRelocatedHeaderPrefix) + orig_name] = Review Comment: **Redundant Logic**: Because the earlier logic already relocated the original `Authorization` header to `Original-Authorization` in `signing_headers`, this loop will copy `Original-Authorization` once when it encounters it in `aws_request->GetHeaders()`. Later, when it encounters the newly generated `Authorization` header, this inner loop condition `orig_value != value` will match again, causing `Original-Authorization` to be redundantly overwritten with the same value. While functionally harmless, this double-insertion is unnecessary. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
