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

@@ -139,6 +139,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
@@ -288,5 +302,6 @@ pub fn iceberg_catalog_rest::UpdateNamespacePropertiesResponse::serialize<__S>(&
 impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::UpdateNamespacePropertiesResponse
 pub fn iceberg_catalog_rest::UpdateNamespacePropertiesResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
 pub const iceberg_catalog_rest::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

crates/catalog/rest/src/catalog.rs

@@ -54,6 +54,8 @@ pub const REST_CATALOG_PROP_URI: &str = "uri";
 pub const REST_CATALOG_PROP_WAREHOUSE: &str = "warehouse";
 /// Disable header redaction in error logs (defaults to false for security)
 pub const REST_CATALOG_PROP_DISABLE_HEADER_REDACTION: &str = "disable-header-redaction";
+/// Identifier for a server-side scan plan associated with credential requests.
+pub const REST_CATALOG_PROP_SCAN_PLAN_ID: &str = "rest.scan.plan-id";
 
 const ICEBERG_REST_SPEC_VERSION: &str = "0.14.1";
 const CARGO_PKG_VERSION: &str = env!("CARGO_PKG_VERSION");
@@ -164,7 +166,7 @@ impl RestCatalogBuilder {
 }
 
 /// Rest catalog configuration.
-#[derive(Clone, Debug, TypedBuilder)]
+#[derive(Clone, TypedBuilder)]
 pub(crate) struct RestCatalogConfig {
     #[builder(default, setter(strip_option))]
     name: Option<String>,
@@ -181,6 +183,19 @@ pub(crate) struct RestCatalogConfig {
     client: Option<Client>,
 }
 
+impl std::fmt::Debug for RestCatalogConfig {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        // Catalog and table properties may contain OAuth credentials, delegated
+        // storage credentials, or arbitrary secret-bearing `header.*` values.
+        f.debug_struct("RestCatalogConfig")
+            .field("name", &self.name)
+            .field("uri", &self.uri)
+            .field("warehouse", &self.warehouse)
+            .field("property_keys", &self.props.keys().collect::<Vec<_>>())
+            .finish_non_exhaustive()
+    }
+}
+
 impl RestCatalogConfig {
     fn url_prefixed(&self, parts: &[&str]) -> String {
         [&self.uri, PATH_V1]
@@ -358,7 +373,7 @@ impl RestCatalogConfig {
 
 #[derive(Debug)]
 struct RestContext {
-    client: HttpClient,
+    client: Arc<HttpClient>,
     /// Runtime config is fetched from rest server and stored here.
     ///
     /// It's could be different from the user config.
@@ -447,7 +462,7 @@ impl RestCatalog {
 
                 Ok(RestContext {
                     config,
-                    client,
+                    client: Arc::new(client),
                     endpoints,
                 })
             })
@@ -510,17 +525,17 @@ impl RestCatalog {
         metadata_location: Option<&str>,
         extra_config: Option<HashMap<String, String>>,
     ) -> Result<FileIO> {
-        let mut props = self.context().await?.config.props.clone();
+        let context = self.context().await?;
+        let mut props = context.config.props.clone();
         if let Some(config) = extra_config {
             props.extend(config);
         }
 
         // If the warehouse is a logical identifier instead of a URL we don't want
         // to raise an exception
-        let warehouse_path = match self.context().await?.config.warehouse.as_deref() {
+        let warehouse_path = match context.config.warehouse.as_deref() {
             Some(url) if Url::parse(url).is_ok() => Some(url),
-            Some(_) => None,
-            None => None,
+            _ => None,
         };
 
         if metadata_location.or(warehouse_path).is_none() {
@@ -541,9 +556,20 @@ impl RestCatalog {
                 )
             })?;
 
-        let file_io = FileIOBuilder::new(factory).with_props(props).build();
+        // If the catalog vends refreshable credentials for this table's storage,
+        // attach a provider so the backend re-fetches them before they expire.
+        let credential_provider = crate::credential::build_vended_credential_provider(
+            context.client.clone(),
+            &context.config.uri,
+            &props,
+        )?;
+
+        let mut builder = FileIOBuilder::new(factory).with_props(props);
+        if let Some(provider) = credential_provider {
+            builder = builder.with_credential_provider(provider);
+        }
 
-        Ok(file_io)
+        Ok(builder.build())
     }
 
     /// Invalidate the current token without generating a new one. On the next request, the client
@@ -1048,7 +1074,14 @@ impl Catalog for RestCatalog {
             "Metadata location missing in `register_table` response!",
         ))?;
 
-        let file_io = self.load_file_io(Some(metadata_location), None).await?;
+        let config = response
+            .config
+            .into_iter()
+            .chain(self.user_config.props.clone())
+            .collect();
+        let file_io = self
+            .load_file_io(Some(metadata_location), Some(config))
+            .await?;
 
         let mut table_builder = Table::builder()
             .identifier(table_ident.clone())

crates/catalog/rest/src/client.rs

@@ -49,9 +49,13 @@ pub(crate) struct HttpClient {
 
 impl Debug for HttpClient {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
+        // Omit the reqwest client: injected clients may carry secret default
+        // headers. Explicit headers use the same redaction policy as errors.
         f.debug_struct("HttpClient")
-            .field("client", &self.client)
-            .field("extra_headers", &self.extra_headers)
+            .field(
+                "extra_headers",
+                &format_headers_redacted(&self.extra_headers, self.disable_header_redaction),
+            )
             .finish_non_exhaustive()
     }
 }
@@ -71,6 +75,24 @@ impl HttpClient {
         })
     }
 
+    /// Create a client for table-scoped resources while reusing this client's
+    /// underlying connection pool.
+    ///
+    /// A load-table response may supply a table token or `header.*` values that
+    /// must be used for subsequent table requests such as credential refresh.
+    pub(crate) fn for_table(
+        &self,
+        catalog_uri: &str,
+        props: HashMap<String, String>,
+    ) -> Result<Self> {
+        let cfg = RestCatalogConfig::builder()
+            .uri(catalog_uri.to_string())
+            .props(props)
+            .client(Some(self.client.clone()))
+            .build();
+        Self::new(&cfg)
+    }
+
     /// Update the http client with new configuration.
     ///
     /// If cfg carries new value, we will use cfg instead.
@@ -156,7 +178,6 @@ impl HttpClient {
                 )
                 .with_context("operation", "auth")
                 .with_context("url", auth_url.to_string())
-                .with_context("json", String::from_utf8_lossy(&text))
                 .with_source(e)
             })?)
         } else {
@@ -170,7 +191,6 @@ impl HttpClient {
                     .with_context("code", code.to_string())
                     .with_context("operation", "auth")
                     .with_context("url", auth_url.to_string())
-                    .with_context("json", String::from_utf8_lossy(&text))
                     .with_source(e)
             })?;
             Err(Error::from(e))
@@ -278,29 +298,30 @@ pub(crate) async fn deserialize_catalog_response<R: DeserializeOwned>(
     let bytes = response.bytes().await?;
 
     serde_json::from_slice::<R>(&bytes).map_err(|e| {
+        // Successful REST responses can contain OAuth tokens and delegated
+        // storage credentials. Never copy an unparsable response into an error.
         Error::new(
             ErrorKind::Unexpected,
             "Failed to parse response from rest catalog server",
         )
-        .with_context("json", String::from_utf8_lossy(&bytes))
         .with_source(e)
     })
 }
 
-/// Headers that contain sensitive information and should be excluded from logs.
-const SENSITIVE_HEADERS: &[&str] = &[
-    "authorization",
-    "proxy-authorization",
-    "set-cookie",
-    "cookie",
-    "x-api-key",
-    "x-auth-token",
-];
-
-/// Returns true if the header name is considered sensitive.
+/// Returns true if the header may carry a secret.
 fn is_sensitive_header(name: &str) -> bool {
     let name_lower = name.to_lowercase();
-    SENSITIVE_HEADERS.iter().any(|h| name_lower == *h)
+    [
+        "auth",
+        "token",
+        "secret",
+        "key",
+        "password",
+        "cookie",
+        "credential",
+    ]
+    .iter()
+    .any(|pattern| name_lower.contains(pattern))
 }
 
 /// Redacts sensitive headers and returns a debug-formatted string.
@@ -386,6 +407,8 @@ mod tests {
     fn test_format_headers_redacted_filters_sensitive() {
         let mut headers = HeaderMap::new();
         headers.insert("authorization", "Bearer secret-token".parse().unwrap());
+        headers.insert("x-client-secret", "private-value".parse().unwrap());
+        headers.insert("x-client-credential", "credential-value".parse().unwrap());
         headers.insert("content-type", "application/json".parse().unwrap());
 
         let result = format_headers_redacted(&headers, false);
@@ -395,6 +418,8 @@ mod tests {
         assert!(result.contains("[REDACTED]"));
         // Sensitive value should NOT be present
         assert!(!result.contains("secret-token"));
+        assert!(!result.contains("private-value"));
+        assert!(!result.contains("credential-value"));
         // Non-sensitive header should be present with actual value
         assert!(result.contains("content-type"));
         assert!(result.contains("application/json"));