feat: add restcatalog authentication api

Author: shuxu.li

src/iceberg/catalog/rest/CMakeLists.txt

@@ -18,6 +18,9 @@
 add_subdirectory(auth)
 
 set(ICEBERG_REST_SOURCES
+    auth/auth_manager.cc
+    auth/auth_managers.cc
+    auth/auth_session.cc
     catalog_properties.cc
     endpoint.cc
     error_handlers.cc
@@ -26,10 +29,7 @@ set(ICEBERG_REST_SOURCES
     resource_paths.cc
     rest_catalog.cc
     rest_util.cc
-    types.cc
-    auth/auth_manager.cc
-    auth/auth_managers.cc
-    auth/auth_session.cc)
+    types.cc)
 
 set(ICEBERG_REST_STATIC_BUILD_INTERFACE_LIBS)
 set(ICEBERG_REST_SHARED_BUILD_INTERFACE_LIBS)

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

@@ -23,19 +23,26 @@
 
 namespace iceberg::rest::auth {
 
-Result<std::unique_ptr<AuthSession>> AuthManager::InitSession(
+Result<std::shared_ptr<AuthSession>> AuthManager::InitSession(
     HttpClient& init_client,
     const std::unordered_map<std::string, std::string>& properties) {
   // By default, use the catalog session for initialization
   return CatalogSession(init_client, properties);
 }
 
-Result<std::unique_ptr<AuthSession>> AuthManager::TableSession(
+Result<std::shared_ptr<AuthSession>> AuthManager::ContextualSession(
+    [[maybe_unused]] const std::unordered_map<std::string, std::string>& context,
+    const std::shared_ptr<AuthSession>& parent) {
+  // By default, return the parent session as-is
+  return parent;
+}
+
+Result<std::shared_ptr<AuthSession>> AuthManager::TableSession(
     [[maybe_unused]] const TableIdentifier& table,
     [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties,
-    [[maybe_unused]] const AuthSession& parent) {
-  // By default, return nullptr to indicate the parent session should be reused.
-  return nullptr;
+    const std::shared_ptr<AuthSession>& parent) {
+  // By default, return the parent session as-is
+  return parent;
 }
 
 }  // namespace iceberg::rest::auth

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

@@ -26,21 +26,14 @@
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
 #include "iceberg/catalog/rest/type_fwd.h"
 #include "iceberg/result.h"
-#include "iceberg/table_identifier.h"
+#include "iceberg/type_fwd.h"
 
 /// \file iceberg/catalog/rest/auth/auth_manager.h
 /// \brief Authentication manager interface for REST catalog.
 
 namespace iceberg::rest::auth {
 
 /// \brief Produces authentication sessions for catalog and table requests.
-///
-/// AuthManager is responsible for creating authentication sessions at different scopes:
-/// - InitSession: Short-lived session for catalog initialization (optional)
-/// - CatalogSession: Long-lived session for catalog-level operations (required)
-/// - TableSession: Optional table-specific session or reuse of catalog session
-///
-/// Implementations are registered via AuthManagers::Register() and loaded by auth type.
 class ICEBERG_REST_EXPORT AuthManager {
  public:
   virtual ~AuthManager() = default;
@@ -54,37 +47,52 @@ class ICEBERG_REST_EXPORT AuthManager {
   /// \param init_client HTTP client used for initialization requests.
   /// \param properties Client configuration supplied by the catalog.
   /// \return Session for initialization or an error if credentials cannot be acquired.
-  virtual Result<std::unique_ptr<AuthSession>> InitSession(
+  virtual Result<std::shared_ptr<AuthSession>> InitSession(
       HttpClient& init_client,
       const std::unordered_map<std::string, std::string>& properties);
 
   /// \brief Create the long-lived catalog session that acts as the parent session.
   ///
   /// This session is used for all catalog-level operations (list namespaces, list tables,
-  /// etc.) and serves as the parent session for table-specific operations. It is owned
-  /// by the catalog and reused throughout the catalog's lifetime.
+  /// etc.) and serves as the parent session for contextual and table-specific sessions.
+  /// It is owned by the catalog and reused throughout the catalog's lifetime.
   ///
   /// \param shared_client HTTP client owned by the catalog and reused for auth calls.
   /// \param properties Catalog properties (client config + server defaults).
   /// \return Session for catalog operations or an error if authentication cannot be set
   /// up.
-  virtual Result<std::unique_ptr<AuthSession>> CatalogSession(
+  virtual Result<std::shared_ptr<AuthSession>> CatalogSession(
       HttpClient& shared_client,
       const std::unordered_map<std::string, std::string>& properties) = 0;
 
+  /// \brief Create or reuse a session for a specific context.
+  ///
+  /// This method is used by SessionCatalog to create sessions for different contexts
+  /// (e.g., different users or tenants). Each REST endpoint call should use the
+  /// appropriate contextual session before sending the HTTP request.
+  ///
+  /// \param context Context properties (e.g., user credentials, tenant info).
+  /// \param parent Catalog session to inherit from or return as-is.
+  /// \return A context-specific session, or the parent session if no context-specific
+  /// session is needed, or an error if session creation fails.
+  virtual Result<std::shared_ptr<AuthSession>> ContextualSession(
+      const std::unordered_map<std::string, std::string>& context,
+      const std::shared_ptr<AuthSession>& parent);
+
   /// \brief Create or reuse a session scoped to a single table/view.
   ///
-  /// This method can return a new table-specific session or indicate that the parent
-  /// catalog session should be reused by returning nullptr.
+  /// This method is called when loading a table that may have table-specific auth
+  /// properties returned by the server.
   ///
   /// \param table Target table identifier.
   /// \param properties Table-specific auth properties returned by the server.
-  /// \param parent Catalog session to inherit from or extract information from.
-  /// \return A new session for the table, nullptr to reuse parent session, or an error.
-  virtual Result<std::unique_ptr<AuthSession>> TableSession(
+  /// \param parent Catalog or contextual session to inherit from or return as-is.
+  /// \return A table-specific session, or the parent session if no table-specific
+  /// session is needed, or an error if session creation fails.
+  virtual Result<std::shared_ptr<AuthSession>> TableSession(
       const TableIdentifier& table,
       const std::unordered_map<std::string, std::string>& properties,
-      const AuthSession& parent);
+      const std::shared_ptr<AuthSession>& parent);
 
   /// \brief Release resources held by the manager.
   ///

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

@@ -29,6 +29,10 @@ namespace iceberg::rest::auth {
 
 namespace {
 
+/// \brief Registry type for AuthManager factories with heterogeneous lookup support.
+using AuthManagerRegistry =
+    std::unordered_map<std::string, AuthManagerFactory, StringHash, StringEqual>;
+
 /// \brief Infer the authentication type from properties.
 ///
 /// If no explicit auth type is set, this function tries to infer it from
@@ -38,22 +42,20 @@ namespace {
 /// This behavior is consistent with Java Iceberg's AuthManagers.
 std::string InferAuthType(
     const std::unordered_map<std::string, std::string>& properties) {
-  // Check for explicit auth type first
-  auto it = properties.find(std::string(AuthProperties::kAuthType));
+  auto it = properties.find(AuthProperties::kAuthType);
   if (it != properties.end() && !it->second.empty()) {
     return StringUtils::ToLower(it->second);
   }
 
   // Infer from OAuth2 properties (credential or token)
-  bool has_credential =
-      properties.contains(std::string(AuthProperties::kOAuth2Credential));
-  bool has_token = properties.contains(std::string(AuthProperties::kOAuth2Token));
+  bool has_credential = properties.contains(AuthProperties::kOAuth2Credential);
+  bool has_token = properties.contains(AuthProperties::kOAuth2Token);
   if (has_credential || has_token) {
-    return std::string(AuthProperties::kAuthTypeOAuth2);
+    return AuthProperties::kAuthTypeOAuth2;
   }
 
   // Default to no authentication
-  return std::string(AuthProperties::kAuthTypeNone);
+  return AuthProperties::kAuthTypeNone;
 }
 
 /// \brief Get the global registry of auth manager factories.
@@ -65,7 +67,7 @@ AuthManagerRegistry& GetRegistry() {
 }  // namespace
 
 void AuthManagers::Register(std::string_view auth_type, AuthManagerFactory factory) {
-  GetRegistry()[StringUtils::ToLower(std::string(auth_type))] = std::move(factory);
+  GetRegistry()[StringUtils::ToLower(auth_type)] = std::move(factory);
 }
 
 Result<std::unique_ptr<AuthManager>> AuthManagers::Load(
@@ -76,6 +78,7 @@ Result<std::unique_ptr<AuthManager>> AuthManagers::Load(
   auto& registry = GetRegistry();
   auto it = registry.find(auth_type);
   if (it == registry.end()) {
+    // TODO: Fallback to default auth manager implementations
     return NotImplemented("Authentication type '{}' is not supported", auth_type);
   }
 

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

@@ -28,32 +28,27 @@
 #include "iceberg/catalog/rest/auth/auth_manager.h"
 #include "iceberg/catalog/rest/iceberg_rest_export.h"
 #include "iceberg/result.h"
-#include "iceberg/util/string_util.h"
 
 /// \file iceberg/catalog/rest/auth/auth_managers.h
 /// \brief Factory for creating authentication managers.
 
 namespace iceberg::rest::auth {
 
-/// \brief Function that builds an AuthManager for a given catalog.
+/// \brief Function that creates an AuthManager from its name.
 ///
-/// \param name Catalog name passed to the manager.
-/// \param properties Consolidated catalog configuration.
-/// \return Newly created manager instance.
-using AuthManagerFactory = std::function<std::unique_ptr<AuthManager>(
+/// \param name Name of the auth manager.
+/// \param properties Properties required by the auth manager.
+/// \return Newly created manager instance or an error if creation fails.
+using AuthManagerFactory = std::function<Result<std::unique_ptr<AuthManager>>(
     std::string_view name,
     const std::unordered_map<std::string, std::string>& properties)>;
 
-/// \brief Registry type for AuthManager factories with heterogeneous lookup support.
-using AuthManagerRegistry =
-    std::unordered_map<std::string, AuthManagerFactory, StringHash, StringEqual>;
-
 /// \brief Registry-backed factory for AuthManager implementations.
 class ICEBERG_REST_EXPORT AuthManagers {
  public:
   /// \brief Load a manager by consulting the "rest.auth.type" configuration.
   ///
-  /// \param name Catalog name passed to the manager.
+  /// \param name Name of the auth manager.
   /// \param properties Catalog properties used to determine auth type.
   /// \return Manager instance or an error if no factory matches.
   static Result<std::unique_ptr<AuthManager>> Load(

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

@@ -19,6 +19,7 @@
 
 #pragma once
 
+#include <string>
 #include <string_view>
 
 /// \file iceberg/catalog/rest/auth/auth_properties.h
@@ -32,37 +33,40 @@ namespace iceberg::rest::auth {
 /// for the REST catalog. It follows the same naming conventions as Java Iceberg.
 struct AuthProperties {
   /// \brief Property key for specifying the authentication type.
-  static constexpr std::string_view kAuthType = "rest.auth.type";
+  inline static const std::string kAuthType = "rest.auth.type";
   /// \brief Authentication type: no authentication.
-  static constexpr std::string_view kAuthTypeNone = "none";
+  inline static const std::string kAuthTypeNone = "none";
   /// \brief Authentication type: HTTP Basic authentication.
-  static constexpr std::string_view kAuthTypeBasic = "basic";
+  inline static const std::string kAuthTypeBasic = "basic";
   /// \brief Authentication type: OAuth2 authentication.
-  static constexpr std::string_view kAuthTypeOAuth2 = "oauth2";
+  inline static const std::string kAuthTypeOAuth2 = "oauth2";
   /// \brief Authentication type: AWS SigV4 authentication.
-  static constexpr std::string_view kAuthTypeSigV4 = "sigv4";
+  inline static const std::string kAuthTypeSigV4 = "sigv4";
+
   /// \brief Property key for Basic auth username.
-  static constexpr std::string_view kBasicUsername = "rest.auth.basic.username";
+  inline static const std::string kBasicUsername = "rest.auth.basic.username";
   /// \brief Property key for Basic auth password.
-  static constexpr std::string_view kBasicPassword = "rest.auth.basic.password";
+  inline static const std::string kBasicPassword = "rest.auth.basic.password";
+
   /// \brief Property key for OAuth2 token (bearer token).
-  static constexpr std::string_view kOAuth2Token = "token";
+  inline static const std::string kOAuth2Token = "token";
   /// \brief Property key for OAuth2 credential (client_id:client_secret).
-  static constexpr std::string_view kOAuth2Credential = "credential";
+  inline static const std::string kOAuth2Credential = "credential";
   /// \brief Property key for OAuth2 scope.
-  static constexpr std::string_view kOAuth2Scope = "scope";
+  inline static const std::string kOAuth2Scope = "scope";
   /// \brief Property key for OAuth2 server URI.
-  static constexpr std::string_view kOAuth2ServerUri = "oauth2-server-uri";
+  inline static const std::string kOAuth2ServerUri = "oauth2-server-uri";
   /// \brief Property key for enabling token refresh.
-  static constexpr std::string_view kOAuth2TokenRefreshEnabled = "token-refresh-enabled";
+  inline static const std::string kOAuth2TokenRefreshEnabled = "token-refresh-enabled";
   /// \brief Default OAuth2 scope for catalog operations.
-  static constexpr std::string_view kOAuth2DefaultScope = "catalog";
+  inline static const std::string kOAuth2DefaultScope = "catalog";
+
   /// \brief Property key for SigV4 region.
-  static constexpr std::string_view kSigV4Region = "rest.auth.sigv4.region";
+  inline static const std::string kSigV4Region = "rest.auth.sigv4.region";
   /// \brief Property key for SigV4 service name.
-  static constexpr std::string_view kSigV4Service = "rest.auth.sigv4.service";
+  inline static const std::string kSigV4Service = "rest.auth.sigv4.service";
   /// \brief Property key for SigV4 delegate auth type.
-  static constexpr std::string_view kSigV4DelegateAuthType =
+  inline static const std::string kSigV4DelegateAuthType =
       "rest.auth.sigv4.delegate-auth-type";
 };
 

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

@@ -23,21 +23,30 @@
 
 namespace iceberg::rest::auth {
 
-DefaultAuthSession::DefaultAuthSession(
-    std::unordered_map<std::string, std::string> headers)
-    : headers_(std::move(headers)) {}
-
-Status DefaultAuthSession::Authenticate(
-    std::unordered_map<std::string, std::string>& headers) {
-  for (const auto& [key, value] : headers_) {
-    headers.try_emplace(key, value);
+namespace {
+
+/// \brief Default implementation that adds static headers to requests.
+class DefaultAuthSession : public AuthSession {
+ public:
+  explicit DefaultAuthSession(std::unordered_map<std::string, std::string> headers)
+      : headers_(std::move(headers)) {}
+
+  Status Authenticate(std::unordered_map<std::string, std::string>& headers) override {
+    for (const auto& [key, value] : headers_) {
+      headers.try_emplace(key, value);
+    }
+    return {};
   }
-  return {};
-}
 
-std::unique_ptr<DefaultAuthSession> DefaultAuthSession::Make(
+ private:
+  std::unordered_map<std::string, std::string> headers_;
+};
+
+}  // namespace
+
+std::shared_ptr<AuthSession> AuthSession::Make(
     std::unordered_map<std::string, std::string> headers) {
-  return std::make_unique<DefaultAuthSession>(std::move(headers));
+  return std::make_shared<DefaultAuthSession>(std::move(headers));
 }
 
 }  // namespace iceberg::rest::auth