This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 7f9691fc7 feat(rest_catalog::auth) Contextual Sessions (#3081)
7f9691fc7 is described below
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]
