Meson: wire SigV4 behind a feature option

plusplusjiajia · plusplusjiajia · commit 6075248ac6c8 · 2026-05-25T17:16:19.000+08:00
diff --git a/meson.options b/meson.options
@@ -44,4 +44,11 @@ option(
     value: 'disabled',
 )
 
+option(
+    'sigv4',
+    type: 'feature',
+    description: 'Build AWS SigV4 authentication support for rest catalog',
+    value: 'disabled',
+)
+
 option('tests', type: 'feature', description: 'Build tests', value: 'enabled')
diff --git a/src/iceberg/catalog/rest/auth/sigv4_auth_manager.cc b/src/iceberg/catalog/rest/auth/sigv4_auth_manager.cc
@@ -20,6 +20,7 @@
 #include "iceberg/catalog/rest/auth/sigv4_auth_manager.h"
 
 #include <cstdlib>
+#include <mutex>
 #include <sstream>
 
 #include <aws/core/Aws.h>
@@ -107,21 +108,22 @@ class RestSigV4Signer : public Aws::Client::AWSAuthV4Signer {
 SigV4AuthSession::SigV4AuthSession(
     std::shared_ptr<AuthSession> delegate, std::string signing_region,
     std::string signing_name,
-    std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider)
+    std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider,
+    std::unordered_map<std::string, std::string> effective_properties)
     : 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())) {}
+          credentials_provider_, signing_name_.c_str(), signing_region_.c_str())),
+      effective_properties_(std::move(effective_properties)) {}
 
 SigV4AuthSession::~SigV4AuthSession() = default;
 
 Result<HTTPRequest> SigV4AuthSession::Authenticate(const HTTPRequest& request) {
   ICEBERG_ASSIGN_OR_RAISE(auto delegate_request, delegate_->Authenticate(request));
   const auto& original_headers = delegate_request.headers;
 
-  // Relocate any delegate-set Authorization so SigV4 takes precedence.
   std::unordered_map<std::string, std::string> signing_headers;
   for (const auto& [name, value] : original_headers) {
     if (StringUtils::EqualsIgnoreCase(name, "Authorization")) {
@@ -138,8 +140,8 @@ Result<HTTPRequest> SigV4AuthSession::Authenticate(const HTTPRequest& request) {
     aws_request->SetHeaderValue(Aws::String(name.c_str()), Aws::String(value.c_str()));
   }
 
-  // Empty body uses hex EMPTY_BODY_SHA256 (Java workaround for the signer
-  // producing an invalid checksum on empty bodies); non-empty uses Base64.
+  // 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 {
@@ -152,12 +154,14 @@ Result<HTTPRequest> SigV4AuthSession::Authenticate(const HTTPRequest& request) {
                                 Aws::Utils::HashingUtils::Base64Encode(sha256));
   }
 
-  if (!signer_->SignRequest(*aws_request)) {
-    return std::unexpected<Error>(
-        Error{ErrorKind::kAuthenticationFailed, "SigV4 signing failed"});
+  {
+    std::lock_guard<std::mutex> lock(signing_mutex_);
+    if (!signer_->SignRequest(*aws_request)) {
+      return std::unexpected<Error>(
+          Error{ErrorKind::kAuthenticationFailed, "SigV4 signing failed"});
+    }
   }
 
-  // Fill headers with the signed set, relocating any conflicting originals.
   HTTPRequest signed_request{.method = delegate_request.method,
                              .url = std::move(delegate_request.url),
                              .headers = {},
@@ -200,7 +204,6 @@ Result<std::shared_ptr<AuthSession>> SigV4AuthManager::CatalogSession(
     HttpClient& shared_client,
     const std::unordered_map<std::string, std::string>& properties) {
   AwsSdkGuard::EnsureInitialized();
-  catalog_properties_ = properties;
   ICEBERG_ASSIGN_OR_RAISE(auto delegate_session,
                           delegate_->CatalogSession(shared_client, properties));
   return WrapSession(std::move(delegate_session), properties);
@@ -214,8 +217,8 @@ Result<std::shared_ptr<AuthSession>> SigV4AuthManager::ContextualSession(
   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);
+  auto merged = MergeProperties(sigv4_parent->effective_properties(), context);
+  return WrapSession(std::move(delegate_session), std::move(merged));
 }
 
 Result<std::shared_ptr<AuthSession>> SigV4AuthManager::TableSession(
@@ -228,8 +231,8 @@ Result<std::shared_ptr<AuthSession>> SigV4AuthManager::TableSession(
       auto delegate_session,
       delegate_->TableSession(table, properties, sigv4_parent->delegate()));
 
-  auto merged = MergeProperties(catalog_properties_, properties);
-  return WrapSession(std::move(delegate_session), merged);
+  auto merged = MergeProperties(sigv4_parent->effective_properties(), properties);
+  return WrapSession(std::move(delegate_session), std::move(merged));
 }
 
 Status SigV4AuthManager::Close() { return delegate_->Close(); }
@@ -285,13 +288,13 @@ std::string SigV4AuthManager::ResolveSigningName(
 
 Result<std::shared_ptr<AuthSession>> SigV4AuthManager::WrapSession(
     std::shared_ptr<AuthSession> delegate_session,
-    const std::unordered_map<std::string, std::string>& properties) {
+    std::unordered_map<std::string, std::string> properties) {
   auto region = ResolveSigningRegion(properties);
   auto service = ResolveSigningName(properties);
   ICEBERG_ASSIGN_OR_RAISE(auto credentials, MakeCredentialsProvider(properties));
-  return std::make_shared<SigV4AuthSession>(std::move(delegate_session),
-                                            std::move(region), std::move(service),
-                                            std::move(credentials));
+  return std::make_shared<SigV4AuthSession>(
+      std::move(delegate_session), std::move(region), std::move(service),
+      std::move(credentials), std::move(properties));
 }
 
 Result<std::unique_ptr<AuthManager>> MakeSigV4AuthManager(
diff --git a/src/iceberg/catalog/rest/auth/sigv4_auth_manager.h b/src/iceberg/catalog/rest/auth/sigv4_auth_manager.h
@@ -20,6 +20,7 @@
 #pragma once
 
 #include <memory>
+#include <mutex>
 #include <string>
 #include <unordered_map>
 
@@ -47,8 +48,8 @@ namespace iceberg::rest::auth {
 ///
 /// See https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_sigv.html
 ///
-/// Thread safety: Authenticate() is NOT thread-safe. Each session should be used
-/// from a single thread, or callers must synchronize externally.
+/// Thread safety: Authenticate() is thread-safe; concurrent signing calls are
+/// serialized by an internal mutex.
 class ICEBERG_REST_EXPORT SigV4AuthSession : public AuthSession {
  public:
   /// SHA-256 hash of empty string, used for requests with no body.
@@ -61,7 +62,8 @@ class ICEBERG_REST_EXPORT SigV4AuthSession : public AuthSession {
   SigV4AuthSession(
       std::shared_ptr<AuthSession> delegate, std::string signing_region,
       std::string signing_name,
-      std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider);
+      std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider,
+      std::unordered_map<std::string, std::string> effective_properties);
 
   ~SigV4AuthSession() override;
 
@@ -71,20 +73,27 @@ class ICEBERG_REST_EXPORT SigV4AuthSession : public AuthSession {
 
   const std::shared_ptr<AuthSession>& delegate() const { return delegate_; }
 
+  /// Merged properties this session was built from. Child sessions inherit
+  /// from this (not the catalog's) so contextual overrides propagate into
+  /// table sessions.
+  const std::unordered_map<std::string, std::string>& effective_properties() const {
+    return effective_properties_;
+  }
+
  private:
   std::shared_ptr<AuthSession> delegate_;
   std::string signing_region_;
   std::string signing_name_;
   std::shared_ptr<Aws::Auth::AWSCredentialsProvider> credentials_provider_;
   std::unique_ptr<Aws::Client::AWSAuthV4Signer> signer_;
+  std::unordered_map<std::string, std::string> effective_properties_;
+  // AWSAuthV4Signer::SignRequest mutates shared signer state.
+  mutable std::mutex signing_mutex_;
 };
 
 /// \brief An AuthManager that produces SigV4AuthSession instances.
 ///
 /// Wraps a delegate AuthManager to handle double authentication (e.g., OAuth2 + SigV4).
-///
-/// Thread safety: CatalogSession() must be called before ContextualSession() or
-/// TableSession(). Concurrent calls are NOT safe — callers must synchronize externally.
 class ICEBERG_REST_EXPORT SigV4AuthManager : public AuthManager {
  public:
   explicit SigV4AuthManager(std::unique_ptr<AuthManager> delegate);
@@ -118,10 +127,9 @@ class ICEBERG_REST_EXPORT SigV4AuthManager : public AuthManager {
       const std::unordered_map<std::string, std::string>& properties);
   Result<std::shared_ptr<AuthSession>> WrapSession(
       std::shared_ptr<AuthSession> delegate_session,
-      const std::unordered_map<std::string, std::string>& properties);
+      std::unordered_map<std::string, std::string> properties);
 
   std::unique_ptr<AuthManager> delegate_;
-  std::unordered_map<std::string, std::string> catalog_properties_;
 };
 
 }  // namespace iceberg::rest::auth
diff --git a/src/iceberg/catalog/rest/http_client.cc b/src/iceberg/catalog/rest/http_client.cc
@@ -81,21 +81,8 @@ std::unordered_map<std::string, std::string> MergeHeaders(
   return merged;
 }
 
-/// \brief Authenticate the request and return the final cpr::Header.
-Result<cpr::Header> AuthenticateRequest(const auth::HTTPRequest& request,
-                                        auth::AuthSession& session) {
-  ICEBERG_ASSIGN_OR_RAISE(auto authenticated, session.Authenticate(request));
-  return cpr::Header(authenticated.headers.begin(), authenticated.headers.end());
-}
-
-/// \brief Converts a map of string key-value pairs to cpr::Parameters.
-cpr::Parameters GetParameters(
-    const std::unordered_map<std::string, std::string>& params) {
-  cpr::Parameters cpr_params;
-  for (const auto& [key, val] : params) {
-    cpr_params.Add({key, val});
-  }
-  return cpr_params;
+cpr::Header ToCprHeader(const auth::HTTPRequest& request) {
+  return cpr::Header(request.headers.begin(), request.headers.end());
 }
 
 /// \brief Append URL-encoded query parameters to a URL, sorted by key.
@@ -175,14 +162,13 @@ Result<HttpResponse> HttpClient::Get(
     const std::unordered_map<std::string, std::string>& headers,
     const ErrorHandler& error_handler, auth::AuthSession& session) {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto all_headers,
-      AuthenticateRequest({.method = HttpMethod::kGet,
-                           .url = AppendQueryString(path, params),
-                           .headers = MergeHeaders(default_headers_, headers),
-                           .body = ""},
-                          session));
-  cpr::Response response =
-      cpr::Get(cpr::Url{path}, GetParameters(params), all_headers, *connection_pool_);
+      auto authenticated,
+      session.Authenticate({.method = HttpMethod::kGet,
+                            .url = AppendQueryString(path, params),
+                            .headers = MergeHeaders(default_headers_, headers),
+                            .body = ""}));
+  cpr::Response response = cpr::Get(cpr::Url{authenticated.url},
+                                    ToCprHeader(authenticated), *connection_pool_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
   HttpResponse http_response;
@@ -195,14 +181,14 @@ Result<HttpResponse> HttpClient::Post(
     const std::unordered_map<std::string, std::string>& headers,
     const ErrorHandler& error_handler, auth::AuthSession& session) {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto all_headers,
-      AuthenticateRequest({.method = HttpMethod::kPost,
-                           .url = path,
-                           .headers = MergeHeaders(default_headers_, headers),
-                           .body = body},
-                          session));
+      auto authenticated,
+      session.Authenticate({.method = HttpMethod::kPost,
+                            .url = path,
+                            .headers = MergeHeaders(default_headers_, headers),
+                            .body = body}));
   cpr::Response response =
-      cpr::Post(cpr::Url{path}, cpr::Body{body}, all_headers, *connection_pool_);
+      cpr::Post(cpr::Url{authenticated.url}, cpr::Body{authenticated.body},
+                ToCprHeader(authenticated), *connection_pool_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
   HttpResponse http_response;
@@ -222,19 +208,18 @@ Result<HttpResponse> HttpClient::PostForm(
   for (const auto& [key, val] : form_data) {
     pair_list.emplace_back(key, val);
   }
-  // Use cpr's own encoding as the signing body to ensure consistency with the
-  // actual payload sent over the wire.
-  cpr::Payload payload(pair_list.begin(), pair_list.end());
-  std::string encoded_body = payload.GetContent();
+  // Sign the exact bytes cpr will put on the wire.
+  std::string encoded_body =
+      cpr::Payload(pair_list.begin(), pair_list.end()).GetContent();
   ICEBERG_ASSIGN_OR_RAISE(
-      auto all_headers,
-      AuthenticateRequest({.method = HttpMethod::kPost,
-                           .url = path,
-                           .headers = MergeHeaders(default_headers_, form_headers),
-                           .body = encoded_body},
-                          session));
+      auto authenticated,
+      session.Authenticate({.method = HttpMethod::kPost,
+                            .url = path,
+                            .headers = MergeHeaders(default_headers_, form_headers),
+                            .body = std::move(encoded_body)}));
   cpr::Response response =
-      cpr::Post(cpr::Url{path}, std::move(payload), all_headers, *connection_pool_);
+      cpr::Post(cpr::Url{authenticated.url}, cpr::Body{authenticated.body},
+                ToCprHeader(authenticated), *connection_pool_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
   HttpResponse http_response;
@@ -246,13 +231,13 @@ Result<HttpResponse> HttpClient::Head(
     const std::string& path, const std::unordered_map<std::string, std::string>& headers,
     const ErrorHandler& error_handler, auth::AuthSession& session) {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto all_headers,
-      AuthenticateRequest({.method = HttpMethod::kHead,
-                           .url = path,
-                           .headers = MergeHeaders(default_headers_, headers),
-                           .body = ""},
-                          session));
-  cpr::Response response = cpr::Head(cpr::Url{path}, all_headers, *connection_pool_);
+      auto authenticated,
+      session.Authenticate({.method = HttpMethod::kHead,
+                            .url = path,
+                            .headers = MergeHeaders(default_headers_, headers),
+                            .body = ""}));
+  cpr::Response response = cpr::Head(cpr::Url{authenticated.url},
+                                     ToCprHeader(authenticated), *connection_pool_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
   HttpResponse http_response;
@@ -265,14 +250,13 @@ Result<HttpResponse> HttpClient::Delete(
     const std::unordered_map<std::string, std::string>& headers,
     const ErrorHandler& error_handler, auth::AuthSession& session) {
   ICEBERG_ASSIGN_OR_RAISE(
-      auto all_headers,
-      AuthenticateRequest({.method = HttpMethod::kDelete,
-                           .url = AppendQueryString(path, params),
-                           .headers = MergeHeaders(default_headers_, headers),
-                           .body = ""},
-                          session));
-  cpr::Response response =
-      cpr::Delete(cpr::Url{path}, GetParameters(params), all_headers, *connection_pool_);
+      auto authenticated,
+      session.Authenticate({.method = HttpMethod::kDelete,
+                            .url = AppendQueryString(path, params),
+                            .headers = MergeHeaders(default_headers_, headers),
+                            .body = ""}));
+  cpr::Response response = cpr::Delete(cpr::Url{authenticated.url},
+                                       ToCprHeader(authenticated), *connection_pool_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
   HttpResponse http_response;
diff --git a/src/iceberg/catalog/rest/meson.build b/src/iceberg/catalog/rest/meson.build
@@ -41,16 +41,26 @@ cpr_needs_static = (
 cpr_dep = dependency('cpr', static: cpr_needs_static)
 
 iceberg_rest_build_deps = [iceberg_dep, cpr_dep]
+iceberg_rest_compile_defs = []
+
+sigv4_opt = get_option('sigv4')
+aws_sdk_core_dep = dependency('aws-cpp-sdk-core', required: sigv4_opt)
+if aws_sdk_core_dep.found()
+    iceberg_rest_sources += files('auth/sigv4_auth_manager.cc')
+    iceberg_rest_build_deps += aws_sdk_core_dep
+    iceberg_rest_compile_defs += '-DICEBERG_BUILD_SIGV4'
+endif
+
 iceberg_rest_lib = library(
     'iceberg_rest',
     sources: iceberg_rest_sources,
     dependencies: iceberg_rest_build_deps,
     gnu_symbol_visibility: 'hidden',
-    cpp_shared_args: ['-DICEBERG_REST_EXPORTING'],
-    cpp_static_args: ['-DICEBERG_REST_STATIC'],
+    cpp_shared_args: ['-DICEBERG_REST_EXPORTING'] + iceberg_rest_compile_defs,
+    cpp_static_args: ['-DICEBERG_REST_STATIC'] + iceberg_rest_compile_defs,
 )
 
-iceberg_rest_compile_args = []
+iceberg_rest_compile_args = iceberg_rest_compile_defs
 if get_option('default_library') == 'static'
     iceberg_rest_compile_args += ['-DICEBERG_REST_STATIC']
 endif
@@ -80,13 +90,17 @@ install_headers(
     subdir: 'iceberg/catalog/rest',
 )
 
+iceberg_rest_auth_headers = [
+    'auth/auth_manager.h',
+    'auth/auth_managers.h',
+    'auth/auth_properties.h',
+    'auth/auth_session.h',
+    'auth/oauth2_util.h',
+]
+if aws_sdk_core_dep.found()
+    iceberg_rest_auth_headers += ['auth/sigv4_auth_manager.h']
+endif
 install_headers(
-    [
-        'auth/auth_manager.h',
-        'auth/auth_managers.h',
-        'auth/auth_properties.h',
-        'auth/auth_session.h',
-        'auth/oauth2_util.h',
-    ],
+    iceberg_rest_auth_headers,
     subdir: 'iceberg/catalog/rest/auth',
 )
diff --git a/src/iceberg/test/meson.build b/src/iceberg/test/meson.build
diff --git a/src/iceberg/test/sigv4_auth_test.cc b/src/iceberg/test/sigv4_auth_test.cc
