This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3081-776648bee4c52ece8a057cfcf0d7756416299ba8 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit 7f9691fc7ed395ecb062a14ae918cb3707f22d30 Author: Jannik Steinmann <[email protected]> AuthorDate: Thu Oct 8 23:55:44 2026 +0000 feat(rest_catalog::auth) Contextual Sessions (#3081) * Add trait method `AuthManager::contextual_session` * Implement contextual session handling * Produce an auth_session per operation and re-use it for multiple Http requests * Rename client to RestCatalogClient * Mark context in contextual_session as used in API docs * Drop Java-derived comment on releasing resources * Revert "Produce an auth_session per operation and re-use it for multiple Http" This reverts commit 32a9dc5f2f1e35012bdcc48fcd9167af0fcc585b. * Expand test coverage * Pass auth session to client on each call This prevents cloning the HeaderMap of the client because a new session is passed for every HTTP request anyway * Leave post_form unauthenticated * Implementor guidance --- crates/catalog/rest/public-api.txt | 27 +-- crates/catalog/rest/src/auth/mod.rs | 39 +++- crates/catalog/rest/src/catalog.rs | 404 ++++++++++++++++++++++++++++++------ crates/catalog/rest/src/client.rs | 120 ++++------- 4 files changed, 422 insertions(+), 168 deletions(-) diff --git a/crates/catalog/rest/public-api.txt b/crates/catalog/rest/public-api.txt index 8afd581bc..47393b573 100644 --- a/crates/catalog/rest/public-api.txt +++ b/crates/catalog/rest/public-api.txt @@ -127,8 +127,6 @@ pub fn iceberg_catalog_rest::ErrorResponse::deserialize<__D>(__deserializer: __D 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<iceberg_catalog_rest::HttpResponse> -pub fn iceberg_catalog_rest::HttpClient::with_auth_session(&self, auth_session: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>) -> Self -pub fn iceberg_catalog_rest::HttpClient::without_auth_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 @@ -222,6 +220,7 @@ 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, 'l [...] +pub fn iceberg_catalog_rest::NoopAuthManager::contextual_session<'life0, 'life1, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, catalog_session: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>) -> 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, 'l [...] 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, 'lif [...] pub struct iceberg_catalog_rest::OAuth2Manager impl iceberg_catalog_rest::OAuth2Manager @@ -234,6 +233,7 @@ 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 [...] +pub fn iceberg_catalog_rest::OAuth2Manager::contextual_session<'life0, 'life1, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, catalog_session: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>) -> 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, 'lif [...] 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: [...] pub struct iceberg_catalog_rest::RegisterTableRequest pub iceberg_catalog_rest::RegisterTableRequest::metadata_location: alloc::string::String @@ -304,21 +304,21 @@ pub struct iceberg_catalog_rest::RestSessionCatalog impl core::fmt::Debug for iceberg_catalog_rest::RestSessionCatalog pub fn iceberg_catalog_rest::RestSessionCatalog::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl iceberg::catalog::session::SessionCatalog for iceberg_catalog_rest::RestSessionCatalog -pub fn iceberg_catalog_rest::RestSessionCatalog::create_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent, properties: 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<iceberg::catalog::Namespace>> + core::marker::Send + 'async_tr [...] -pub fn iceberg_catalog_rest::RestSessionCatalog::create_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent, creation: iceberg::catalog::TableCreation) -> 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, 'life1: [...] -pub fn iceberg_catalog_rest::RestSessionCatalog::drop_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent) -> 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, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::create_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent, properties: 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<iceberg::catalog::Namespace>> + core::marker::Send + 'async_tra [...] +pub fn iceberg_catalog_rest::RestSessionCatalog::create_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent, creation: iceberg::catalog::TableCreation) -> 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, 'life1: ' [...] +pub fn iceberg_catalog_rest::RestSessionCatalog::drop_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent) -> 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, 'life2: 'async_trait pub fn iceberg_catalog_rest::RestSessionCatalog::drop_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, table: &'life2 iceberg::catalog::TableIdent) -> 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, 'life2: 'async_trait -pub fn iceberg_catalog_rest::RestSessionCatalog::get_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<iceberg::catalog::Namespace>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait -pub fn iceberg_catalog_rest::RestSessionCatalog::list_namespaces<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, parent: core::option::Option<&'life2 iceberg::catalog::NamespaceIdent>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::vec::Vec<iceberg::catalog::NamespaceIdent>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, ' [...] -pub fn iceberg_catalog_rest::RestSessionCatalog::list_tables<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::vec::Vec<iceberg::catalog::TableIdent>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2 [...] -pub fn iceberg_catalog_rest::RestSessionCatalog::load_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, table_ident: &'life2 iceberg::catalog::TableIdent) -> 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, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::get_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<iceberg::catalog::Namespace>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::list_namespaces<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, parent: core::option::Option<&'life2 iceberg::catalog::NamespaceIdent>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::vec::Vec<iceberg::catalog::NamespaceIdent>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'l [...] +pub fn iceberg_catalog_rest::RestSessionCatalog::list_tables<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, namespace: &'life2 iceberg::catalog::NamespaceIdent) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::vec::Vec<iceberg::catalog::TableIdent>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: [...] +pub fn iceberg_catalog_rest::RestSessionCatalog::load_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, table_ident: &'life2 iceberg::catalog::TableIdent) -> 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, 'life1: 'async_trait, 'life2: 'async_trait pub fn iceberg_catalog_rest::RestSessionCatalog::namespace_exists<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, ns: &'life2 iceberg::catalog::NamespaceIdent) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<bool>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait pub fn iceberg_catalog_rest::RestSessionCatalog::purge_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, table: &'life2 iceberg::catalog::TableIdent) -> 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, 'life2: 'async_trait -pub fn iceberg_catalog_rest::RestSessionCatalog::register_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, table_ident: &'life2 iceberg::catalog::TableIdent, metadata_location: alloc::string::String) -> 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, 'life1: ' [...] -pub fn iceberg_catalog_rest::RestSessionCatalog::rename_table<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, src: &'life2 iceberg::catalog::TableIdent, dest: &'life3 iceberg::catalog::TableIdent) -> 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, 'life2 [...] +pub fn iceberg_catalog_rest::RestSessionCatalog::register_table<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, table_ident: &'life2 iceberg::catalog::TableIdent, metadata_location: alloc::string::String) -> 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, 'life1: 'a [...] +pub fn iceberg_catalog_rest::RestSessionCatalog::rename_table<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, src: &'life2 iceberg::catalog::TableIdent, dest: &'life3 iceberg::catalog::TableIdent) -> 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, 'life2: [...] pub fn iceberg_catalog_rest::RestSessionCatalog::table_exists<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, table: &'life2 iceberg::catalog::TableIdent) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<bool>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait pub fn iceberg_catalog_rest::RestSessionCatalog::update_namespace<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, _namespace: &'life2 iceberg::catalog::NamespaceIdent, _properties: 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<()>> + core::marker::Send + 'async_trait)>> where Self: 'asy [...] -pub fn iceberg_catalog_rest::RestSessionCatalog::update_table<'life0, 'life1, 'async_trait>(&'life0 self, _context: &'life1 iceberg::catalog::session::SessionContext, 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, 'life1: 'async_trait +pub fn iceberg_catalog_rest::RestSessionCatalog::update_table<'life0, 'life1, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, 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, 'life1: 'async_trait pub struct iceberg_catalog_rest::RestSessionCatalogBuilder impl iceberg_catalog_rest::RestSessionCatalogBuilder pub fn iceberg_catalog_rest::RestSessionCatalogBuilder::load(self, name: impl core::convert::Into<alloc::string::String>, props: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> impl core::future::future::Future<Output = iceberg::error::Result<iceberg_catalog_rest::RestSessionCatalog>> + core::marker::Send @@ -385,12 +385,15 @@ 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: [...] +pub fn iceberg_catalog_rest::AuthManager::contextual_session<'life0, 'life1, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, catalog_session: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>) -> 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 [...] 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: 'a [...] 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, 'l [...] +pub fn iceberg_catalog_rest::NoopAuthManager::contextual_session<'life0, 'life1, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, catalog_session: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>) -> 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, 'l [...] 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, 'lif [...] 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 [...] +pub fn iceberg_catalog_rest::OAuth2Manager::contextual_session<'life0, 'life1, 'async_trait>(&'life0 self, context: &'life1 iceberg::catalog::session::SessionContext, catalog_session: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>) -> 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, 'lif [...] 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: [...] 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 diff --git a/crates/catalog/rest/src/auth/mod.rs b/crates/catalog/rest/src/auth/mod.rs index b15dcdb48..b4c0cf0be 100644 --- a/crates/catalog/rest/src/auth/mod.rs +++ b/crates/catalog/rest/src/auth/mod.rs @@ -25,7 +25,7 @@ use std::fmt::Debug; use std::sync::Arc; use async_trait::async_trait; -use iceberg::Result; +use iceberg::{Result, SessionContext}; pub use oauth2::OAuth2Manager; use crate::client::HttpClient; @@ -44,11 +44,13 @@ pub const AUTH_TYPE_OAUTH2: &str = "oauth2"; /// [`RestCatalogBuilder::with_auth_manager`](crate::RestCatalogBuilder::with_auth_manager) or /// [`RestSessionCatalogBuilder::with_auth_manager`](crate::RestSessionCatalogBuilder::with_auth_manager). /// Catalog initialization calls [`AuthManager::catalog_session`] exactly once; -/// later sessions may rely on the state established by that call. +/// the context-specific sessions that [`AuthManager::contextual_session`] +/// derives from it may rely on the state established by that call. /// -/// 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. +/// [`Self::init_session`] and [`Self::catalog_session`] are handed the +/// catalog's [`HttpClient`], which an implementation may clone and store 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 @@ -73,6 +75,33 @@ pub trait AuthManager: Debug + Send + Sync { client: &HttpClient, props: &HashMap<String, String>, ) -> Result<Arc<dyn AuthSession>>; + + /// Returns the authentication session for a specific context. + /// + /// The catalog calls this method only after [`Self::catalog_session`] has + /// succeeded. `catalog_session` is the catalog session returned by this + /// manager. If the context does not require different authentication, + /// implementations should return `catalog_session` unchanged. + /// + /// The catalog does not cache the returned session. Implementations should + /// cache context-specific sessions internally using [`SessionContext::session_id`] + /// and are responsible for eviction. Reusing a session ID with different + /// context may therefore return the previously cached session. + /// + /// The catalog calls this method for every request without serializing + /// calls that share a session ID. Caching implementations must guard + /// against concurrently creating multiple sessions for the same context. + /// + /// No [`HttpClient`] is passed. Implementations that need one, e.g. for a + /// token exchange, should store the client handed to [`Self::catalog_session`]. + async fn contextual_session( + &self, + context: &SessionContext, + catalog_session: Arc<dyn AuthSession>, + ) -> Result<Arc<dyn AuthSession>> { + let _ = context; + Ok(catalog_session) + } } /// Authenticates outgoing REST catalog requests. diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index cc5973316..dcc89823f 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -39,7 +39,9 @@ use reqwest::{Client, Method, StatusCode, Url}; use tokio::sync::OnceCell; use typed_builder::TypedBuilder; -use crate::auth::{AUTH_TYPE_NONE, AUTH_TYPE_OAUTH2, AuthManager, NoopAuthManager, OAuth2Manager}; +use crate::auth::{ + AUTH_TYPE_NONE, AUTH_TYPE_OAUTH2, AuthManager, AuthSession, NoopAuthManager, OAuth2Manager, +}; use crate::client::{ HttpClient, deserialize_catalog_response, deserialize_unexpected_catalog_error, }; @@ -416,10 +418,14 @@ pub(crate) fn oauth_params_from_props(props: &HashMap<String, String>) -> HashMa params } -#[derive(Debug)] -struct RestClient { - /// Carries the session the auth manager derived from the merged - /// configuration, so every request below is authenticated. +struct RestCatalogClient { + /// The manager that created `catalog_session`; retained so each request can + /// derive authentication for its [`SessionContext`]. + auth_manager: Arc<dyn AuthManager>, + /// The catalog-wide session passed to [`AuthManager::contextual_session`]. + catalog_session: Arc<dyn AuthSession>, + /// Shared HTTP transport and configuration. It remains unauthenticated; + /// request-time clones attach the contextual session. http_client: HttpClient, /// Runtime config is fetched from rest server and stored here. /// @@ -429,9 +435,21 @@ struct RestClient { endpoints: HashSet<Endpoint>, } -impl RestClient { - /// Initializes the runtime config, advertised endpoints, and authentication - /// sessions shared by one REST catalog instance. +impl Debug for RestCatalogClient { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + // Auth managers and sessions may contain secrets, so keep them out of + // the catalog's derived Debug output just as HttpClient does. + f.debug_struct("RestCatalogClient") + .field("http_client", &self.http_client) + .field("config", &self.config) + .field("endpoints", &self.endpoints) + .finish_non_exhaustive() + } +} + +impl RestCatalogClient { + /// Initializes the runtime config, advertised endpoints, auth manager, and + /// catalog authentication session shared by one REST catalog instance. async fn init( user_config: &RestCatalogConfig, auth_manager: Arc<dyn AuthManager>, @@ -442,16 +460,9 @@ impl RestClient { // it before deriving the catalog session. let catalog_config = { let init_session = auth_manager - .init_session( - &http_client.without_auth_session(), - &Self::auth_props(user_config), - ) + .init_session(&http_client, &Self::auth_props(user_config)) .await?; - Self::load_config( - &http_client.with_auth_session(Arc::from(init_session)), - user_config, - ) - .await? + Self::load_config(&http_client, init_session.as_ref(), user_config).await? }; // Use the advertised endpoints as-is, falling back to // `DEFAULT_ENDPOINTS` when absent or empty. @@ -463,16 +474,15 @@ impl RestClient { let http_client = http_client.update_with(&config)?; // The manager is handed an unauthenticated client: its own // requests must not be signed by the session it is deriving. - let session = auth_manager - .catalog_session( - &http_client.without_auth_session(), - &Self::auth_props(&config), - ) + let catalog_session = auth_manager + .catalog_session(&http_client, &Self::auth_props(&config)) .await?; Ok(Self { + auth_manager, + catalog_session, config, - http_client: http_client.with_auth_session(session), + http_client, endpoints, }) } @@ -480,12 +490,22 @@ impl RestClient { /// Testing only: the bearer token the catalog session would attach. #[cfg(test)] async fn token(&self) -> Option<String> { - self.http_client.token().await + self.http_client.token(self.catalog_session.as_ref()).await } - /// Sends `request`, authenticated by the client's session. - async fn query_catalog(&self, request: HttpRequest) -> Result<HttpResponse> { - self.http_client.query_catalog(request).await + /// Sends `request` with the authentication derived for `context`. + async fn query_catalog( + &self, + context: &SessionContext, + request: HttpRequest, + ) -> Result<HttpResponse> { + let session = self + .auth_manager + .contextual_session(context, Arc::clone(&self.catalog_session)) + .await?; + self.http_client + .query_catalog(session.as_ref(), request) + .await } /// The properties handed to the [`AuthManager`], with the catalog `uri` @@ -512,6 +532,7 @@ impl RestClient { /// It's required for a REST catalog to update its config after creation. async fn load_config( http_client: &HttpClient, + auth_session: &dyn AuthSession, user_config: &RestCatalogConfig, ) -> Result<CatalogConfig> { let mut request_builder = http_client.request(Method::GET, user_config.config_endpoint()); @@ -522,7 +543,7 @@ impl RestClient { let request = HttpRequest::build(request_builder)?; - let http_response = http_client.query_catalog(request).await?; + let http_response = http_client.query_catalog(auth_session, request).await?; match http_response.status() { StatusCode::OK => deserialize_catalog_response(http_response), @@ -575,7 +596,7 @@ impl RestCatalog { } #[cfg(test)] - async fn client(&self) -> Result<&RestClient> { + async fn client(&self) -> Result<&RestCatalogClient> { self.inner.client().await } } @@ -687,18 +708,20 @@ impl Catalog for RestCatalog { /// REST catalog implementation of [`SessionCatalog`]. /// -/// Each operation accepts a [`SessionContext`]. REST configuration, authentication sessions, -/// and the HTTP client are initialized lazily once per catalog and shared across all operations. +/// Each operation accepts a [`SessionContext`]. REST configuration, the auth manager, the catalog +/// authentication session, and the HTTP client are initialized lazily once per catalog. Each +/// REST request uses the contextual authentication session returned by the manager. #[derive(Debug)] pub struct RestSessionCatalog { - /// Injected through [`RestSessionCatalogBuilder::with_auth_manager`]; otherwise - /// one is resolved from `rest.auth.type` when the client is built. - auth_manager: Option<Arc<dyn AuthManager>>, + /// Builder-supplied override retained so lazy client initialization can + /// clone it into the runtime state. When absent, a manager is resolved from + /// `rest.auth.type` during initialization. + auth_manager_override: Option<Arc<dyn AuthManager>>, /// User config is stored as-is and never changed. /// /// It could be different from the config fetched from the server and used at runtime. user_config: RestCatalogConfig, - client: OnceCell<RestClient>, + client: OnceCell<RestCatalogClient>, /// Storage factory for creating FileIO instances. storage_factory: Option<Arc<dyn StorageFactory>>, runtime: Runtime, @@ -710,13 +733,13 @@ impl RestSessionCatalog { /// Creates a `RestSessionCatalog` from a [`RestCatalogConfig`]. fn new( config: RestCatalogConfig, - auth_manager: Option<Box<dyn AuthManager>>, + auth_manager_override: Option<Box<dyn AuthManager>>, storage_factory: Option<Arc<dyn StorageFactory>>, runtime: Runtime, kms_client: Option<Arc<dyn KeyManagementClient>>, ) -> Self { Self { - auth_manager: auth_manager.map(Arc::from), + auth_manager_override: auth_manager_override.map(Arc::from), user_config: config, client: OnceCell::new(), storage_factory, @@ -728,7 +751,7 @@ impl RestSessionCatalog { /// Sends a DELETE request for the given table, optionally requesting purge. async fn delete_table( &self, - _context: &SessionContext, + context: &SessionContext, table: &TableIdent, purge: bool, ) -> Result<()> { @@ -743,7 +766,7 @@ impl RestSessionCatalog { } let request = HttpRequest::build(request_builder)?; - let http_response = client.query_catalog(request).await?; + let http_response = client.query_catalog(context, request).await?; match http_response.status() { StatusCode::NO_CONTENT | StatusCode::OK => Ok(()), @@ -783,7 +806,7 @@ impl RestSessionCatalog { /// Resolves the auth manager: a `with_auth_manager` override wins, /// otherwise one is built from the `rest.auth.type` configuration. fn resolve_auth_manager(&self) -> Result<Arc<dyn AuthManager>> { - if let Some(auth_manager) = &self.auth_manager { + if let Some(auth_manager) = &self.auth_manager_override { return Ok(auth_manager.clone()); } let config = &self.user_config; @@ -811,11 +834,11 @@ impl RestSessionCatalog { } } - /// Gets the [`RestClient`] from the catalog. - async fn client(&self) -> Result<&RestClient> { + /// Gets the [`RestCatalogClient`] from the catalog. + async fn client(&self) -> Result<&RestCatalogClient> { self.client .get_or_try_init(|| async { - RestClient::init(&self.user_config, self.resolve_auth_manager()?).await + RestCatalogClient::init(&self.user_config, self.resolve_auth_manager()?).await }) .await } @@ -829,9 +852,14 @@ impl RestSessionCatalog { /// Issue a `HEAD` request to `url` and interpret it as an existence check: /// `2xx` means it exists, `404` means it doesn't. - async fn check_exists_via_head(&self, client: &RestClient, url: String) -> Result<bool> { + async fn check_exists_via_head( + &self, + context: &SessionContext, + client: &RestCatalogClient, + url: String, + ) -> Result<bool> { let request = HttpRequest::build(client.http_client.request(Method::HEAD, url))?; - let http_response = client.query_catalog(request).await?; + let http_response = client.query_catalog(context, request).await?; match http_response.status() { StatusCode::NO_CONTENT | StatusCode::OK => Ok(true), @@ -891,7 +919,7 @@ impl RestSessionCatalog { impl SessionCatalog for RestSessionCatalog { async fn list_namespaces( &self, - _context: &SessionContext, + context: &SessionContext, parent: Option<&NamespaceIdent>, ) -> Result<Vec<NamespaceIdent>> { let client = self.client().await?; @@ -911,7 +939,9 @@ impl SessionCatalog for RestSessionCatalog { request = request.query(&[("pageToken", token)]); } - let http_response = client.query_catalog(HttpRequest::build(request)?).await?; + let http_response = client + .query_catalog(context, HttpRequest::build(request)?) + .await?; match http_response.status() { StatusCode::OK => { @@ -945,7 +975,7 @@ impl SessionCatalog for RestSessionCatalog { async fn create_namespace( &self, - _context: &SessionContext, + context: &SessionContext, namespace: &NamespaceIdent, properties: HashMap<String, String>, ) -> Result<Namespace> { @@ -961,7 +991,7 @@ impl SessionCatalog for RestSessionCatalog { }), )?; - let http_response = client.query_catalog(request).await?; + let http_response = client.query_catalog(context, request).await?; match http_response.status() { StatusCode::OK => { @@ -981,7 +1011,7 @@ impl SessionCatalog for RestSessionCatalog { async fn get_namespace( &self, - _context: &SessionContext, + context: &SessionContext, namespace: &NamespaceIdent, ) -> Result<Namespace> { let client = self.client().await?; @@ -992,7 +1022,7 @@ impl SessionCatalog for RestSessionCatalog { .request(Method::GET, client.config.namespace_endpoint(namespace)), )?; - let http_response = client.query_catalog(request).await?; + let http_response = client.query_catalog(context, request).await?; match http_response.status() { StatusCode::OK => { @@ -1028,7 +1058,7 @@ impl SessionCatalog for RestSessionCatalog { } let client = self.client().await?; - self.check_exists_via_head(client, client.config.namespace_endpoint(ns)) + self.check_exists_via_head(context, client, client.config.namespace_endpoint(ns)) .await } @@ -1046,7 +1076,7 @@ impl SessionCatalog for RestSessionCatalog { async fn drop_namespace( &self, - _context: &SessionContext, + context: &SessionContext, namespace: &NamespaceIdent, ) -> Result<()> { let client = self.client().await?; @@ -1057,7 +1087,7 @@ impl SessionCatalog for RestSessionCatalog { .request(Method::DELETE, client.config.namespace_endpoint(namespace)), )?; - let http_response = client.query_catalog(request).await?; + let http_response = client.query_catalog(context, request).await?; match http_response.status() { StatusCode::NO_CONTENT | StatusCode::OK => Ok(()), @@ -1074,7 +1104,7 @@ impl SessionCatalog for RestSessionCatalog { async fn list_tables( &self, - _context: &SessionContext, + context: &SessionContext, namespace: &NamespaceIdent, ) -> Result<Vec<TableIdent>> { let client = self.client().await?; @@ -1089,7 +1119,9 @@ impl SessionCatalog for RestSessionCatalog { request = request.query(&[("pageToken", token)]); } - let http_response = client.query_catalog(HttpRequest::build(request)?).await?; + let http_response = client + .query_catalog(context, HttpRequest::build(request)?) + .await?; match http_response.status() { StatusCode::OK => { @@ -1129,7 +1161,7 @@ impl SessionCatalog for RestSessionCatalog { /// the value provided locally to the `RestSessionCatalog` will take precedence. async fn create_table( &self, - _context: &SessionContext, + context: &SessionContext, namespace: &NamespaceIdent, creation: TableCreation, ) -> Result<Table> { @@ -1152,7 +1184,7 @@ impl SessionCatalog for RestSessionCatalog { }), )?; - let http_response = client.query_catalog(request).await?; + let http_response = client.query_catalog(context, request).await?; let response = match http_response.status() { StatusCode::OK => deserialize_catalog_response::<LoadTableResult>(http_response)?, @@ -1214,7 +1246,7 @@ impl SessionCatalog for RestSessionCatalog { /// value provided locally to the `RestSessionCatalog` will take precedence. async fn load_table( &self, - _context: &SessionContext, + context: &SessionContext, table_ident: &TableIdent, ) -> Result<Table> { let client = self.client().await?; @@ -1225,7 +1257,7 @@ impl SessionCatalog for RestSessionCatalog { .request(Method::GET, client.config.table_endpoint(table_ident)), )?; - let http_response = client.query_catalog(request).await?; + let http_response = client.query_catalog(context, request).await?; let response = match http_response.status() { StatusCode::OK | StatusCode::NOT_MODIFIED => { @@ -1296,14 +1328,14 @@ impl SessionCatalog for RestSessionCatalog { } let client = self.client().await?; - self.check_exists_via_head(client, client.config.table_endpoint(table)) + self.check_exists_via_head(context, client, client.config.table_endpoint(table)) .await } /// Rename a table in the catalog. async fn rename_table( &self, - _context: &SessionContext, + context: &SessionContext, src: &TableIdent, dest: &TableIdent, ) -> Result<()> { @@ -1319,7 +1351,7 @@ impl SessionCatalog for RestSessionCatalog { }), )?; - let http_response = client.query_catalog(request).await?; + let http_response = client.query_catalog(context, request).await?; match http_response.status() { StatusCode::NO_CONTENT | StatusCode::OK => Ok(()), @@ -1340,7 +1372,7 @@ impl SessionCatalog for RestSessionCatalog { async fn register_table( &self, - _context: &SessionContext, + context: &SessionContext, table_ident: &TableIdent, metadata_location: String, ) -> Result<Table> { @@ -1362,7 +1394,7 @@ impl SessionCatalog for RestSessionCatalog { }), )?; - let http_response = client.query_catalog(request).await?; + let http_response = client.query_catalog(context, request).await?; let response: LoadTableResult = match http_response.status() { StatusCode::OK => deserialize_catalog_response::<LoadTableResult>(http_response)?, @@ -1407,7 +1439,7 @@ impl SessionCatalog for RestSessionCatalog { async fn update_table( &self, - _context: &SessionContext, + context: &SessionContext, mut commit: TableCommit, ) -> Result<Table> { let client = self.client().await?; @@ -1426,7 +1458,7 @@ impl SessionCatalog for RestSessionCatalog { }), )?; - let http_response = client.query_catalog(request).await?; + let http_response = client.query_catalog(context, request).await?; let response: CommitTableResponse = match http_response.status() { StatusCode::OK => deserialize_catalog_response(http_response)?, @@ -1664,7 +1696,7 @@ impl RestSessionCatalogBuilder { mod tests { use std::fs::File; use std::io::BufReader; - use std::sync::Arc; + use std::sync::{Arc, Mutex}; use chrono::{TimeZone, Utc}; use iceberg::io::LocalFsStorageFactory; @@ -1683,6 +1715,75 @@ mod tests { use crate::auth::AuthSession; use crate::request::HttpRequest; + #[derive(Debug)] + struct PlainContextSession; + + #[async_trait] + impl AuthSession for PlainContextSession { + async fn authenticate(&self, _request: &mut HttpRequest) -> Result<()> { + Ok(()) + } + } + + #[derive(Debug)] + struct ContextSession(String); + + #[async_trait] + impl AuthSession for ContextSession { + async fn authenticate(&self, request: &mut HttpRequest) -> Result<()> { + request.headers_mut().insert( + "x-session-id", + HeaderValue::from_str(&self.0).expect("valid test session ID"), + ); + Ok(()) + } + } + + #[derive(Debug)] + struct ContextManager { + catalog_session: Arc<dyn AuthSession>, + seen_session_ids: Arc<Mutex<Vec<String>>>, + fail_contextual_session: bool, + } + + #[async_trait] + impl AuthManager for ContextManager { + async fn init_session( + &self, + _client: &HttpClient, + _props: &HashMap<String, String>, + ) -> Result<Box<dyn AuthSession>> { + Ok(Box::new(PlainContextSession)) + } + + async fn catalog_session( + &self, + _client: &HttpClient, + _props: &HashMap<String, String>, + ) -> Result<Arc<dyn AuthSession>> { + Ok(self.catalog_session.clone()) + } + + async fn contextual_session( + &self, + context: &SessionContext, + catalog_session: Arc<dyn AuthSession>, + ) -> Result<Arc<dyn AuthSession>> { + assert!(Arc::ptr_eq(&catalog_session, &self.catalog_session)); + self.seen_session_ids + .lock() + .unwrap() + .push(context.session_id().to_string()); + if self.fail_contextual_session { + return Err(Error::new( + ErrorKind::Unexpected, + "contextual session failure", + )); + } + Ok(Arc::new(ContextSession(context.session_id().to_string()))) + } + } + fn test_catalog(config: RestCatalogConfig) -> RestSessionCatalog { RestSessionCatalog::new(config, None, None, Runtime::current(), None) } @@ -1698,6 +1799,20 @@ mod tests { ) } + fn context_catalog( + config: RestCatalogConfig, + fail_contextual_session: bool, + ) -> (RestSessionCatalog, Arc<Mutex<Vec<String>>>) { + let seen_session_ids = Arc::new(Mutex::new(Vec::new())); + let catalog_session: Arc<dyn AuthSession> = Arc::new(PlainContextSession); + let catalog = test_catalog_with(config, ContextManager { + catalog_session, + seen_session_ids: seen_session_ids.clone(), + fail_contextual_session, + }); + (catalog, seen_session_ids) + } + fn test_client() -> HttpClient { HttpClient::new( &RestCatalogConfig::builder() @@ -2278,7 +2393,7 @@ mod tests { bootstrap_oauth_mock.assert_async().await; // The catalog session's endpoint follows the overridden URI (visible // via the session's Debug, which prints its token endpoint). - let session_debug = format!("{:?}", client.http_client.auth_session()); + let session_debug = format!("{:?}", client.catalog_session); assert!(session_debug.contains(&format!("{}/v1/oauth/tokens", overridden.url()))); } @@ -2790,6 +2905,157 @@ mod tests { assert!(dropped.load(Ordering::SeqCst)); } + #[tokio::test] + async fn test_contextual_session_authenticates_each_catalog_request() { + let mut server = Server::new_async().await; + let config_mock = create_config_mock(&mut server).await; + let first_page = server + .mock("GET", "/v1/namespaces") + .match_header("x-session-id", "session-123") + .with_body(r#"{"namespaces": [["ns1"]], "next-page-token": "next"}"#) + .create_async() + .await; + let second_page = server + .mock("GET", "/v1/namespaces?pageToken=next") + .match_header("x-session-id", "session-123") + .with_body(r#"{"namespaces": [["ns2"]]}"#) + .create_async() + .await; + + let (catalog, seen_session_ids) = context_catalog( + RestCatalogConfig::builder().uri(server.url()).build(), + false, + ); + let context = SessionContext::builder() + .session_id("session-123".to_string()) + .build(); + + let namespaces = catalog.list_namespaces(&context, None).await.unwrap(); + + assert_eq!(namespaces, vec![ + NamespaceIdent::new("ns1".to_string()), + NamespaceIdent::new("ns2".to_string()), + ]); + assert_eq!(*seen_session_ids.lock().unwrap(), vec![ + "session-123", + "session-123" + ]); + config_mock.assert_async().await; + first_page.assert_async().await; + second_page.assert_async().await; + } + + #[tokio::test] + async fn test_contextual_session_authenticates_namespace_exists_head() { + let mut server = Server::new_async().await; + let config_mock = create_config_mock_with_exists_endpoints(&mut server).await; + let exists_mock = server + .mock("HEAD", "/v1/namespaces/ns1") + .match_header("x-session-id", "namespace-session") + .with_status(204) + .create_async() + .await; + let (catalog, seen_session_ids) = context_catalog( + RestCatalogConfig::builder().uri(server.url()).build(), + false, + ); + let context = SessionContext::builder() + .session_id("namespace-session".to_string()) + .build(); + + assert!( + catalog + .namespace_exists(&context, &NamespaceIdent::new("ns1".to_string())) + .await + .unwrap() + ); + assert_eq!(*seen_session_ids.lock().unwrap(), vec!["namespace-session"]); + config_mock.assert_async().await; + exists_mock.assert_async().await; + } + + #[tokio::test] + async fn test_contextual_session_authenticates_table_exists_head() { + let mut server = Server::new_async().await; + let config_mock = create_config_mock_with_exists_endpoints(&mut server).await; + let exists_mock = server + .mock("HEAD", "/v1/namespaces/ns1/tables/table1") + .match_header("x-session-id", "table-session") + .with_status(204) + .create_async() + .await; + let (catalog, seen_session_ids) = context_catalog( + RestCatalogConfig::builder().uri(server.url()).build(), + false, + ); + let context = SessionContext::builder() + .session_id("table-session".to_string()) + .build(); + + assert!( + catalog + .table_exists( + &context, + &TableIdent::new(NamespaceIdent::new("ns1".to_string()), "table1".to_string(),), + ) + .await + .unwrap() + ); + assert_eq!(*seen_session_ids.lock().unwrap(), vec!["table-session"]); + config_mock.assert_async().await; + exists_mock.assert_async().await; + } + + #[tokio::test] + async fn test_contextual_session_authenticates_write_operation() { + let mut server = Server::new_async().await; + let config_mock = create_config_mock(&mut server).await; + let drop_mock = server + .mock("DELETE", "/v1/namespaces/ns1") + .match_header("x-session-id", "write-session") + .with_status(204) + .create_async() + .await; + let (catalog, seen_session_ids) = context_catalog( + RestCatalogConfig::builder().uri(server.url()).build(), + false, + ); + let context = SessionContext::builder() + .session_id("write-session".to_string()) + .build(); + + catalog + .drop_namespace(&context, &NamespaceIdent::new("ns1".to_string())) + .await + .unwrap(); + assert_eq!(*seen_session_ids.lock().unwrap(), vec!["write-session"]); + config_mock.assert_async().await; + drop_mock.assert_async().await; + } + + #[tokio::test] + async fn test_contextual_session_error_prevents_operation_request() { + let mut server = Server::new_async().await; + let config_mock = create_config_mock(&mut server).await; + let list_mock = server + .mock("GET", "/v1/namespaces") + .expect(0) + .create_async() + .await; + let (catalog, seen_session_ids) = + context_catalog(RestCatalogConfig::builder().uri(server.url()).build(), true); + let context = SessionContext::builder() + .session_id("failing-session".to_string()) + .build(); + + let error = catalog.list_namespaces(&context, None).await.unwrap_err(); + assert_eq!(error.kind(), ErrorKind::Unexpected); + assert_eq!(error.message(), "contextual session failure"); + assert_eq!(*seen_session_ids.lock().unwrap(), vec!["failing-session"]); + config_mock.assert_async().await; + list_mock.assert_async().await; + } + #[test] fn test_config_debug_redacts_secrets() { let config = RestCatalogConfig::builder() diff --git a/crates/catalog/rest/src/client.rs b/crates/catalog/rest/src/client.rs index 6de740a52..f81f86574 100644 --- a/crates/catalog/rest/src/client.rs +++ b/crates/catalog/rest/src/client.rs @@ -17,7 +17,6 @@ use std::collections::HashMap; use std::fmt::{Debug, Formatter}; -use std::sync::Arc; use iceberg::{Error, ErrorKind, Result}; use reqwest::header::HeaderMap; @@ -25,7 +24,7 @@ use reqwest::{Client, IntoUrl, Method, RequestBuilder}; use serde::de::DeserializeOwned; use crate::RestCatalogConfig; -use crate::auth::{AuthSession, NoopSession}; +use crate::auth::AuthSession; use crate::request::HttpRequest; use crate::response::HttpResponse; @@ -41,12 +40,6 @@ pub struct HttpClient { extra_headers: HeaderMap, /// Whether to disable header redaction in error logs (defaults to false for security). disable_header_redaction: bool, - /// Authenticates everything this client sends. A client handed to an - /// [`AuthManager`] carries no authentication, since that is what the - /// manager is about to create. - /// - /// [`AuthManager`]: crate::auth::AuthManager - auth_session: Arc<dyn AuthSession>, } impl Debug for HttpClient { @@ -64,30 +57,13 @@ impl Debug for HttpClient { } impl HttpClient { - /// The same client authenticating with `auth_session` instead: a derived - /// auth session reuses the connection pool and headers of the client it - /// came from, carrying its own authentication. - pub fn with_auth_session(&self, auth_session: Arc<dyn AuthSession>) -> Self { - Self { - auth_session, - ..self.clone() - } - } - - /// The same client with no authentication, for the requests that must not - /// carry it — a session refreshing its own token over the client it - /// authenticates would otherwise recurse. - pub fn without_auth_session(&self) -> Self { - self.with_auth_session(Arc::new(NoopSession)) - } - - /// Sends a form-encoded POST and returns the response status and body, + /// Sends an unauthenticated form-encoded POST and returns the response, /// which is what an [`AuthManager`] needs to exchange a credential for a - /// token. Only `headers` are sent; the catalog's own extra headers are - /// not merged in. + /// token. /// - /// Like every request, it carries this client's session; call - /// [`Self::without_auth_session`] first to send it unauthenticated. + /// Only `headers` are sent: neither the catalog's extra headers nor + /// any session's authentication is applied, so a token request carries + /// exactly the credentials its caller puts into `headers` or `form`. /// /// [`AuthManager`]: crate::auth::AuthManager pub async fn post_form( @@ -108,7 +84,7 @@ impl HttpClient { reqwest::header::CONTENT_TYPE, reqwest::header::HeaderValue::from_static("application/x-www-form-urlencoded"), ); - self.auth_session.authenticate(&mut request).await?; + let response = self.client.execute(request.into_inner()).await?; HttpResponse::read(response).await } @@ -119,7 +95,6 @@ impl HttpClient { client: cfg.client(), extra_headers: cfg.extra_headers()?, disable_header_redaction: cfg.disable_header_redaction(), - auth_session: Arc::new(NoopSession), }) } @@ -137,28 +112,21 @@ impl HttpClient { client: cfg.client(), extra_headers, disable_header_redaction: cfg.disable_header_redaction(), - auth_session: self.auth_session, }) } - /// Testing only: the session authenticating this client's requests. - #[cfg(test)] - pub(crate) fn auth_session(&self) -> &Arc<dyn AuthSession> { - &self.auth_session - } - /// Testing only: the bearer token `session` would attach. /// /// Authenticates a throwaway request (never sent) and reads the header /// back, so it works for any [`AuthSession`]. #[cfg(test)] - pub(crate) async fn token(&self) -> Option<String> { + pub(crate) async fn token(&self, auth_session: &dyn AuthSession) -> Option<String> { let mut request = HttpRequest::build( self.client .request(Method::GET, "http://localhost/token-probe";), ) .ok()?; - self.auth_session.authenticate(&mut request).await.ok()?; + auth_session.authenticate(&mut request).await.ok()?; let request = request.into_inner(); request .headers() @@ -178,10 +146,15 @@ impl HttpClient { /// Sends `request` to the Iceberg REST catalog, authenticated by this /// client's session. - pub(crate) async fn query_catalog(&self, mut request: HttpRequest) -> Result<HttpResponse> { + pub(crate) async fn query_catalog( + &self, + auth_session: &dyn AuthSession, + mut request: HttpRequest, + ) -> Result<HttpResponse> { // Authenticate first, then apply extra headers, so a configured // `header.authorization` keeps overriding a token (unchanged behavior). - self.auth_session.authenticate(&mut request).await?; + auth_session.authenticate(&mut request).await?; + let mut request = request.into_inner(); request.headers_mut().extend(self.extra_headers.clone()); HttpResponse::read(self.client.execute(request).await?).await @@ -282,19 +255,6 @@ pub(crate) fn deserialize_unexpected_catalog_error( mod tests { use super::*; - #[derive(Debug)] - struct StaticSession; - - #[async_trait::async_trait] - impl AuthSession for StaticSession { - async fn authenticate(&self, request: &mut HttpRequest) -> Result<()> { - request - .headers_mut() - .insert("authorization", "Bearer tok".parse().unwrap()); - Ok(()) - } - } - #[tokio::test] async fn test_a_truncated_body_error_names_the_url() { // `bytes()` builds its error without a URL, so `read` attaches the one @@ -376,40 +336,36 @@ mod tests { } #[tokio::test] - async fn test_post_form_carries_the_session_until_it_is_removed() { - // Every request a client sends carries its session; a caller that - // needs an unauthenticated one removes the session first. + async fn test_post_form_is_never_authenticated() { + // A token request carries no authentication, not even a + // user-configured `header.authorization` meant for the catalog. let mut server = mockito::Server::new_async().await; - let signed = server - .mock("POST", "/token") - .match_header("authorization", "Bearer tok") - .with_status(200) - .create_async() - .await; - let unsigned = server + let mock = server .mock("POST", "/token") .match_header("authorization", mockito::Matcher::Missing) .with_status(200) .create_async() .await; - let client = HttpClient::new(&RestCatalogConfig::builder().uri(server.url()).build()) - .unwrap() - .with_auth_session(Arc::new(StaticSession)); - let url = format!("{}/token", server.url()); - - client - .post_form(&url, &HeaderMap::new(), &HashMap::new()) - .await - .unwrap(); - signed.assert_async().await; + HttpClient::new( + &RestCatalogConfig::builder() + .uri(server.url()) + .props(HashMap::from([( + "header.authorization".to_string(), + "Basic xyz".to_string(), + )])) + .build(), + ) + .unwrap() + .post_form( + &format!("{}/token", server.url()), + &HeaderMap::new(), + &HashMap::new(), + ) + .await + .unwrap(); - client - .without_auth_session() - .post_form(&url, &HeaderMap::new(), &HashMap::new()) - .await - .unwrap(); - unsigned.assert_async().await; + mock.assert_async().await; } #[test]
