feat(rest): introduce AuthManager/AuthSession and rework SigV4 as a wrapping manager

Mirrors Java's AuthManager API: Noop/OAuth2/SigV4 managers selected via
rest.auth.type (or the legacy rest.sigv4-enabled), injectable through
RestCatalogBuilder::with_auth_manager. OAuth2 token handling moves out of
HttpClient into OAuth2Manager with the cached token surviving the config
handshake. SigV4 now wraps a delegate session, relocating its Authorization
header to X-Iceberg-Authorization before signing, so token auth composes
with SigV4 instead of being mutually exclusive.

Author: plusplusjiajia

crates/catalog/rest/public-api.txt

@@ -193,6 +193,18 @@ impl serde_core::ser::Serialize for iceberg_catalog_rest::NamespaceResponse
 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, 'async_trait>(&'life0 self, _props: &'life1 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
+pub fn iceberg_catalog_rest::NoopAuthManager::init_session<'life0, 'async_trait>(&'life0 self) -> 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
+pub struct iceberg_catalog_rest::OAuth2Manager
+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, 'async_trait>(&'life0 self, props: &'life1 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
+pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'async_trait>(&'life0 self) -> 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
 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
@@ -248,8 +260,8 @@ pub fn iceberg_catalog_rest::RestCatalog::update_namespace<'life0, 'life1, 'asyn
 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
-pub fn iceberg_catalog_rest::RestCatalogBuilder::with_signer(self, signer: alloc::sync::Arc<dyn iceberg_catalog_rest::HttpRequestSigner>) -> Self
 impl core::default::Default for iceberg_catalog_rest::RestCatalogBuilder
 pub fn iceberg_catalog_rest::RestCatalogBuilder::default() -> Self
 impl core::fmt::Debug for iceberg_catalog_rest::RestCatalogBuilder
@@ -260,15 +272,22 @@ pub fn iceberg_catalog_rest::RestCatalogBuilder::load(self, name: impl core::con
 pub fn iceberg_catalog_rest::RestCatalogBuilder::with_kms_client_factory(self, kms_client_factory: alloc::sync::Arc<dyn iceberg::encryption::kms::factory::KmsClientFactory>) -> Self
 pub fn iceberg_catalog_rest::RestCatalogBuilder::with_runtime(self, runtime: iceberg::runtime::Runtime) -> Self
 pub fn iceberg_catalog_rest::RestCatalogBuilder::with_storage_factory(self, storage_factory: alloc::sync::Arc<dyn iceberg::io::storage::StorageFactory>) -> Self
+pub struct iceberg_catalog_rest::SigV4AuthManager
+impl iceberg_catalog_rest::SigV4AuthManager
+pub fn iceberg_catalog_rest::SigV4AuthManager::new(delegate: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthManager>, signer: iceberg_catalog_rest::SigV4Signer) -> Self
+impl core::fmt::Debug for iceberg_catalog_rest::SigV4AuthManager
+pub fn iceberg_catalog_rest::SigV4AuthManager::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::SigV4AuthManager
+pub fn iceberg_catalog_rest::SigV4AuthManager::catalog_session<'life0, 'life1, 'async_trait>(&'life0 self, props: &'life1 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
+pub fn iceberg_catalog_rest::SigV4AuthManager::init_session<'life0, 'async_trait>(&'life0 self) -> 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
 pub struct iceberg_catalog_rest::SigV4Signer
 impl iceberg_catalog_rest::SigV4Signer
 pub fn iceberg_catalog_rest::SigV4Signer::new(credentials: iceberg_catalog_rest::AwsCredentials, region: alloc::string::String, service: alloc::string::String, mode: iceberg_catalog_rest::PayloadHashMode) -> Self
+pub fn iceberg_catalog_rest::SigV4Signer::sign(&self, request: &mut reqwest::async_impl::request::Request) -> iceberg::error::Result<()>
 impl core::clone::Clone for iceberg_catalog_rest::SigV4Signer
 pub fn iceberg_catalog_rest::SigV4Signer::clone(&self) -> iceberg_catalog_rest::SigV4Signer
 impl core::fmt::Debug for iceberg_catalog_rest::SigV4Signer
 pub fn iceberg_catalog_rest::SigV4Signer::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
-impl iceberg_catalog_rest::HttpRequestSigner for iceberg_catalog_rest::SigV4Signer
-pub fn iceberg_catalog_rest::SigV4Signer::sign(&self, request: &mut reqwest::async_impl::request::Request) -> iceberg::error::Result<()>
 pub struct iceberg_catalog_rest::StorageCredential
 pub iceberg_catalog_rest::StorageCredential::config: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>
 pub iceberg_catalog_rest::StorageCredential::prefix: alloc::string::String
@@ -315,6 +334,10 @@ impl serde_core::ser::Serialize for iceberg_catalog_rest::UpdateNamespacePropert
 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::AUTH_TYPE_SIGV4: &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_SIGNING_ACCESS_KEY: &str
 pub const iceberg_catalog_rest::REST_CATALOG_PROP_SIGNING_NAME: &str
@@ -324,7 +347,19 @@ pub const iceberg_catalog_rest::REST_CATALOG_PROP_SIGNING_SESSION_TOKEN: &str
 pub const iceberg_catalog_rest::REST_CATALOG_PROP_SIGV4_ENABLED: &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::HttpRequestSigner: core::marker::Send + core::marker::Sync + core::fmt::Debug
-pub fn iceberg_catalog_rest::HttpRequestSigner::sign(&self, request: &mut reqwest::async_impl::request::Request) -> iceberg::error::Result<()>
-impl iceberg_catalog_rest::HttpRequestSigner for iceberg_catalog_rest::SigV4Signer
-pub fn iceberg_catalog_rest::SigV4Signer::sign(&self, request: &mut reqwest::async_impl::request::Request) -> iceberg::error::Result<()>
+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, 'async_trait>(&'life0 self, props: &'life1 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
+pub fn iceberg_catalog_rest::AuthManager::init_session<'life0, 'async_trait>(&'life0 self) -> 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
+impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::NoopAuthManager
+pub fn iceberg_catalog_rest::NoopAuthManager::catalog_session<'life0, 'life1, 'async_trait>(&'life0 self, _props: &'life1 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
+pub fn iceberg_catalog_rest::NoopAuthManager::init_session<'life0, 'async_trait>(&'life0 self) -> 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
+impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::OAuth2Manager
+pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'async_trait>(&'life0 self, props: &'life1 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
+pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'async_trait>(&'life0 self) -> 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
+impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::SigV4AuthManager
+pub fn iceberg_catalog_rest::SigV4AuthManager::catalog_session<'life0, 'life1, 'async_trait>(&'life0 self, props: &'life1 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
+pub fn iceberg_catalog_rest::SigV4AuthManager::init_session<'life0, 'async_trait>(&'life0 self) -> 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
+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 reqwest::async_impl::request::Request) -> 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
+pub fn iceberg_catalog_rest::AuthSession::invalidate<'life0, 'async_trait>(&'life0 self) -> 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
+pub fn iceberg_catalog_rest::AuthSession::refresh<'life0, 'async_trait>(&'life0 self) -> 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

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

@@ -0,0 +1,108 @@
+// 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;
+mod sigv4;
+
+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 reqwest::Request;
+pub use sigv4::SigV4AuthManager;
+
+/// `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";
+/// `rest.auth.type` value selecting AWS SigV4 request signing.
+pub const AUTH_TYPE_SIGV4: &str = "sigv4";
+
+/// 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`, and
+/// lives for the lifetime of the catalog.
+#[async_trait]
+pub trait AuthManager: Debug + Send + Sync {
+    /// Session used for the initial `/v1/config` handshake, built from the
+    /// user-supplied configuration.
+    async fn init_session(&self) -> Result<Arc<dyn AuthSession>>;
+
+    /// Session used for all subsequent catalog requests, given the properties
+    /// merged from the user configuration and the server's config response.
+    ///
+    /// Implementations may carry state (e.g. a cached token) over from the
+    /// init session.
+    async fn catalog_session(
+        &self,
+        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 Request) -> Result<()>;
+
+    /// Drops any cached credentials so the next request re-authenticates.
+    async fn invalidate(&self) -> Result<()> {
+        Ok(())
+    }
+
+    /// Proactively refreshes cached credentials (e.g. re-exchanges an OAuth2
+    /// client credential for a new token), leaving them intact on failure.
+    async fn refresh(&self) -> Result<()> {
+        Ok(())
+    }
+}
+
+/// [`AuthManager`] that performs no authentication.
+#[derive(Debug)]
+pub struct NoopAuthManager;
+
+/// [`AuthSession`] that performs no authentication.
+#[derive(Debug)]
+struct NoopSession;
+
+#[async_trait]
+impl AuthManager for NoopAuthManager {
+    async fn init_session(&self) -> Result<Arc<dyn AuthSession>> {
+        Ok(Arc::new(NoopSession))
+    }
+
+    async fn catalog_session(
+        &self,
+        _props: &HashMap<String, String>,
+    ) -> Result<Arc<dyn AuthSession>> {
+        Ok(Arc::new(NoopSession))
+    }
+}
+
+#[async_trait]
+impl AuthSession for NoopSession {
+    async fn authenticate(&self, _request: &mut Request) -> Result<()> {
+        Ok(())
+    }
+}