adopt request-in/request-out Authenticate interface

Author: plusplusjiajia

src/iceberg/catalog/rest/auth/auth_properties.h

@@ -54,23 +54,13 @@ class ICEBERG_REST_EXPORT AuthProperties : public ConfigBase<AuthProperties> {
 
   // ---- SigV4 entries ----
 
-  inline static const std::string kSigV4Region = "rest.auth.sigv4.region";
-  inline static const std::string kSigV4Service = "rest.auth.sigv4.service";
   inline static const std::string kSigV4DelegateAuthType =
       "rest.auth.sigv4.delegate-auth-type";
-
-  // ---- SigV4 AWS credential entries ----
-
-  /// AWS region for SigV4 signing.
   inline static const std::string kSigV4SigningRegion = "rest.signing-region";
-  /// AWS service name for SigV4 signing.
   inline static const std::string kSigV4SigningName = "rest.signing-name";
   inline static const std::string kSigV4SigningNameDefault = "execute-api";
-  /// Static access key ID for SigV4 signing.
   inline static const std::string kSigV4AccessKeyId = "rest.access-key-id";
-  /// Static secret access key for SigV4 signing.
   inline static const std::string kSigV4SecretAccessKey = "rest.secret-access-key";
-  /// Optional session token for SigV4 signing.
   inline static const std::string kSigV4SessionToken = "rest.session-token";
 
   // ---- OAuth2 entries ----

src/iceberg/catalog/rest/auth/auth_session.cc

@@ -33,13 +33,12 @@ class DefaultAuthSession : public AuthSession {
   explicit DefaultAuthSession(std::unordered_map<std::string, std::string> headers)
       : headers_(std::move(headers)) {}
 
-  Status Authenticate(
-      std::unordered_map<std::string, std::string>& headers,
-      [[maybe_unused]] const HTTPRequestContext& request_context) override {
+  Result<HTTPRequest> Authenticate(const HTTPRequest& request) override {
+    HTTPRequest authenticated = request;
     for (const auto& [key, value] : headers_) {
-      headers.try_emplace(key, value);
+      authenticated.headers.try_emplace(key, value);
     }
-    return {};
+    return authenticated;
   }
 
  private:

src/iceberg/catalog/rest/auth/auth_session.h

@@ -33,10 +33,14 @@
 
 namespace iceberg::rest::auth {
 
-/// \brief Context about the HTTP request being authenticated.
-struct ICEBERG_REST_EXPORT HTTPRequestContext {
+/// \brief An outgoing HTTP request passed through an AuthSession. Mirrors the
+/// HTTPRequest type used by the Java reference implementation so signing
+/// implementations like SigV4 can operate on method, url, headers, and body
+/// as a single value.
+struct ICEBERG_REST_EXPORT HTTPRequest {
   HttpMethod method = HttpMethod::kGet;
   std::string url;
+  std::unordered_map<std::string, std::string> headers;
   std::string body;
 };
 
@@ -45,26 +49,21 @@ class ICEBERG_REST_EXPORT AuthSession {
  public:
   virtual ~AuthSession() = default;
 
-  /// \brief Authenticate the given request headers.
+  /// \brief Authenticate an outgoing HTTP request.
   ///
-  /// This method adds authentication information (e.g., Authorization header)
-  /// to the provided headers map. The implementation should be idempotent.
+  /// Returns a new request with authentication information (e.g., an
+  /// Authorization header) added. Implementations must be idempotent and must
+  /// not mutate the input request.
   ///
-  /// \param[in,out] headers The headers map to add authentication information to.
-  /// \return Status indicating success or one of the following errors:
+  /// \param request The request to authenticate.
+  /// \return The authenticated request on success, or one of:
   ///         - AuthenticationFailed: General authentication failure (invalid credentials,
   ///         etc.)
   ///         - TokenExpired: Authentication token has expired and needs refresh
   ///         - NotAuthorized: Not authenticated (401)
   ///         - IOError: Network or connection errors when reaching auth server
   ///         - RestError: HTTP errors from authentication service
-  virtual Status Authenticate(std::unordered_map<std::string, std::string>& headers,
-                              const HTTPRequestContext& request_context) = 0;
-
-  /// \brief Convenience overload for callers that don't need a request context.
-  Status Authenticate(std::unordered_map<std::string, std::string>& headers) {
-    return Authenticate(headers, HTTPRequestContext{});
-  }
+  virtual Result<HTTPRequest> Authenticate(const HTTPRequest& request) = 0;
 
   /// \brief Close the session and release any resources.
   ///

src/iceberg/catalog/rest/auth/sigv4_auth_manager.cc

@@ -82,21 +82,18 @@ std::unordered_map<std::string, std::string> MergeProperties(
   return merged;
 }
 
-/// SigV4 signer reproducing Java RESTSigV4AuthSession's
-/// SignerChecksumParams(SHA256, X_AMZ_CONTENT_SHA256) output: canonical
-/// headers carry Base64(SHA256(body)), canonical request trailer uses hex.
+/// 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) {
-    // AWSAuthV4Signer normally overwrites x-amz-content-sha256 with the hex
-    // body hash right before canonicalization, which would clobber the Base64
-    // value the caller pre-sets. Clearing this flag skips that overwrite so
-    // canonical headers see the caller's Base64, while ComputePayloadHash
-    // still feeds hex into the canonical request trailer.
+    // 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;
   }
 };
@@ -118,41 +115,37 @@ SigV4AuthSession::SigV4AuthSession(
 
 SigV4AuthSession::~SigV4AuthSession() = default;
 
-Status SigV4AuthSession::Authenticate(
-    std::unordered_map<std::string, std::string>& headers,
-    const HTTPRequestContext& request_context) {
-  ICEBERG_RETURN_UNEXPECTED(delegate_->Authenticate(headers, request_context));
-  auto original_headers = headers;
+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] : 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(request_context.url.c_str());
+  Aws::Http::URI aws_uri(delegate_request.url.c_str());
   auto aws_request = std::make_shared<Aws::Http::Standard::StandardHttpRequest>(
-      aws_uri, ToAwsMethod(request_context.method));
+      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 uses hex EMPTY_BODY_SHA256 (Java workaround for the signer
-  // producing an invalid checksum for empty bodies); non-empty body uses
-  // Base64(SHA256(body)). See RestSigV4Signer doc for why this value survives
-  // signing to land in the canonical headers unchanged.
-  if (request_context.body.empty()) {
+  // producing an invalid checksum on empty bodies); non-empty uses 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", request_context.body);
+        Aws::MakeShared<std::stringstream>("SigV4Body", delegate_request.body);
     aws_request->AddContentBody(body_stream);
     auto sha256 = Aws::Utils::HashingUtils::CalculateSHA256(
-        Aws::String(request_context.body.data(), request_context.body.size()));
+        Aws::String(delegate_request.body.data(), delegate_request.body.size()));
     aws_request->SetHeaderValue("x-amz-content-sha256",
                                 Aws::Utils::HashingUtils::Base64Encode(sha256));
   }
@@ -162,21 +155,25 @@ Status SigV4AuthSession::Authenticate(
         Error{ErrorKind::kAuthenticationFailed, "SigV4 signing failed"});
   }
 
-  // Merge signed headers back; relocate any original value that conflicts.
-  headers.clear();
+  // Fill headers with the signed set, relocating any conflicting originals.
+  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) {
-        headers[std::string(kRelocatedHeaderPrefix) + orig_name] = orig_value;
+        signed_request.headers[std::string(kRelocatedHeaderPrefix) + orig_name] =
+            orig_value;
         break;
       }
     }
-    headers[std::move(name)] = std::move(value);
+    signed_request.headers[std::move(name)] = std::move(value);
   }
 
-  return {};
+  return signed_request;
 }
 
 Status SigV4AuthSession::Close() { return delegate_->Close(); }
@@ -243,7 +240,6 @@ SigV4AuthManager::MakeCredentialsProvider(
   bool has_ak = access_key_it != properties.end() && !access_key_it->second.empty();
   bool has_sk = secret_key_it != properties.end() && !secret_key_it->second.empty();
 
-  // Reject partial credentials — providing only one of AK/SK is a misconfiguration.
   ICEBERG_PRECHECK(
       has_ak == has_sk, "Both '{}' and '{}' must be set together, or neither",
       AuthProperties::kSigV4AccessKeyId, AuthProperties::kSigV4SecretAccessKey);
@@ -263,14 +259,10 @@ SigV4AuthManager::MakeCredentialsProvider(
 
 std::string SigV4AuthManager::ResolveSigningRegion(
     const std::unordered_map<std::string, std::string>& properties) {
-  auto it = properties.find(AuthProperties::kSigV4SigningRegion);
-  if (it != properties.end() && !it->second.empty()) {
+  if (auto it = properties.find(AuthProperties::kSigV4SigningRegion);
+      it != properties.end() && !it->second.empty()) {
     return it->second;
   }
-  auto legacy_it = properties.find(AuthProperties::kSigV4Region);
-  if (legacy_it != properties.end() && !legacy_it->second.empty()) {
-    return legacy_it->second;
-  }
   if (const char* env = std::getenv("AWS_REGION")) {
     return std::string(env);
   }
@@ -282,14 +274,10 @@ std::string SigV4AuthManager::ResolveSigningRegion(
 
 std::string SigV4AuthManager::ResolveSigningName(
     const std::unordered_map<std::string, std::string>& properties) {
-  auto it = properties.find(AuthProperties::kSigV4SigningName);
-  if (it != properties.end() && !it->second.empty()) {
+  if (auto it = properties.find(AuthProperties::kSigV4SigningName);
+      it != properties.end() && !it->second.empty()) {
     return it->second;
   }
-  auto legacy_it = properties.find(AuthProperties::kSigV4Service);
-  if (legacy_it != properties.end() && !legacy_it->second.empty()) {
-    return legacy_it->second;
-  }
   return AuthProperties::kSigV4SigningNameDefault;
 }
 

src/iceberg/catalog/rest/auth/sigv4_auth_manager.h

@@ -65,8 +65,7 @@ class ICEBERG_REST_EXPORT SigV4AuthSession : public AuthSession {
 
   ~SigV4AuthSession() override;
 
-  Status Authenticate(std::unordered_map<std::string, std::string>& headers,
-                      const HTTPRequestContext& request_context) override;
+  Result<HTTPRequest> Authenticate(const HTTPRequest& request) override;
 
   Status Close() override;
 

src/iceberg/catalog/rest/http_client.cc

@@ -70,17 +70,22 @@ namespace {
 /// \brief Default error type for unparseable REST responses.
 constexpr std::string_view kRestExceptionType = "RESTException";
 
-/// \brief Prepare headers for an HTTP request.
-Result<cpr::Header> BuildHeaders(
-    const std::unordered_map<std::string, std::string>& request_headers,
+/// \brief Merge default headers with per-request headers (per-request wins).
+std::unordered_map<std::string, std::string> MergeHeaders(
     const std::unordered_map<std::string, std::string>& default_headers,
-    auth::AuthSession& session, const auth::HTTPRequestContext& request_context) {
-  std::unordered_map<std::string, std::string> headers(default_headers);
+    const std::unordered_map<std::string, std::string>& request_headers) {
+  std::unordered_map<std::string, std::string> merged(default_headers);
   for (const auto& [key, val] : request_headers) {
-    headers.insert_or_assign(key, val);
+    merged.insert_or_assign(key, val);
   }
-  ICEBERG_RETURN_UNEXPECTED(session.Authenticate(headers, request_context));
-  return cpr::Header(headers.begin(), headers.end());
+  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.
@@ -171,8 +176,11 @@ Result<HttpResponse> HttpClient::Get(
     const ErrorHandler& error_handler, auth::AuthSession& session) {
   ICEBERG_ASSIGN_OR_RAISE(
       auto all_headers,
-      BuildHeaders(headers, default_headers_, session,
-                   {HttpMethod::kGet, AppendQueryString(path, params), ""}));
+      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_);
 
@@ -188,7 +196,11 @@ Result<HttpResponse> HttpClient::Post(
     const ErrorHandler& error_handler, auth::AuthSession& session) {
   ICEBERG_ASSIGN_OR_RAISE(
       auto all_headers,
-      BuildHeaders(headers, default_headers_, session, {HttpMethod::kPost, path, body}));
+      AuthenticateRequest({.method = HttpMethod::kPost,
+                           .url = path,
+                           .headers = MergeHeaders(default_headers_, headers),
+                           .body = body},
+                          session));
   cpr::Response response =
       cpr::Post(cpr::Url{path}, cpr::Body{body}, all_headers, *connection_pool_);
 
@@ -214,9 +226,13 @@ Result<HttpResponse> HttpClient::PostForm(
   // actual payload sent over the wire.
   cpr::Payload payload(pair_list.begin(), pair_list.end());
   std::string encoded_body = payload.GetContent();
-  ICEBERG_ASSIGN_OR_RAISE(auto all_headers,
-                          BuildHeaders(form_headers, default_headers_, session,
-                                       {HttpMethod::kPost, path, encoded_body}));
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto all_headers,
+      AuthenticateRequest({.method = HttpMethod::kPost,
+                           .url = path,
+                           .headers = MergeHeaders(default_headers_, form_headers),
+                           .body = encoded_body},
+                          session));
   cpr::Response response =
       cpr::Post(cpr::Url{path}, std::move(payload), all_headers, *connection_pool_);
 
@@ -231,7 +247,11 @@ Result<HttpResponse> HttpClient::Head(
     const ErrorHandler& error_handler, auth::AuthSession& session) {
   ICEBERG_ASSIGN_OR_RAISE(
       auto all_headers,
-      BuildHeaders(headers, default_headers_, session, {HttpMethod::kHead, path, ""}));
+      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_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
@@ -246,8 +266,11 @@ Result<HttpResponse> HttpClient::Delete(
     const ErrorHandler& error_handler, auth::AuthSession& session) {
   ICEBERG_ASSIGN_OR_RAISE(
       auto all_headers,
-      BuildHeaders(headers, default_headers_, session,
-                   {HttpMethod::kDelete, AppendQueryString(path, params), ""}));
+      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_);