feat(rest): add initial oauth2 support to rest catalog (#577)

Add OAuth2 authentication support for the REST catalog, including:
- OAuth2Manager with static token and client_credentials grant flows

TODO:
- RefreshToken and ExchangeToken will be supported later

Author: lishuxu

src/iceberg/catalog/rest/CMakeLists.txt

@@ -20,7 +20,9 @@ add_subdirectory(auth)
 set(ICEBERG_REST_SOURCES
     auth/auth_manager.cc
     auth/auth_managers.cc
+    auth/auth_properties.cc
     auth/auth_session.cc
+    auth/oauth2_util.cc
     catalog_properties.cc
     endpoint.cc
     error_handlers.cc

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

@@ -19,9 +19,12 @@
 
 #include "iceberg/catalog/rest/auth/auth_manager.h"
 
+#include <optional>
+
 #include "iceberg/catalog/rest/auth/auth_manager_internal.h"
 #include "iceberg/catalog/rest/auth/auth_properties.h"
 #include "iceberg/catalog/rest/auth/auth_session.h"
+#include "iceberg/catalog/rest/auth/oauth2_util.h"
 #include "iceberg/util/macros.h"
 #include "iceberg/util/transform_util.h"
 
@@ -80,7 +83,8 @@ class BasicAuthManager : public AuthManager {
                      "Missing required property '{}'", AuthProperties::kBasicPassword);
     std::string credential = username_it->second + ":" + password_it->second;
     return AuthSession::MakeDefault(
-        {{"Authorization", "Basic " + TransformUtil::Base64Encode(credential)}});
+        {{std::string(kAuthorizationHeader),
+          "Basic " + TransformUtil::Base64Encode(credential)}});
   }
 };
 
@@ -90,4 +94,76 @@ Result<std::unique_ptr<AuthManager>> MakeBasicAuthManager(
   return std::make_unique<BasicAuthManager>();
 }
 
+/// \brief OAuth2 authentication manager.
+class OAuth2Manager : public AuthManager {
+ public:
+  Result<std::shared_ptr<AuthSession>> InitSession(
+      HttpClient& init_client,
+      const std::unordered_map<std::string, std::string>& properties) override {
+    ICEBERG_ASSIGN_OR_RAISE(auto config, AuthProperties::FromProperties(properties));
+    // No token refresh during init (short-lived session).
+    config.Set(AuthProperties::kKeepRefreshed, false);
+
+    // Credential takes priority: fetch a fresh token for the config request.
+    if (!config.credential().empty()) {
+      auto init_session = AuthSession::MakeDefault(AuthHeaders(config.token()));
+      ICEBERG_ASSIGN_OR_RAISE(init_token_response_,
+                              FetchToken(init_client, *init_session, config));
+      return AuthSession::MakeDefault(AuthHeaders(init_token_response_->access_token));
+    }
+
+    if (!config.token().empty()) {
+      return AuthSession::MakeDefault(AuthHeaders(config.token()));
+    }
+
+    return AuthSession::MakeDefault({});
+  }
+
+  Result<std::shared_ptr<AuthSession>> CatalogSession(
+      HttpClient& client,
+      const std::unordered_map<std::string, std::string>& properties) override {
+    ICEBERG_ASSIGN_OR_RAISE(auto config, AuthProperties::FromProperties(properties));
+
+    // Reuse token from init phase.
+    if (init_token_response_.has_value()) {
+      auto token_response = std::move(*init_token_response_);
+      init_token_response_.reset();
+      return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(),
+                                     config.client_id(), config.client_secret(),
+                                     config.scope(), client);
+    }
+
+    // If token is provided, use it directly.
+    if (!config.token().empty()) {
+      return AuthSession::MakeDefault(AuthHeaders(config.token()));
+    }
+
+    // Fetch a new token using client_credentials grant.
+    if (!config.credential().empty()) {
+      auto base_session = AuthSession::MakeDefault(AuthHeaders(config.token()));
+      OAuthTokenResponse token_response;
+      ICEBERG_ASSIGN_OR_RAISE(token_response, FetchToken(client, *base_session, config));
+      // TODO(lishuxu): should we directly pass config to the MakeOAuth2 call?
+      return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(),
+                                     config.client_id(), config.client_secret(),
+                                     config.scope(), client);
+    }
+
+    return AuthSession::MakeDefault({});
+  }
+
+  // TODO(lishuxu): Override TableSession() for token exchange (RFC 8693).
+  // TODO(lishuxu): Override ContextualSession() for per-context exchange.
+
+ private:
+  /// Cached token from InitSession
+  std::optional<OAuthTokenResponse> init_token_response_;
+};
+
+Result<std::unique_ptr<AuthManager>> MakeOAuth2Manager(
+    [[maybe_unused]] std::string_view name,
+    [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties) {
+  return std::make_unique<OAuth2Manager>();
+}
+
 }  // namespace iceberg::rest::auth

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

@@ -42,4 +42,9 @@ Result<std::unique_ptr<AuthManager>> MakeBasicAuthManager(
     std::string_view name,
     const std::unordered_map<std::string, std::string>& properties);
 
+/// \brief Create an OAuth2 authentication manager.
+Result<std::unique_ptr<AuthManager>> MakeOAuth2Manager(
+    std::string_view name,
+    const std::unordered_map<std::string, std::string>& properties);
+
 }  // namespace iceberg::rest::auth

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

@@ -52,8 +52,8 @@ std::string InferAuthType(
   }
 
   // Infer from OAuth2 properties (credential or token)
-  bool has_credential = properties.contains(AuthProperties::kOAuth2Credential);
-  bool has_token = properties.contains(AuthProperties::kOAuth2Token);
+  bool has_credential = properties.contains(AuthProperties::kCredential.key());
+  bool has_token = properties.contains(AuthProperties::kToken.key());
   if (has_credential || has_token) {
     return AuthProperties::kAuthTypeOAuth2;
   }
@@ -65,6 +65,7 @@ AuthManagerRegistry CreateDefaultRegistry() {
   return {
       {AuthProperties::kAuthTypeNone, MakeNoopAuthManager},
       {AuthProperties::kAuthTypeBasic, MakeBasicAuthManager},
+      {AuthProperties::kAuthTypeOAuth2, MakeOAuth2Manager},
   };
 }
 

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

@@ -0,0 +1,83 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "iceberg/catalog/rest/auth/auth_properties.h"
+
+#include <utility>
+
+#include "iceberg/catalog/rest/catalog_properties.h"
+
+namespace iceberg::rest::auth {
+
+namespace {
+
+std::pair<std::string, std::string> ParseCredential(const std::string& credential) {
+  auto colon_pos = credential.find(':');
+  if (colon_pos == std::string::npos) {
+    return {"", credential};
+  }
+  return {credential.substr(0, colon_pos), credential.substr(colon_pos + 1)};
+}
+
+}  // namespace
+
+std::unordered_map<std::string, std::string> AuthProperties::optional_oauth_params()
+    const {
+  std::unordered_map<std::string, std::string> params;
+  if (auto audience = Get(kAudience); !audience.empty()) {
+    params.emplace(kAudience.key(), std::move(audience));
+  }
+  if (auto resource = Get(kResource); !resource.empty()) {
+    params.emplace(kResource.key(), std::move(resource));
+  }
+  return params;
+}
+
+Result<AuthProperties> AuthProperties::FromProperties(
+    const std::unordered_map<std::string, std::string>& properties) {
+  AuthProperties config;
+  config.configs_ = properties;
+
+  // Parse client_id/client_secret from credential
+  if (auto cred = config.credential(); !cred.empty()) {
+    auto [id, secret] = ParseCredential(cred);
+    config.client_id_ = std::move(id);
+    config.client_secret_ = std::move(secret);
+  }
+
+  // Resolve token endpoint: if not explicitly set, derive from catalog URI
+  if (properties.find(kOAuth2ServerUri.key()) == properties.end() ||
+      properties.at(kOAuth2ServerUri.key()).empty()) {
+    auto uri_it = properties.find(RestCatalogProperties::kUri.key());
+    if (uri_it != properties.end() && !uri_it->second.empty()) {
+      std::string_view base = uri_it->second;
+      while (!base.empty() && base.back() == '/') {
+        base.remove_suffix(1);
+      }
+      config.Set(kOAuth2ServerUri,
+                 std::string(base) + "/" + std::string(kOAuth2ServerUri.value()));
+    }
+  }
+
+  // TODO(lishuxu): Parse JWT exp claim from token to set expires_at_millis_.
+
+  return config;
+}
+
+}  // namespace iceberg::rest::auth

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

@@ -19,55 +19,88 @@
 
 #pragma once
 
+#include <cstdint>
+#include <optional>
 #include <string>
-#include <string_view>
+#include <unordered_map>
+
+#include "iceberg/catalog/rest/iceberg_rest_export.h"
+#include "iceberg/result.h"
+#include "iceberg/util/config.h"
 
 /// \file iceberg/catalog/rest/auth/auth_properties.h
-/// \brief Property keys and constants for REST catalog authentication.
+/// \brief Property keys and configuration for REST catalog authentication.
 
 namespace iceberg::rest::auth {
 
-/// \brief Property keys and constants for authentication configuration.
-///
-/// This struct defines all the property keys used to configure authentication
-/// for the REST catalog. It follows the same naming conventions as Java Iceberg.
-struct AuthProperties {
-  /// \brief Property key for specifying the authentication type.
+/// \brief Authentication properties
+class ICEBERG_REST_EXPORT AuthProperties : public ConfigBase<AuthProperties> {
+ public:
+  template <typename T>
+  using Entry = const ConfigBase<AuthProperties>::Entry<T>;
+
+  // ---- Authentication type constants (not Entry-based) ----
+
   inline static const std::string kAuthType = "rest.auth.type";
-  /// \brief Authentication type: no authentication.
   inline static const std::string kAuthTypeNone = "none";
-  /// \brief Authentication type: HTTP Basic authentication.
   inline static const std::string kAuthTypeBasic = "basic";
-  /// \brief Authentication type: OAuth2 authentication.
   inline static const std::string kAuthTypeOAuth2 = "oauth2";
-  /// \brief Authentication type: AWS SigV4 authentication.
   inline static const std::string kAuthTypeSigV4 = "sigv4";
 
-  /// \brief Property key for Basic auth username.
+  // ---- Basic auth entries ----
+
   inline static const std::string kBasicUsername = "rest.auth.basic.username";
-  /// \brief Property key for Basic auth password.
   inline static const std::string kBasicPassword = "rest.auth.basic.password";
 
-  /// \brief Property key for OAuth2 token (bearer token).
-  inline static const std::string kOAuth2Token = "token";
-  /// \brief Property key for OAuth2 credential (client_id:client_secret).
-  inline static const std::string kOAuth2Credential = "credential";
-  /// \brief Property key for OAuth2 scope.
-  inline static const std::string kOAuth2Scope = "scope";
-  /// \brief Property key for OAuth2 server URI.
-  inline static const std::string kOAuth2ServerUri = "oauth2-server-uri";
-  /// \brief Property key for enabling token refresh.
-  inline static const std::string kOAuth2TokenRefreshEnabled = "token-refresh-enabled";
-  /// \brief Default OAuth2 scope for catalog operations.
-  inline static const std::string kOAuth2DefaultScope = "catalog";
-
-  /// \brief Property key for SigV4 region.
+  // ---- SigV4 entries ----
+
   inline static const std::string kSigV4Region = "rest.auth.sigv4.region";
-  /// \brief Property key for SigV4 service name.
   inline static const std::string kSigV4Service = "rest.auth.sigv4.service";
-  /// \brief Property key for SigV4 delegate auth type.
   inline static const std::string kSigV4DelegateAuthType =
       "rest.auth.sigv4.delegate-auth-type";
+
+  // ---- OAuth2 entries ----
+
+  inline static Entry<std::string> kToken{"token", ""};
+  inline static Entry<std::string> kCredential{"credential", ""};
+  inline static Entry<std::string> kScope{"scope", "catalog"};
+  inline static Entry<std::string> kOAuth2ServerUri{"oauth2-server-uri",
+                                                    "v1/oauth/tokens"};
+  inline static Entry<bool> kKeepRefreshed{"token-refresh-enabled", true};
+  inline static Entry<bool> kExchangeEnabled{"token-exchange-enabled", true};
+  inline static Entry<std::string> kAudience{"audience", ""};
+  inline static Entry<std::string> kResource{"resource", ""};
+
+  /// \brief Build an AuthProperties from a properties map.
+  static Result<AuthProperties> FromProperties(
+      const std::unordered_map<std::string, std::string>& properties);
+
+  /// \brief Get the bearer token.
+  std::string token() const { return Get(kToken); }
+  /// \brief Get the raw credential string.
+  std::string credential() const { return Get(kCredential); }
+  /// \brief Get the OAuth2 scope.
+  std::string scope() const { return Get(kScope); }
+  /// \brief Get the token endpoint URI.
+  std::string oauth2_server_uri() const { return Get(kOAuth2ServerUri); }
+  /// \brief Whether token refresh is enabled.
+  bool keep_refreshed() const { return Get(kKeepRefreshed); }
+  /// \brief Whether token exchange is enabled.
+  bool exchange_enabled() const { return Get(kExchangeEnabled); }
+
+  /// \brief Parsed client_id from credential (empty if no colon).
+  const std::string& client_id() const { return client_id_; }
+  /// \brief Parsed client_secret from credential.
+  const std::string& client_secret() const { return client_secret_; }
+
+  /// \brief Build optional OAuth params (audience, resource) from config.
+  std::unordered_map<std::string, std::string> optional_oauth_params() const;
+
+ private:
+  std::string client_id_;
+  std::string client_secret_;
+  std::string token_type_;
+  std::optional<int64_t> expires_at_millis_;
 };
 
 }  // namespace iceberg::rest::auth

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

@@ -21,6 +21,8 @@
 
 #include <utility>
 
+#include "iceberg/catalog/rest/auth/oauth2_util.h"
+
 namespace iceberg::rest::auth {
 
 namespace {
@@ -49,4 +51,13 @@ std::shared_ptr<AuthSession> AuthSession::MakeDefault(
   return std::make_shared<DefaultAuthSession>(std::move(headers));
 }
 
+std::shared_ptr<AuthSession> AuthSession::MakeOAuth2(
+    const OAuthTokenResponse& initial_token, const std::string& /*token_endpoint*/,
+    const std::string& /*client_id*/, const std::string& /*client_secret*/,
+    const std::string& /*scope*/, HttpClient& /*client*/) {
+  // TODO(lishuxu): Create OAuth2AuthSession with auto-refresh support.
+  return MakeDefault({{std::string(kAuthorizationHeader),
+                       std::string(kBearerPrefix) + initial_token.access_token}});
+}
+
 }  // namespace iceberg::rest::auth