feat(rest): Support refreshing vended storage credentials

crates/catalog/rest/Cargo.toml

@@ -35,6 +35,7 @@ chrono = { workspace = true }
 http = { workspace = true }
 iceberg = { workspace = true }
 itertools = { workspace = true }
+rand = { workspace = true }
 reqwest = { workspace = true }
 serde = { workspace = true }
 serde_derive = { workspace = true }

crates/catalog/rest/public-api.txt

@@ -171,6 +171,20 @@ impl serde_core::ser::Serialize for iceberg_catalog_rest::ListTablesResponse
 pub fn iceberg_catalog_rest::ListTablesResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
 impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::ListTablesResponse
 pub fn iceberg_catalog_rest::ListTablesResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
+pub struct iceberg_catalog_rest::LoadCredentialsResponse
+pub iceberg_catalog_rest::LoadCredentialsResponse::storage_credentials: alloc::vec::Vec<iceberg_catalog_rest::StorageCredential>
+impl core::clone::Clone for iceberg_catalog_rest::LoadCredentialsResponse
+pub fn iceberg_catalog_rest::LoadCredentialsResponse::clone(&self) -> iceberg_catalog_rest::LoadCredentialsResponse
+impl core::cmp::Eq for iceberg_catalog_rest::LoadCredentialsResponse
+impl core::cmp::PartialEq for iceberg_catalog_rest::LoadCredentialsResponse
+pub fn iceberg_catalog_rest::LoadCredentialsResponse::eq(&self, other: &iceberg_catalog_rest::LoadCredentialsResponse) -> bool
+impl core::fmt::Debug for iceberg_catalog_rest::LoadCredentialsResponse
+pub fn iceberg_catalog_rest::LoadCredentialsResponse::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
+impl core::marker::StructuralPartialEq for iceberg_catalog_rest::LoadCredentialsResponse
+impl serde_core::ser::Serialize for iceberg_catalog_rest::LoadCredentialsResponse
+pub fn iceberg_catalog_rest::LoadCredentialsResponse::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer
+impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::LoadCredentialsResponse
+pub fn iceberg_catalog_rest::LoadCredentialsResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
 pub struct iceberg_catalog_rest::LoadTableResult
 pub iceberg_catalog_rest::LoadTableResult::config: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>
 pub iceberg_catalog_rest::LoadTableResult::metadata: iceberg::spec::table_metadata::TableMetadata
@@ -213,6 +227,7 @@ pub fn iceberg_catalog_rest::NoopAuthManager::fmt(&self, f: &mut core::fmt::Form
 impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::NoopAuthManager
 pub fn iceberg_catalog_rest::NoopAuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg_catalog_rest::NoopAuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::NoopAuthManager::table_session<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _table: &'life2 iceberg::catalog::TableIdent, _props: &'life3 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>, parent: 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: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait
 pub struct iceberg_catalog_rest::OAuth2Manager
 impl iceberg_catalog_rest::OAuth2Manager
 pub fn iceberg_catalog_rest::OAuth2Manager::new(token_endpoint: impl core::convert::Into<alloc::string::String>) -> Self
@@ -225,6 +240,7 @@ pub fn iceberg_catalog_rest::OAuth2Manager::fmt(&self, f: &mut core::fmt::Format
 impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::OAuth2Manager
 pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::OAuth2Manager::table_session<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _table: &'life2 iceberg::catalog::TableIdent, props: &'life3 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>, parent: 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: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait
 pub struct iceberg_catalog_rest::RegisterTableRequest
 pub iceberg_catalog_rest::RegisterTableRequest::metadata_location: alloc::string::String
 pub iceberg_catalog_rest::RegisterTableRequest::name: alloc::string::String
@@ -339,16 +355,20 @@ pub const iceberg_catalog_rest::AUTH_TYPE_NONE: &str
 pub const iceberg_catalog_rest::AUTH_TYPE_OAUTH2: &str
 pub const iceberg_catalog_rest::REST_CATALOG_PROP_AUTH_TYPE: &str
 pub const iceberg_catalog_rest::REST_CATALOG_PROP_DISABLE_HEADER_REDACTION: &str
+pub const iceberg_catalog_rest::REST_CATALOG_PROP_SCAN_PLAN_ID: &str
 pub const iceberg_catalog_rest::REST_CATALOG_PROP_URI: &str
 pub const iceberg_catalog_rest::REST_CATALOG_PROP_WAREHOUSE: &str
 pub trait iceberg_catalog_rest::AuthManager: core::fmt::Debug + core::marker::Send + core::marker::Sync
 pub fn iceberg_catalog_rest::AuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg_catalog_rest::AuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::AuthManager::table_session<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _table: &'life2 iceberg::catalog::TableIdent, _props: &'life3 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>, parent: 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: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait
 impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::NoopAuthManager
 pub fn iceberg_catalog_rest::NoopAuthManager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg_catalog_rest::NoopAuthManager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::NoopAuthManager::table_session<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _table: &'life2 iceberg::catalog::TableIdent, _props: &'life3 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>, parent: 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: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait
 impl iceberg_catalog_rest::AuthManager for iceberg_catalog_rest::OAuth2Manager
 pub fn iceberg_catalog_rest::OAuth2Manager::catalog_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
 pub fn iceberg_catalog_rest::OAuth2Manager::init_session<'life0, 'life1, 'life2, 'async_trait>(&'life0 self, client: &'life1 iceberg_catalog_rest::HttpClient, props: &'life2 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<alloc::boxed::Box<dyn iceberg_catalog_rest::AuthSession>>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait
+pub fn iceberg_catalog_rest::OAuth2Manager::table_session<'life0, 'life1, 'life2, 'life3, 'async_trait>(&'life0 self, _client: &'life1 iceberg_catalog_rest::HttpClient, _table: &'life2 iceberg::catalog::TableIdent, props: &'life3 std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>, parent: 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: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait
 pub trait iceberg_catalog_rest::AuthSession: core::fmt::Debug + core::marker::Send + core::marker::Sync
 pub fn iceberg_catalog_rest::AuthSession::authenticate<'life0, 'life1, 'async_trait>(&'life0 self, request: &'life1 mut iceberg_catalog_rest::HttpRequest) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::error::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait

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

@@ -25,7 +25,7 @@ use std::fmt::Debug;
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use iceberg::Result;
+use iceberg::{Result, TableIdent};
 pub use oauth2::OAuth2Manager;
 
 use crate::client::HttpClient;
@@ -42,9 +42,9 @@ pub const AUTH_TYPE_OAUTH2: &str = "oauth2";
 /// property or injected through `RestCatalogBuilder::with_auth_manager`. It
 /// builds the sessions the catalog then keeps.
 ///
-/// Both methods are handed the catalog's [`HttpClient`], which an
-/// implementation may reuse for its own requests (e.g. a token exchange) so
-/// that they share the catalog's connection pool and configuration.
+/// Session-construction methods are handed the catalog's [`HttpClient`], which
+/// an implementation may reuse for its own requests (e.g. a token exchange)
+/// so that they share the catalog's connection pool and configuration.
 #[async_trait]
 pub trait AuthManager: Debug + Send + Sync {
     /// Session used for the initial `/v1/config` handshake, given the
@@ -69,6 +69,22 @@ pub trait AuthManager: Debug + Send + Sync {
         client: &HttpClient,
         props: &HashMap<String, String>,
     ) -> Result<Arc<dyn AuthSession>>;
+
+    /// Returns a session for requests associated with `table`.
+    ///
+    /// `props` are the unmerged properties returned by the table endpoint.
+    /// The default preserves the catalog session; managers should return a
+    /// child session only when the table properties contain an authentication
+    /// override.
+    async fn table_session(
+        &self,
+        _client: &HttpClient,
+        _table: &TableIdent,
+        _props: &HashMap<String, String>,
+        parent: Arc<dyn AuthSession>,
+    ) -> Result<Arc<dyn AuthSession>> {
+        Ok(parent)
+    }
 }
 
 /// Authenticates outgoing REST catalog requests.

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

@@ -21,7 +21,7 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 use http::StatusCode;
-use iceberg::{Credential, Error, ErrorKind, Result};
+use iceberg::{Credential, Error, ErrorKind, Result, TableIdent};
 use reqwest::header::HeaderMap;
 use tokio::sync::Mutex;
 
@@ -45,8 +45,9 @@ struct OAuth2Params {
 /// Iceberg REST catalogs.
 ///
 /// A configured `token` is used directly; otherwise `credential` is exchanged
-/// for a token at the token endpoint and cached. The cached token is shared
-/// across sessions so it survives the config handshake.
+/// for a token at the token endpoint and cached. The cached token is shared by
+/// the init and catalog sessions so it survives the config handshake;
+/// table-specific tokens use isolated sessions.
 pub struct OAuth2Manager {
     token: Arc<Mutex<Option<Credential>>>,
     init_params: OAuth2Params,
@@ -143,13 +144,30 @@ impl AuthManager for OAuth2Manager {
     ) -> Result<Arc<dyn AuthSession>> {
         Ok(Arc::new(self.session_from(client, props).await?))
     }
+
+    async fn table_session(
+        &self,
+        _client: &HttpClient,
+        _table: &TableIdent,
+        props: &HashMap<String, String>,
+        parent: Arc<dyn AuthSession>,
+    ) -> Result<Arc<dyn AuthSession>> {
+        let Some(token) = props.get("token") else {
+            return Ok(parent);
+        };
+
+        Ok(Arc::new(OAuth2Session {
+            token: Arc::new(Mutex::new(Some(Credential::from(token.clone())))),
+            token_source: TokenSource::StaticToken,
+        }))
+    }
 }
 
 impl OAuth2Manager {
     /// Builds a session from the manager's options with `props` merged onto
     /// them, so an injected manager keeps whatever a property doesn't
-    /// override. The manager's token cell is shared with every session it
-    /// builds, so a token cached during the handshake survives it.
+    /// override. The manager's token cell is shared by the init and catalog
+    /// sessions, so a token cached during the handshake survives it.
     async fn session_from(
         &self,
         client: &HttpClient,
@@ -376,4 +394,56 @@ mod tests {
             "Bearer tok-static"
         );
     }
+
+    #[tokio::test]
+    async fn test_table_session_inherits_parent_unless_token_is_overridden() {
+        let manager = OAuth2Manager::new("http://localhost/unused").with_token("catalog-token");
+        let client = test_client();
+        let parent = manager
+            .catalog_session(&client, &HashMap::new())
+            .await
+            .unwrap();
+        let table = TableIdent::from_strs(["namespace", "table"]).unwrap();
+
+        let inherited = manager
+            .table_session(&client, &table, &HashMap::new(), Arc::clone(&parent))
+            .await
+            .unwrap();
+        assert!(Arc::ptr_eq(&parent, &inherited));
+
+        let overridden = manager
+            .table_session(
+                &client,
+                &table,
+                &HashMap::from([("token".to_string(), "table-token".to_string())]),
+                Arc::clone(&parent),
+            )
+            .await
+            .unwrap();
+        assert!(!Arc::ptr_eq(&parent, &overridden));
+
+        let mut parent_request = HttpRequest::new(
+            Client::new()
+                .get("https://rest.example.com/catalog")
+                .build()
+                .unwrap(),
+        );
+        parent.authenticate(&mut parent_request).await.unwrap();
+        assert_eq!(
+            parent_request.headers().get("authorization").unwrap(),
+            "Bearer catalog-token"
+        );
+
+        let mut table_request = HttpRequest::new(
+            Client::new()
+                .get("https://rest.example.com/table")
+                .build()
+                .unwrap(),
+        );
+        overridden.authenticate(&mut table_request).await.unwrap();
+        assert_eq!(
+            table_request.headers().get("authorization").unwrap(),
+            "Bearer table-token"
+        );
+    }
 }