feat!(rest): introduce AuthManager/AuthSession and migrate OAuth2 #2838

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

CTTY merged 9 commits into apache:main from plusplusjiajia:feat/rest-auth-manager-core

Aug 13, 2026

Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

crates/catalog/rest/Cargo.toml

-Original file line number
+Diff line change
@@ Expand Up / @@ -40,12 +40,17 @@ serde = { workspace = true } @@
     serde_derive = { workspace = true }
     serde_json = { workspace = true }
     tokio = { workspace = true }
+    tracing = { workspace = true }
     typed-builder = { workspace = true }
     uuid = { workspace = true, features = ["v4"] }
     [dev-dependencies]
+    bytes = { workspace = true }
+    futures = { workspace = true }
     iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }
     mockito = { workspace = true }
+    # `stream` lets tests build a streaming body to exercise HttpRequestBody::Streaming.
+    reqwest = { workspace = true, features = ["stream"] }
     tokio = { workspace = true }
     [lints]
@@ Expand Down @@

crates/catalog/rest/public-api.txt

-Original file line number
+Diff line change
@@ -1,4 +1,19 @@
     pub mod iceberg_catalog_rest
+    pub enum iceberg_catalog_rest::HttpRequestBody<'a>
+    pub iceberg_catalog_rest::HttpRequestBody::Buffered(&'a [u8])
+    pub iceberg_catalog_rest::HttpRequestBody::Empty
+    pub iceberg_catalog_rest::HttpRequestBody::Streaming
+    impl<'a> iceberg_catalog_rest::HttpRequestBody<'a>
+    pub fn iceberg_catalog_rest::HttpRequestBody<'a>::as_bytes(&self) -> core::option::Option<&'a [u8]>
+    impl<'a> core::clone::Clone for iceberg_catalog_rest::HttpRequestBody<'a>
+    pub fn iceberg_catalog_rest::HttpRequestBody<'a>::clone(&self) -> iceberg_catalog_rest::HttpRequestBody<'a>
+    impl<'a> core::cmp::Eq for iceberg_catalog_rest::HttpRequestBody<'a>
+    impl<'a> core::cmp::PartialEq for iceberg_catalog_rest::HttpRequestBody<'a>
+    pub fn iceberg_catalog_rest::HttpRequestBody<'a>::eq(&self, other: &iceberg_catalog_rest::HttpRequestBody<'a>) -> bool
+    impl<'a> core::fmt::Debug for iceberg_catalog_rest::HttpRequestBody<'a>
+    pub fn iceberg_catalog_rest::HttpRequestBody<'a>::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+    impl<'a> core::marker::Copy for iceberg_catalog_rest::HttpRequestBody<'a>
+    impl<'a> core::marker::StructuralPartialEq for iceberg_catalog_rest::HttpRequestBody<'a>
     pub struct iceberg_catalog_rest::CommitTableRequest
     pub iceberg_catalog_rest::CommitTableRequest::identifier: core::option::Option<iceberg::catalog::TableIdent>
     pub iceberg_catalog_rest::CommitTableRequest::requirements: alloc::vec::Vec<iceberg::catalog::TableRequirement>
@@ Expand Down Expand Up @@
     pub fn iceberg_catalog_rest::ErrorResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
     impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::ErrorResponse
     pub fn iceberg_catalog_rest::ErrorResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
+    pub struct iceberg_catalog_rest::HttpClient
+    impl iceberg_catalog_rest::HttpClient
+    pub async fn iceberg_catalog_rest::HttpClient::post_form(&self, url: &str, headers: &http::header::map::HeaderMap, form: &std::collections::hash::map::HashMap<&str, &str>) -> iceberg::error::Result<(http::status::StatusCode, alloc::vec::Vec<u8>)>
+    pub fn iceberg_catalog_rest::HttpClient::with_session(&self, session: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>) -> Self
+    pub fn iceberg_catalog_rest::HttpClient::without_session(&self) -> Self
+    impl core::clone::Clone for iceberg_catalog_rest::HttpClient
+    pub fn iceberg_catalog_rest::HttpClient::clone(&self) -> iceberg_catalog_rest::HttpClient
+    impl core::fmt::Debug for iceberg_catalog_rest::HttpClient
+    pub fn iceberg_catalog_rest::HttpClient::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+    pub struct iceberg_catalog_rest::HttpRequest
+    impl iceberg_catalog_rest::HttpRequest
+    pub fn iceberg_catalog_rest::HttpRequest::body(&self) -> iceberg_catalog_rest::HttpRequestBody<'_>
+    pub fn iceberg_catalog_rest::HttpRequest::headers(&self) -> &http::header::map::HeaderMap
+    pub fn iceberg_catalog_rest::HttpRequest::headers_mut(&mut self) -> &mut http::header::map::HeaderMap
+    pub fn iceberg_catalog_rest::HttpRequest::method(&self) -> &http::method::Method
+    pub fn iceberg_catalog_rest::HttpRequest::new(inner: reqwest::async_impl::request::Request) -> Self
+    pub fn iceberg_catalog_rest::HttpRequest::url_str(&self) -> &str
     pub struct iceberg_catalog_rest::ListNamespaceResponse
     pub iceberg_catalog_rest::ListNamespaceResponse::namespaces: alloc::vec::Vec<iceberg::catalog::NamespaceIdent>
     pub iceberg_catalog_rest::ListNamespaceResponse::next_page_token: core::option::Option<alloc::string::String>
@@ Expand Down Expand Up @@
     pub fn iceberg_catalog_rest::NamespaceResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
     impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::NamespaceResponse
     pub fn iceberg_catalog_rest::NamespaceResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
+    pub struct iceberg_catalog_rest::NoopAuthManager
+    impl core::fmt::Debug for iceberg_catalog_rest::NoopAuthManager
+    pub fn iceberg_catalog_rest::NoopAuthManager::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+    impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::NoopAuthManager
+    pub fn iceberg_catalog_rest::NoopAuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+    pub fn iceberg_catalog_rest::NoopAuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+    pub struct iceberg_catalog_rest::OAuth2Manager
+    impl iceberg_catalog_rest::OAuth2Manager
+    pub fn iceberg_catalog_rest::OAuth2Manager::new(token_endpoint: impl core::convert::Into<alloc::string::String>) -> Self
+    pub fn iceberg_catalog_rest::OAuth2Manager::with_credential(self, client_id: core::option::Option<alloc::string::String>, client_secret: alloc::string::String) -> Self
+    pub fn iceberg_catalog_rest::OAuth2Manager::with_extra_headers(self, headers: http::header::map::HeaderMap) -> Self
+    pub fn iceberg_catalog_rest::OAuth2Manager::with_extra_oauth_params(self, params: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> Self
+    pub fn iceberg_catalog_rest::OAuth2Manager::with_token(self, token: impl core::convert::Into<alloc::string::String>) -> Self
+    impl core::fmt::Debug for iceberg_catalog_rest::OAuth2Manager
+    pub fn iceberg_catalog_rest::OAuth2Manager::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+    impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::OAuth2Manager
+    pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+    pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
     pub struct iceberg_catalog_rest::RegisterTableRequest
     pub iceberg_catalog_rest::RegisterTableRequest::metadata_location: alloc::string::String
     pub iceberg_catalog_rest::RegisterTableRequest::name: alloc::string::String
@@ Expand Down Expand Up @@
     impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::RenameTableRequest
     pub fn iceberg_catalog_rest::RenameTableRequest::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
     pub struct iceberg_catalog_rest::RestCatalog
-    impl iceberg_catalog_rest::RestCatalog
-    pub async fn iceberg_catalog_rest::RestCatalog::invalidate_token(&self) -> iceberg::error::Result<()>
-    pub async fn iceberg_catalog_rest::RestCatalog::regenerate_token(&self) -> iceberg::error::Result<()>
     impl core::fmt::Debug for iceberg_catalog_rest::RestCatalog
     pub fn iceberg_catalog_rest::RestCatalog::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
     impl iceberg::catalog::Catalog for iceberg_catalog_rest::RestCatalog
@@ Expand All @@
     pub fn iceberg_catalog_rest::RestCatalog::update_table<'life0, 'async_trait>(&'life0 self, commit: iceberg::catalog::TableCommit) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<iceberg::table::Table>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait
     pub struct iceberg_catalog_rest::RestCatalogBuilder
     impl iceberg_catalog_rest::RestCatalogBuilder
+    pub fn iceberg_catalog_rest::RestCatalogBuilder::with_auth_manager(self, auth_manager: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthManager>) -> Self
     pub fn iceberg_catalog_rest::RestCatalogBuilder::with_client(self, client: reqwest::async_impl::client::Client) -> Self
     impl core::default::Default for iceberg_catalog_rest::RestCatalogBuilder
     pub fn iceberg_catalog_rest::RestCatalogBuilder::default() -> Self
@@ Expand Down Expand Up @@
     pub fn iceberg_catalog_rest::UpdateNamespacePropertiesResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
     impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::UpdateNamespacePropertiesResponse
     pub fn iceberg_catalog_rest::UpdateNamespacePropertiesResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
+    pub const iceberg_catalog_rest::AUTH_TYPE_NONE: &str
+    pub const iceberg_catalog_rest::AUTH_TYPE_OAUTH2: &str
+    pub const iceberg_catalog_rest::REST_CATALOG_PROP_AUTH_TYPE: &str
     pub const iceberg_catalog_rest::REST_CATALOG_PROP_DISABLE_HEADER_REDACTION: &str
     pub const iceberg_catalog_rest::REST_CATALOG_PROP_URI: &str
     pub const iceberg_catalog_rest::REST_CATALOG_PROP_WAREHOUSE: &str
+    pub trait iceberg_catalog_rest::AuthManager: core::fmt::Debug + core::marker::Send + core::marker::Sync
+    pub fn iceberg_catalog_rest::AuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+    pub fn iceberg_catalog_rest::AuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+    impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::NoopAuthManager
+    pub fn iceberg_catalog_rest::NoopAuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+    pub fn iceberg_catalog_rest::NoopAuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+    impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::OAuth2Manager
+    pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+    pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+    pub trait iceberg_catalog_rest::AuthSession: core::fmt::Debug + core::marker::Send + core::marker::Sync
+    pub fn iceberg_catalog_rest::AuthSession::authenticate<'life0, 'life1, 'async_trait>(&'life0 self, request: &'life1 mut iceberg_catalog_rest::HttpRequest) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait

crates/catalog/rest/src/auth/mod.rs

-Original file line number
+Diff line change
@@ -0,0 +1,113 @@
+    // 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.
+    //! Pluggable authentication for the REST catalog, mirroring Iceberg Java's
+    //! `AuthManager`/`AuthSession` API.
+    mod oauth2;
+    use std::collections::HashMap;
+    use std::fmt::Debug;
+    use std::sync::Arc;
+    use async_trait::async_trait;
+    use iceberg::Result;
+    pub use oauth2::OAuth2Manager;
+    use crate::client::HttpClient;
+    use crate::request::HttpRequest;
+    /// `rest.auth.type` value disabling authentication.
+    pub const AUTH_TYPE_NONE: &str = "none";
+    /// `rest.auth.type` value selecting OAuth2 token authentication.
+    pub const AUTH_TYPE_OAUTH2: &str = "oauth2";
+    /// Creates the [`AuthSession`]s used to authenticate REST catalog requests.
+    ///
+    /// A manager is created once per catalog, either from the `rest.auth.type`
+    /// property or injected through `RestCatalogBuilder::with_auth_manager`. It
+    /// builds the sessions the catalog then keeps.
+    ///
+    /// Both methods are handed the catalog's [`HttpClient`], which an
+    /// implementation may reuse for its own requests (e.g. a token exchange) so
+    /// that they share the catalog's connection pool and configuration.
+    #[async_trait]
+    pub trait AuthManager: Debug + Send + Sync {
+        /// Session used for the initial `/v1/config` handshake, given the
+        /// user-supplied properties.
+        ///
+        /// Returns a [`Box`]: an init session is used once and released, unlike
+        /// the shared [`AuthManager::catalog_session`].
+        async fn init_session(
+            &self,
+            client: &HttpClient,
+            props: &HashMap<String, String>,
+        ) -> Result<Box<dyn AuthSession>>;
+        /// Session used for all subsequent catalog requests, given the properties
+        /// merged from the user configuration and the server's config response.
+        ///
+        /// Returns an [`Arc`]: this session is shared by concurrent requests for
+        /// the rest of the catalog's lifetime. Implementations may carry state
+        /// (e.g. a cached token) over from the init session.
+        async fn catalog_session(
+            &self,
+            client: &HttpClient,
+            props: &HashMap<String, String>,
+        ) -> Result<Arc<dyn AuthSession>>;
+    }
+    /// Authenticates outgoing REST catalog requests.
+    #[async_trait]
+    pub trait AuthSession: Debug + Send + Sync {
+        /// Applies authentication to the request (adds headers, signs, ...).
+        async fn authenticate(&self, request: &mut HttpRequest) -> Result<()>;
+    }
+    /// [`AuthManager`] that performs no authentication.
+    #[derive(Debug)]
+    pub struct NoopAuthManager;
+    /// [`AuthSession`] that performs no authentication.
+    #[derive(Debug)]
+    pub(crate) struct NoopSession;
+    #[async_trait]
+    impl AuthManager for NoopAuthManager {
+        async fn init_session(
+            &self,
+            _client: &HttpClient,
+            _props: &HashMap<String, String>,
+        ) -> Result<Box<dyn AuthSession>> {
+            Ok(Box::new(NoopSession))
+        }
+        async fn catalog_session(
+            &self,
+            _client: &HttpClient,
+            _props: &HashMap<String, String>,
+        ) -> Result<Arc<dyn AuthSession>> {
+            Ok(Arc::new(NoopSession))
+        }
+    }
+    #[async_trait]
+    impl AuthSession for NoopSession {
+        async fn authenticate(&self, _request: &mut HttpRequest) -> Result<()> {
+            Ok(())
+        }
+    }

Navigation Menu

Uh oh!

feat!(rest): introduce AuthManager/AuthSession and migrate OAuth2 #2838

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

Uh oh!