feat: Implement NoopAuthManager and integrate it with RestCatalog (#544)

- Add NoopAuthManager for "none" authentication type
- Register "none" auth type in static registry with case-insensitive matching
- Add KnownAuthTypes() to distinguish known-but-unimplemented types
(NotImplemented) from unknown types (InvalidArgument)
- Integrate AuthManager into RestCatalog

Author: lishuxu

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

@@ -19,7 +19,10 @@
 
 #include "iceberg/catalog/rest/auth/auth_managers.h"
 
+#include <unordered_set>
+
 #include "iceberg/catalog/rest/auth/auth_properties.h"
+#include "iceberg/catalog/rest/auth/auth_session.h"
 #include "iceberg/util/string_util.h"
 
 namespace iceberg::rest::auth {
@@ -30,6 +33,16 @@ namespace {
 using AuthManagerRegistry =
     std::unordered_map<std::string, AuthManagerFactory, StringHash, StringEqual>;
 
+const std::unordered_set<std::string, StringHash, StringEqual>& KnownAuthTypes() {
+  static const std::unordered_set<std::string, StringHash, StringEqual> kAuthTypes = {
+      AuthProperties::kAuthTypeNone,
+      AuthProperties::kAuthTypeBasic,
+      AuthProperties::kAuthTypeOAuth2,
+      AuthProperties::kAuthTypeSigV4,
+  };
+  return kAuthTypes;
+}
+
 // Infer the authentication type from properties.
 std::string InferAuthType(
     const std::unordered_map<std::string, std::string>& properties) {
@@ -48,9 +61,39 @@ std::string InferAuthType(
   return AuthProperties::kAuthTypeNone;
 }
 
+/// \brief Authentication manager that performs no authentication.
+class NoopAuthManager : public AuthManager {
+ public:
+  static Result<std::unique_ptr<AuthManager>> Make(
+      [[maybe_unused]] std::string_view name,
+      [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties) {
+    return std::make_unique<NoopAuthManager>();
+  }
+
+  Result<std::shared_ptr<AuthSession>> CatalogSession(
+      [[maybe_unused]] HttpClient& client,
+      [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties)
+      override {
+    return AuthSession::MakeDefault({});
+  }
+};
+
+template <typename T>
+AuthManagerFactory MakeAuthFactory() {
+  return
+      [](std::string_view name, const std::unordered_map<std::string, std::string>& props)
+          -> Result<std::unique_ptr<AuthManager>> { return T::Make(name, props); };
+}
+
+AuthManagerRegistry CreateDefaultRegistry() {
+  return {
+      {AuthProperties::kAuthTypeNone, MakeAuthFactory<NoopAuthManager>()},
+  };
+}
+
 // Get the global registry of auth manager factories.
 AuthManagerRegistry& GetRegistry() {
-  static AuthManagerRegistry registry;
+  static AuthManagerRegistry registry = CreateDefaultRegistry();
   return registry;
 }
 
@@ -68,8 +111,10 @@ Result<std::unique_ptr<AuthManager>> AuthManagers::Load(
   auto& registry = GetRegistry();
   auto it = registry.find(auth_type);
   if (it == registry.end()) {
-    // TODO(Li Shuxu): Fallback to default auth manager implementations
-    return NotImplemented("Authentication type '{}' is not supported", auth_type);
+    if (KnownAuthTypes().contains(auth_type)) {
+      return NotImplemented("Authentication type '{}' is not yet supported", auth_type);
+    }
+    return InvalidArgument("Unknown authentication type: '{}'", auth_type);
   }
 
   return it->second(name, properties);

src/iceberg/catalog/rest/http_client.cc

@@ -22,6 +22,7 @@
 #include <cpr/cpr.h>
 #include <nlohmann/json.hpp>
 
+#include "iceberg/catalog/rest/auth/auth_session.h"
 #include "iceberg/catalog/rest/constant.h"
 #include "iceberg/catalog/rest/error_handlers.h"
 #include "iceberg/catalog/rest/json_serde_internal.h"
@@ -67,19 +68,17 @@ namespace {
 /// \brief Default error type for unparseable REST responses.
 constexpr std::string_view kRestExceptionType = "RESTException";
 
-/// \brief Merges global default headers with request-specific headers.
-///
-/// Combines the global headers derived from RestCatalogProperties with the headers
-/// passed in the specific request. Request-specific headers have higher priority
-/// and will override global defaults if the keys conflict (e.g., overriding
-/// the default "Content-Type").
-cpr::Header MergeHeaders(const std::unordered_map<std::string, std::string>& defaults,
-                         const std::unordered_map<std::string, std::string>& overrides) {
-  cpr::Header combined_headers = {defaults.begin(), defaults.end()};
-  for (const auto& [key, val] : overrides) {
-    combined_headers.insert_or_assign(key, val);
+/// \brief Prepare headers for an HTTP request.
+Result<cpr::Header> BuildHeaders(
+    const std::unordered_map<std::string, std::string>& request_headers,
+    const std::unordered_map<std::string, std::string>& default_headers,
+    auth::AuthSession& session) {
+  std::unordered_map<std::string, std::string> headers(default_headers);
+  for (const auto& [key, val] : request_headers) {
+    headers.emplace(key, val);
   }
-  return combined_headers;
+  ICEBERG_RETURN_UNEXPECTED(session.Authenticate(headers));
+  return cpr::Header(headers.begin(), headers.end());
 }
 
 /// \brief Converts a map of string key-value pairs to cpr::Parameters.
@@ -149,10 +148,11 @@ HttpClient::~HttpClient() = default;
 Result<HttpResponse> HttpClient::Get(
     const std::string& path, const std::unordered_map<std::string, std::string>& params,
     const std::unordered_map<std::string, std::string>& headers,
-    const ErrorHandler& error_handler) {
-  auto final_headers = MergeHeaders(default_headers_, headers);
+    const ErrorHandler& error_handler, auth::AuthSession& session) {
+  ICEBERG_ASSIGN_OR_RAISE(auto all_headers,
+                          BuildHeaders(headers, default_headers_, session));
   cpr::Response response =
-      cpr::Get(cpr::Url{path}, GetParameters(params), final_headers, *connection_pool_);
+      cpr::Get(cpr::Url{path}, GetParameters(params), all_headers, *connection_pool_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
   HttpResponse http_response;
@@ -163,10 +163,11 @@ Result<HttpResponse> HttpClient::Get(
 Result<HttpResponse> HttpClient::Post(
     const std::string& path, const std::string& body,
     const std::unordered_map<std::string, std::string>& headers,
-    const ErrorHandler& error_handler) {
-  auto final_headers = MergeHeaders(default_headers_, headers);
+    const ErrorHandler& error_handler, auth::AuthSession& session) {
+  ICEBERG_ASSIGN_OR_RAISE(auto all_headers,
+                          BuildHeaders(headers, default_headers_, session));
   cpr::Response response =
-      cpr::Post(cpr::Url{path}, cpr::Body{body}, final_headers, *connection_pool_);
+      cpr::Post(cpr::Url{path}, cpr::Body{body}, all_headers, *connection_pool_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
   HttpResponse http_response;
@@ -178,17 +179,19 @@ Result<HttpResponse> HttpClient::PostForm(
     const std::string& path,
     const std::unordered_map<std::string, std::string>& form_data,
     const std::unordered_map<std::string, std::string>& headers,
-    const ErrorHandler& error_handler) {
-  auto final_headers = MergeHeaders(default_headers_, headers);
-  final_headers.insert_or_assign(kHeaderContentType, kMimeTypeFormUrlEncoded);
+    const ErrorHandler& error_handler, auth::AuthSession& session) {
+  std::unordered_map<std::string, std::string> form_headers(headers);
+  form_headers.insert_or_assign(kHeaderContentType, kMimeTypeFormUrlEncoded);
+  ICEBERG_ASSIGN_OR_RAISE(auto all_headers,
+                          BuildHeaders(form_headers, default_headers_, session));
   std::vector<cpr::Pair> pair_list;
   pair_list.reserve(form_data.size());
   for (const auto& [key, val] : form_data) {
     pair_list.emplace_back(key, val);
   }
   cpr::Response response =
       cpr::Post(cpr::Url{path}, cpr::Payload(pair_list.begin(), pair_list.end()),
-                final_headers, *connection_pool_);
+                all_headers, *connection_pool_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
   HttpResponse http_response;
@@ -198,9 +201,10 @@ Result<HttpResponse> HttpClient::PostForm(
 
 Result<HttpResponse> HttpClient::Head(
     const std::string& path, const std::unordered_map<std::string, std::string>& headers,
-    const ErrorHandler& error_handler) {
-  auto final_headers = MergeHeaders(default_headers_, headers);
-  cpr::Response response = cpr::Head(cpr::Url{path}, final_headers, *connection_pool_);
+    const ErrorHandler& error_handler, auth::AuthSession& session) {
+  ICEBERG_ASSIGN_OR_RAISE(auto all_headers,
+                          BuildHeaders(headers, default_headers_, session));
+  cpr::Response response = cpr::Head(cpr::Url{path}, all_headers, *connection_pool_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
   HttpResponse http_response;
@@ -211,10 +215,11 @@ Result<HttpResponse> HttpClient::Head(
 Result<HttpResponse> HttpClient::Delete(
     const std::string& path, const std::unordered_map<std::string, std::string>& params,
     const std::unordered_map<std::string, std::string>& headers,
-    const ErrorHandler& error_handler) {
-  auto final_headers = MergeHeaders(default_headers_, headers);
-  cpr::Response response = cpr::Delete(cpr::Url{path}, GetParameters(params),
-                                       final_headers, *connection_pool_);
+    const ErrorHandler& error_handler, auth::AuthSession& session) {
+  ICEBERG_ASSIGN_OR_RAISE(auto all_headers,
+                          BuildHeaders(headers, default_headers_, session));
+  cpr::Response response =
+      cpr::Delete(cpr::Url{path}, GetParameters(params), all_headers, *connection_pool_);
 
   ICEBERG_RETURN_UNEXPECTED(HandleFailureResponse(response, error_handler));
   HttpResponse http_response;

src/iceberg/catalog/rest/http_client.h

@@ -82,30 +82,33 @@ class ICEBERG_REST_EXPORT HttpClient {
   Result<HttpResponse> Get(const std::string& path,
                            const std::unordered_map<std::string, std::string>& params,
                            const std::unordered_map<std::string, std::string>& headers,
-                           const ErrorHandler& error_handler);
+                           const ErrorHandler& error_handler, auth::AuthSession& session);
 
   /// \brief Sends a POST request.
   Result<HttpResponse> Post(const std::string& path, const std::string& body,
                             const std::unordered_map<std::string, std::string>& headers,
-                            const ErrorHandler& error_handler);
+                            const ErrorHandler& error_handler,
+                            auth::AuthSession& session);
 
   /// \brief Sends a POST request with form data.
   Result<HttpResponse> PostForm(
       const std::string& path,
       const std::unordered_map<std::string, std::string>& form_data,
       const std::unordered_map<std::string, std::string>& headers,
-      const ErrorHandler& error_handler);
+      const ErrorHandler& error_handler, auth::AuthSession& session);
 
   /// \brief Sends a HEAD request.
   Result<HttpResponse> Head(const std::string& path,
                             const std::unordered_map<std::string, std::string>& headers,
-                            const ErrorHandler& error_handler);
+                            const ErrorHandler& error_handler,
+                            auth::AuthSession& session);
 
   /// \brief Sends a DELETE request.
   Result<HttpResponse> Delete(const std::string& path,
                               const std::unordered_map<std::string, std::string>& params,
                               const std::unordered_map<std::string, std::string>& headers,
-                              const ErrorHandler& error_handler);
+                              const ErrorHandler& error_handler,
+                              auth::AuthSession& session);
 
  private:
   std::unordered_map<std::string, std::string> default_headers_;