From 2ec2d4fcdb9a6270ca1126737ed03a142bade6af Mon Sep 17 00:00:00 2001 From: Patrick Butler Date: Wed, 9 Sep 2026 13:57:39 -0400 Subject: [PATCH 1/5] support for vended creds with GCP biglake --- Cargo.lock | 1 + src/storage-types/src/connections.rs | 191 ++++++++--- .../src/connections/iceberg_credentials.rs | 322 ++++++++++++++---- 3 files changed, 410 insertions(+), 104 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3052342c2636f..728ffa7969f86 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5021,6 +5021,7 @@ dependencies = [ "opendal", "reqsign-aws-v4", "reqsign-core", + "reqsign-google", "serde", "typetag", "url", diff --git a/src/storage-types/src/connections.rs b/src/storage-types/src/connections.rs index 628af8478aba5..036536d0abb62 100644 --- a/src/storage-types/src/connections.rs +++ b/src/storage-types/src/connections.rs @@ -37,7 +37,8 @@ use iceberg_catalog_rest::{ RestCatalogBuilder, TokenProvider, }; use iceberg_storage_opendal::{ - AwsCredential, CustomAwsCredentialLoader, OpenDalStorageFactory, ProvideCredential, + AwsCredential, CustomAwsCredentialLoader, CustomGcsCredentialLoader, OpenDalStorageFactory, + ProvideCredential, }; use itertools::Itertools; use mz_ccsr::tls::{Certificate, Identity}; @@ -103,9 +104,11 @@ pub mod string_or_secret; const OAUTH2_PARAM_SCOPE: &str = "scope"; const REST_CATALOG_PROP_OAUTH2_SERVER_URI: &str = "oauth2-server-uri"; -/// Requests catalog-vended storage credentials. `iceberg-rust` turns `header.*` catalog -/// properties into headers on every REST request, the same way the Iceberg Java client -/// carries this one. +/// The prefix marking a catalog property that `iceberg-rust` turns into a header on every REST +/// request, the same convention the Iceberg Java client uses. +const REST_CATALOG_HEADER_PROP_PREFIX: &str = "header."; +/// Requests catalog-vended storage credentials, carried as a header by way of +/// [`REST_CATALOG_HEADER_PROP_PREFIX`]. const REST_CATALOG_PROP_ACCESS_DELEGATION: &str = "header.X-Iceberg-Access-Delegation"; /// A credential loader that wraps an aws-sdk-rust credentials provider for use with @@ -1018,6 +1021,79 @@ impl IcebergCatalogConnection { Ok(Arc::new(catalog)) } + /// Collects the headers `iceberg-rust` puts on every catalog request out of the `header.*` + /// props it takes them from. + /// + /// The credential endpoints Materialize calls directly bypass the catalog client, so without + /// this they would reach the same server missing headers it may require, `x-goog-user-project` + /// on a GCP-hosted catalog among them. + fn catalog_headers(props: &BTreeMap) -> Result { + props + .iter() + .filter_map(|(k, v)| { + k.strip_prefix(REST_CATALOG_HEADER_PROP_PREFIX) + .map(|name| (name, v)) + }) + .map(|(name, value)| { + let name = HeaderName::try_from(name) + .with_context(|| format!("invalid Iceberg catalog header name: {name}"))?; + let value = HeaderValue::try_from(value) + .with_context(|| format!("invalid Iceberg catalog header value for {name}"))?; + Ok((name, value)) + }) + .collect() + } + + /// Resolves the endpoint that vends storage credentials for `table`, or `None` if this + /// connection has no use for one. + /// + /// A loader built on the returned endpoint takes sole responsibility for storage credentials, + /// so `None` is the signal to leave the catalog's static `storage-credentials` props in force. + /// It means either that the connection did not ask for delegation, or that the caller named no + /// table, and the specification scopes vended credentials to a single table. + async fn vended_credential_endpoint( + &self, + rest: &RestIcebergCatalog, + client: &reqwest::Client, + token: &Arc, + headers: &HeaderMap, + table: Option<&TableIdent>, + ) -> Result, anyhow::Error> { + match (&rest.access_delegation, table) { + (Some(IcebergAccessDelegation::VendedCredentials), Some(table)) => Ok(Some( + iceberg_credentials::table_credentials_endpoint( + &self.uri, + client, + token, + headers, + rest.warehouse.as_deref(), + table, + ) + .await?, + )), + _ => Ok(None), + } + } + + /// Builds a GCS storage factory, refreshing vended credentials from `endpoint` if there is one. + fn gcs_storage_factory( + endpoint: Option, + client: &reqwest::Client, + token: &Arc, + headers: &HeaderMap, + ) -> OpenDalStorageFactory { + OpenDalStorageFactory::Gcs { + customized_credential_load: endpoint.map(|endpoint| { + CustomGcsCredentialLoader::new(iceberg_credentials::VendedCredentialLoader::new( + client.clone(), + endpoint, + Arc::clone(token), + headers.clone(), + )) + }), + } + } + async fn connect_rest( &self, rest: &RestIcebergCatalog, @@ -1118,31 +1194,10 @@ impl IcebergCatalogConnection { oauth_params.into_iter().collect(), )); - // Installing a loader hands it sole responsibility for S3 credentials: OpenDAL - // replaces its whole provider chain, including the static keys parsed out of the - // catalog's vended `storage-credentials` props. So only install one when the - // connection asked for delegation and we know which table to refresh, and let the - // static props serve every other case. - let customized_credential_load = match (&rest.access_delegation, table) { - (Some(IcebergAccessDelegation::VendedCredentials), Some(table)) => { - let endpoint = iceberg_credentials::table_credentials_endpoint( - &self.uri, - &client, - &token, - rest.warehouse.as_deref(), - table, - ) - .await?; - Some(CustomAwsCredentialLoader::new( - iceberg_credentials::VendedCredentialLoader::new( - client.clone(), - endpoint, - Arc::clone(&token), - ), - )) - } - _ => None, - }; + let headers = Self::catalog_headers(&props)?; + let endpoint = self + .vended_credential_endpoint(rest, &client, &token, &headers, table) + .await?; ( // The catalog tells us where the data lives but not what @@ -1155,14 +1210,26 @@ impl IcebergCatalogConnection { // `iceberg-rust` wires into the same FileIO. // N.B. This is not confirmed to work with other catalog & storage implementations. IcebergStorageProvider::S3 => OpenDalStorageFactory::S3 { - customized_credential_load, + customized_credential_load: endpoint.map(|endpoint| { + CustomAwsCredentialLoader::new( + iceberg_credentials::VendedCredentialLoader::new( + client.clone(), + endpoint, + Arc::clone(&token), + headers.clone(), + ), + ) + }), }, - // Both take their credentials from the catalog's - // config, which `iceberg-rust` forwards to `opendal` - // the same way. Neither has an equivalent of the S3 - // credential loader, so vended credentials for these - // stores only work through those props. - IcebergStorageProvider::Gcs => OpenDalStorageFactory::Gcs, + IcebergStorageProvider::Gcs => { + Self::gcs_storage_factory(endpoint, &client, &token, &headers) + } + // ADLS takes its credentials from the catalog's config, which + // `iceberg-rust` forwards to `opendal` the same way. OpenDAL's Azure + // service does not go through reqsign's credential provider + // abstraction, so there is no hook to wrap: vended credentials for + // ADLS work only through those static props, and stop working when + // they expire. IcebergStorageProvider::Adls => OpenDalStorageFactory::Azdls, }, // NOTE: We construct our own OAuth authenticator for the Catalog client instead of using the one built in. @@ -1193,18 +1260,31 @@ impl IcebergCatalogConnection { ); } + // The service account authenticates catalog requests whether or not the catalog + // vends storage credentials, and doubles as the token source for refreshing them. + let token: Arc = Arc::new(GcpTokenProvider { service_account }); + let headers = Self::catalog_headers(&props)?; + let endpoint = self + .vended_credential_endpoint(rest, &client, &token, &headers, table) + .await?; + ( - OpenDalStorageFactory::Gcs, - Some(iceberg_catalog_rest::BearerTokenAuthenticator::new( - Arc::new(GcpTokenProvider { service_account }), - )), + // A GCP-hosted catalog vends GCS credentials, so the storage provider is not + // in question here the way it is for a generic REST catalog. + // + // NOTE: with delegation the service account key above stops governing storage + // and only authenticates the catalog. That is not a change in precedence: + // OpenDAL already preferred the vended token over a credential file, and a + // loader only keeps that token from expiring. + Self::gcs_storage_factory(endpoint, &client, &token, &headers), + Some(iceberg_catalog_rest::BearerTokenAuthenticator::new(token)), ) } }; - // `iceberg-rust` turns `header.*` props into headers on every REST request, so - // connection asked for delegation. - // than falling back to their configured storage credentials. + // Inserted after the storage factory is built, so that the loaders above carry only the + // headers the catalog client would send on an ordinary request. Each adds this one itself, + // since the credentials endpoint is the one request that always asks for delegation. if let Some(delegation) = &rest.access_delegation { props.insert( REST_CATALOG_PROP_ACCESS_DELEGATION.to_string(), @@ -3540,6 +3620,31 @@ impl AwsPrivatelinkConnection { mod tests { use super::*; + #[mz_ore::test] + fn test_catalog_headers() { + let props = BTreeMap::from_iter( + [ + (REST_CATALOG_PROP_URI, "https://catalog.example"), + (REST_CATALOG_PROP_WAREHOUSE, "wh"), + ("header.x-goog-user-project", "some-project"), + (REST_CATALOG_PROP_ACCESS_DELEGATION, "vended-credentials"), + ] + .map(|(k, v)| (k.to_string(), v.to_string())), + ); + + // Only `header.*` props become headers, and the prefix is stripped. Header names are + // matched case-insensitively, so the delegation prop's mixed-case spelling still lands. + let headers = IcebergCatalogConnection::catalog_headers(&props).expect("valid headers"); + assert_eq!(headers.len(), 2); + assert_eq!(headers["x-goog-user-project"], "some-project"); + assert_eq!(headers["x-iceberg-access-delegation"], "vended-credentials"); + + // A prop whose name is not a legal header is an error rather than a dropped header: it + // would otherwise mean silently talking to the catalog differently than asked. + let props = BTreeMap::from([("header.bad name".to_string(), "v".to_string())]); + assert!(IcebergCatalogConnection::catalog_headers(&props).is_err()); + } + #[mz_ore::test] fn test_check_service_name() { // Customer-owned and AWS-managed endpoint services are both accepted, diff --git a/src/storage-types/src/connections/iceberg_credentials.rs b/src/storage-types/src/connections/iceberg_credentials.rs index 0c446e8890bcc..bc839385e8240 100644 --- a/src/storage-types/src/connections/iceberg_credentials.rs +++ b/src/storage-types/src/connections/iceberg_credentials.rs @@ -21,14 +21,16 @@ //! builds the catalog but keeps it private, so this module asks the server directly. use std::collections::BTreeMap; +use std::fmt::Debug; use std::sync::Arc; use std::time::{Duration, Instant, SystemTime}; use anyhow::{Context, anyhow}; +use http::{HeaderMap, HeaderName, HeaderValue}; use iceberg::TableIdent; -use iceberg::io::{S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN}; +use iceberg::io::{GCS_TOKEN, S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY, S3_SESSION_TOKEN}; use iceberg_catalog_rest::{StorageCredential, TokenProvider}; -use iceberg_storage_opendal::{AwsCredential, ProvideCredential}; +use iceberg_storage_opendal::{AwsCredential, GcsCredential, GcsToken, ProvideCredential}; use mz_ore::error::ErrorExt; use reqsign_core::time::Timestamp; use reqwest::StatusCode; @@ -41,49 +43,152 @@ use crate::connections::IcebergAccessDelegation; /// The `X-Iceberg-Access-Delegation` header, spelled for requests Materialize issues itself /// rather than through the catalog client, which takes it as a `header.*` catalog property. -const ICEBERG_ACCESS_DELEGATION_HEADER: &str = "X-Iceberg-Access-Delegation"; +const ICEBERG_ACCESS_DELEGATION_HEADER: HeaderName = + HeaderName::from_static("x-iceberg-access-delegation"); /// Property name for the most common way catalogs report when vended S3 credentials expire. /// Catalogs are not required to send it. const S3_SESSION_TOKEN_EXPIRES_AT_MS: &str = "s3.session-token-expires-at-ms"; +/// Property name for when a vended GCS token expires, in epoch milliseconds. +/// +/// NOTE: no `-ms` suffix, unlike [`S3_SESSION_TOKEN_EXPIRES_AT_MS`], even though both carry +/// milliseconds. The spelling comes from Iceberg's `GCPProperties`, not from analogy with the S3 +/// property. +const GCS_TOKEN_EXPIRES_AT: &str = "gcs.oauth2.token-expires-at"; + /// How far ahead of a reported expiry to re-fetch a vended credential. +/// +/// NOTE: must stay comfortably above reqsign's own refresh buffers, which are 120s for a GCS +/// token. A GCS token handed back inside that window fails signing outright, because a vended +/// credential carries no service account for reqsign to fall back on. const VENDED_CREDENTIAL_REFRESH_BUFFER: Duration = Duration::from_secs(900); /// How long to trust a vended credential that reports no expiry. /// -/// A catalog that omits [`S3_SESSION_TOKEN_EXPIRES_AT_MS`] leaves nothing to schedule against, and -/// a credential held past its real lifetime fails every S3 request until the dataflow restarts. So +/// A catalog that omits the expiry property leaves nothing to schedule against, and a credential +/// held past its real lifetime fails every storage request until the dataflow restarts. So /// re-fetch on a short interval instead: each one is a single REST call against a credential the /// sink is already using. // TODO SS-449: make this a dyncfg const VENDED_CREDENTIAL_DEFAULT_TTL: Duration = Duration::from_secs(300); +/// A storage credential that a REST catalog can vend. +/// +/// Implementors carry the object store's notion of a credential and know which +/// `storage-credentials` properties encode it. Everything else about vending, including the fetch, +/// the caching, and the refresh schedule, is shared by [`VendedCredentialLoader`]. +pub(super) trait VendedCredential: Clone + Debug + Send + Sync + Unpin + 'static { + /// The object store this credential authenticates against, for diagnostics. + const STORE: &'static str; + + /// Builds the credential from one entry of a catalog's `storage-credentials`. + fn from_vended(vended: &StorageCredential) -> reqsign_core::Result; + + /// When the credential expires, if the catalog reported it. + fn expires_at(&self) -> Option; +} + +impl VendedCredential for AwsCredential { + const STORE: &'static str = "S3"; + + fn from_vended(vended: &StorageCredential) -> reqsign_core::Result { + Ok(AwsCredential { + access_key_id: required(vended, S3_ACCESS_KEY_ID)?, + secret_access_key: required(vended, S3_SECRET_ACCESS_KEY)?, + session_token: vended.config.get(S3_SESSION_TOKEN).cloned(), + expires_in: vended_expires_at(vended, S3_SESSION_TOKEN_EXPIRES_AT_MS)?, + }) + } + + fn expires_at(&self) -> Option { + self.expires_in + } +} + +impl VendedCredential for GcsCredential { + const STORE: &'static str = "GCS"; + + fn from_vended(vended: &StorageCredential) -> reqsign_core::Result { + Ok(GcsCredential::with_token(GcsToken { + access_token: required(vended, GCS_TOKEN)?, + expires_at: vended_expires_at(vended, GCS_TOKEN_EXPIRES_AT)?, + })) + } + + fn expires_at(&self) -> Option { + self.token.as_ref().and_then(|token| token.expires_at) + } +} + +/// Reads a property a vended credential cannot do without. +fn required(vended: &StorageCredential, prop: &str) -> reqsign_core::Result { + vended.config.get(prop).cloned().ok_or_else(|| { + reqsign_core::Error::credential_invalid(format!( + "vended Iceberg storage credential for prefix {} is missing {prop}", + vended.prefix + )) + }) +} + +/// Reads an expiry property, which catalogs are not required to send. +fn vended_expires_at( + vended: &StorageCredential, + prop: &str, +) -> reqsign_core::Result> { + let Some(raw) = vended.config.get(prop) else { + debug!( + prefix = vended.prefix, + prop, "Iceberg catalog vended a storage credential with no expiry" + ); + return Ok(None); + }; + let millis = raw.parse::().map_err(|e| { + reqsign_core::Error::credential_invalid(format!( + "vended Iceberg storage credential for prefix {} has an unparseable {prop}", + vended.prefix + )) + .with_source(e) + })?; + Ok(Some(Timestamp::from_millisecond(millis)?)) +} + #[derive(Debug)] -pub(super) struct VendedCredentialLoader { +pub(super) struct VendedCredentialLoader { client: reqwest::Client, credential_endpoint: Url, token: Arc, - cached: Mutex>, + /// Every header this request needs beyond the bearer token, resolved once. + headers: HeaderMap, + cached: Mutex>, } -impl VendedCredentialLoader { +impl VendedCredentialLoader { + /// `headers` are the headers the catalog client puts on its own requests, which this one + /// bypasses. The delegation header is added here rather than expected in `headers`, since + /// asking the catalog to vend is the whole point of this request. pub(super) fn new( client: reqwest::Client, credential_endpoint: Url, token: Arc, + mut headers: HeaderMap, ) -> Self { + headers.insert( + ICEBERG_ACCESS_DELEGATION_HEADER, + HeaderValue::from_static(IcebergAccessDelegation::VendedCredentials.as_header_value()), + ); Self { client, credential_endpoint, token, + headers, cached: Mutex::new(None), } } /// Fetches a fresh credential from the catalog, paired with the instant at which it should be /// re-fetched. - async fn fetch(&self) -> reqsign_core::Result<(AwsCredential, Instant)> { + async fn fetch(&self) -> reqsign_core::Result<(C, Instant)> { let token = self.token.token().await.map_err(|e| { reqsign_core::Error::credential_invalid( "failed to obtain a catalog token for vended Iceberg storage credentials", @@ -95,10 +200,7 @@ impl VendedCredentialLoader { .client .get(self.credential_endpoint.clone()) .bearer_auth(token) - .header( - ICEBERG_ACCESS_DELEGATION_HEADER, - IcebergAccessDelegation::VendedCredentials.as_header_value(), - ) + .headers(self.headers.clone()) .send() .await .map_err(|e| { @@ -149,7 +251,7 @@ impl VendedCredentialLoader { "Iceberg catalog vended multiple storage credentials; using the longest prefix" ); } - let credential = response + let vended = response .storage_credentials .into_iter() .max_by_key(|credential| credential.prefix.len()) @@ -160,53 +262,14 @@ impl VendedCredentialLoader { )) })?; - let missing = |prop: &str| { - reqsign_core::Error::credential_invalid(format!( - "vended Iceberg storage credential for prefix {} is missing {prop}", - credential.prefix - )) - }; - let access_key_id = credential - .config - .get(S3_ACCESS_KEY_ID) - .ok_or_else(|| missing(S3_ACCESS_KEY_ID))? - .clone(); - let secret_access_key = credential - .config - .get(S3_SECRET_ACCESS_KEY) - .ok_or_else(|| missing(S3_SECRET_ACCESS_KEY))? - .clone(); - - let expires_in = credential - .config - .get(S3_SESSION_TOKEN_EXPIRES_AT_MS) - .map(|raw| { - let millis = raw.parse::().map_err(|e| { - reqsign_core::Error::credential_invalid(format!( - "vended Iceberg storage credential for prefix {} has an unparseable \ - {S3_SESSION_TOKEN_EXPIRES_AT_MS}", - credential.prefix - )) - .with_source(e) - })?; - Timestamp::from_millisecond(millis) - }) - .transpose()?; - - Ok(( - AwsCredential { - access_key_id, - secret_access_key, - session_token: credential.config.get(S3_SESSION_TOKEN).cloned(), - expires_in, - }, - refresh_deadline(expires_in), - )) + let credential = C::from_vended(&vended)?; + let refresh_at = refresh_deadline(credential.expires_at()); + Ok((credential, refresh_at)) } } -impl ProvideCredential for VendedCredentialLoader { - type Credential = AwsCredential; +impl ProvideCredential for VendedCredentialLoader { + type Credential = C; async fn provide_credential( &self, @@ -215,7 +278,7 @@ impl ProvideCredential for VendedCredentialLoader { // The lock is deliberately held across the fetch. `create_operator` builds a fresh // OpenDAL `Operator` for every file operation, so reqsign's own credential cache never // outlives a single call and this cache is all that stands between the sink and one - // catalog round trip per S3 request. Serializing here means a stale entry costs one + // catalog round trip per storage request. Serializing here means a stale entry costs one // refetch rather than one per in-flight operation. let mut cached = self.cached.lock().await; @@ -225,7 +288,16 @@ impl ProvideCredential for VendedCredentialLoader { return Ok(Some(credential.clone())); } - let (credential, refresh_at) = self.fetch().await?; + // The chain reqsign consults reports only that credential loading failed, without a + // cause, so log the real one here or it is lost. + let (credential, refresh_at) = self.fetch().await.inspect_err(|e| { + warn!( + store = C::STORE, + endpoint = %self.credential_endpoint, + error = %e, + "failed to refresh vended Iceberg storage credentials" + ); + })?; *cached = Some((credential.clone(), refresh_at)); Ok(Some(credential)) } @@ -327,11 +399,13 @@ fn table_credentials_url( /// Resolves the REST endpoint that vends storage credentials for `table`. /// /// Takes a round trip to the catalog's `config` endpoint, because the resource path carries a -/// request prefix that only the server knows. +/// request prefix that only the server knows. `headers` are the headers the catalog client puts +/// on its own requests, which this one bypasses. pub(super) async fn table_credentials_endpoint( uri: &Url, client: &reqwest::Client, token: &Arc, + headers: &HeaderMap, warehouse: Option<&str>, table: &TableIdent, ) -> Result { @@ -344,6 +418,7 @@ pub(super) async fn table_credentials_endpoint( let response = client .get(config_endpoint.clone()) .bearer_auth(bearer) + .headers(headers.clone()) .send() .await .with_context(|| { @@ -389,6 +464,131 @@ mod tests { } } + fn vended(prefix: &str, config: &[(&str, &str)]) -> StorageCredential { + StorageCredential { + prefix: prefix.to_string(), + config: config + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + } + } + + #[mz_ore::test] + fn test_aws_credential_from_vended() { + let credential = AwsCredential::from_vended(&vended( + "s3://bucket/warehouse", + &[ + (S3_ACCESS_KEY_ID, "key"), + (S3_SECRET_ACCESS_KEY, "secret"), + (S3_SESSION_TOKEN, "session"), + (S3_SESSION_TOKEN_EXPIRES_AT_MS, "1700000000000"), + ], + )) + .expect("complete credential"); + assert_eq!(credential.access_key_id, "key"); + assert_eq!(credential.secret_access_key, "secret"); + assert_eq!(credential.session_token.as_deref(), Some("session")); + assert_eq!( + credential.expires_at(), + Some(Timestamp::from_millisecond(1700000000000).unwrap()) + ); + + // A catalog need not report an expiry, and long-lived static keys have no session token. + let credential = AwsCredential::from_vended(&vended( + "s3://bucket/warehouse", + &[(S3_ACCESS_KEY_ID, "key"), (S3_SECRET_ACCESS_KEY, "secret")], + )) + .expect("credential without expiry"); + assert_eq!(credential.session_token, None); + assert_eq!(credential.expires_at(), None); + + // A credential missing either key is unusable, and says which one. + let err = AwsCredential::from_vended(&vended( + "s3://bucket/warehouse", + &[(S3_ACCESS_KEY_ID, "key")], + )) + .expect_err("incomplete credential"); + assert!( + err.to_string().contains(S3_SECRET_ACCESS_KEY), + "unexpected error: {err}" + ); + + // A malformed expiry is an error rather than a silently ignored property: treating it as + // absent would fall back to the default TTL and hide a catalog bug. + let err = AwsCredential::from_vended(&vended( + "s3://bucket/warehouse", + &[ + (S3_ACCESS_KEY_ID, "key"), + (S3_SECRET_ACCESS_KEY, "secret"), + (S3_SESSION_TOKEN_EXPIRES_AT_MS, "soon"), + ], + )) + .expect_err("unparseable expiry"); + assert!( + err.to_string().contains("unparseable"), + "unexpected error: {err}" + ); + } + + #[mz_ore::test] + fn test_gcs_credential_from_vended() { + let credential = GcsCredential::from_vended(&vended( + "gs://bucket/warehouse", + &[ + (GCS_TOKEN, "ya29.token"), + (GCS_TOKEN_EXPIRES_AT, "1700000000000"), + ], + )) + .expect("complete credential"); + // A vended credential is a bare token: reqsign has no service account to fall back on. + assert!(credential.service_account.is_none()); + let token = credential.token.as_ref().expect("token"); + assert_eq!(token.access_token, "ya29.token"); + assert_eq!( + credential.expires_at(), + Some(Timestamp::from_millisecond(1700000000000).unwrap()) + ); + + let credential = + GcsCredential::from_vended(&vended("gs://bucket/warehouse", &[(GCS_TOKEN, "tok")])) + .expect("credential without expiry"); + assert_eq!(credential.expires_at(), None); + + let err = GcsCredential::from_vended(&vended( + "gs://bucket/warehouse", + &[(GCS_TOKEN_EXPIRES_AT, "1700000000000")], + )) + .expect_err("credential without a token"); + assert!( + err.to_string().contains(GCS_TOKEN), + "unexpected error: {err}" + ); + } + + #[mz_ore::test] + fn test_refresh_deadline() { + // No reported expiry falls back to the fixed interval. + let deadline = refresh_deadline(None); + assert!(deadline > Instant::now()); + assert!(deadline <= Instant::now() + VENDED_CREDENTIAL_DEFAULT_TTL); + + // A comfortably distant expiry is refreshed a buffer ahead of it. + let lifetime = Duration::from_secs(3600); + let deadline = refresh_deadline(Some(Timestamp::now() + lifetime)); + let expected = Instant::now() + lifetime - VENDED_CREDENTIAL_REFRESH_BUFFER; + assert!( + deadline <= expected && deadline > expected - Duration::from_secs(60), + "deadline is not a buffer ahead of the expiry" + ); + + // An expiry inside the buffer, or already past, is refreshed on the next call rather + // than yielding a deadline reqsign would reject anyway. + for offset in [VENDED_CREDENTIAL_REFRESH_BUFFER / 2, Duration::from_secs(0)] { + assert!(refresh_deadline(Some(Timestamp::now() + offset)) <= Instant::now()); + } + } + #[mz_ore::test] fn test_announced_prefix() { // Overrides win, matching how the catalog client merges the two. From 1f3321bd819c24b8a9adf052afc82a80e315c4ed Mon Sep 17 00:00:00 2001 From: Patrick Butler Date: Wed, 9 Sep 2026 16:19:58 -0400 Subject: [PATCH 2/5] remove restriction on access delegation option for GCP catalog connection --- src/sql/src/plan/statement/ddl/connection.rs | 5 ----- 1 file changed, 5 deletions(-) diff --git a/src/sql/src/plan/statement/ddl/connection.rs b/src/sql/src/plan/statement/ddl/connection.rs index eccbfed25b732..1ec8ed6b178f4 100644 --- a/src/sql/src/plan/statement/ddl/connection.rs +++ b/src/sql/src/plan/statement/ddl/connection.rs @@ -756,11 +756,6 @@ impl ConnectionOptionExtracted { "invalid CONNECTION: OAUTH2 SERVER URL applies to CREDENTIAL auth, not GCP CONNECTION" ); } - if self.access_delegation.is_some() { - sql_bail!( - "invalid CONNECTION: ICEBERG GCP CONNECTION does not support ACCESS DELEGATION" - ); - } // A GCP connection authenticates to GCS, so the // store is already determined. Reject a // contradicting value rather than silently From 69e7418a94bdae7c2b2b9c9671de26416587f525 Mon Sep 17 00:00:00 2001 From: Patrick Butler Date: Wed, 9 Sep 2026 16:34:48 -0400 Subject: [PATCH 3/5] update cargo --- Cargo.lock | 3 --- Cargo.toml | 6 +++--- 2 files changed, 3 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 728ffa7969f86..b7d9d56ea21f1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4934,7 +4934,6 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.1" -source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=958ac5b3c318026aeb362485c593e4956aeabe95#958ac5b3c318026aeb362485c593e4956aeabe95" dependencies = [ "aes-gcm", "anyhow", @@ -4991,7 +4990,6 @@ dependencies = [ [[package]] name = "iceberg-catalog-rest" version = "0.10.1" -source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=958ac5b3c318026aeb362485c593e4956aeabe95#958ac5b3c318026aeb362485c593e4956aeabe95" dependencies = [ "async-trait", "chrono", @@ -5010,7 +5008,6 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.1" -source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=958ac5b3c318026aeb362485c593e4956aeabe95#958ac5b3c318026aeb362485c593e4956aeabe95" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 4da5cc64393e3..604daba080d41 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -670,9 +670,9 @@ async-compression = { git = "https://github.com/MaterializeInc/async-compression # NOTE: The `[workspace.dependencies]` version requirement above must stay # semver-compatible with this revision's crate version, otherwise Cargo drops # these patches into `[[patch.unused]]` and silently builds against crates.io. -iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "958ac5b3c318026aeb362485c593e4956aeabe95" } -iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "958ac5b3c318026aeb362485c593e4956aeabe95" } -iceberg-storage-opendal = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "958ac5b3c318026aeb362485c593e4956aeabe95" } +iceberg = { path = "/Users/patrickbutler/Documents/code/iceberg-rust/crates/iceberg" } +iceberg-catalog-rest = { path = "/Users/patrickbutler/Documents/code/iceberg-rust/crates/catalog/rest" } +iceberg-storage-opendal = { path = "/Users/patrickbutler/Documents/code/iceberg-rust/crates/storage/opendal" } # Custom duckdb crate to support mz needs # All changes should go to the `mz_changes` branch. From eba7f20c2bdf39e0253acf07bcfb4683c664c54d Mon Sep 17 00:00:00 2001 From: Patrick Butler Date: Thu, 10 Sep 2026 13:01:18 -0400 Subject: [PATCH 4/5] update cargo to use iceberg fork rev --- Cargo.lock | 3 +++ Cargo.toml | 6 +++--- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index b7d9d56ea21f1..a919d246e4f62 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4934,6 +4934,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.1" +source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=f789d90ff4b516331ba07afc2b9790b20fb3db2a#f789d90ff4b516331ba07afc2b9790b20fb3db2a" dependencies = [ "aes-gcm", "anyhow", @@ -4990,6 +4991,7 @@ dependencies = [ [[package]] name = "iceberg-catalog-rest" version = "0.10.1" +source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=f789d90ff4b516331ba07afc2b9790b20fb3db2a#f789d90ff4b516331ba07afc2b9790b20fb3db2a" dependencies = [ "async-trait", "chrono", @@ -5008,6 +5010,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.1" +source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=f789d90ff4b516331ba07afc2b9790b20fb3db2a#f789d90ff4b516331ba07afc2b9790b20fb3db2a" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 604daba080d41..dc89234e75eec 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -670,9 +670,9 @@ async-compression = { git = "https://github.com/MaterializeInc/async-compression # NOTE: The `[workspace.dependencies]` version requirement above must stay # semver-compatible with this revision's crate version, otherwise Cargo drops # these patches into `[[patch.unused]]` and silently builds against crates.io. -iceberg = { path = "/Users/patrickbutler/Documents/code/iceberg-rust/crates/iceberg" } -iceberg-catalog-rest = { path = "/Users/patrickbutler/Documents/code/iceberg-rust/crates/catalog/rest" } -iceberg-storage-opendal = { path = "/Users/patrickbutler/Documents/code/iceberg-rust/crates/storage/opendal" } +iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "f789d90ff4b516331ba07afc2b9790b20fb3db2a" } +iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "f789d90ff4b516331ba07afc2b9790b20fb3db2a" } +iceberg-storage-opendal = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "f789d90ff4b516331ba07afc2b9790b20fb3db2a" } # Custom duckdb crate to support mz needs # All changes should go to the `mz_changes` branch. From 48cb251847d53fee9b03b9dcab1ae027b81c29d5 Mon Sep 17 00:00:00 2001 From: Patrick Butler Date: Thu, 10 Sep 2026 16:15:42 -0400 Subject: [PATCH 5/5] updated iceberg-rust revision after merging PR in fork --- Cargo.lock | 6 +++--- Cargo.toml | 6 +++--- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a919d246e4f62..3c48348701ebf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4934,7 +4934,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.1" -source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=f789d90ff4b516331ba07afc2b9790b20fb3db2a#f789d90ff4b516331ba07afc2b9790b20fb3db2a" +source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=985929c6c5cc7083bac30c881e7ec43ec4d57b64#985929c6c5cc7083bac30c881e7ec43ec4d57b64" dependencies = [ "aes-gcm", "anyhow", @@ -4991,7 +4991,7 @@ dependencies = [ [[package]] name = "iceberg-catalog-rest" version = "0.10.1" -source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=f789d90ff4b516331ba07afc2b9790b20fb3db2a#f789d90ff4b516331ba07afc2b9790b20fb3db2a" +source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=985929c6c5cc7083bac30c881e7ec43ec4d57b64#985929c6c5cc7083bac30c881e7ec43ec4d57b64" dependencies = [ "async-trait", "chrono", @@ -5010,7 +5010,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.1" -source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=f789d90ff4b516331ba07afc2b9790b20fb3db2a#f789d90ff4b516331ba07afc2b9790b20fb3db2a" +source = "git+https://github.com/MaterializeInc/iceberg-rust.git?rev=985929c6c5cc7083bac30c881e7ec43ec4d57b64#985929c6c5cc7083bac30c881e7ec43ec4d57b64" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index dc89234e75eec..abc56811708d3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -670,9 +670,9 @@ async-compression = { git = "https://github.com/MaterializeInc/async-compression # NOTE: The `[workspace.dependencies]` version requirement above must stay # semver-compatible with this revision's crate version, otherwise Cargo drops # these patches into `[[patch.unused]]` and silently builds against crates.io. -iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "f789d90ff4b516331ba07afc2b9790b20fb3db2a" } -iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "f789d90ff4b516331ba07afc2b9790b20fb3db2a" } -iceberg-storage-opendal = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "f789d90ff4b516331ba07afc2b9790b20fb3db2a" } +iceberg = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "985929c6c5cc7083bac30c881e7ec43ec4d57b64" } +iceberg-catalog-rest = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "985929c6c5cc7083bac30c881e7ec43ec4d57b64" } +iceberg-storage-opendal = { git = "https://github.com/MaterializeInc/iceberg-rust.git", rev = "985929c6c5cc7083bac30c881e7ec43ec4d57b64" } # Custom duckdb crate to support mz needs # All changes should go to the `mz_changes` branch.