From 0c45d70a8d6e13cfcea7d906573e584736947244 Mon Sep 17 00:00:00 2001 From: "shuxu.li" Date: Sun, 2 Aug 2026 22:15:48 +0800 Subject: [PATCH 1/3] feat(rest): support OAuth token exchange sessions Add RFC 8693 token exchange support, including token type helpers, request form construction, OAuth endpoint normalization, and response handling. Preserve OAuth metadata in auth sessions and create contextual and table-scoped child sessions from direct tokens, credentials, or typed tokens. Disable child refresh until session lifecycle management is available. --- src/iceberg/catalog/rest/auth/auth_manager.cc | 112 ++++- .../catalog/rest/auth/auth_properties.cc | 40 +- .../catalog/rest/auth/auth_properties.h | 14 + src/iceberg/catalog/rest/auth/auth_session.cc | 55 ++- src/iceberg/catalog/rest/auth/auth_session.h | 14 + src/iceberg/catalog/rest/auth/oauth2_util.cc | 114 ++++- src/iceberg/catalog/rest/auth/oauth2_util.h | 48 ++ src/iceberg/catalog/rest/resource_paths.cc | 2 +- src/iceberg/catalog/rest/resource_paths.h | 2 +- src/iceberg/test/auth_manager_test.cc | 426 ++++++++++++++++++ .../test/rest_catalog_integration_test.cc | 103 +++++ src/iceberg/test/rest_util_test.cc | 8 + 12 files changed, 890 insertions(+), 48 deletions(-) diff --git a/src/iceberg/catalog/rest/auth/auth_manager.cc b/src/iceberg/catalog/rest/auth/auth_manager.cc index 10290489a..6b2a289ba 100644 --- a/src/iceberg/catalog/rest/auth/auth_manager.cc +++ b/src/iceberg/catalog/rest/auth/auth_manager.cc @@ -25,6 +25,7 @@ #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/catalog/session_context.h" #include "iceberg/util/base64.h" #include "iceberg/util/macros.h" @@ -121,6 +122,7 @@ class OAuth2Manager : public AuthManager { HttpClient& client, const std::unordered_map& properties) override { ICEBERG_ASSIGN_OR_RAISE(auto config, AuthProperties::FromProperties(properties)); + shared_client_ = &client; // Reuse token from init phase. if (init_token_response_.has_value()) { @@ -134,7 +136,15 @@ class OAuth2Manager : public AuthManager { // If token is provided, use it directly. if (!config.token().empty()) { - return AuthSession::MakeDefault(AuthHeaders(config.token())); + OAuthTokenResponse token_response{ + .access_token = config.token(), + .token_type = "bearer", + .issued_token_type = AuthProperties::kAccessTokenType, + }; + return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(), + config.client_id(), config.client_secret(), + config.scope(), /*keep_refreshed=*/false, + config.optional_oauth_params(), client); } // Fetch a new token using client_credentials grant. @@ -148,15 +158,109 @@ class OAuth2Manager : public AuthManager { config.optional_oauth_params(), client); } - return AuthSession::MakeDefault({}); + return MakeSession(AccessTokenResponse(""), config, /*keep_refreshed=*/false); + } + + Result> ContextualSession( + const SessionContext& context, std::shared_ptr parent) override { + return MaybeCreateChildSession(context.credentials, /*allow_credential=*/true, + std::move(parent)); } - // TODO(lishuxu): Override TableSession() for token exchange (RFC 8693). - // TODO(lishuxu): Override ContextualSession() for per-context exchange. + Result> TableSession( + [[maybe_unused]] const TableIdentifier& table, + const std::unordered_map& properties, + std::shared_ptr parent) override { + return MaybeCreateChildSession(FilterTableSessionProperties(properties), + /*allow_credential=*/false, std::move(parent)); + } private: + static OAuthTokenResponse AccessTokenResponse(std::string token) { + return { + .access_token = std::move(token), + .token_type = "bearer", + .issued_token_type = AuthProperties::kAccessTokenType, + }; + } + + static Result ChildConfig(const OAuth2SessionInfo& parent_info, + const std::string& credential) { + auto properties = parent_info.optional_oauth_params; + properties[AuthProperties::kCredential.key()] = credential; + properties[AuthProperties::kScope.key()] = parent_info.scope; + properties[AuthProperties::kOAuth2ServerUri.key()] = parent_info.oauth2_server_uri; + return AuthProperties::FromProperties(properties); + } + + Result> MakeSession( + const OAuthTokenResponse& token_response, const AuthProperties& config, + bool keep_refreshed) const { + ICEBERG_PRECHECK(shared_client_ != nullptr, + "OAuth2 catalog session must be initialized before child sessions"); + return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(), + config.client_id(), config.client_secret(), + config.scope(), keep_refreshed, + config.optional_oauth_params(), *shared_client_); + } + + Result> MaybeCreateChildSession( + const std::unordered_map& credentials, + bool allow_credential, std::shared_ptr parent) const { + auto token_it = credentials.find(AuthProperties::kToken.key()); + auto credential_it = credentials.find(AuthProperties::kCredential.key()); + auto typed_token = FindPreferredTypedToken(credentials); + if (token_it == credentials.end() && + (!allow_credential || credential_it == credentials.end()) && + !typed_token.has_value()) { + return parent; + } + + ICEBERG_PRECHECK(shared_client_ != nullptr, + "OAuth2 catalog session must be initialized before child sessions"); + auto parent_info = parent->OAuth2Info(); + ICEBERG_PRECHECK(parent_info.has_value(), + "OAuth2 child session requires OAuth2 parent metadata"); + + if (token_it != credentials.end()) { + ICEBERG_ASSIGN_OR_RAISE(auto config, + ChildConfig(*parent_info, parent_info->credential)); + return MakeSession(AccessTokenResponse(token_it->second), config, + /*keep_refreshed=*/false); + } + + if (allow_credential && credential_it != credentials.end()) { + ICEBERG_ASSIGN_OR_RAISE(auto config, + ChildConfig(*parent_info, credential_it->second)); + ICEBERG_ASSIGN_OR_RAISE(auto response, + FetchToken(*shared_client_, *parent, config)); + return MakeSession(response, config, /*keep_refreshed=*/false); + } + + std::optional actor; + if (!parent_info->token.empty()) { + actor = OAuth2Token{ + .token_type = parent_info->issued_token_type, + .token = parent_info->token, + }; + } + TokenExchangeRequest request{ + .oauth2_server_uri = parent_info->oauth2_server_uri, + .subject = std::move(*typed_token), + .actor = std::move(actor), + .scope = parent_info->scope, + .optional_oauth_params = parent_info->optional_oauth_params, + }; + ICEBERG_ASSIGN_OR_RAISE(auto response, + ExchangeToken(*shared_client_, *parent, {}, request)); + ICEBERG_ASSIGN_OR_RAISE(auto config, + ChildConfig(*parent_info, parent_info->credential)); + return MakeSession(response, config, /*keep_refreshed=*/false); + } + /// Cached token from InitSession std::optional init_token_response_; + HttpClient* shared_client_ = nullptr; }; Result> MakeOAuth2Manager( diff --git a/src/iceberg/catalog/rest/auth/auth_properties.cc b/src/iceberg/catalog/rest/auth/auth_properties.cc index dcf16782c..67d9319d8 100644 --- a/src/iceberg/catalog/rest/auth/auth_properties.cc +++ b/src/iceberg/catalog/rest/auth/auth_properties.cc @@ -22,6 +22,7 @@ #include #include "iceberg/catalog/rest/catalog_properties.h" +#include "iceberg/catalog/rest/rest_util.h" namespace iceberg::rest::auth { @@ -35,6 +36,30 @@ std::pair ParseCredential(const std::string& credentia return {credential.substr(0, colon_pos), credential.substr(colon_pos + 1)}; } +Result ResolveOAuth2ServerUri( + const std::unordered_map& properties) { + auto endpoint_it = properties.find(AuthProperties::kOAuth2ServerUri.key()); + std::string endpoint = endpoint_it == properties.end() || endpoint_it->second.empty() + ? AuthProperties::kOAuth2ServerUri.value() + : endpoint_it->second; + + if (endpoint.starts_with("http://") || endpoint.starts_with("https://")) { + return endpoint; + } + if (endpoint.starts_with('/')) { + return InvalidArgument("OAuth2 server URI path must not start with '/': {}", + endpoint); + } + + auto uri_it = properties.find(RestCatalogProperties::kUri.key()); + if (uri_it == properties.end() || uri_it->second.empty()) { + return endpoint; + } + + return std::string(TrimTrailingSlash(uri_it->second)) + "/" + + std::string(TrimTrailingSlash(endpoint)); +} + } // namespace std::unordered_map AuthProperties::optional_oauth_params() @@ -61,19 +86,8 @@ Result AuthProperties::FromProperties( 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())); - } - } + ICEBERG_ASSIGN_OR_RAISE(auto oauth2_server_uri, ResolveOAuth2ServerUri(properties)); + config.Set(kOAuth2ServerUri, std::move(oauth2_server_uri)); // TODO(lishuxu): Parse JWT exp claim from token to set expires_at_millis_. diff --git a/src/iceberg/catalog/rest/auth/auth_properties.h b/src/iceberg/catalog/rest/auth/auth_properties.h index a699569c1..8784194cc 100644 --- a/src/iceberg/catalog/rest/auth/auth_properties.h +++ b/src/iceberg/catalog/rest/auth/auth_properties.h @@ -82,6 +82,20 @@ class ICEBERG_REST_EXPORT AuthProperties : public ConfigBase { inline static Entry kAudience{"audience", ""}; inline static Entry kResource{"resource", ""}; + // ---- OAuth2 token type constants ---- + + inline static const std::string kAccessTokenType = + "urn:ietf:params:oauth:token-type:access_token"; + inline static const std::string kRefreshTokenType = + "urn:ietf:params:oauth:token-type:refresh_token"; + inline static const std::string kIdTokenType = + "urn:ietf:params:oauth:token-type:id_token"; + inline static const std::string kSaml1TokenType = + "urn:ietf:params:oauth:token-type:saml1"; + inline static const std::string kSaml2TokenType = + "urn:ietf:params:oauth:token-type:saml2"; + inline static const std::string kJwtTokenType = "urn:ietf:params:oauth:token-type:jwt"; + /// \brief Build an AuthProperties from a properties map. static Result FromProperties( const std::unordered_map& properties); diff --git a/src/iceberg/catalog/rest/auth/auth_session.cc b/src/iceberg/catalog/rest/auth/auth_session.cc index 545ee00b1..fbb1899d2 100644 --- a/src/iceberg/catalog/rest/auth/auth_session.cc +++ b/src/iceberg/catalog/rest/auth/auth_session.cc @@ -85,6 +85,18 @@ class OAuth2AuthSession : public AuthSession, return request; } + std::optional OAuth2Info() const override { + std::shared_lock lock(mutex_); + return OAuth2SessionInfo{ + .token = token_, + .issued_token_type = issued_token_type_, + .credential = Credential(config_), + .scope = config_.scope, + .oauth2_server_uri = config_.token_endpoint, + .optional_oauth_params = config_.optional_oauth_params, + }; + } + Status Close() override { return CloseImpl(); } ~OAuth2AuthSession() override { std::ignore = CloseImpl(); } @@ -107,12 +119,15 @@ class OAuth2AuthSession : public AuthSession, return {}; } + static std::string Credential(const Config& config) { + return config.client_id.empty() ? config.client_secret + : config.client_id + ":" + config.client_secret; + } + static Result MakeRefreshProperties(const Config& config) { std::unordered_map properties = config.optional_oauth_params; - properties[AuthProperties::kCredential.key()] = - config.client_id.empty() ? config.client_secret - : config.client_id + ":" + config.client_secret; + properties[AuthProperties::kCredential.key()] = Credential(config); properties[AuthProperties::kScope.key()] = config.scope; properties[AuthProperties::kOAuth2ServerUri.key()] = config.token_endpoint; @@ -141,11 +156,14 @@ class OAuth2AuthSession : public AuthSession, OAuth2AuthSession& session_; }; - void SetInitialToken(const OAuthTokenResponse& token_response) { + void UpdateTokenState(const OAuthTokenResponse& token_response) { token_ = token_response.access_token; - headers_ = {{std::string(kAuthorizationHeader), std::string(kBearerPrefix) + token_}}; + issued_token_type_ = token_response.issued_token_type.empty() + ? AuthProperties::kAccessTokenType + : token_response.issued_token_type; + headers_ = AuthHeaders(token_); - // Determine expiration time + expires_at_ = std::chrono::steady_clock::time_point{}; if (token_response.expires_in_secs.has_value()) { expires_at_ = std::chrono::steady_clock::now() + std::chrono::seconds(*token_response.expires_in_secs); @@ -157,6 +175,10 @@ class OAuth2AuthSession : public AuthSession, std::chrono::system_clock::time_point(std::chrono::milliseconds(*exp_ms)); expires_at_ = now_steady + (exp_sys - now_sys); } + } + + void SetInitialToken(const OAuthTokenResponse& token_response) { + UpdateTokenState(token_response); if (config_.keep_refreshed && expires_at_ != std::chrono::steady_clock::time_point{}) { @@ -184,23 +206,7 @@ class OAuth2AuthSession : public AuthSession, auto& response = result.value(); { std::unique_lock lock(mutex_); - token_ = response.access_token; - headers_ = { - {std::string(kAuthorizationHeader), std::string(kBearerPrefix) + token_}}; - - // Reset before deriving new expiry - expires_at_ = std::chrono::steady_clock::time_point{}; - - if (response.expires_in_secs.has_value()) { - expires_at_ = std::chrono::steady_clock::now() + - std::chrono::seconds(*response.expires_in_secs); - } else if (auto exp_ms = ExpiresAtMillis(token_); exp_ms.has_value()) { - auto now_sys = std::chrono::system_clock::now(); - auto now_steady = std::chrono::steady_clock::now(); - auto exp_sys = - std::chrono::system_clock::time_point(std::chrono::milliseconds(*exp_ms)); - expires_at_ = now_steady + (exp_sys - now_sys); - } + UpdateTokenState(response); } // Note: ScheduleRefresh must be called outside the lock. ScheduleRefresh(); @@ -262,8 +268,9 @@ class OAuth2AuthSession : public AuthSession, return std::max(wait_time, std::chrono::milliseconds(10)); } - mutable std::shared_mutex mutex_; // protects token_, headers_, expires_at_ + mutable std::shared_mutex mutex_; // protects token state, headers, and expiration std::string token_; + std::string issued_token_type_; std::unordered_map headers_; std::chrono::steady_clock::time_point expires_at_{}; diff --git a/src/iceberg/catalog/rest/auth/auth_session.h b/src/iceberg/catalog/rest/auth/auth_session.h index 3d0063a04..cfd32355a 100644 --- a/src/iceberg/catalog/rest/auth/auth_session.h +++ b/src/iceberg/catalog/rest/auth/auth_session.h @@ -20,6 +20,7 @@ #pragma once #include +#include #include #include @@ -33,6 +34,16 @@ namespace iceberg::rest::auth { +/// \brief OAuth2 metadata used to derive child authentication sessions. +struct ICEBERG_REST_EXPORT OAuth2SessionInfo { + std::string token; + std::string issued_token_type; + std::string credential; + std::string scope; + std::string oauth2_server_uri; + std::unordered_map optional_oauth_params; +}; + /// \brief An authentication session that can authenticate outgoing HTTP requests. class ICEBERG_REST_EXPORT AuthSession { public: @@ -54,6 +65,9 @@ class ICEBERG_REST_EXPORT AuthSession { /// - RestError: HTTP errors from authentication service virtual Result Authenticate(HttpRequest request) = 0; + /// \brief Return OAuth2 metadata when this is an OAuth2 session. + virtual std::optional OAuth2Info() const { return std::nullopt; } + /// \brief Close the session and release any resources. /// /// This method is called when the session is no longer needed. For stateful diff --git a/src/iceberg/catalog/rest/auth/oauth2_util.cc b/src/iceberg/catalog/rest/auth/oauth2_util.cc index d5e94821c..1f9da22b5 100644 --- a/src/iceberg/catalog/rest/auth/oauth2_util.cc +++ b/src/iceberg/catalog/rest/auth/oauth2_util.cc @@ -36,9 +36,22 @@ namespace { constexpr std::string_view kGrantType = "grant_type"; constexpr std::string_view kClientCredentials = "client_credentials"; +constexpr std::string_view kTokenExchange = + "urn:ietf:params:oauth:grant-type:token-exchange"; constexpr std::string_view kClientId = "client_id"; constexpr std::string_view kClientSecret = "client_secret"; constexpr std::string_view kScope = "scope"; +constexpr std::string_view kSubjectToken = "subject_token"; +constexpr std::string_view kSubjectTokenType = "subject_token_type"; +constexpr std::string_view kActorToken = "actor_token"; +constexpr std::string_view kActorTokenType = "actor_token_type"; + +Result ParseTokenResponse(const std::string& response_body) { + ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response_body)); + ICEBERG_ASSIGN_OR_RAISE(auto token_response, FromJson(json)); + ICEBERG_RETURN_UNEXPECTED(token_response.Validate()); + return token_response; +} } // namespace @@ -49,6 +62,101 @@ std::unordered_map AuthHeaders(const std::string& toke return {}; } +bool IsValidTokenType(std::string_view token_type) { + return token_type == AuthProperties::kAccessTokenType || + token_type == AuthProperties::kRefreshTokenType || + token_type == AuthProperties::kIdTokenType || + token_type == AuthProperties::kSaml1TokenType || + token_type == AuthProperties::kSaml2TokenType || + token_type == AuthProperties::kJwtTokenType; +} + +std::array TokenPreferenceOrder() { + return { + std::string_view(AuthProperties::kIdTokenType), + std::string_view(AuthProperties::kAccessTokenType), + std::string_view(AuthProperties::kJwtTokenType), + std::string_view(AuthProperties::kSaml2TokenType), + std::string_view(AuthProperties::kSaml1TokenType), + }; +} + +std::optional FindPreferredTypedToken( + const std::unordered_map& credentials) { + for (std::string_view token_type : TokenPreferenceOrder()) { + auto token_it = credentials.find(std::string(token_type)); + if (token_it != credentials.end()) { + return OAuth2Token{ + .token_type = token_it->first, + .token = token_it->second, + }; + } + } + return std::nullopt; +} + +std::unordered_map FilterTableSessionProperties( + const std::unordered_map& properties) { + std::unordered_map filtered; + auto token_it = properties.find(AuthProperties::kToken.key()); + if (token_it != properties.end()) { + filtered.emplace(token_it->first, token_it->second); + } + for (std::string_view token_type : TokenPreferenceOrder()) { + auto token_it = properties.find(std::string(token_type)); + if (token_it != properties.end()) { + filtered.emplace(token_it->first, token_it->second); + } + } + return filtered; +} + +Result> BuildTokenExchangeForm( + const TokenExchangeRequest& request) { + if (request.subject.token.empty()) { + return InvalidArgument("OAuth2 subject token must not be empty"); + } + if (!IsValidTokenType(request.subject.token_type)) { + return InvalidArgument("Invalid OAuth2 subject token type: '{}'", + request.subject.token_type); + } + if (request.actor.has_value()) { + if (request.actor->token.empty()) { + return InvalidArgument("OAuth2 actor token must not be empty"); + } + if (!IsValidTokenType(request.actor->token_type)) { + return InvalidArgument("Invalid OAuth2 actor token type: '{}'", + request.actor->token_type); + } + } + + std::unordered_map form_data{ + {std::string(kGrantType), std::string(kTokenExchange)}, + {std::string(kScope), request.scope}, + {std::string(kSubjectToken), request.subject.token}, + {std::string(kSubjectTokenType), request.subject.token_type}, + }; + if (request.actor.has_value()) { + form_data.emplace(kActorToken, request.actor->token); + form_data.emplace(kActorTokenType, request.actor->token_type); + } + for (const auto& [key, value] : request.optional_oauth_params) { + form_data.insert_or_assign(key, value); + } + return form_data; +} + +Result ExchangeToken( + HttpClient& client, AuthSession& session, + const std::unordered_map& extra_headers, + const TokenExchangeRequest& request) { + ICEBERG_ASSIGN_OR_RAISE(auto form_data, BuildTokenExchangeForm(request)); + ICEBERG_ASSIGN_OR_RAISE( + auto response, client.PostForm(request.oauth2_server_uri, form_data, extra_headers, + *OAuthErrorHandler::Instance(), session)); + return ParseTokenResponse(response.body()); +} + Result FetchToken(HttpClient& client, AuthSession& session, const AuthProperties& properties) { std::unordered_map form_data{ @@ -67,11 +175,7 @@ Result FetchToken(HttpClient& client, AuthSession& session, auto response, client.PostForm(properties.oauth2_server_uri(), form_data, /*headers=*/{}, *OAuthErrorHandler::Instance(), session)); - - ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body())); - ICEBERG_ASSIGN_OR_RAISE(auto token_response, FromJson(json)); - ICEBERG_RETURN_UNEXPECTED(token_response.Validate()); - return token_response; + return ParseTokenResponse(response.body()); } std::optional ExpiresAtMillis(std::string_view token) { diff --git a/src/iceberg/catalog/rest/auth/oauth2_util.h b/src/iceberg/catalog/rest/auth/oauth2_util.h index 428ebc385..fe4f6723a 100644 --- a/src/iceberg/catalog/rest/auth/oauth2_util.h +++ b/src/iceberg/catalog/rest/auth/oauth2_util.h @@ -19,6 +19,7 @@ #pragma once +#include #include #include #include @@ -38,6 +39,19 @@ namespace iceberg::rest::auth { inline constexpr std::string_view kAuthorizationHeader = "Authorization"; inline constexpr std::string_view kBearerPrefix = "Bearer "; +struct ICEBERG_REST_EXPORT OAuth2Token { + std::string token_type; + std::string token; +}; + +struct ICEBERG_REST_EXPORT TokenExchangeRequest { + std::string oauth2_server_uri; + OAuth2Token subject; + std::optional actor; + std::string scope; + std::unordered_map optional_oauth_params; +}; + /// \brief Fetch an OAuth2 token using the client_credentials grant type. /// /// \param client HTTP client to use for the request. @@ -55,6 +69,40 @@ ICEBERG_REST_EXPORT Result FetchToken( ICEBERG_REST_EXPORT std::unordered_map AuthHeaders( const std::string& token); +/// \brief Return whether a token type is a supported RFC token type. +ICEBERG_REST_EXPORT bool IsValidTokenType(std::string_view token_type); + +/// \brief Return the preferred order for typed OAuth tokens. +ICEBERG_REST_EXPORT std::array TokenPreferenceOrder(); + +/// \brief Find the highest-preference typed OAuth token in credentials. +ICEBERG_REST_EXPORT std::optional FindPreferredTypedToken( + const std::unordered_map& credentials); + +/// \brief Filter table session properties to allowed OAuth credentials. +ICEBERG_REST_EXPORT std::unordered_map +FilterTableSessionProperties( + const std::unordered_map& properties); + +/// \brief Build RFC 8693 token exchange form data. +/// +/// \param request Token exchange request values. +/// \return Form data or an error if token values are invalid. +ICEBERG_REST_EXPORT Result> +BuildTokenExchangeForm(const TokenExchangeRequest& request); + +/// \brief Exchange an OAuth2 token using the RFC 8693 grant type. +/// +/// \param client HTTP client to use for the request. +/// \param session Auth session for the request headers. +/// \param extra_headers Request headers applied before session authentication. +/// \param request Token exchange endpoint and form values. +/// \return The token response or an error. +ICEBERG_REST_EXPORT Result ExchangeToken( + HttpClient& client, AuthSession& session, + const std::unordered_map& extra_headers, + const TokenExchangeRequest& request); + /// \brief Extract expiration time from a JWT token. /// /// Decodes the JWT payload (base64url) and reads the "exp" claim. diff --git a/src/iceberg/catalog/rest/resource_paths.cc b/src/iceberg/catalog/rest/resource_paths.cc index d18dd4636..3a70eb113 100644 --- a/src/iceberg/catalog/rest/resource_paths.cc +++ b/src/iceberg/catalog/rest/resource_paths.cc @@ -51,7 +51,7 @@ Result ResourcePaths::Config() const { } Result ResourcePaths::OAuth2Tokens() const { - return std::format("{}/v1/{}oauth/tokens", base_uri_, prefix_); + return std::format("{}/v1/oauth/tokens", base_uri_); } Result ResourcePaths::Namespaces() const { diff --git a/src/iceberg/catalog/rest/resource_paths.h b/src/iceberg/catalog/rest/resource_paths.h index 27135bb22..99e748231 100644 --- a/src/iceberg/catalog/rest/resource_paths.h +++ b/src/iceberg/catalog/rest/resource_paths.h @@ -49,7 +49,7 @@ class ICEBERG_REST_EXPORT ResourcePaths { /// \brief Get the /v1/config endpoint path. Result Config() const; - /// \brief Get the /v1/{prefix}/oauth/tokens endpoint path. + /// \brief Get the /v1/oauth/tokens endpoint path. Result OAuth2Tokens() const; /// \brief Get the /v1/{prefix}/namespaces endpoint path. diff --git a/src/iceberg/test/auth_manager_test.cc b/src/iceberg/test/auth_manager_test.cc index 19526b7e3..c9a03e91c 100644 --- a/src/iceberg/test/auth_manager_test.cc +++ b/src/iceberg/test/auth_manager_test.cc @@ -37,11 +37,13 @@ #include "iceberg/catalog/rest/auth/auth_session.h" #include "iceberg/catalog/rest/auth/oauth2_util.h" #include "iceberg/catalog/rest/auth/token_refresh_scheduler.h" +#include "iceberg/catalog/rest/catalog_properties.h" #include "iceberg/catalog/rest/error_handlers.h" #include "iceberg/catalog/rest/http_client.h" #include "iceberg/catalog/rest/json_serde_internal.h" #include "iceberg/catalog/session_context.h" #include "iceberg/json_serde_internal.h" +#include "iceberg/table_identifier.h" #include "iceberg/test/matchers.h" #include "iceberg/util/base64.h" @@ -69,6 +71,258 @@ class AuthManagerTest : public ::testing::Test { HttpClient client_{{}}; }; +TEST(OAuth2UtilTest, TokenTypeConstantsUseRfcUrns) { + EXPECT_EQ(AuthProperties::kAccessTokenType, + "urn:ietf:params:oauth:token-type:access_token"); + EXPECT_EQ(AuthProperties::kRefreshTokenType, + "urn:ietf:params:oauth:token-type:refresh_token"); + EXPECT_EQ(AuthProperties::kIdTokenType, "urn:ietf:params:oauth:token-type:id_token"); + EXPECT_EQ(AuthProperties::kSaml1TokenType, "urn:ietf:params:oauth:token-type:saml1"); + EXPECT_EQ(AuthProperties::kSaml2TokenType, "urn:ietf:params:oauth:token-type:saml2"); + EXPECT_EQ(AuthProperties::kJwtTokenType, "urn:ietf:params:oauth:token-type:jwt"); +} + +TEST(OAuth2UtilTest, ValidTokenTypesIncludeRefreshToken) { + EXPECT_TRUE(IsValidTokenType(AuthProperties::kAccessTokenType)); + EXPECT_TRUE(IsValidTokenType(AuthProperties::kRefreshTokenType)); + EXPECT_TRUE(IsValidTokenType(AuthProperties::kIdTokenType)); + EXPECT_TRUE(IsValidTokenType(AuthProperties::kSaml1TokenType)); + EXPECT_TRUE(IsValidTokenType(AuthProperties::kSaml2TokenType)); + EXPECT_TRUE(IsValidTokenType(AuthProperties::kJwtTokenType)); + EXPECT_FALSE(IsValidTokenType("urn:ietf:params:oauth:token-type:unknown")); +} + +TEST(OAuth2UtilTest, TokenPreferenceOrder) { + auto order = TokenPreferenceOrder(); + ASSERT_EQ(order.size(), 5); + EXPECT_EQ(order[0], AuthProperties::kIdTokenType); + EXPECT_EQ(order[1], AuthProperties::kAccessTokenType); + EXPECT_EQ(order[2], AuthProperties::kJwtTokenType); + EXPECT_EQ(order[3], AuthProperties::kSaml2TokenType); + EXPECT_EQ(order[4], AuthProperties::kSaml1TokenType); +} + +TEST(OAuth2UtilTest, FindPreferredTypedTokenUsesPreferenceOrder) { + std::unordered_map credentials = { + {AuthProperties::kAccessTokenType, "access-token"}, + {AuthProperties::kJwtTokenType, "jwt-token"}, + {AuthProperties::kIdTokenType, "id-token"}, + {AuthProperties::kSaml2TokenType, "saml2-token"}, + {AuthProperties::kSaml1TokenType, "saml1-token"}, + }; + + auto token = FindPreferredTypedToken(credentials); + ASSERT_TRUE(token.has_value()); + EXPECT_EQ(token->token_type, AuthProperties::kIdTokenType); + EXPECT_EQ(token->token, "id-token"); + + credentials.erase(AuthProperties::kIdTokenType); + token = FindPreferredTypedToken(credentials); + ASSERT_TRUE(token.has_value()); + EXPECT_EQ(token->token_type, AuthProperties::kAccessTokenType); + EXPECT_EQ(token->token, "access-token"); + + credentials.clear(); + EXPECT_FALSE(FindPreferredTypedToken(credentials).has_value()); +} + +TEST(OAuth2UtilTest, FilterTableSessionPropertiesUsesAllowList) { + std::unordered_map properties = { + {AuthProperties::kToken.key(), "bearer-token"}, + {AuthProperties::kCredential.key(), "client:secret"}, + {AuthProperties::kScope.key(), "catalog"}, + {AuthProperties::kAccessTokenType, "access-token"}, + {AuthProperties::kRefreshTokenType, "refresh-token"}, + {AuthProperties::kIdTokenType, "id-token"}, + {AuthProperties::kJwtTokenType, "jwt-token"}, + {AuthProperties::kSaml2TokenType, "saml2-token"}, + {AuthProperties::kSaml1TokenType, "saml1-token"}, + {"unrelated", "value"}, + }; + + auto filtered = FilterTableSessionProperties(properties); + EXPECT_EQ(filtered.size(), 6); + EXPECT_EQ(filtered.at(AuthProperties::kToken.key()), "bearer-token"); + EXPECT_EQ(filtered.at(AuthProperties::kAccessTokenType), "access-token"); + EXPECT_EQ(filtered.at(AuthProperties::kIdTokenType), "id-token"); + EXPECT_EQ(filtered.at(AuthProperties::kJwtTokenType), "jwt-token"); + EXPECT_EQ(filtered.at(AuthProperties::kSaml2TokenType), "saml2-token"); + EXPECT_EQ(filtered.at(AuthProperties::kSaml1TokenType), "saml1-token"); + EXPECT_FALSE(filtered.contains(AuthProperties::kCredential.key())); + EXPECT_FALSE(filtered.contains(AuthProperties::kScope.key())); + EXPECT_FALSE(filtered.contains(AuthProperties::kRefreshTokenType)); + EXPECT_FALSE(filtered.contains("unrelated")); +} + +TEST(OAuth2UtilTest, BuildsTokenExchangeFormWithoutActor) { + TokenExchangeRequest request{ + .oauth2_server_uri = "https://auth.example.com/token", + .subject = + { + .token_type = AuthProperties::kIdTokenType, + .token = "subject-token", + }, + .scope = "catalog", + .optional_oauth_params = + { + {AuthProperties::kAudience.key(), "catalog-audience"}, + {AuthProperties::kResource.key(), "catalog-resource"}, + }, + }; + + ICEBERG_UNWRAP_OR_FAIL(auto form_data, BuildTokenExchangeForm(request)); + EXPECT_EQ(form_data.size(), 6); + EXPECT_EQ(form_data.at("grant_type"), + "urn:ietf:params:oauth:grant-type:token-exchange"); + EXPECT_EQ(form_data.at("scope"), "catalog"); + EXPECT_EQ(form_data.at("subject_token"), "subject-token"); + EXPECT_EQ(form_data.at("subject_token_type"), AuthProperties::kIdTokenType); + EXPECT_EQ(form_data.at("audience"), "catalog-audience"); + EXPECT_EQ(form_data.at("resource"), "catalog-resource"); + EXPECT_FALSE(form_data.contains("actor_token")); + EXPECT_FALSE(form_data.contains("actor_token_type")); +} + +TEST(OAuth2UtilTest, BuildsTokenExchangeFormWithActor) { + TokenExchangeRequest request{ + .subject = + { + .token_type = AuthProperties::kJwtTokenType, + .token = "subject-token", + }, + .actor = + OAuth2Token{ + .token_type = AuthProperties::kAccessTokenType, + .token = "actor-token", + }, + .scope = "catalog", + }; + + ICEBERG_UNWRAP_OR_FAIL(auto form_data, BuildTokenExchangeForm(request)); + EXPECT_EQ(form_data.size(), 6); + EXPECT_EQ(form_data.at("actor_token"), "actor-token"); + EXPECT_EQ(form_data.at("actor_token_type"), AuthProperties::kAccessTokenType); +} + +TEST(OAuth2UtilTest, TokenExchangeOptionalParamsUseLastValue) { + TokenExchangeRequest request{ + .subject = + { + .token_type = AuthProperties::kAccessTokenType, + .token = "subject-token", + }, + .scope = "catalog", + .optional_oauth_params = {{"scope", "custom-scope"}}, + }; + + ICEBERG_UNWRAP_OR_FAIL(auto form_data, BuildTokenExchangeForm(request)); + EXPECT_EQ(form_data.at("scope"), "custom-scope"); +} + +TEST(OAuth2UtilTest, RejectsInvalidTokenExchangeSubject) { + TokenExchangeRequest request{ + .subject = {.token_type = "invalid-token-type", .token = "subject-token"}, + }; + + auto invalid_type = BuildTokenExchangeForm(request); + EXPECT_THAT(invalid_type, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(invalid_type, HasErrorMessage("Invalid OAuth2 subject token type")); + + request.subject = { + .token_type = AuthProperties::kAccessTokenType, + .token = "", + }; + auto empty_token = BuildTokenExchangeForm(request); + EXPECT_THAT(empty_token, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(empty_token, HasErrorMessage("subject token must not be empty")); +} + +TEST(OAuth2UtilTest, RejectsInvalidTokenExchangeActor) { + TokenExchangeRequest request{ + .subject = + { + .token_type = AuthProperties::kAccessTokenType, + .token = "subject-token", + }, + .actor = OAuth2Token{.token_type = "invalid-token-type", .token = "actor-token"}, + }; + + auto invalid_type = BuildTokenExchangeForm(request); + EXPECT_THAT(invalid_type, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(invalid_type, HasErrorMessage("Invalid OAuth2 actor token type")); + + request.actor = OAuth2Token{ + .token_type = AuthProperties::kAccessTokenType, + .token = "", + }; + auto empty_token = BuildTokenExchangeForm(request); + EXPECT_THAT(empty_token, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(empty_token, HasErrorMessage("actor token must not be empty")); +} + +TEST(AuthPropertiesTest, ResolvesDefaultOAuth2ServerUri) { + ICEBERG_UNWRAP_OR_FAIL( + auto config, + AuthProperties::FromProperties({ + {RestCatalogProperties::kUri.key(), "https://catalog.example.com/api/"}, + {RestCatalogProperties::kPrefix.key(), "warehouse"}, + })); + + EXPECT_EQ(config.oauth2_server_uri(), + "https://catalog.example.com/api/v1/oauth/tokens"); +} + +TEST(AuthPropertiesTest, ResolvesEmptyOAuth2ServerUriToDefault) { + ICEBERG_UNWRAP_OR_FAIL( + auto config, AuthProperties::FromProperties({ + {RestCatalogProperties::kUri.key(), "https://catalog.example.com"}, + {AuthProperties::kOAuth2ServerUri.key(), ""}, + })); + + EXPECT_EQ(config.oauth2_server_uri(), "https://catalog.example.com/v1/oauth/tokens"); +} + +TEST(AuthPropertiesTest, ResolvesExplicitRelativeOAuth2ServerUri) { + ICEBERG_UNWRAP_OR_FAIL( + auto config, + AuthProperties::FromProperties({ + {RestCatalogProperties::kUri.key(), "https://catalog.example.com/api/"}, + {AuthProperties::kOAuth2ServerUri.key(), "oauth/token/"}, + })); + + EXPECT_EQ(config.oauth2_server_uri(), "https://catalog.example.com/api/oauth/token"); +} + +TEST(AuthPropertiesTest, PreservesExplicitAbsoluteOAuth2ServerUri) { + ICEBERG_UNWRAP_OR_FAIL( + auto config, + AuthProperties::FromProperties({ + {RestCatalogProperties::kUri.key(), "https://catalog.example.com"}, + {AuthProperties::kOAuth2ServerUri.key(), "https://auth.example.com/token/"}, + })); + + EXPECT_EQ(config.oauth2_server_uri(), "https://auth.example.com/token/"); +} + +TEST(AuthPropertiesTest, PreservesRelativeOAuth2ServerUriWithoutCatalogUri) { + ICEBERG_UNWRAP_OR_FAIL(auto config, + AuthProperties::FromProperties({ + {AuthProperties::kOAuth2ServerUri.key(), "oauth/token"}, + })); + + EXPECT_EQ(config.oauth2_server_uri(), "oauth/token"); +} + +TEST(AuthPropertiesTest, RejectsOAuth2ServerUriWithLeadingSlash) { + auto result = AuthProperties::FromProperties({ + {RestCatalogProperties::kUri.key(), "https://catalog.example.com"}, + {AuthProperties::kOAuth2ServerUri.key(), "/v1/oauth/tokens"}, + }); + + EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); + EXPECT_THAT(result, HasErrorMessage("must not start with '/'")); +} + // Verifies loading NoopAuthManager with explicit "none" auth type TEST_F(AuthManagerTest, LoadNoopAuthManagerExplicit) { std::unordered_map properties = { @@ -115,6 +369,61 @@ TEST_F(AuthManagerTest, HttpHeadersAreCaseInsensitiveSingleValueMap) { EXPECT_EQ(headers.at("AUTHORIZATION"), "Bearer first"); } +TEST_F(AuthManagerTest, DefaultSessionPreservesRequestAuthorizationHeader) { + auto session = AuthSession::MakeDefault(AuthHeaders("parent-token")); + + ICEBERG_UNWRAP_OR_FAIL( + auto authenticated, + session->Authenticate({.headers = {{"Authorization", "Basic credentials"}}})); + + EXPECT_EQ(authenticated.headers.at("Authorization"), "Basic credentials"); + EXPECT_FALSE(session->OAuth2Info().has_value()); +} + +TEST_F(AuthManagerTest, OAuth2SessionPreservesRequestAuthorizationHeader) { + OAuthTokenResponse token_response{ + .access_token = "parent-token", + .token_type = "bearer", + }; + ICEBERG_UNWRAP_OR_FAIL( + auto session, + AuthSession::MakeOAuth2(token_response, "https://auth.example.com/token", "", "", + "catalog", /*keep_refreshed=*/false, {}, client_)); + + ICEBERG_UNWRAP_OR_FAIL( + auto authenticated, + session->Authenticate({.headers = {{"Authorization", "Basic credentials"}}})); + + EXPECT_EQ(authenticated.headers.at("Authorization"), "Basic credentials"); + + ASSERT_TRUE(session->OAuth2Info().has_value()); + EXPECT_EQ(session->OAuth2Info()->issued_token_type, AuthProperties::kAccessTokenType); +} + +TEST_F(AuthManagerTest, OAuth2SessionExposesMetadata) { + OAuthTokenResponse token_response{ + .access_token = "parent-token", + .token_type = "bearer", + .issued_token_type = AuthProperties::kJwtTokenType, + }; + ICEBERG_UNWRAP_OR_FAIL( + auto session, + AuthSession::MakeOAuth2( + token_response, "https://auth.example.com/token", "client-id", "client-secret", + "catalog", /*keep_refreshed=*/false, + {{AuthProperties::kAudience.key(), "catalog-audience"}}, client_)); + + auto info = session->OAuth2Info(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->token, "parent-token"); + EXPECT_EQ(info->issued_token_type, AuthProperties::kJwtTokenType); + EXPECT_EQ(info->credential, "client-id:client-secret"); + EXPECT_EQ(info->scope, "catalog"); + EXPECT_EQ(info->oauth2_server_uri, "https://auth.example.com/token"); + EXPECT_EQ(info->optional_oauth_params.at(AuthProperties::kAudience.key()), + "catalog-audience"); +} + TEST_F(AuthManagerTest, HttpClientRejectsParamsWhenUrlAlreadyHasQuery) { auto session = AuthSession::MakeDefault({}); auto result = @@ -266,6 +575,11 @@ TEST_F(AuthManagerTest, OAuth2StaticToken) { std::unordered_map properties = { {AuthProperties::kAuthType, "oauth2"}, {AuthProperties::kToken.key(), "my-static-token"}, + {AuthProperties::kCredential.key(), "client-id:client-secret"}, + {AuthProperties::kScope.key(), "catalog"}, + {AuthProperties::kOAuth2ServerUri.key(), "https://auth.example.com/token"}, + {AuthProperties::kAudience.key(), "catalog-audience"}, + {AuthProperties::kResource.key(), "catalog-resource"}, }; auto manager_result = AuthManagers::Load("test-catalog", properties); @@ -277,6 +591,18 @@ TEST_F(AuthManagerTest, OAuth2StaticToken) { auto auth_result = session_result.value()->Authenticate({}); ASSERT_THAT(auth_result, IsOk()); EXPECT_EQ(auth_result.value().headers["Authorization"], "Bearer my-static-token"); + + auto info = session_result.value()->OAuth2Info(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->token, "my-static-token"); + EXPECT_EQ(info->issued_token_type, AuthProperties::kAccessTokenType); + EXPECT_EQ(info->credential, "client-id:client-secret"); + EXPECT_EQ(info->scope, "catalog"); + EXPECT_EQ(info->oauth2_server_uri, "https://auth.example.com/token"); + EXPECT_EQ(info->optional_oauth_params.at(AuthProperties::kAudience.key()), + "catalog-audience"); + EXPECT_EQ(info->optional_oauth_params.at(AuthProperties::kResource.key()), + "catalog-resource"); } // Verifies OAuth2 type is inferred from token property @@ -314,6 +640,106 @@ TEST_F(AuthManagerTest, OAuth2MissingCredentials) { ASSERT_TRUE(auth_result.has_value()); EXPECT_EQ(auth_result.value().headers.find("Authorization"), auth_result.value().headers.end()); + + auto info = session_result.value()->OAuth2Info(); + ASSERT_TRUE(info.has_value()); + EXPECT_TRUE(info->token.empty()); + EXPECT_EQ(info->issued_token_type, AuthProperties::kAccessTokenType); +} + +TEST_F(AuthManagerTest, OAuth2ContextTokenCreatesChildAndHasPriority) { + std::unordered_map properties = { + {AuthProperties::kAuthType, "oauth2"}, + {AuthProperties::kScope.key(), "catalog"}, + {AuthProperties::kOAuth2ServerUri.key(), "https://auth.example.com/token"}, + {AuthProperties::kAudience.key(), "catalog-audience"}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog", properties)); + ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client_, properties)); + + SessionContext context{ + .session_id = "tenant-a", + .credentials = + { + {AuthProperties::kToken.key(), "context-token"}, + {AuthProperties::kCredential.key(), "unused-credential"}, + {AuthProperties::kIdTokenType, "unused-id-token"}, + }, + }; + ICEBERG_UNWRAP_OR_FAIL(auto child, manager->ContextualSession(context, parent)); + + EXPECT_NE(child, parent); + ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({})); + EXPECT_EQ(authenticated.headers.at("Authorization"), "Bearer context-token"); + + auto info = child->OAuth2Info(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->token, "context-token"); + EXPECT_EQ(info->issued_token_type, AuthProperties::kAccessTokenType); + EXPECT_TRUE(info->credential.empty()); + EXPECT_EQ(info->scope, "catalog"); + EXPECT_EQ(info->oauth2_server_uri, "https://auth.example.com/token"); + EXPECT_EQ(info->optional_oauth_params.at(AuthProperties::kAudience.key()), + "catalog-audience"); +} + +TEST_F(AuthManagerTest, OAuth2ContextTypedTokenOnlyUsesCredentials) { + std::unordered_map properties = { + {AuthProperties::kAuthType, "oauth2"}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog", properties)); + ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client_, properties)); + + SessionContext context{ + .session_id = "tenant-a", + .credentials = {{"unrelated", "value"}}, + .properties = {{AuthProperties::kIdTokenType, "property-id-token"}}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto child, manager->ContextualSession(context, parent)); + + EXPECT_EQ(child, parent); +} + +TEST_F(AuthManagerTest, OAuth2TableTokenCreatesChild) { + std::unordered_map properties = { + {AuthProperties::kAuthType, "oauth2"}, + {AuthProperties::kScope.key(), "catalog"}, + {AuthProperties::kOAuth2ServerUri.key(), "https://auth.example.com/token"}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog", properties)); + ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client_, properties)); + TableIdentifier table{.ns = Namespace{{"db"}}, .name = "table"}; + + ICEBERG_UNWRAP_OR_FAIL( + auto child, + manager->TableSession(table, + {{AuthProperties::kToken.key(), "table-token"}, + {AuthProperties::kCredential.key(), "ignored-credential"}}, + parent)); + + EXPECT_NE(child, parent); + ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({})); + EXPECT_EQ(authenticated.headers.at("Authorization"), "Bearer table-token"); + auto info = child->OAuth2Info(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->issued_token_type, AuthProperties::kAccessTokenType); + EXPECT_TRUE(info->credential.empty()); +} + +TEST_F(AuthManagerTest, OAuth2TableIgnoresCredential) { + std::unordered_map properties = { + {AuthProperties::kAuthType, "oauth2"}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog", properties)); + ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client_, properties)); + TableIdentifier table{.ns = Namespace{{"db"}}, .name = "table"}; + + ICEBERG_UNWRAP_OR_FAIL( + auto child, + manager->TableSession( + table, {{AuthProperties::kCredential.key(), "ignored-credential"}}, parent)); + + EXPECT_EQ(child, parent); } // Verifies that when both token and credential are provided, token takes priority diff --git a/src/iceberg/test/rest_catalog_integration_test.cc b/src/iceberg/test/rest_catalog_integration_test.cc index 96f392533..feeee2c3a 100644 --- a/src/iceberg/test/rest_catalog_integration_test.cc +++ b/src/iceberg/test/rest_catalog_integration_test.cc @@ -34,6 +34,8 @@ #include #include +#include "iceberg/catalog/rest/auth/auth_managers.h" +#include "iceberg/catalog/rest/auth/auth_properties.h" #include "iceberg/catalog/rest/auth/auth_session.h" #include "iceberg/catalog/rest/catalog_properties.h" #include "iceberg/catalog/rest/error_handlers.h" @@ -101,6 +103,8 @@ bool CheckServiceReady(uint16_t port) { std::string CatalogUri() { return std::format("{}:{}", kLocalhostUri, kRestCatalogPort); } +std::string OAuthTokenUri() { return CatalogUri() + "/v1/oauth/tokens"; } + } // namespace /// \brief Integration test fixture for REST catalog with Docker Compose. @@ -211,6 +215,105 @@ TEST_F(RestCatalogIntegrationTest, MakeCatalogSuccess) { EXPECT_THAT(root->WithContext(SessionContext{}), IsError(ErrorKind::kInvalidArgument)); } +TEST_F(RestCatalogIntegrationTest, OAuthContextCredentialEndToEnd) { + HttpClient client; + std::unordered_map properties = { + {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2}, + {auth::AuthProperties::kToken.key(), "catalog-token"}, + {auth::AuthProperties::kOAuth2ServerUri.key(), OAuthTokenUri()}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto manager, + auth::AuthManagers::Load("test-catalog", properties)); + ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client, properties)); + SessionContext context{ + .session_id = "tenant-context-credential", + .credentials = {{auth::AuthProperties::kCredential.key(), "context-client:secret"}}, + }; + + ICEBERG_UNWRAP_OR_FAIL(auto child, manager->ContextualSession(context, parent)); + ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({})); + + EXPECT_EQ(authenticated.headers.at("Authorization"), + "Bearer client-credentials-token:sub=context-client"); + auto info = child->OAuth2Info(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->issued_token_type, auth::AuthProperties::kAccessTokenType); +} + +TEST_F(RestCatalogIntegrationTest, OAuthContextTypedTokenEndToEnd) { + HttpClient client; + std::unordered_map properties = { + {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2}, + {auth::AuthProperties::kToken.key(), "catalog-token"}, + {auth::AuthProperties::kOAuth2ServerUri.key(), OAuthTokenUri()}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto manager, + auth::AuthManagers::Load("test-catalog", properties)); + ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client, properties)); + SessionContext context{ + .session_id = "tenant-context-token", + .credentials = {{auth::AuthProperties::kIdTokenType, "context-id-token"}}, + }; + + ICEBERG_UNWRAP_OR_FAIL(auto child, manager->ContextualSession(context, parent)); + ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({})); + + EXPECT_EQ(authenticated.headers.at("Authorization"), + "Bearer token-exchange-token:sub=context-id-token,act=catalog-token"); + auto info = child->OAuth2Info(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->issued_token_type, auth::AuthProperties::kAccessTokenType); +} + +TEST_F(RestCatalogIntegrationTest, OAuthTableTypedTokenEndToEnd) { + HttpClient client; + std::unordered_map properties = { + {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2}, + {auth::AuthProperties::kToken.key(), "catalog-token"}, + {auth::AuthProperties::kOAuth2ServerUri.key(), OAuthTokenUri()}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto manager, + auth::AuthManagers::Load("test-catalog", properties)); + ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client, properties)); + TableIdentifier table{.ns = Namespace{{"db"}}, .name = "events"}; + + ICEBERG_UNWRAP_OR_FAIL( + auto child, + manager->TableSession( + table, {{auth::AuthProperties::kJwtTokenType, "table-jwt-token"}}, parent)); + ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({})); + + EXPECT_EQ(authenticated.headers.at("Authorization"), + "Bearer token-exchange-token:sub=table-jwt-token,act=catalog-token"); + auto info = child->OAuth2Info(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->issued_token_type, auth::AuthProperties::kAccessTokenType); +} + +TEST_F(RestCatalogIntegrationTest, OAuthTokenExchangeWithoutActorEndToEnd) { + HttpClient client; + std::unordered_map properties = { + {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2}, + {auth::AuthProperties::kOAuth2ServerUri.key(), OAuthTokenUri()}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto manager, + auth::AuthManagers::Load("test-catalog", properties)); + ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client, properties)); + SessionContext context{ + .session_id = "tenant-no-actor", + .credentials = {{auth::AuthProperties::kIdTokenType, "context-id-token"}}, + }; + + ICEBERG_UNWRAP_OR_FAIL(auto child, manager->ContextualSession(context, parent)); + ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({})); + + EXPECT_EQ(authenticated.headers.at("Authorization"), + "Bearer token-exchange-token:sub=context-id-token"); + auto info = child->OAuth2Info(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->issued_token_type, auth::AuthProperties::kAccessTokenType); +} + TEST_F(RestCatalogIntegrationTest, LoadsConfiguredMetricsReporter) { auto loaded = std::make_shared>(false); ASSERT_THAT(MetricsReporters::Register( diff --git a/src/iceberg/test/rest_util_test.cc b/src/iceberg/test/rest_util_test.cc index 0035afca0..6af3772d6 100644 --- a/src/iceberg/test/rest_util_test.cc +++ b/src/iceberg/test/rest_util_test.cc @@ -103,6 +103,14 @@ TEST(RestUtilTest, ResourcePathsRejectsEmptyNamespaceSeparator) { EXPECT_THAT(result, HasErrorMessage("REST namespace separator cannot be empty")); } +TEST(RestUtilTest, OAuth2TokensPathDoesNotUseCatalogPrefix) { + ICEBERG_UNWRAP_OR_FAIL( + auto paths, ResourcePaths::Make("https://catalog.example.com", "warehouse", "%1F")); + + EXPECT_THAT(paths->OAuth2Tokens(), + HasValue(::testing::Eq("https://catalog.example.com/v1/oauth/tokens"))); +} + TEST(RestUtilTest, EncodeString) { // RFC 3986 unreserved characters should not be encoded EXPECT_THAT(EncodeString("abc123XYZ"), HasValue(::testing::Eq("abc123XYZ"))); From 6ac2495461f03f419cf95a623f2974fb84ac1652 Mon Sep 17 00:00:00 2001 From: "shuxu.li" Date: Sat, 22 Aug 2026 10:37:23 +0800 Subject: [PATCH 2/3] feat(rest): support OAuth token exchange sessions Add RFC 8693 token exchange support, including token type helpers, request form construction, OAuth endpoint normalization, and response handling. Preserve OAuth metadata in auth sessions and create contextual and table-scoped child sessions from direct tokens, credentials, or typed tokens. Disable child refresh until session lifecycle management is available. --- src/iceberg/catalog/rest/auth/auth_properties.cc | 12 +++++------- src/iceberg/test/auth_manager_test.cc | 15 ++++++++------- 2 files changed, 13 insertions(+), 14 deletions(-) diff --git a/src/iceberg/catalog/rest/auth/auth_properties.cc b/src/iceberg/catalog/rest/auth/auth_properties.cc index 67d9319d8..0e7a8a2df 100644 --- a/src/iceberg/catalog/rest/auth/auth_properties.cc +++ b/src/iceberg/catalog/rest/auth/auth_properties.cc @@ -46,18 +46,16 @@ Result ResolveOAuth2ServerUri( if (endpoint.starts_with("http://") || endpoint.starts_with("https://")) { return endpoint; } - if (endpoint.starts_with('/')) { - return InvalidArgument("OAuth2 server URI path must not start with '/': {}", - endpoint); - } - auto uri_it = properties.find(RestCatalogProperties::kUri.key()); if (uri_it == properties.end() || uri_it->second.empty()) { return endpoint; } - return std::string(TrimTrailingSlash(uri_it->second)) + "/" + - std::string(TrimTrailingSlash(endpoint)); + auto base_uri = std::string(TrimTrailingSlash(uri_it->second)); + if (endpoint.starts_with('/')) { + return base_uri + endpoint; + } + return base_uri + "/" + std::string(TrimTrailingSlash(endpoint)); } } // namespace diff --git a/src/iceberg/test/auth_manager_test.cc b/src/iceberg/test/auth_manager_test.cc index c9a03e91c..4eb167bb4 100644 --- a/src/iceberg/test/auth_manager_test.cc +++ b/src/iceberg/test/auth_manager_test.cc @@ -313,14 +313,15 @@ TEST(AuthPropertiesTest, PreservesRelativeOAuth2ServerUriWithoutCatalogUri) { EXPECT_EQ(config.oauth2_server_uri(), "oauth/token"); } -TEST(AuthPropertiesTest, RejectsOAuth2ServerUriWithLeadingSlash) { - auto result = AuthProperties::FromProperties({ - {RestCatalogProperties::kUri.key(), "https://catalog.example.com"}, - {AuthProperties::kOAuth2ServerUri.key(), "/v1/oauth/tokens"}, - }); +TEST(AuthPropertiesTest, ResolvesExplicitAbsolutePathOAuth2ServerUri) { + ICEBERG_UNWRAP_OR_FAIL( + auto config, + AuthProperties::FromProperties({ + {RestCatalogProperties::kUri.key(), "https://catalog.example.com/"}, + {AuthProperties::kOAuth2ServerUri.key(), "/v1/oauth/tokens"}, + })); - EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); - EXPECT_THAT(result, HasErrorMessage("must not start with '/'")); + EXPECT_EQ(config.oauth2_server_uri(), "https://catalog.example.com/v1/oauth/tokens"); } // Verifies loading NoopAuthManager with explicit "none" auth type From 198ef6d8c7a866c2609ae6df76fb7f0205ee6920 Mon Sep 17 00:00:00 2001 From: Gang Wu Date: Mon, 24 Aug 2026 23:53:01 +0800 Subject: [PATCH 3/3] align OAuth2 session lifecycle with Java - propagate shared HttpClient ownership through REST auth managers - preserve OAuth2 session state and request-start expiry semantics - expose OAuth2 metadata through a synchronized OAuth2Info snapshot - encapsulate OAuth2 utilities and normalize token endpoints - update REST auth and integration tests --- src/iceberg/catalog/rest/auth/auth_manager.cc | 154 +++++++--- src/iceberg/catalog/rest/auth/auth_manager.h | 4 +- .../catalog/rest/auth/auth_properties.cc | 7 +- src/iceberg/catalog/rest/auth/auth_session.cc | 252 ++-------------- src/iceberg/catalog/rest/auth/auth_session.h | 8 +- .../catalog/rest/auth/auth_session_internal.h | 282 ++++++++++++++++++ src/iceberg/catalog/rest/auth/oauth2_util.cc | 112 +++---- src/iceberg/catalog/rest/auth/oauth2_util.h | 135 ++++----- .../rest/auth/sigv4_auth_manager_internal.h | 4 +- .../catalog/rest/auth/sigv4_manager.cc | 5 +- .../rest/auth/token_refresh_scheduler.h | 2 +- src/iceberg/catalog/rest/rest_catalog.cc | 4 +- src/iceberg/test/auth_manager_test.cc | 272 ++++------------- .../test/rest_catalog_integration_test.cc | 33 +- src/iceberg/test/rest_util_test.cc | 8 - src/iceberg/test/sigv4_auth_test.cc | 3 +- 16 files changed, 621 insertions(+), 664 deletions(-) create mode 100644 src/iceberg/catalog/rest/auth/auth_session_internal.h diff --git a/src/iceberg/catalog/rest/auth/auth_manager.cc b/src/iceberg/catalog/rest/auth/auth_manager.cc index 6b2a289ba..5facfa3eb 100644 --- a/src/iceberg/catalog/rest/auth/auth_manager.cc +++ b/src/iceberg/catalog/rest/auth/auth_manager.cc @@ -19,11 +19,16 @@ #include "iceberg/catalog/rest/auth/auth_manager.h" +#include +#include #include +#include +#include #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/auth_session_internal.h" #include "iceberg/catalog/rest/auth/oauth2_util.h" #include "iceberg/catalog/session_context.h" #include "iceberg/util/base64.h" @@ -31,11 +36,50 @@ namespace iceberg::rest::auth { +namespace { + +constexpr std::string_view kAuthorizationHeader = "Authorization"; + +const std::array kTokenPreferenceOrder = { + AuthProperties::kIdTokenType, AuthProperties::kAccessTokenType, + AuthProperties::kJwtTokenType, AuthProperties::kSaml2TokenType, + AuthProperties::kSaml1TokenType, +}; + +std::optional> FindPreferredTypedToken( + const std::unordered_map& credentials) { + for (std::string_view token_type : kTokenPreferenceOrder) { + auto token_it = credentials.find(std::string(token_type)); + if (token_it != credentials.end()) { + return std::pair{token_it->first, token_it->second}; + } + } + return std::nullopt; +} + +std::unordered_map FilterTableSessionProperties( + const std::unordered_map& properties) { + std::unordered_map filtered; + if (auto token_it = properties.find(AuthProperties::kToken.key()); + token_it != properties.end()) { + filtered.emplace(token_it->first, token_it->second); + } + for (std::string_view token_type : kTokenPreferenceOrder) { + auto token_it = properties.find(std::string(token_type)); + if (token_it != properties.end()) { + filtered.emplace(token_it->first, token_it->second); + } + } + return filtered; +} + +} // namespace + Result> AuthManager::InitSession( - HttpClient& init_client, + std::shared_ptr init_client, const std::unordered_map& properties) { // By default, use the catalog session for initialization - return CatalogSession(init_client, properties); + return CatalogSession(std::move(init_client), properties); } Result> AuthManager::ContextualSession( @@ -56,7 +100,7 @@ Result> AuthManager::TableSession( class NoopAuthManager : public AuthManager { public: Result> CatalogSession( - [[maybe_unused]] HttpClient& client, + [[maybe_unused]] std::shared_ptr client, [[maybe_unused]] const std::unordered_map& properties) override { return AuthSession::MakeDefault({}); @@ -73,7 +117,7 @@ Result> MakeNoopAuthManager( class BasicAuthManager : public AuthManager { public: Result> CatalogSession( - [[maybe_unused]] HttpClient& client, + [[maybe_unused]] std::shared_ptr client, const std::unordered_map& properties) override { auto username_it = properties.find(AuthProperties::kBasicUsername); ICEBERG_PRECHECK(username_it != properties.end() && !username_it->second.empty(), @@ -97,44 +141,51 @@ Result> MakeBasicAuthManager( class OAuth2Manager : public AuthManager { public: Result> InitSession( - HttpClient& init_client, + std::shared_ptr init_client, const std::unordered_map& properties) override { + ICEBERG_PRECHECK(init_client != nullptr, + "OAuth2 initialization HTTP client must not be null"); 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)); + auto init_session = + AuthSession::MakeDefault(OAuth2Util::AuthHeaders(config.token())); + start_time_ = std::chrono::steady_clock::now(); + ICEBERG_ASSIGN_OR_RAISE( + auth_response_, OAuth2Util::FetchToken(*init_client, *init_session, config)); + // TODO(lishuxu): Match Java OAuth2Util.AuthSession.fromTokenResponse here. + return AuthSession::MakeDefault( + OAuth2Util::AuthHeaders(auth_response_->access_token)); } if (!config.token().empty()) { - return AuthSession::MakeDefault(AuthHeaders(config.token())); + // TODO(lishuxu): Match Java OAuth2Util.AuthSession.fromAccessToken here. + return AuthSession::MakeDefault(OAuth2Util::AuthHeaders(config.token())); } return AuthSession::MakeDefault({}); } Result> CatalogSession( - HttpClient& client, + std::shared_ptr shared_client, const std::unordered_map& properties) override { ICEBERG_ASSIGN_OR_RAISE(auto config, AuthProperties::FromProperties(properties)); - shared_client_ = &client; - - // 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(), config.keep_refreshed(), - config.optional_oauth_params(), client); + ICEBERG_PRECHECK(shared_client != nullptr, + "OAuth2 catalog session HTTP client must not be null"); + refresh_client_ = std::move(shared_client); + // Reuse the token response and start time from the init phase. + if (auth_response_.has_value()) { + return internal::MakeOAuth2Session( + *auth_response_, config.oauth2_server_uri(), config.client_id(), + config.client_secret(), config.scope(), config.keep_refreshed(), + config.optional_oauth_params(), refresh_client_, start_time_); } - // If token is provided, use it directly. + // TODO(lishuxu): Honor token-refresh-enabled for catalog bearer tokens, matching + // Java. If token is provided, use it directly. if (!config.token().empty()) { OAuthTokenResponse token_response{ .access_token = config.token(), @@ -144,18 +195,21 @@ class OAuth2Manager : public AuthManager { return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(), config.client_id(), config.client_secret(), config.scope(), /*keep_refreshed=*/false, - config.optional_oauth_params(), client); + config.optional_oauth_params(), refresh_client_); } // Fetch a new token using client_credentials grant. if (!config.credential().empty()) { - auto base_session = AuthSession::MakeDefault(AuthHeaders(config.token())); + auto base_session = + AuthSession::MakeDefault(OAuth2Util::AuthHeaders(config.token())); OAuthTokenResponse token_response; - ICEBERG_ASSIGN_OR_RAISE(token_response, FetchToken(client, *base_session, config)); + ICEBERG_ASSIGN_OR_RAISE( + token_response, + OAuth2Util::FetchToken(*refresh_client_, *base_session, config)); return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(), config.client_id(), config.client_secret(), config.scope(), config.keep_refreshed(), - config.optional_oauth_params(), client); + config.optional_oauth_params(), refresh_client_); } return MakeSession(AccessTokenResponse(""), config, /*keep_refreshed=*/false); @@ -163,6 +217,8 @@ class OAuth2Manager : public AuthManager { Result> ContextualSession( const SessionContext& context, std::shared_ptr parent) override { + // TODO(lishuxu): Add child-session caching and refresh, matching Java + // AuthSessionCache. return MaybeCreateChildSession(context.credentials, /*allow_credential=*/true, std::move(parent)); } @@ -175,6 +231,11 @@ class OAuth2Manager : public AuthManager { /*allow_credential=*/false, std::move(parent)); } + Status Close() override { + refresh_client_.reset(); + return {}; + } + private: static OAuthTokenResponse AccessTokenResponse(std::string token) { return { @@ -196,17 +257,17 @@ class OAuth2Manager : public AuthManager { Result> MakeSession( const OAuthTokenResponse& token_response, const AuthProperties& config, bool keep_refreshed) const { - ICEBERG_PRECHECK(shared_client_ != nullptr, + ICEBERG_PRECHECK(refresh_client_ != nullptr, "OAuth2 catalog session must be initialized before child sessions"); return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(), config.client_id(), config.client_secret(), config.scope(), keep_refreshed, - config.optional_oauth_params(), *shared_client_); + config.optional_oauth_params(), refresh_client_); } Result> MaybeCreateChildSession( const std::unordered_map& credentials, - bool allow_credential, std::shared_ptr parent) const { + bool allow_credential, std::shared_ptr parent) { auto token_it = credentials.find(AuthProperties::kToken.key()); auto credential_it = credentials.find(AuthProperties::kCredential.key()); auto typed_token = FindPreferredTypedToken(credentials); @@ -216,7 +277,7 @@ class OAuth2Manager : public AuthManager { return parent; } - ICEBERG_PRECHECK(shared_client_ != nullptr, + ICEBERG_PRECHECK(refresh_client_ != nullptr, "OAuth2 catalog session must be initialized before child sessions"); auto parent_info = parent->OAuth2Info(); ICEBERG_PRECHECK(parent_info.has_value(), @@ -233,34 +294,31 @@ class OAuth2Manager : public AuthManager { ICEBERG_ASSIGN_OR_RAISE(auto config, ChildConfig(*parent_info, credential_it->second)); ICEBERG_ASSIGN_OR_RAISE(auto response, - FetchToken(*shared_client_, *parent, config)); + OAuth2Util::FetchToken(*refresh_client_, *parent, config)); return MakeSession(response, config, /*keep_refreshed=*/false); } - std::optional actor; + std::optional actor_token; + std::optional actor_token_type; if (!parent_info->token.empty()) { - actor = OAuth2Token{ - .token_type = parent_info->issued_token_type, - .token = parent_info->token, - }; + actor_token = parent_info->token; + actor_token_type = parent_info->issued_token_type; } - TokenExchangeRequest request{ - .oauth2_server_uri = parent_info->oauth2_server_uri, - .subject = std::move(*typed_token), - .actor = std::move(actor), - .scope = parent_info->scope, - .optional_oauth_params = parent_info->optional_oauth_params, - }; - ICEBERG_ASSIGN_OR_RAISE(auto response, - ExchangeToken(*shared_client_, *parent, {}, request)); + ICEBERG_ASSIGN_OR_RAISE( + auto response, + OAuth2Util::ExchangeToken(*refresh_client_, *parent, {}, typed_token->second, + typed_token->first, actor_token, actor_token_type, + parent_info->scope, parent_info->oauth2_server_uri, + parent_info->optional_oauth_params)); ICEBERG_ASSIGN_OR_RAISE(auto config, ChildConfig(*parent_info, parent_info->credential)); return MakeSession(response, config, /*keep_refreshed=*/false); } - /// Cached token from InitSession - std::optional init_token_response_; - HttpClient* shared_client_ = nullptr; + /// Token response and start time captured by InitSession. + std::optional auth_response_; + std::optional start_time_; + std::shared_ptr refresh_client_; }; Result> MakeOAuth2Manager( diff --git a/src/iceberg/catalog/rest/auth/auth_manager.h b/src/iceberg/catalog/rest/auth/auth_manager.h index 0a97c9b2a..6ba25c284 100644 --- a/src/iceberg/catalog/rest/auth/auth_manager.h +++ b/src/iceberg/catalog/rest/auth/auth_manager.h @@ -48,7 +48,7 @@ class ICEBERG_REST_EXPORT AuthManager { /// \param properties Client configuration supplied by the catalog. /// \return Session for initialization or an error if credentials cannot be acquired. virtual Result> InitSession( - HttpClient& init_client, + std::shared_ptr init_client, const std::unordered_map& properties); /// \brief Create the long-lived catalog session that acts as the parent session. @@ -62,7 +62,7 @@ class ICEBERG_REST_EXPORT AuthManager { /// \return Session for catalog operations or an error if authentication cannot be set /// up. virtual Result> CatalogSession( - HttpClient& shared_client, + std::shared_ptr shared_client, const std::unordered_map& properties) = 0; /// \brief Create or reuse a session for a specific context. diff --git a/src/iceberg/catalog/rest/auth/auth_properties.cc b/src/iceberg/catalog/rest/auth/auth_properties.cc index 0e7a8a2df..f373df617 100644 --- a/src/iceberg/catalog/rest/auth/auth_properties.cc +++ b/src/iceberg/catalog/rest/auth/auth_properties.cc @@ -39,13 +39,16 @@ std::pair ParseCredential(const std::string& credentia Result ResolveOAuth2ServerUri( const std::unordered_map& properties) { auto endpoint_it = properties.find(AuthProperties::kOAuth2ServerUri.key()); - std::string endpoint = endpoint_it == properties.end() || endpoint_it->second.empty() + std::string endpoint = endpoint_it == properties.end() ? AuthProperties::kOAuth2ServerUri.value() : endpoint_it->second; if (endpoint.starts_with("http://") || endpoint.starts_with("https://")) { return endpoint; } + if (endpoint.empty()) { + return endpoint; + } auto uri_it = properties.find(RestCatalogProperties::kUri.key()); if (uri_it == properties.end() || uri_it->second.empty()) { return endpoint; @@ -55,7 +58,7 @@ Result ResolveOAuth2ServerUri( if (endpoint.starts_with('/')) { return base_uri + endpoint; } - return base_uri + "/" + std::string(TrimTrailingSlash(endpoint)); + return base_uri + "/" + endpoint; } } // namespace diff --git a/src/iceberg/catalog/rest/auth/auth_session.cc b/src/iceberg/catalog/rest/auth/auth_session.cc index fbb1899d2..60aa0a3b4 100644 --- a/src/iceberg/catalog/rest/auth/auth_session.cc +++ b/src/iceberg/catalog/rest/auth/auth_session.cc @@ -28,6 +28,7 @@ #include #include "iceberg/catalog/rest/auth/auth_properties.h" +#include "iceberg/catalog/rest/auth/auth_session_internal.h" #include "iceberg/catalog/rest/auth/oauth2_util.h" #include "iceberg/catalog/rest/auth/token_refresh_scheduler.h" #include "iceberg/catalog/rest/http_client.h" @@ -54,236 +55,6 @@ class DefaultAuthSession : public AuthSession { std::unordered_map headers_; }; -/// \brief OAuth2 session with automatic token refresh. -class OAuth2AuthSession : public AuthSession, - public std::enable_shared_from_this { - public: - struct Config { - std::string token_endpoint; - std::string client_id; - std::string client_secret; - std::string scope; - std::unordered_map optional_oauth_params; - bool keep_refreshed; - }; - - /// \brief Create an OAuth2 session and optionally schedule refresh. - static Result> Make( - const OAuthTokenResponse& initial_token, Config config, HttpClient& client) { - ICEBERG_ASSIGN_OR_RAISE(auto refresh_properties, MakeRefreshProperties(config)); - auto session = std::shared_ptr( - new OAuth2AuthSession(std::move(config), std::move(refresh_properties), client)); - session->SetInitialToken(initial_token); - return session; - } - - Result Authenticate(HttpRequest request) override { - std::shared_lock lock(mutex_); - for (const auto& [key, value] : headers_) { - request.headers.try_emplace(key, value); - } - return request; - } - - std::optional OAuth2Info() const override { - std::shared_lock lock(mutex_); - return OAuth2SessionInfo{ - .token = token_, - .issued_token_type = issued_token_type_, - .credential = Credential(config_), - .scope = config_.scope, - .oauth2_server_uri = config_.token_endpoint, - .optional_oauth_params = config_.optional_oauth_params, - }; - } - - Status Close() override { return CloseImpl(); } - - ~OAuth2AuthSession() override { std::ignore = CloseImpl(); } - - private: - OAuth2AuthSession(Config config, AuthProperties refresh_properties, HttpClient& client) - : config_(std::move(config)), - refresh_properties_(std::move(refresh_properties)), - client_(client) {} - - Status CloseImpl() { - bool expected = false; - if (!closed_.compare_exchange_strong(expected, true)) { - return {}; // Already closed - } - TokenRefreshScheduler::Instance().Cancel(scheduled_task_id_.exchange(0)); - std::unique_lock lock(refresh_mutex_); - refresh_cv_.wait(lock, [this] { return active_refresh_count_ == 0; }); - TokenRefreshScheduler::Instance().Cancel(scheduled_task_id_.exchange(0)); - return {}; - } - - static std::string Credential(const Config& config) { - return config.client_id.empty() ? config.client_secret - : config.client_id + ":" + config.client_secret; - } - - static Result MakeRefreshProperties(const Config& config) { - std::unordered_map properties = - config.optional_oauth_params; - properties[AuthProperties::kCredential.key()] = Credential(config); - properties[AuthProperties::kScope.key()] = config.scope; - properties[AuthProperties::kOAuth2ServerUri.key()] = config.token_endpoint; - - return AuthProperties::FromProperties(properties); - } - - class RefreshAttemptGuard { - public: - explicit RefreshAttemptGuard(OAuth2AuthSession& session) : session_(session) { - std::lock_guard lock(session_.refresh_mutex_); - ++session_.active_refresh_count_; - } - - ~RefreshAttemptGuard() { - bool notify = false; - { - std::lock_guard lock(session_.refresh_mutex_); - notify = --session_.active_refresh_count_ == 0; - } - if (notify) { - session_.refresh_cv_.notify_all(); - } - } - - private: - OAuth2AuthSession& session_; - }; - - void UpdateTokenState(const OAuthTokenResponse& token_response) { - token_ = token_response.access_token; - issued_token_type_ = token_response.issued_token_type.empty() - ? AuthProperties::kAccessTokenType - : token_response.issued_token_type; - headers_ = AuthHeaders(token_); - - expires_at_ = std::chrono::steady_clock::time_point{}; - if (token_response.expires_in_secs.has_value()) { - expires_at_ = std::chrono::steady_clock::now() + - std::chrono::seconds(*token_response.expires_in_secs); - } else if (auto exp_ms = ExpiresAtMillis(token_); exp_ms.has_value()) { - // Convert absolute epoch millis to steady_clock time_point - auto now_sys = std::chrono::system_clock::now(); - auto now_steady = std::chrono::steady_clock::now(); - auto exp_sys = - std::chrono::system_clock::time_point(std::chrono::milliseconds(*exp_ms)); - expires_at_ = now_steady + (exp_sys - now_sys); - } - } - - void SetInitialToken(const OAuthTokenResponse& token_response) { - UpdateTokenState(token_response); - - if (config_.keep_refreshed && - expires_at_ != std::chrono::steady_clock::time_point{}) { - ScheduleRefresh(); - } - } - - void DoRefresh() { DoRefreshAttempt(0, std::chrono::milliseconds(200)); } - - /// \brief Single refresh attempt. On failure, schedules a retry via the - /// scheduler (non-blocking) instead of sleeping on the worker thread. - void DoRefreshAttempt(int attempt, std::chrono::milliseconds backoff) { - static constexpr int kMaxRetries = 5; - static constexpr auto kMaxBackoff = std::chrono::milliseconds(10'000); - - RefreshAttemptGuard guard(*this); - if (closed_.load()) return; - - // Use an empty session for the refresh request (no auth headers — - // avoids circular dependency of using an expired token to refresh itself) - auto empty_session = AuthSession::MakeDefault({}); - - auto result = FetchToken(client_, *empty_session, refresh_properties_); - if (result.has_value()) { - auto& response = result.value(); - { - std::unique_lock lock(mutex_); - UpdateTokenState(response); - } - // Note: ScheduleRefresh must be called outside the lock. - ScheduleRefresh(); - return; // Success - } - - // Schedule retry with exponential backoff (non-blocking) - if (attempt + 1 < kMaxRetries && !closed_.load()) { - auto next_backoff = - std::min(std::chrono::duration_cast(backoff * 2), - kMaxBackoff); - std::weak_ptr weak_self = shared_from_this(); - auto retry_id = TokenRefreshScheduler::Instance().Schedule( - backoff, - [weak_self = std::move(weak_self), next_attempt = attempt + 1, next_backoff] { - if (auto self = weak_self.lock()) { - self->DoRefreshAttempt(next_attempt, next_backoff); - } - }); - scheduled_task_id_.store(retry_id); - } - // All retries exhausted — stop refreshing silently. - // Next request will use the expired token; server returns 401. - } - - /// \brief Schedule the next token refresh based on expiration time. - /// - /// Must be called outside any lock on mutex_ (CalculateRefreshDelay - /// acquires shared_lock internally). - void ScheduleRefresh() { - if (!config_.keep_refreshed || closed_.load()) return; - - auto delay = CalculateRefreshDelay(); - if (delay < std::chrono::milliseconds::zero()) return; - - std::weak_ptr weak_self = shared_from_this(); - auto new_id = TokenRefreshScheduler::Instance().Schedule( - delay, [weak_self = std::move(weak_self)] { - if (auto self = weak_self.lock()) { - self->DoRefresh(); - } - }); - scheduled_task_id_.store(new_id); - } - - std::chrono::milliseconds CalculateRefreshDelay() const { - std::shared_lock lock(mutex_); - auto now = std::chrono::steady_clock::now(); - if (expires_at_ == std::chrono::steady_clock::time_point{}) { - return std::chrono::milliseconds(-1); - } - if (expires_at_ <= now) return std::chrono::milliseconds::zero(); - - auto expires_in = - std::chrono::duration_cast(expires_at_ - now); - // Refresh window: 10% of remaining time, capped at 5 minutes - auto refresh_window = std::min(expires_in / 10, std::chrono::milliseconds(300'000)); - auto wait_time = expires_in - refresh_window; - return std::max(wait_time, std::chrono::milliseconds(10)); - } - - mutable std::shared_mutex mutex_; // protects token state, headers, and expiration - std::string token_; - std::string issued_token_type_; - std::unordered_map headers_; - std::chrono::steady_clock::time_point expires_at_{}; - - Config config_; - AuthProperties refresh_properties_; - HttpClient& client_; // It should outlive the session - std::atomic scheduled_task_id_{0}; - std::atomic closed_{false}; - std::mutex refresh_mutex_; - std::condition_variable refresh_cv_; - int active_refresh_count_ = 0; -}; - } // namespace std::shared_ptr AuthSession::MakeDefault( @@ -296,8 +67,20 @@ Result> AuthSession::MakeOAuth2( const std::string& client_id, const std::string& client_secret, const std::string& scope, bool keep_refreshed, const std::unordered_map& optional_oauth_params, - HttpClient& client) { - OAuth2AuthSession::Config config{ + std::shared_ptr client) { + return internal::MakeOAuth2Session( + initial_token, token_endpoint, client_id, client_secret, scope, keep_refreshed, + optional_oauth_params, std::move(client), std::nullopt); +} + +Result> internal::MakeOAuth2Session( + const OAuthTokenResponse& initial_token, const std::string& token_endpoint, + const std::string& client_id, const std::string& client_secret, + const std::string& scope, bool keep_refreshed, + const std::unordered_map& optional_oauth_params, + std::shared_ptr client, + std::optional token_request_started_at) { + internal::OAuth2Session::Config config{ .token_endpoint = token_endpoint, .client_id = client_id, .client_secret = client_secret, @@ -305,8 +88,9 @@ Result> AuthSession::MakeOAuth2( .optional_oauth_params = optional_oauth_params, .keep_refreshed = keep_refreshed, }; - ICEBERG_ASSIGN_OR_RAISE( - auto session, OAuth2AuthSession::Make(initial_token, std::move(config), client)); + ICEBERG_ASSIGN_OR_RAISE(auto session, internal::OAuth2Session::Make( + initial_token, std::move(config), + std::move(client), token_request_started_at)); return std::static_pointer_cast(std::move(session)); } diff --git a/src/iceberg/catalog/rest/auth/auth_session.h b/src/iceberg/catalog/rest/auth/auth_session.h index cfd32355a..bdc77ebc4 100644 --- a/src/iceberg/catalog/rest/auth/auth_session.h +++ b/src/iceberg/catalog/rest/auth/auth_session.h @@ -77,7 +77,7 @@ class ICEBERG_REST_EXPORT AuthSession { /// \return Status indicating success or failure of closing the session. virtual Status Close() { return {}; } - /// \brief Create a default session with static headers. + /// \brief Create a session with static headers. /// /// This factory method creates a session that adds a fixed set of headers to each /// request. It is suitable for authentication methods that use static credentials, @@ -102,15 +102,15 @@ class ICEBERG_REST_EXPORT AuthSession { /// \param scope OAuth2 scope for refresh requests. /// \param keep_refreshed Whether to schedule automatic token refresh. /// \param optional_oauth_params Optional OAuth params (audience, resource) for refresh. - /// \param client HTTP client for making refresh requests. The caller owns the - /// client and must keep it alive until the session is closed. + /// \param client HTTP client for making refresh requests. The session retains + /// ownership of the client. /// \return A new session that manages token lifecycle automatically. static Result> MakeOAuth2( const OAuthTokenResponse& initial_token, const std::string& token_endpoint, const std::string& client_id, const std::string& client_secret, const std::string& scope, bool keep_refreshed, const std::unordered_map& optional_oauth_params, - HttpClient& client); + std::shared_ptr client); }; } // namespace iceberg::rest::auth diff --git a/src/iceberg/catalog/rest/auth/auth_session_internal.h b/src/iceberg/catalog/rest/auth/auth_session_internal.h new file mode 100644 index 000000000..b8db3836e --- /dev/null +++ b/src/iceberg/catalog/rest/auth/auth_session_internal.h @@ -0,0 +1,282 @@ +/* + * 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. + */ + +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#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/catalog/rest/auth/token_refresh_scheduler.h" +#include "iceberg/catalog/rest/http_client.h" +#include "iceberg/util/macros.h" + +namespace iceberg::rest::auth::internal { + +inline std::optional TokenExpirationTime( + const OAuthTokenResponse& response, + std::chrono::steady_clock::time_point request_started_at, + std::chrono::system_clock::time_point now_system = std::chrono::system_clock::now(), + std::chrono::steady_clock::time_point now_steady = std::chrono::steady_clock::now()) { + if (auto exp_ms = OAuth2Util::ExpiresAtMillis(response.access_token); + exp_ms.has_value()) { + auto expiration_system = + std::chrono::system_clock::time_point(std::chrono::milliseconds(*exp_ms)); + return now_steady + (expiration_system - now_system); + } + if (response.expires_in_secs.has_value()) { + return request_started_at + std::chrono::seconds(*response.expires_in_secs); + } + return std::nullopt; +} + +/// \brief Internal OAuth2 authentication session. +class OAuth2Session final : public AuthSession, + public std::enable_shared_from_this { + public: + struct Config { + std::string token_endpoint; + std::string client_id; + std::string client_secret; + std::string scope; + std::unordered_map optional_oauth_params; + bool keep_refreshed; + }; + + static Result> Make( + const OAuthTokenResponse& initial_token, Config config, + std::shared_ptr client, + std::optional token_request_started_at) { + ICEBERG_PRECHECK(client != nullptr, "OAuth2 session HTTP client must not be null"); + ICEBERG_ASSIGN_OR_RAISE(auto refresh_properties, MakeRefreshProperties(config)); + auto session = std::shared_ptr(new OAuth2Session( + std::move(config), std::move(refresh_properties), std::move(client))); + session->SetInitialToken(initial_token, token_request_started_at); + return session; + } + + Result Authenticate(HttpRequest request) override { + std::shared_lock lock(mutex_); + for (const auto& [key, value] : headers_) { + request.headers.try_emplace(key, value); + } + return request; + } + + std::optional OAuth2Info() const override { + std::shared_lock lock(mutex_); + return OAuth2SessionInfo{ + .token = token_, + .issued_token_type = issued_token_type_, + .credential = Credential(config_), + .scope = config_.scope, + .oauth2_server_uri = config_.token_endpoint, + .optional_oauth_params = config_.optional_oauth_params, + }; + } + + Status Close() override { return CloseImpl(); } + + ~OAuth2Session() override { std::ignore = CloseImpl(); } + + private: + OAuth2Session(Config config, AuthProperties refresh_properties, + std::shared_ptr client) + : config_(std::move(config)), + refresh_properties_(std::move(refresh_properties)), + client_(std::move(client)) {} + + Status CloseImpl() { + bool expected = false; + if (!closed_.compare_exchange_strong(expected, true)) { + return {}; + } + TokenRefreshScheduler::Instance().Cancel(scheduled_task_id_.exchange(0)); + std::unique_lock lock(refresh_mutex_); + refresh_cv_.wait(lock, [this] { return active_refresh_count_ == 0; }); + TokenRefreshScheduler::Instance().Cancel(scheduled_task_id_.exchange(0)); + return {}; + } + + static std::string Credential(const Config& config) { + return config.client_id.empty() ? config.client_secret + : config.client_id + ":" + config.client_secret; + } + + static Result MakeRefreshProperties(const Config& config) { + std::unordered_map properties = + config.optional_oauth_params; + properties[AuthProperties::kCredential.key()] = Credential(config); + properties[AuthProperties::kScope.key()] = config.scope; + properties[AuthProperties::kOAuth2ServerUri.key()] = config.token_endpoint; + return AuthProperties::FromProperties(properties); + } + + class RefreshAttemptGuard { + public: + explicit RefreshAttemptGuard(OAuth2Session& session) : session_(session) { + std::lock_guard lock(session_.refresh_mutex_); + ++session_.active_refresh_count_; + } + + ~RefreshAttemptGuard() { + bool notify = false; + { + std::lock_guard lock(session_.refresh_mutex_); + notify = --session_.active_refresh_count_ == 0; + } + if (notify) { + session_.refresh_cv_.notify_all(); + } + } + + private: + OAuth2Session& session_; + }; + + void UpdateTokenState(const OAuthTokenResponse& token_response, + std::optional + token_request_started_at = std::nullopt) { + token_ = token_response.access_token; + issued_token_type_ = token_response.issued_token_type.empty() + ? AuthProperties::kAccessTokenType + : token_response.issued_token_type; + headers_ = OAuth2Util::AuthHeaders(token_); + expires_at_ = std::chrono::steady_clock::time_point{}; + auto request_started_at = + token_request_started_at.value_or(std::chrono::steady_clock::now()); + if (auto expiration = TokenExpirationTime(token_response, request_started_at); + expiration.has_value()) { + expires_at_ = *expiration; + } + } + + void SetInitialToken( + const OAuthTokenResponse& token_response, + std::optional token_request_started_at) { + UpdateTokenState(token_response, token_request_started_at); + if (config_.keep_refreshed && + expires_at_ != std::chrono::steady_clock::time_point{}) { + ScheduleRefresh(); + } + } + + void DoRefresh() { + DoRefreshAttempt(0, std::chrono::milliseconds(200), std::chrono::steady_clock::now()); + } + + void DoRefreshAttempt(int attempt, std::chrono::milliseconds backoff, + std::chrono::steady_clock::time_point refresh_started_at) { + static constexpr int kMaxRetries = 5; + static constexpr auto kMaxBackoff = std::chrono::milliseconds(10'000); + RefreshAttemptGuard guard(*this); + if (closed_.load()) return; + + auto empty_session = AuthSession::MakeDefault({}); + // TODO(lishuxu): Honor token-exchange-enabled and refresh via token exchange, + // matching Java. + auto result = OAuth2Util::FetchToken(*client_, *empty_session, refresh_properties_); + if (result.has_value()) { + auto& response = result.value(); + { + std::unique_lock lock(mutex_); + UpdateTokenState(response, refresh_started_at); + } + ScheduleRefresh(); + return; + } + + if (attempt + 1 < kMaxRetries && !closed_.load()) { + auto next_backoff = + std::min(std::chrono::duration_cast(backoff * 2), + kMaxBackoff); + std::weak_ptr weak_self = shared_from_this(); + auto retry_id = TokenRefreshScheduler::Instance().Schedule( + backoff, [weak_self = std::move(weak_self), next_attempt = attempt + 1, + next_backoff, refresh_started_at] { + if (auto self = weak_self.lock()) { + self->DoRefreshAttempt(next_attempt, next_backoff, refresh_started_at); + } + }); + scheduled_task_id_.store(retry_id); + } + } + + void ScheduleRefresh() { + if (!config_.keep_refreshed || closed_.load()) return; + auto delay = CalculateRefreshDelay(); + if (delay < std::chrono::milliseconds::zero()) return; + + std::weak_ptr weak_self = shared_from_this(); + auto new_id = TokenRefreshScheduler::Instance().Schedule( + delay, [weak_self = std::move(weak_self)] { + if (auto self = weak_self.lock()) self->DoRefresh(); + }); + scheduled_task_id_.store(new_id); + } + + std::chrono::milliseconds CalculateRefreshDelay() const { + std::shared_lock lock(mutex_); + auto now = std::chrono::steady_clock::now(); + if (expires_at_ == std::chrono::steady_clock::time_point{}) { + return std::chrono::milliseconds(-1); + } + if (expires_at_ <= now) return std::chrono::milliseconds::zero(); + auto expires_in = + std::chrono::duration_cast(expires_at_ - now); + auto refresh_window = std::min(expires_in / 10, std::chrono::milliseconds(300'000)); + auto wait_time = expires_in - refresh_window; + return std::max(wait_time, std::chrono::milliseconds(10)); + } + + mutable std::shared_mutex mutex_; + std::string token_; + std::string issued_token_type_; + std::unordered_map headers_; + std::chrono::steady_clock::time_point expires_at_{}; + Config config_; + AuthProperties refresh_properties_; + std::shared_ptr client_; + std::atomic scheduled_task_id_{0}; + std::atomic closed_{false}; + std::mutex refresh_mutex_; + std::condition_variable refresh_cv_; + int active_refresh_count_ = 0; +}; + +Result> MakeOAuth2Session( + const OAuthTokenResponse& initial_token, const std::string& token_endpoint, + const std::string& client_id, const std::string& client_secret, + const std::string& scope, bool keep_refreshed, + const std::unordered_map& optional_oauth_params, + std::shared_ptr client, + std::optional token_request_started_at); + +} // namespace iceberg::rest::auth::internal diff --git a/src/iceberg/catalog/rest/auth/oauth2_util.cc b/src/iceberg/catalog/rest/auth/oauth2_util.cc index 1f9da22b5..d62ac7e1a 100644 --- a/src/iceberg/catalog/rest/auth/oauth2_util.cc +++ b/src/iceberg/catalog/rest/auth/oauth2_util.cc @@ -45,6 +45,8 @@ constexpr std::string_view kSubjectToken = "subject_token"; constexpr std::string_view kSubjectTokenType = "subject_token_type"; constexpr std::string_view kActorToken = "actor_token"; constexpr std::string_view kActorTokenType = "actor_token_type"; +constexpr std::string_view kAuthorizationHeader = "Authorization"; +constexpr std::string_view kBearerPrefix = "Bearer "; Result ParseTokenResponse(const std::string& response_body) { ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response_body)); @@ -53,15 +55,6 @@ Result ParseTokenResponse(const std::string& response_body) return token_response; } -} // namespace - -std::unordered_map AuthHeaders(const std::string& token) { - if (!token.empty()) { - return {{std::string(kAuthorizationHeader), std::string(kBearerPrefix) + token}}; - } - return {}; -} - bool IsValidTokenType(std::string_view token_type) { return token_type == AuthProperties::kAccessTokenType || token_type == AuthProperties::kRefreshTokenType || @@ -71,94 +64,73 @@ bool IsValidTokenType(std::string_view token_type) { token_type == AuthProperties::kJwtTokenType; } -std::array TokenPreferenceOrder() { - return { - std::string_view(AuthProperties::kIdTokenType), - std::string_view(AuthProperties::kAccessTokenType), - std::string_view(AuthProperties::kJwtTokenType), - std::string_view(AuthProperties::kSaml2TokenType), - std::string_view(AuthProperties::kSaml1TokenType), - }; -} - -std::optional FindPreferredTypedToken( - const std::unordered_map& credentials) { - for (std::string_view token_type : TokenPreferenceOrder()) { - auto token_it = credentials.find(std::string(token_type)); - if (token_it != credentials.end()) { - return OAuth2Token{ - .token_type = token_it->first, - .token = token_it->second, - }; - } - } - return std::nullopt; -} +} // namespace -std::unordered_map FilterTableSessionProperties( - const std::unordered_map& properties) { - std::unordered_map filtered; - auto token_it = properties.find(AuthProperties::kToken.key()); - if (token_it != properties.end()) { - filtered.emplace(token_it->first, token_it->second); - } - for (std::string_view token_type : TokenPreferenceOrder()) { - auto token_it = properties.find(std::string(token_type)); - if (token_it != properties.end()) { - filtered.emplace(token_it->first, token_it->second); - } +std::unordered_map OAuth2Util::AuthHeaders( + const std::string& token) { + if (!token.empty()) { + return {{std::string(kAuthorizationHeader), std::string(kBearerPrefix) + token}}; } - return filtered; + return {}; } -Result> BuildTokenExchangeForm( - const TokenExchangeRequest& request) { - if (request.subject.token.empty()) { +Result> OAuth2Util::TokenExchangeRequest( + const std::string& subject_token, const std::string& subject_token_type, + const std::optional& actor_token, + const std::optional& actor_token_type, const std::string& scope, + const std::unordered_map& optional_params) { + if (subject_token.empty()) { return InvalidArgument("OAuth2 subject token must not be empty"); } - if (!IsValidTokenType(request.subject.token_type)) { - return InvalidArgument("Invalid OAuth2 subject token type: '{}'", - request.subject.token_type); + if (!IsValidTokenType(subject_token_type)) { + return InvalidArgument("Invalid OAuth2 subject token type: '{}'", subject_token_type); } - if (request.actor.has_value()) { - if (request.actor->token.empty()) { + if (actor_token.has_value()) { + if (actor_token->empty()) { return InvalidArgument("OAuth2 actor token must not be empty"); } - if (!IsValidTokenType(request.actor->token_type)) { + if (!actor_token_type.has_value() || !IsValidTokenType(*actor_token_type)) { return InvalidArgument("Invalid OAuth2 actor token type: '{}'", - request.actor->token_type); + actor_token_type.value_or("")); } } std::unordered_map form_data{ {std::string(kGrantType), std::string(kTokenExchange)}, - {std::string(kScope), request.scope}, - {std::string(kSubjectToken), request.subject.token}, - {std::string(kSubjectTokenType), request.subject.token_type}, + {std::string(kScope), scope}, + {std::string(kSubjectToken), subject_token}, + {std::string(kSubjectTokenType), subject_token_type}, }; - if (request.actor.has_value()) { - form_data.emplace(kActorToken, request.actor->token); - form_data.emplace(kActorTokenType, request.actor->token_type); + if (actor_token.has_value()) { + form_data.emplace(kActorToken, *actor_token); + form_data.emplace(kActorTokenType, *actor_token_type); } - for (const auto& [key, value] : request.optional_oauth_params) { + for (const auto& [key, value] : optional_params) { form_data.insert_or_assign(key, value); } return form_data; } -Result ExchangeToken( +Result OAuth2Util::ExchangeToken( HttpClient& client, AuthSession& session, const std::unordered_map& extra_headers, - const TokenExchangeRequest& request) { - ICEBERG_ASSIGN_OR_RAISE(auto form_data, BuildTokenExchangeForm(request)); + const std::string& subject_token, const std::string& subject_token_type, + const std::optional& actor_token, + const std::optional& actor_token_type, const std::string& scope, + const std::string& oauth2_server_uri, + const std::unordered_map& optional_params) { ICEBERG_ASSIGN_OR_RAISE( - auto response, client.PostForm(request.oauth2_server_uri, form_data, extra_headers, - *OAuthErrorHandler::Instance(), session)); + auto form_data, TokenExchangeRequest(subject_token, subject_token_type, actor_token, + actor_token_type, scope, optional_params)); + ICEBERG_ASSIGN_OR_RAISE(auto response, + client.PostForm(oauth2_server_uri, form_data, extra_headers, + *OAuthErrorHandler::Instance(), session)); return ParseTokenResponse(response.body()); } -Result FetchToken(HttpClient& client, AuthSession& session, - const AuthProperties& properties) { +Result OAuth2Util::FetchToken(HttpClient& client, + AuthSession& session, + const AuthProperties& properties) { std::unordered_map form_data{ {std::string(kGrantType), std::string(kClientCredentials)}, {std::string(kClientSecret), properties.client_secret()}, @@ -178,7 +150,7 @@ Result FetchToken(HttpClient& client, AuthSession& session, return ParseTokenResponse(response.body()); } -std::optional ExpiresAtMillis(std::string_view token) { +std::optional OAuth2Util::ExpiresAtMillis(std::string_view token) { if (token.empty()) { return std::nullopt; } diff --git a/src/iceberg/catalog/rest/auth/oauth2_util.h b/src/iceberg/catalog/rest/auth/oauth2_util.h index fe4f6723a..78a29050b 100644 --- a/src/iceberg/catalog/rest/auth/oauth2_util.h +++ b/src/iceberg/catalog/rest/auth/oauth2_util.h @@ -19,7 +19,6 @@ #pragma once -#include #include #include #include @@ -36,81 +35,65 @@ namespace iceberg::rest::auth { -inline constexpr std::string_view kAuthorizationHeader = "Authorization"; -inline constexpr std::string_view kBearerPrefix = "Bearer "; - -struct ICEBERG_REST_EXPORT OAuth2Token { - std::string token_type; - std::string token; -}; - -struct ICEBERG_REST_EXPORT TokenExchangeRequest { - std::string oauth2_server_uri; - OAuth2Token subject; - std::optional actor; - std::string scope; - std::unordered_map optional_oauth_params; +/// \brief OAuth2 token and authentication utilities. +class ICEBERG_REST_EXPORT OAuth2Util { + public: + OAuth2Util() = delete; + + /// \brief Fetch an OAuth2 token using the client_credentials grant type. + /// + /// \param client HTTP client to use for the request. + /// \param session Auth session for the request headers. + /// \param properties Auth configuration containing credential, scope, + /// token endpoint, and optional OAuth params. + /// \return The token response or an error. + static Result FetchToken(HttpClient& client, AuthSession& session, + const AuthProperties& properties); + + /// \brief Build auth headers from a token string. + /// + /// \param token Bearer token string (may be empty). + /// \return Headers map with Authorization header if token is non-empty. + static std::unordered_map AuthHeaders( + const std::string& token); + + /// \brief Exchange an OAuth2 token using the RFC 8693 grant type. + /// + /// \param client HTTP client to use for the request. + /// \param session Auth session for the request headers. + /// \param extra_headers Request headers applied before session authentication. + /// \param subject_token Subject token to exchange. + /// \param subject_token_type Subject token type. + /// \param actor_token Optional actor token. + /// \param actor_token_type Optional actor token type. + /// \param scope OAuth2 scope. + /// \param oauth2_server_uri Token exchange endpoint. + /// \param optional_params Optional OAuth parameters. + /// \return The token response or an error. + static Result ExchangeToken( + HttpClient& client, AuthSession& session, + const std::unordered_map& extra_headers, + const std::string& subject_token, const std::string& subject_token_type, + const std::optional& actor_token, + const std::optional& actor_token_type, const std::string& scope, + const std::string& oauth2_server_uri, + const std::unordered_map& optional_params); + + /// \brief Extract expiration time from a JWT token. + /// + /// Decodes the JWT payload (base64url) and reads the "exp" claim. + /// Returns std::nullopt if the token is not a valid JWT or has no "exp" claim. + /// + /// \param token A token string containing three dot-separated JWT segments. + /// \return Expiration time as milliseconds since epoch, or std::nullopt. + static std::optional ExpiresAtMillis(std::string_view token); + + private: + static Result> TokenExchangeRequest( + const std::string& subject_token, const std::string& subject_token_type, + const std::optional& actor_token, + const std::optional& actor_token_type, const std::string& scope, + const std::unordered_map& optional_params); }; -/// \brief Fetch an OAuth2 token using the client_credentials grant type. -/// -/// \param client HTTP client to use for the request. -/// \param session Auth session for the request headers. -/// \param properties Auth configuration containing credential, scope, -/// token endpoint, and optional OAuth params. -/// \return The token response or an error. -ICEBERG_REST_EXPORT Result FetchToken( - HttpClient& client, AuthSession& session, const AuthProperties& properties); - -/// \brief Build auth headers from a token string. -/// -/// \param token Bearer token string (may be empty). -/// \return Headers map with Authorization header if token is non-empty. -ICEBERG_REST_EXPORT std::unordered_map AuthHeaders( - const std::string& token); - -/// \brief Return whether a token type is a supported RFC token type. -ICEBERG_REST_EXPORT bool IsValidTokenType(std::string_view token_type); - -/// \brief Return the preferred order for typed OAuth tokens. -ICEBERG_REST_EXPORT std::array TokenPreferenceOrder(); - -/// \brief Find the highest-preference typed OAuth token in credentials. -ICEBERG_REST_EXPORT std::optional FindPreferredTypedToken( - const std::unordered_map& credentials); - -/// \brief Filter table session properties to allowed OAuth credentials. -ICEBERG_REST_EXPORT std::unordered_map -FilterTableSessionProperties( - const std::unordered_map& properties); - -/// \brief Build RFC 8693 token exchange form data. -/// -/// \param request Token exchange request values. -/// \return Form data or an error if token values are invalid. -ICEBERG_REST_EXPORT Result> -BuildTokenExchangeForm(const TokenExchangeRequest& request); - -/// \brief Exchange an OAuth2 token using the RFC 8693 grant type. -/// -/// \param client HTTP client to use for the request. -/// \param session Auth session for the request headers. -/// \param extra_headers Request headers applied before session authentication. -/// \param request Token exchange endpoint and form values. -/// \return The token response or an error. -ICEBERG_REST_EXPORT Result ExchangeToken( - HttpClient& client, AuthSession& session, - const std::unordered_map& extra_headers, - const TokenExchangeRequest& request); - -/// \brief Extract expiration time from a JWT token. -/// -/// Decodes the JWT payload (base64url) and reads the "exp" claim. -/// Returns std::nullopt if the token is not a valid JWT or has no "exp" claim. -/// -/// \param token A token string. If it is a JWT (three dot-separated base64url -/// segments), the "exp" claim is extracted from the payload. -/// \return Expiration time as milliseconds since epoch, or std::nullopt. -ICEBERG_REST_EXPORT std::optional ExpiresAtMillis(std::string_view token); - } // namespace iceberg::rest::auth diff --git a/src/iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h b/src/iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h index 53ec0853a..7bb7a3aa3 100644 --- a/src/iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h +++ b/src/iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h @@ -118,11 +118,11 @@ class ICEBERG_REST_EXPORT SigV4AuthManager : public AuthManager { ~SigV4AuthManager() override; Result> InitSession( - HttpClient& init_client, + std::shared_ptr init_client, const std::unordered_map& properties) override; Result> CatalogSession( - HttpClient& shared_client, + std::shared_ptr shared_client, const std::unordered_map& properties) override; Result> ContextualSession( diff --git a/src/iceberg/catalog/rest/auth/sigv4_manager.cc b/src/iceberg/catalog/rest/auth/sigv4_manager.cc index 6678f1b3f..6e0dce86b 100644 --- a/src/iceberg/catalog/rest/auth/sigv4_manager.cc +++ b/src/iceberg/catalog/rest/auth/sigv4_manager.cc @@ -51,6 +51,7 @@ namespace iceberg::rest::auth { namespace { +constexpr std::string_view kAuthorizationHeader = "Authorization"; constexpr std::string_view kAmzContentSha256Header = "x-amz-content-sha256"; class AwsSdkLifecycle { @@ -379,7 +380,7 @@ SigV4AuthManager::SigV4AuthManager(std::unique_ptr delegate) SigV4AuthManager::~SigV4AuthManager() = default; Result> SigV4AuthManager::InitSession( - HttpClient& init_client, + std::shared_ptr init_client, const std::unordered_map& properties) { ICEBERG_RETURN_UNEXPECTED(AwsSdkLifecycle::Instance().EnsureInitialized()); ICEBERG_ASSIGN_OR_RAISE(auto delegate_session, @@ -389,7 +390,7 @@ Result> SigV4AuthManager::InitSession( } Result> SigV4AuthManager::CatalogSession( - HttpClient& shared_client, + std::shared_ptr shared_client, const std::unordered_map& properties) { ICEBERG_RETURN_UNEXPECTED(AwsSdkLifecycle::Instance().EnsureInitialized()); catalog_properties_ = properties; diff --git a/src/iceberg/catalog/rest/auth/token_refresh_scheduler.h b/src/iceberg/catalog/rest/auth/token_refresh_scheduler.h index 02dc0e14f..5ef20ed94 100644 --- a/src/iceberg/catalog/rest/auth/token_refresh_scheduler.h +++ b/src/iceberg/catalog/rest/auth/token_refresh_scheduler.h @@ -37,7 +37,7 @@ namespace iceberg::rest::auth { /// \brief A process-global scheduler for delayed token refresh tasks. /// /// Uses a single background thread that sleeps until the next task is due. -/// All OAuth2AuthSession instances share this scheduler. Tasks are lightweight +/// All OAuth2Session instances share this scheduler. Tasks are lightweight /// (a single HTTP POST to refresh a token), so one thread is sufficient. /// /// Thread safety: All public methods are thread-safe. diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc index 349071f42..4a4f990ea 100644 --- a/src/iceberg/catalog/rest/rest_catalog.cc +++ b/src/iceberg/catalog/rest/rest_catalog.cc @@ -399,7 +399,7 @@ Result> RestCatalog::Make( config.Get(RestCatalogProperties::kNamespaceSeparator))); // Create init session for fetching server configuration - HttpClient init_client(config.ExtractHeaders()); + auto init_client = std::make_shared(config.ExtractHeaders()); ICEBERG_ASSIGN_OR_RAISE(auto init_session, auth_manager->InitSession(init_client, config.configs())); ICEBERG_ASSIGN_OR_RAISE(auto server_config, @@ -432,7 +432,7 @@ Result> RestCatalog::Make( auto client = std::make_shared(final_config.ExtractHeaders()); ICEBERG_ASSIGN_OR_RAISE(auto catalog_session, - auth_manager->CatalogSession(*client, final_config.configs())); + auth_manager->CatalogSession(client, final_config.configs())); // Create FileIO with the final configuration ICEBERG_ASSIGN_OR_RAISE(auto file_io, MakeCatalogFileIO(final_config)); diff --git a/src/iceberg/test/auth_manager_test.cc b/src/iceberg/test/auth_manager_test.cc index 4eb167bb4..61962d1c3 100644 --- a/src/iceberg/test/auth_manager_test.cc +++ b/src/iceberg/test/auth_manager_test.cc @@ -22,6 +22,7 @@ #include #include #include +#include #include #include #include @@ -35,6 +36,7 @@ #include "iceberg/catalog/rest/auth/auth_managers.h" #include "iceberg/catalog/rest/auth/auth_properties.h" #include "iceberg/catalog/rest/auth/auth_session.h" +#include "iceberg/catalog/rest/auth/auth_session_internal.h" #include "iceberg/catalog/rest/auth/oauth2_util.h" #include "iceberg/catalog/rest/auth/token_refresh_scheduler.h" #include "iceberg/catalog/rest/catalog_properties.h" @@ -68,198 +70,9 @@ std::string MakeJwt(const std::string& payload_json) { class AuthManagerTest : public ::testing::Test { protected: - HttpClient client_{{}}; + std::shared_ptr client_ = std::make_shared(); }; -TEST(OAuth2UtilTest, TokenTypeConstantsUseRfcUrns) { - EXPECT_EQ(AuthProperties::kAccessTokenType, - "urn:ietf:params:oauth:token-type:access_token"); - EXPECT_EQ(AuthProperties::kRefreshTokenType, - "urn:ietf:params:oauth:token-type:refresh_token"); - EXPECT_EQ(AuthProperties::kIdTokenType, "urn:ietf:params:oauth:token-type:id_token"); - EXPECT_EQ(AuthProperties::kSaml1TokenType, "urn:ietf:params:oauth:token-type:saml1"); - EXPECT_EQ(AuthProperties::kSaml2TokenType, "urn:ietf:params:oauth:token-type:saml2"); - EXPECT_EQ(AuthProperties::kJwtTokenType, "urn:ietf:params:oauth:token-type:jwt"); -} - -TEST(OAuth2UtilTest, ValidTokenTypesIncludeRefreshToken) { - EXPECT_TRUE(IsValidTokenType(AuthProperties::kAccessTokenType)); - EXPECT_TRUE(IsValidTokenType(AuthProperties::kRefreshTokenType)); - EXPECT_TRUE(IsValidTokenType(AuthProperties::kIdTokenType)); - EXPECT_TRUE(IsValidTokenType(AuthProperties::kSaml1TokenType)); - EXPECT_TRUE(IsValidTokenType(AuthProperties::kSaml2TokenType)); - EXPECT_TRUE(IsValidTokenType(AuthProperties::kJwtTokenType)); - EXPECT_FALSE(IsValidTokenType("urn:ietf:params:oauth:token-type:unknown")); -} - -TEST(OAuth2UtilTest, TokenPreferenceOrder) { - auto order = TokenPreferenceOrder(); - ASSERT_EQ(order.size(), 5); - EXPECT_EQ(order[0], AuthProperties::kIdTokenType); - EXPECT_EQ(order[1], AuthProperties::kAccessTokenType); - EXPECT_EQ(order[2], AuthProperties::kJwtTokenType); - EXPECT_EQ(order[3], AuthProperties::kSaml2TokenType); - EXPECT_EQ(order[4], AuthProperties::kSaml1TokenType); -} - -TEST(OAuth2UtilTest, FindPreferredTypedTokenUsesPreferenceOrder) { - std::unordered_map credentials = { - {AuthProperties::kAccessTokenType, "access-token"}, - {AuthProperties::kJwtTokenType, "jwt-token"}, - {AuthProperties::kIdTokenType, "id-token"}, - {AuthProperties::kSaml2TokenType, "saml2-token"}, - {AuthProperties::kSaml1TokenType, "saml1-token"}, - }; - - auto token = FindPreferredTypedToken(credentials); - ASSERT_TRUE(token.has_value()); - EXPECT_EQ(token->token_type, AuthProperties::kIdTokenType); - EXPECT_EQ(token->token, "id-token"); - - credentials.erase(AuthProperties::kIdTokenType); - token = FindPreferredTypedToken(credentials); - ASSERT_TRUE(token.has_value()); - EXPECT_EQ(token->token_type, AuthProperties::kAccessTokenType); - EXPECT_EQ(token->token, "access-token"); - - credentials.clear(); - EXPECT_FALSE(FindPreferredTypedToken(credentials).has_value()); -} - -TEST(OAuth2UtilTest, FilterTableSessionPropertiesUsesAllowList) { - std::unordered_map properties = { - {AuthProperties::kToken.key(), "bearer-token"}, - {AuthProperties::kCredential.key(), "client:secret"}, - {AuthProperties::kScope.key(), "catalog"}, - {AuthProperties::kAccessTokenType, "access-token"}, - {AuthProperties::kRefreshTokenType, "refresh-token"}, - {AuthProperties::kIdTokenType, "id-token"}, - {AuthProperties::kJwtTokenType, "jwt-token"}, - {AuthProperties::kSaml2TokenType, "saml2-token"}, - {AuthProperties::kSaml1TokenType, "saml1-token"}, - {"unrelated", "value"}, - }; - - auto filtered = FilterTableSessionProperties(properties); - EXPECT_EQ(filtered.size(), 6); - EXPECT_EQ(filtered.at(AuthProperties::kToken.key()), "bearer-token"); - EXPECT_EQ(filtered.at(AuthProperties::kAccessTokenType), "access-token"); - EXPECT_EQ(filtered.at(AuthProperties::kIdTokenType), "id-token"); - EXPECT_EQ(filtered.at(AuthProperties::kJwtTokenType), "jwt-token"); - EXPECT_EQ(filtered.at(AuthProperties::kSaml2TokenType), "saml2-token"); - EXPECT_EQ(filtered.at(AuthProperties::kSaml1TokenType), "saml1-token"); - EXPECT_FALSE(filtered.contains(AuthProperties::kCredential.key())); - EXPECT_FALSE(filtered.contains(AuthProperties::kScope.key())); - EXPECT_FALSE(filtered.contains(AuthProperties::kRefreshTokenType)); - EXPECT_FALSE(filtered.contains("unrelated")); -} - -TEST(OAuth2UtilTest, BuildsTokenExchangeFormWithoutActor) { - TokenExchangeRequest request{ - .oauth2_server_uri = "https://auth.example.com/token", - .subject = - { - .token_type = AuthProperties::kIdTokenType, - .token = "subject-token", - }, - .scope = "catalog", - .optional_oauth_params = - { - {AuthProperties::kAudience.key(), "catalog-audience"}, - {AuthProperties::kResource.key(), "catalog-resource"}, - }, - }; - - ICEBERG_UNWRAP_OR_FAIL(auto form_data, BuildTokenExchangeForm(request)); - EXPECT_EQ(form_data.size(), 6); - EXPECT_EQ(form_data.at("grant_type"), - "urn:ietf:params:oauth:grant-type:token-exchange"); - EXPECT_EQ(form_data.at("scope"), "catalog"); - EXPECT_EQ(form_data.at("subject_token"), "subject-token"); - EXPECT_EQ(form_data.at("subject_token_type"), AuthProperties::kIdTokenType); - EXPECT_EQ(form_data.at("audience"), "catalog-audience"); - EXPECT_EQ(form_data.at("resource"), "catalog-resource"); - EXPECT_FALSE(form_data.contains("actor_token")); - EXPECT_FALSE(form_data.contains("actor_token_type")); -} - -TEST(OAuth2UtilTest, BuildsTokenExchangeFormWithActor) { - TokenExchangeRequest request{ - .subject = - { - .token_type = AuthProperties::kJwtTokenType, - .token = "subject-token", - }, - .actor = - OAuth2Token{ - .token_type = AuthProperties::kAccessTokenType, - .token = "actor-token", - }, - .scope = "catalog", - }; - - ICEBERG_UNWRAP_OR_FAIL(auto form_data, BuildTokenExchangeForm(request)); - EXPECT_EQ(form_data.size(), 6); - EXPECT_EQ(form_data.at("actor_token"), "actor-token"); - EXPECT_EQ(form_data.at("actor_token_type"), AuthProperties::kAccessTokenType); -} - -TEST(OAuth2UtilTest, TokenExchangeOptionalParamsUseLastValue) { - TokenExchangeRequest request{ - .subject = - { - .token_type = AuthProperties::kAccessTokenType, - .token = "subject-token", - }, - .scope = "catalog", - .optional_oauth_params = {{"scope", "custom-scope"}}, - }; - - ICEBERG_UNWRAP_OR_FAIL(auto form_data, BuildTokenExchangeForm(request)); - EXPECT_EQ(form_data.at("scope"), "custom-scope"); -} - -TEST(OAuth2UtilTest, RejectsInvalidTokenExchangeSubject) { - TokenExchangeRequest request{ - .subject = {.token_type = "invalid-token-type", .token = "subject-token"}, - }; - - auto invalid_type = BuildTokenExchangeForm(request); - EXPECT_THAT(invalid_type, IsError(ErrorKind::kInvalidArgument)); - EXPECT_THAT(invalid_type, HasErrorMessage("Invalid OAuth2 subject token type")); - - request.subject = { - .token_type = AuthProperties::kAccessTokenType, - .token = "", - }; - auto empty_token = BuildTokenExchangeForm(request); - EXPECT_THAT(empty_token, IsError(ErrorKind::kInvalidArgument)); - EXPECT_THAT(empty_token, HasErrorMessage("subject token must not be empty")); -} - -TEST(OAuth2UtilTest, RejectsInvalidTokenExchangeActor) { - TokenExchangeRequest request{ - .subject = - { - .token_type = AuthProperties::kAccessTokenType, - .token = "subject-token", - }, - .actor = OAuth2Token{.token_type = "invalid-token-type", .token = "actor-token"}, - }; - - auto invalid_type = BuildTokenExchangeForm(request); - EXPECT_THAT(invalid_type, IsError(ErrorKind::kInvalidArgument)); - EXPECT_THAT(invalid_type, HasErrorMessage("Invalid OAuth2 actor token type")); - - request.actor = OAuth2Token{ - .token_type = AuthProperties::kAccessTokenType, - .token = "", - }; - auto empty_token = BuildTokenExchangeForm(request); - EXPECT_THAT(empty_token, IsError(ErrorKind::kInvalidArgument)); - EXPECT_THAT(empty_token, HasErrorMessage("actor token must not be empty")); -} - TEST(AuthPropertiesTest, ResolvesDefaultOAuth2ServerUri) { ICEBERG_UNWRAP_OR_FAIL( auto config, @@ -272,14 +85,14 @@ TEST(AuthPropertiesTest, ResolvesDefaultOAuth2ServerUri) { "https://catalog.example.com/api/v1/oauth/tokens"); } -TEST(AuthPropertiesTest, ResolvesEmptyOAuth2ServerUriToDefault) { +TEST(AuthPropertiesTest, PreservesEmptyOAuth2ServerUri) { ICEBERG_UNWRAP_OR_FAIL( auto config, AuthProperties::FromProperties({ {RestCatalogProperties::kUri.key(), "https://catalog.example.com"}, {AuthProperties::kOAuth2ServerUri.key(), ""}, })); - EXPECT_EQ(config.oauth2_server_uri(), "https://catalog.example.com/v1/oauth/tokens"); + EXPECT_TRUE(config.oauth2_server_uri().empty()); } TEST(AuthPropertiesTest, ResolvesExplicitRelativeOAuth2ServerUri) { @@ -290,7 +103,7 @@ TEST(AuthPropertiesTest, ResolvesExplicitRelativeOAuth2ServerUri) { {AuthProperties::kOAuth2ServerUri.key(), "oauth/token/"}, })); - EXPECT_EQ(config.oauth2_server_uri(), "https://catalog.example.com/api/oauth/token"); + EXPECT_EQ(config.oauth2_server_uri(), "https://catalog.example.com/api/oauth/token/"); } TEST(AuthPropertiesTest, PreservesExplicitAbsoluteOAuth2ServerUri) { @@ -313,15 +126,25 @@ TEST(AuthPropertiesTest, PreservesRelativeOAuth2ServerUriWithoutCatalogUri) { EXPECT_EQ(config.oauth2_server_uri(), "oauth/token"); } -TEST(AuthPropertiesTest, ResolvesExplicitAbsolutePathOAuth2ServerUri) { +TEST(AuthPropertiesTest, PreservesAbsolutePathWithoutCatalogUri) { + ICEBERG_UNWRAP_OR_FAIL(auto config, + AuthProperties::FromProperties({ + {AuthProperties::kOAuth2ServerUri.key(), "/oauth/token"}, + })); + + EXPECT_EQ(config.oauth2_server_uri(), "/oauth/token"); +} + +TEST(AuthPropertiesTest, ResolvesOAuth2ServerUriWithLeadingSlash) { ICEBERG_UNWRAP_OR_FAIL( auto config, AuthProperties::FromProperties({ - {RestCatalogProperties::kUri.key(), "https://catalog.example.com/"}, + {RestCatalogProperties::kUri.key(), "https://catalog.example.com/api/"}, {AuthProperties::kOAuth2ServerUri.key(), "/v1/oauth/tokens"}, })); - EXPECT_EQ(config.oauth2_server_uri(), "https://catalog.example.com/v1/oauth/tokens"); + EXPECT_EQ(config.oauth2_server_uri(), + "https://catalog.example.com/api/v1/oauth/tokens"); } // Verifies loading NoopAuthManager with explicit "none" auth type @@ -371,7 +194,7 @@ TEST_F(AuthManagerTest, HttpHeadersAreCaseInsensitiveSingleValueMap) { } TEST_F(AuthManagerTest, DefaultSessionPreservesRequestAuthorizationHeader) { - auto session = AuthSession::MakeDefault(AuthHeaders("parent-token")); + auto session = AuthSession::MakeDefault(OAuth2Util::AuthHeaders("parent-token")); ICEBERG_UNWRAP_OR_FAIL( auto authenticated, @@ -397,8 +220,9 @@ TEST_F(AuthManagerTest, OAuth2SessionPreservesRequestAuthorizationHeader) { EXPECT_EQ(authenticated.headers.at("Authorization"), "Basic credentials"); - ASSERT_TRUE(session->OAuth2Info().has_value()); - EXPECT_EQ(session->OAuth2Info()->issued_token_type, AuthProperties::kAccessTokenType); + auto info = session->OAuth2Info(); + ASSERT_TRUE(info.has_value()); + EXPECT_EQ(info->issued_token_type, AuthProperties::kAccessTokenType); } TEST_F(AuthManagerTest, OAuth2SessionExposesMetadata) { @@ -428,8 +252,8 @@ TEST_F(AuthManagerTest, OAuth2SessionExposesMetadata) { TEST_F(AuthManagerTest, HttpClientRejectsParamsWhenUrlAlreadyHasQuery) { auto session = AuthSession::MakeDefault({}); auto result = - client_.Get("http://127.0.0.1/v1/config?existing=true", {{"warehouse", "prod"}}, - /*headers=*/{}, *rest::DefaultErrorHandler::Instance(), *session); + client_->Get("http://127.0.0.1/v1/config?existing=true", {{"warehouse", "prod"}}, + /*headers=*/{}, *rest::DefaultErrorHandler::Instance(), *session); EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument)); EXPECT_THAT(result, HasErrorMessage("must not contain a query string")); @@ -550,7 +374,8 @@ TEST_F(AuthManagerTest, RegisterCustomAuthManager) { class CustomAuthManager : public AuthManager { public: Result> CatalogSession( - HttpClient&, const std::unordered_map&) override { + std::shared_ptr, + const std::unordered_map&) override { return AuthSession::MakeDefault({{"X-Custom-Auth", "custom-value"}}); } }; @@ -838,12 +663,12 @@ TEST_F(AuthManagerTest, OAuthTokenResponseNATokenType) { EXPECT_EQ(result->token_type, "N_A"); } -// ---- ExpiresAtMillis tests ---- +// ---- OAuth2Util expiry tests ---- TEST_F(AuthManagerTest, ExpiresAtMillisValidJwt) { std::string token = MakeJwt(R"({"sub":"user","exp":1700000000})"); - auto result = ExpiresAtMillis(token); + auto result = OAuth2Util::ExpiresAtMillis(token); ASSERT_TRUE(result.has_value()); EXPECT_EQ(result.value(), 1700000000LL * 1000); @@ -863,7 +688,7 @@ TEST_F(AuthManagerTest, ExpiresAtMillisInvalidTokensReturnNullopt) { }; for (const auto& token : tokens) { - EXPECT_FALSE(ExpiresAtMillis(token).has_value()) << token; + EXPECT_FALSE(OAuth2Util::ExpiresAtMillis(token).has_value()) << token; } } @@ -942,10 +767,10 @@ TEST(TokenRefreshSchedulerTest, CancelInvalidHandleIsNoop) { scheduler.Shutdown(); } -// ---- OAuth2AuthSession tests ---- +// ---- OAuth2Session tests ---- -TEST(OAuth2AuthSessionTest, InitialTokenIsUsed) { - HttpClient client({}); +TEST(OAuth2SessionTest, InitialTokenIsUsed) { + auto client = std::make_shared(); OAuthTokenResponse token_response; token_response.access_token = "initial-token-123"; token_response.token_type = "bearer"; @@ -966,4 +791,35 @@ TEST(OAuth2AuthSessionTest, InitialTokenIsUsed) { session->Close(); } +TEST(OAuth2SessionTest, InitTokenExpirationUsesRequestStartTime) { + OAuthTokenResponse token_response{ + .access_token = "opaque-token", + .token_type = "bearer", + .expires_in_secs = 60, + }; + auto request_started_at = std::chrono::steady_clock::time_point{}; + + auto expiration = internal::TokenExpirationTime(token_response, request_started_at); + + ASSERT_TRUE(expiration.has_value()); + EXPECT_EQ(*expiration, request_started_at + std::chrono::seconds(60)); +} + +TEST(OAuth2SessionTest, JwtExpirationTakesPriorityOverExpiresIn) { + OAuthTokenResponse token_response{ + .access_token = MakeJwt(R"({"exp":120})"), + .token_type = "bearer", + .expires_in_secs = 60, + }; + auto request_started_at = std::chrono::steady_clock::time_point{}; + auto now_system = std::chrono::system_clock::time_point(std::chrono::seconds(100)); + auto now_steady = std::chrono::steady_clock::time_point(std::chrono::seconds(50)); + + auto expiration = internal::TokenExpirationTime(token_response, request_started_at, + now_system, now_steady); + + ASSERT_TRUE(expiration.has_value()); + EXPECT_EQ(*expiration, std::chrono::steady_clock::time_point(std::chrono::seconds(70))); +} + } // namespace iceberg::rest::auth diff --git a/src/iceberg/test/rest_catalog_integration_test.cc b/src/iceberg/test/rest_catalog_integration_test.cc index feeee2c3a..25e2b1955 100644 --- a/src/iceberg/test/rest_catalog_integration_test.cc +++ b/src/iceberg/test/rest_catalog_integration_test.cc @@ -216,7 +216,7 @@ TEST_F(RestCatalogIntegrationTest, MakeCatalogSuccess) { } TEST_F(RestCatalogIntegrationTest, OAuthContextCredentialEndToEnd) { - HttpClient client; + auto client = std::make_shared(); std::unordered_map properties = { {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2}, {auth::AuthProperties::kToken.key(), "catalog-token"}, @@ -240,8 +240,33 @@ TEST_F(RestCatalogIntegrationTest, OAuthContextCredentialEndToEnd) { EXPECT_EQ(info->issued_token_type, auth::AuthProperties::kAccessTokenType); } +TEST_F(RestCatalogIntegrationTest, OAuthContextCredentialThroughRestCatalog) { + auto config = RestCatalogProperties::default_properties(); + config.Set(RestCatalogProperties::kUri, CatalogUri()) + .Set(RestCatalogProperties::kName, std::string(kCatalogName)) + .Set(RestCatalogProperties::kWarehouse, std::string(kWarehouseName)); + config.mutable_configs()[std::string(RestCatalogProperties::kIOImpl.key())] = + std::string(kStdFileIOImpl); + config.mutable_configs()[auth::AuthProperties::kAuthType] = + auth::AuthProperties::kAuthTypeOAuth2; + config.mutable_configs()[auth::AuthProperties::kToken.key()] = "catalog-token"; + config.mutable_configs()[auth::AuthProperties::kOAuth2ServerUri.key()] = + OAuthTokenUri(); + + ICEBERG_UNWRAP_OR_FAIL(auto root, RestCatalog::Make(config)); + SessionContext context{ + .session_id = "tenant-context-credential", + .credentials = {{auth::AuthProperties::kCredential.key(), "context-client:secret"}}, + }; + ICEBERG_UNWRAP_OR_FAIL(auto catalog, root->WithContext(context)); + ICEBERG_UNWRAP_OR_FAIL(auto namespaces, + catalog->ListNamespaces(Namespace{.levels = {}})); + + EXPECT_TRUE(namespaces.empty()); +} + TEST_F(RestCatalogIntegrationTest, OAuthContextTypedTokenEndToEnd) { - HttpClient client; + auto client = std::make_shared(); std::unordered_map properties = { {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2}, {auth::AuthProperties::kToken.key(), "catalog-token"}, @@ -266,7 +291,7 @@ TEST_F(RestCatalogIntegrationTest, OAuthContextTypedTokenEndToEnd) { } TEST_F(RestCatalogIntegrationTest, OAuthTableTypedTokenEndToEnd) { - HttpClient client; + auto client = std::make_shared(); std::unordered_map properties = { {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2}, {auth::AuthProperties::kToken.key(), "catalog-token"}, @@ -291,7 +316,7 @@ TEST_F(RestCatalogIntegrationTest, OAuthTableTypedTokenEndToEnd) { } TEST_F(RestCatalogIntegrationTest, OAuthTokenExchangeWithoutActorEndToEnd) { - HttpClient client; + auto client = std::make_shared(); std::unordered_map properties = { {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2}, {auth::AuthProperties::kOAuth2ServerUri.key(), OAuthTokenUri()}, diff --git a/src/iceberg/test/rest_util_test.cc b/src/iceberg/test/rest_util_test.cc index 6af3772d6..0035afca0 100644 --- a/src/iceberg/test/rest_util_test.cc +++ b/src/iceberg/test/rest_util_test.cc @@ -103,14 +103,6 @@ TEST(RestUtilTest, ResourcePathsRejectsEmptyNamespaceSeparator) { EXPECT_THAT(result, HasErrorMessage("REST namespace separator cannot be empty")); } -TEST(RestUtilTest, OAuth2TokensPathDoesNotUseCatalogPrefix) { - ICEBERG_UNWRAP_OR_FAIL( - auto paths, ResourcePaths::Make("https://catalog.example.com", "warehouse", "%1F")); - - EXPECT_THAT(paths->OAuth2Tokens(), - HasValue(::testing::Eq("https://catalog.example.com/v1/oauth/tokens"))); -} - TEST(RestUtilTest, EncodeString) { // RFC 3986 unreserved characters should not be encoded EXPECT_THAT(EncodeString("abc123XYZ"), HasValue(::testing::Eq("abc123XYZ"))); diff --git a/src/iceberg/test/sigv4_auth_test.cc b/src/iceberg/test/sigv4_auth_test.cc index 12d18792f..6dad2d2ff 100644 --- a/src/iceberg/test/sigv4_auth_test.cc +++ b/src/iceberg/test/sigv4_auth_test.cc @@ -21,6 +21,7 @@ # include # include +# include # include # include # include @@ -201,7 +202,7 @@ class SigV4AuthTest : public ::testing::Test { return session->Authenticate(std::move(request)); } - HttpClient client_{{}}; + std::shared_ptr client_ = std::make_shared(); }; TEST_F(SigV4AuthTest, LifecycleInitializeIsIdempotent) {