feat: Implement NoopAuthManager and integrate AuthManager into RestCatalog

Author: shuxu.li

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

@@ -21,8 +21,10 @@
 
 #include <algorithm>
 #include <cctype>
+#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 {
@@ -33,6 +35,17 @@ namespace {
 using AuthManagerRegistry =
     std::unordered_map<std::string, AuthManagerFactory, StringHash, StringEqual>;
 
+/// \brief Known authentication types that are defined in the Iceberg spec.
+const std::unordered_set<std::string, StringHash, StringEqual>& KnownAuthTypes() {
+  static const std::unordered_set<std::string, StringHash, StringEqual> types = {
+      AuthProperties::kAuthTypeNone,
+      AuthProperties::kAuthTypeBasic,
+      AuthProperties::kAuthTypeOAuth2,
+      AuthProperties::kAuthTypeSigV4,
+  };
+  return types;
+}
+
 // Infer the authentication type from properties.
 std::string InferAuthType(
     const std::unordered_map<std::string, std::string>& properties) {
@@ -51,9 +64,30 @@ std::string InferAuthType(
   return AuthProperties::kAuthTypeNone;
 }
 
+/// \brief Authentication manager that performs no authentication.
+class NoopAuthManager : public AuthManager {
+ public:
+  Result<std::shared_ptr<AuthSession>> CatalogSession(
+      [[maybe_unused]] HttpClient& shared_client,
+      [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties)
+      override {
+    return AuthSession::MakeDefault({});
+  }
+};
+
 // Get the global registry of auth manager factories.
 AuthManagerRegistry& GetRegistry() {
-  static AuthManagerRegistry registry;
+  static AuthManagerRegistry registry = [] {
+    AuthManagerRegistry r;
+    // Register built-in "none" auth manager
+    r[AuthProperties::kAuthTypeNone] =
+        []([[maybe_unused]] std::string_view name,
+           [[maybe_unused]] const std::unordered_map<std::string, std::string>& props)
+        -> Result<std::unique_ptr<AuthManager>> {
+      return std::make_unique<NoopAuthManager>();
+    };
+    return r;
+  }();
   return registry;
 }
 
@@ -71,8 +105,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/rest_catalog.cc

@@ -26,6 +26,9 @@
 
 #include <nlohmann/json.hpp>
 
+#include "iceberg/catalog/rest/auth/auth_manager.h"
+#include "iceberg/catalog/rest/auth/auth_managers.h"
+#include "iceberg/catalog/rest/auth/auth_session.h"
 #include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/catalog/rest/constant.h"
 #include "iceberg/catalog/rest/endpoint.h"
@@ -66,12 +69,18 @@ std::unordered_set<Endpoint> GetDefaultEndpoints() {
 }
 
 /// \brief Fetch server config and merge it with client config
-Result<CatalogConfig> FetchServerConfig(const ResourcePaths& paths,
-                                        const RestCatalogProperties& current_config) {
+Result<CatalogConfig> FetchServerConfig(
+    const ResourcePaths& paths, const RestCatalogProperties& current_config,
+    const std::shared_ptr<auth::AuthSession>& session) {
   ICEBERG_ASSIGN_OR_RAISE(auto config_path, paths.Config());
   HttpClient client(current_config.ExtractHeaders());
+
+  // Get authentication headers
+  std::unordered_map<std::string, std::string> auth_headers;
+  ICEBERG_RETURN_UNEXPECTED(session->Authenticate(auth_headers));
+
   ICEBERG_ASSIGN_OR_RAISE(const auto response,
-                          client.Get(config_path, /*params=*/{}, /*headers=*/{},
+                          client.Get(config_path, /*params=*/{}, auth_headers,
                                      *DefaultErrorHandler::Instance()));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   return CatalogConfigFromJson(json);
@@ -114,11 +123,23 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
   if (!file_io) {
     return InvalidArgument("FileIO is required to create RestCatalog");
   }
+
+  std::string catalog_name = config.Get(RestCatalogProperties::kName);
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_manager,
+                          auth::AuthManagers::Load(catalog_name, config.configs()));
+
   ICEBERG_ASSIGN_OR_RAISE(
       auto paths, ResourcePaths::Make(std::string(TrimTrailingSlash(uri)),
                                       config.Get(RestCatalogProperties::kPrefix)));
-  ICEBERG_ASSIGN_OR_RAISE(auto server_config, FetchServerConfig(*paths, config));
 
+  // Create init session for fetching server configuration
+  HttpClient init_client(config.ExtractHeaders());
+  ICEBERG_ASSIGN_OR_RAISE(auto init_session,
+                          auth_manager->InitSession(init_client, config.configs()));
+  ICEBERG_ASSIGN_OR_RAISE(auto server_config,
+                          FetchServerConfig(*paths, config, init_session));
+
+  // Merge client config with server defaults and overrides
   std::unique_ptr<RestCatalogProperties> final_config = RestCatalogProperties::FromMap(
       MergeConfigs(server_config.defaults, config.configs(), server_config.overrides));
 
@@ -139,27 +160,43 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
       paths, ResourcePaths::Make(std::string(TrimTrailingSlash(final_uri)),
                                  final_config->Get(RestCatalogProperties::kPrefix)));
 
-  return std::shared_ptr<RestCatalog>(
-      new RestCatalog(std::move(final_config), std::move(file_io), std::move(paths),
-                      std::move(endpoints)));
+  auto client = std::make_unique<HttpClient>(final_config->ExtractHeaders());
+  ICEBERG_ASSIGN_OR_RAISE(auto catalog_session,
+                          auth_manager->CatalogSession(*client, final_config->configs()));
+  return std::shared_ptr<RestCatalog>(new RestCatalog(
+      std::move(final_config), std::move(file_io), std::move(client), std::move(paths),
+      std::move(endpoints), std::move(auth_manager), std::move(catalog_session)));
 }
 
 RestCatalog::RestCatalog(std::unique_ptr<RestCatalogProperties> config,
                          std::shared_ptr<FileIO> file_io,
+                         std::unique_ptr<HttpClient> client,
                          std::unique_ptr<ResourcePaths> paths,
-                         std::unordered_set<Endpoint> endpoints)
+                         std::unordered_set<Endpoint> endpoints,
+                         std::unique_ptr<auth::AuthManager> auth_manager,
+                         std::shared_ptr<auth::AuthSession> catalog_session)
     : config_(std::move(config)),
       file_io_(std::move(file_io)),
-      client_(std::make_unique<HttpClient>(config_->ExtractHeaders())),
+      client_(std::move(client)),
       paths_(std::move(paths)),
       name_(config_->Get(RestCatalogProperties::kName)),
-      supported_endpoints_(std::move(endpoints)) {}
+      supported_endpoints_(std::move(endpoints)),
+      auth_manager_(std::move(auth_manager)),
+      catalog_session_(std::move(catalog_session)) {}
 
 std::string_view RestCatalog::name() const { return name_; }
 
+Result<std::unordered_map<std::string, std::string>> RestCatalog::AuthHeaders() const {
+  std::unordered_map<std::string, std::string> headers;
+  ICEBERG_RETURN_UNEXPECTED(catalog_session_->Authenticate(headers));
+  return headers;
+}
+
 Result<std::vector<Namespace>> RestCatalog::ListNamespaces(const Namespace& ns) const {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::ListNamespaces());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespaces());
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
+
   std::vector<Namespace> result;
   std::string next_token;
   while (true) {
@@ -172,7 +209,7 @@ Result<std::vector<Namespace>> RestCatalog::ListNamespaces(const Namespace& ns)
     }
     ICEBERG_ASSIGN_OR_RAISE(
         const auto response,
-        client_->Get(path, params, /*headers=*/{}, *NamespaceErrorHandler::Instance()));
+        client_->Get(path, params, auth_headers, *NamespaceErrorHandler::Instance()));
     ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
     ICEBERG_ASSIGN_OR_RAISE(auto list_response, ListNamespacesResponseFromJson(json));
     result.insert(result.end(), list_response.namespaces.begin(),
@@ -189,10 +226,12 @@ Status RestCatalog::CreateNamespace(
     const Namespace& ns, const std::unordered_map<std::string, std::string>& properties) {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::CreateNamespace());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespaces());
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
+
   CreateNamespaceRequest request{.namespace_ = ns, .properties = properties};
   ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
   ICEBERG_ASSIGN_OR_RAISE(const auto response,
-                          client_->Post(path, json_request, /*headers=*/{},
+                          client_->Post(path, json_request, auth_headers,
                                         *NamespaceErrorHandler::Instance()));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   ICEBERG_ASSIGN_OR_RAISE(auto create_response, CreateNamespaceResponseFromJson(json));
@@ -203,8 +242,10 @@ Result<std::unordered_map<std::string, std::string>> RestCatalog::GetNamespacePr
     const Namespace& ns) const {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::GetNamespaceProperties());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespace_(ns));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
+
   ICEBERG_ASSIGN_OR_RAISE(const auto response,
-                          client_->Get(path, /*params=*/{}, /*headers=*/{},
+                          client_->Get(path, /*params=*/{}, auth_headers,
                                        *NamespaceErrorHandler::Instance()));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   ICEBERG_ASSIGN_OR_RAISE(auto get_response, GetNamespaceResponseFromJson(json));
@@ -214,8 +255,10 @@ Result<std::unordered_map<std::string, std::string>> RestCatalog::GetNamespacePr
 Status RestCatalog::DropNamespace(const Namespace& ns) {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::DropNamespace());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespace_(ns));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
+
   ICEBERG_ASSIGN_OR_RAISE(const auto response,
-                          client_->Delete(path, /*params=*/{}, /*headers=*/{},
+                          client_->Delete(path, /*params=*/{}, auth_headers,
                                           *DropNamespaceErrorHandler::Instance()));
   return {};
 }
@@ -227,21 +270,25 @@ Result<bool> RestCatalog::NamespaceExists(const Namespace& ns) const {
   }
 
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Namespace_(ns));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
+
   return CaptureNoSuchNamespace(
-      client_->Head(path, /*headers=*/{}, *NamespaceErrorHandler::Instance()));
+      client_->Head(path, auth_headers, *NamespaceErrorHandler::Instance()));
 }
 
 Status RestCatalog::UpdateNamespaceProperties(
     const Namespace& ns, const std::unordered_map<std::string, std::string>& updates,
     const std::unordered_set<std::string>& removals) {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::UpdateNamespace());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->NamespaceProperties(ns));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
+
   UpdateNamespacePropertiesRequest request{
       .removals = std::vector<std::string>(removals.begin(), removals.end()),
       .updates = updates};
   ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
   ICEBERG_ASSIGN_OR_RAISE(const auto response,
-                          client_->Post(path, json_request, /*headers=*/{},
+                          client_->Post(path, json_request, auth_headers,
                                         *NamespaceErrorHandler::Instance()));
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   ICEBERG_ASSIGN_OR_RAISE(auto update_response,
@@ -251,8 +298,9 @@ Status RestCatalog::UpdateNamespaceProperties(
 
 Result<std::vector<TableIdentifier>> RestCatalog::ListTables(const Namespace& ns) const {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::ListTables());
-
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Tables(ns));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
+
   std::vector<TableIdentifier> result;
   std::string next_token;
   while (true) {
@@ -262,7 +310,7 @@ Result<std::vector<TableIdentifier>> RestCatalog::ListTables(const Namespace& ns
     }
     ICEBERG_ASSIGN_OR_RAISE(
         const auto response,
-        client_->Get(path, params, /*headers=*/{}, *TableErrorHandler::Instance()));
+        client_->Get(path, params, auth_headers, *TableErrorHandler::Instance()));
     ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
     ICEBERG_ASSIGN_OR_RAISE(auto list_response, ListTablesResponseFromJson(json));
     result.insert(result.end(), list_response.identifiers.begin(),
@@ -282,6 +330,7 @@ Result<LoadTableResult> RestCatalog::CreateTableInternal(
     const std::unordered_map<std::string, std::string>& properties, bool stage_create) {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::CreateTable());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Tables(identifier.ns));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
 
   CreateTableRequest request{
       .name = identifier.name,
@@ -296,7 +345,7 @@ Result<LoadTableResult> RestCatalog::CreateTableInternal(
   ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
   ICEBERG_ASSIGN_OR_RAISE(
       const auto response,
-      client_->Post(path, json_request, /*headers=*/{}, *TableErrorHandler::Instance()));
+      client_->Post(path, json_request, auth_headers, *TableErrorHandler::Instance()));
 
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   return LoadTableResultFromJson(json);
@@ -320,6 +369,7 @@ Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
     const std::vector<std::unique_ptr<TableUpdate>>& updates) {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::UpdateTable());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
 
   CommitTableRequest request{.identifier = identifier};
   request.requirements.reserve(requirements.size());
@@ -334,7 +384,7 @@ Result<std::shared_ptr<Table>> RestCatalog::UpdateTable(
   ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
   ICEBERG_ASSIGN_OR_RAISE(
       const auto response,
-      client_->Post(path, json_request, /*headers=*/{}, *TableErrorHandler::Instance()));
+      client_->Post(path, json_request, auth_headers, *TableErrorHandler::Instance()));
 
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   ICEBERG_ASSIGN_OR_RAISE(auto commit_response, CommitTableResponseFromJson(json));
@@ -363,14 +413,15 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
 Status RestCatalog::DropTable(const TableIdentifier& identifier, bool purge) {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::DeleteTable());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
 
   std::unordered_map<std::string, std::string> params;
   if (purge) {
     params["purgeRequested"] = "true";
   }
   ICEBERG_ASSIGN_OR_RAISE(
       const auto response,
-      client_->Delete(path, params, /*headers=*/{}, *TableErrorHandler::Instance()));
+      client_->Delete(path, params, auth_headers, *TableErrorHandler::Instance()));
   return {};
 }
 
@@ -381,19 +432,22 @@ Result<bool> RestCatalog::TableExists(const TableIdentifier& identifier) const {
   }
 
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
+
   return CaptureNoSuchTable(
-      client_->Head(path, /*headers=*/{}, *TableErrorHandler::Instance()));
+      client_->Head(path, auth_headers, *TableErrorHandler::Instance()));
 }
 
 Status RestCatalog::RenameTable(const TableIdentifier& from, const TableIdentifier& to) {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::RenameTable());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Rename());
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
 
   RenameTableRequest request{.source = from, .destination = to};
   ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
   ICEBERG_ASSIGN_OR_RAISE(
       const auto response,
-      client_->Post(path, json_request, /*headers=*/{}, *TableErrorHandler::Instance()));
+      client_->Post(path, json_request, auth_headers, *TableErrorHandler::Instance()));
 
   return {};
 }
@@ -402,9 +456,11 @@ Result<std::string> RestCatalog::LoadTableInternal(
     const TableIdentifier& identifier) const {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::LoadTable());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Table(identifier));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
+
   ICEBERG_ASSIGN_OR_RAISE(
       const auto response,
-      client_->Get(path, /*params=*/{}, /*headers=*/{}, *TableErrorHandler::Instance()));
+      client_->Get(path, /*params=*/{}, auth_headers, *TableErrorHandler::Instance()));
   return response.body();
 }
 
@@ -422,6 +478,7 @@ Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
     const TableIdentifier& identifier, const std::string& metadata_file_location) {
   ICEBERG_ENDPOINT_CHECK(supported_endpoints_, Endpoint::RegisterTable());
   ICEBERG_ASSIGN_OR_RAISE(auto path, paths_->Register(identifier.ns));
+  ICEBERG_ASSIGN_OR_RAISE(auto auth_headers, AuthHeaders());
 
   RegisterTableRequest request{
       .name = identifier.name,
@@ -431,7 +488,7 @@ Result<std::shared_ptr<Table>> RestCatalog::RegisterTable(
   ICEBERG_ASSIGN_OR_RAISE(auto json_request, ToJsonString(ToJson(request)));
   ICEBERG_ASSIGN_OR_RAISE(
       const auto response,
-      client_->Post(path, json_request, /*headers=*/{}, *TableErrorHandler::Instance()));
+      client_->Post(path, json_request, auth_headers, *TableErrorHandler::Instance()));
 
   ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
   ICEBERG_ASSIGN_OR_RAISE(auto load_result, LoadTableResultFromJson(json));