diff --git a/Cargo.lock b/Cargo.lock index 859704c5a2..9a28aa0f20 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2989,6 +2989,7 @@ dependencies = [ "futures", "http 1.5.0", "iceberg", + "iceberg-property-macro", "iceberg_test_utils", "itertools 0.13.0", "mockito", diff --git a/crates/catalog/rest/Cargo.toml b/crates/catalog/rest/Cargo.toml index a43970bf7f..d4b734cdaa 100644 --- a/crates/catalog/rest/Cargo.toml +++ b/crates/catalog/rest/Cargo.toml @@ -42,6 +42,7 @@ base64 = { workspace = true, optional = true } chrono = { workspace = true } http = { workspace = true } iceberg = { workspace = true } +iceberg-property-macro = { workspace = true } itertools = { workspace = true } moka = { workspace = true, features = ["sync"] } reqwest = { workspace = true } diff --git a/crates/catalog/rest/public-api.txt b/crates/catalog/rest/public-api.txt index a0c6fd03ce..52074195e6 100644 --- a/crates/catalog/rest/public-api.txt +++ b/crates/catalog/rest/public-api.txt @@ -403,11 +403,18 @@ impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::UpdateNames pub fn iceberg_catalog_rest::UpdateNamespacePropertiesResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> 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_AUDIENCE: &str pub const iceberg_catalog_rest::REST_CATALOG_PROP_AUTH_TYPE: &str +pub const iceberg_catalog_rest::REST_CATALOG_PROP_CREDENTIAL: &str pub const iceberg_catalog_rest::REST_CATALOG_PROP_DISABLE_HEADER_REDACTION: &str +pub const iceberg_catalog_rest::REST_CATALOG_PROP_OAUTH2_SERVER_URI: &str pub const iceberg_catalog_rest::REST_CATALOG_PROP_PAGE_SIZE: &str +pub const iceberg_catalog_rest::REST_CATALOG_PROP_PREFIX: &str +pub const iceberg_catalog_rest::REST_CATALOG_PROP_RESOURCE: &str +pub const iceberg_catalog_rest::REST_CATALOG_PROP_SCOPE: &str pub const iceberg_catalog_rest::REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS: &str pub const iceberg_catalog_rest::REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES: &str +pub const iceberg_catalog_rest::REST_CATALOG_PROP_TOKEN: &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 diff --git a/crates/catalog/rest/src/auth/oauth2.rs b/crates/catalog/rest/src/auth/oauth2.rs index aad91996a8..e417e8f94d 100644 --- a/crates/catalog/rest/src/auth/oauth2.rs +++ b/crates/catalog/rest/src/auth/oauth2.rs @@ -28,8 +28,7 @@ use tokio::sync::Mutex; use super::{AuthManager, AuthSession, HttpRequest}; use crate::catalog::{ - REST_CATALOG_PROP_URI, RestCatalogConfig, credential_from_props, default_token_endpoint, - explicit_headers_from_props, + OAuth2Credential, REST_CATALOG_PROP_URI, RestCatalogProperties, default_token_endpoint, }; use crate::client::HttpClient; use crate::types::{ErrorResponse, TokenResponse}; @@ -38,7 +37,7 @@ use crate::types::{ErrorResponse, TokenResponse}; struct OAuth2Params { extra_headers: HeaderMap, token_endpoint: String, - credential: Option<(Option, SensitiveString)>, + credential: Option, extra_oauth_params: HashMap, } @@ -87,7 +86,7 @@ impl OAuth2Manager { /// Sets the client credential exchanged for a token at the token endpoint. pub fn with_credential(mut self, client_id: Option, client_secret: String) -> Self { - self.init_params.credential = Some((client_id, client_secret.into())); + self.init_params.credential = Some(OAuth2Credential::new(client_id, client_secret.into())); self } @@ -105,17 +104,17 @@ impl OAuth2Manager { self } - pub(crate) fn from_config(cfg: &RestCatalogConfig) -> Result { - Ok(Self { - token: Arc::new(Mutex::new(cfg.token().map(SensitiveString::from))), + pub(crate) fn from_properties(cfg: &RestCatalogProperties) -> Self { + Self { + token: Arc::new(Mutex::new(cfg.token().clone())), init_params: OAuth2Params { - extra_headers: cfg.extra_headers()?, + extra_headers: cfg.extra_headers(), token_endpoint: cfg.get_token_endpoint(), - credential: cfg.credential().map(|(id, secret)| (id, secret.into())), + credential: cfg.credential().clone(), extra_oauth_params: cfg.extra_oauth_params(), }, - endpoint_is_default: cfg.explicit_oauth2_server_uri().is_none(), - }) + endpoint_is_default: cfg.oauth2_server_uri().is_none(), + } } } @@ -156,34 +155,31 @@ impl OAuth2Manager { client: &HttpClient, props: &HashMap, ) -> Result { + let properties = RestCatalogProperties::for_handshake(props)?; // The properties may carry a new token (or restate the user's). - if let Some(token) = props.get("token") { - *self.token.lock().await = Some(SensitiveString::from(token.clone())); + if let Some(token) = properties.token() { + *self.token.lock().await = Some(token.clone()); } let mut extra_headers = self.init_params.extra_headers.clone(); - extra_headers.extend(explicit_headers_from_props(props)?); + extra_headers.extend(properties.headers().0.clone()); let mut extra_oauth_params = self.init_params.extra_oauth_params.clone(); - for key in ["scope", "audience", "resource"] { - if let Some(value) = props.get(key) { - extra_oauth_params.insert(key.to_string(), value.to_string()); - } - } + extra_oauth_params.extend(properties.explicit_oauth_params()); - let token_endpoint = match props.get("oauth2-server-uri") { + let token_endpoint = match properties.oauth2_server_uri() { Some(uri) if !uri.is_empty() => uri.clone(), - // A default endpoint follows the merged catalog URI (which - // `/v1/config` may have overridden); explicit ones are kept. - _ if self.endpoint_is_default => props - .get(REST_CATALOG_PROP_URI) - .map(|uri| default_token_endpoint(uri)) - .unwrap_or_else(|| self.init_params.token_endpoint.clone()), + // A default endpoint follows the merged catalog URI; an injected + // manager keeps its explicit endpoint. + _ if self.endpoint_is_default && props.contains_key(REST_CATALOG_PROP_URI) => { + default_token_endpoint(properties.uri()) + } _ => self.init_params.token_endpoint.clone(), }; - let credential = credential_from_props(props) - .map(|(id, secret)| (id, secret.into())) + let credential = properties + .credential() + .clone() .or_else(|| self.init_params.credential.clone()); Ok(OAuth2Session { @@ -247,7 +243,7 @@ enum TokenSource { struct ClientCredentialsConfig { client: HttpClient, - credential: (Option, SensitiveString), + credential: OAuth2Credential, token_endpoint: String, extra_headers: HeaderMap, extra_oauth_params: HashMap, @@ -265,14 +261,12 @@ impl Debug for OAuth2Session { impl ClientCredentialsConfig { async fn exchange_credential_for_token(&self) -> Result { - let (client_id, client_secret) = &self.credential; - let mut params = HashMap::with_capacity(4); params.insert("grant_type", "client_credentials"); - if let Some(client_id) = client_id { + if let Some(client_id) = self.credential.client_id() { params.insert("client_id", client_id); } - params.insert("client_secret", client_secret.expose()); + params.insert("client_secret", self.credential.client_secret().expose()); params.extend( self.extra_oauth_params .iter() @@ -348,15 +342,14 @@ mod tests { use reqwest::Client; use super::*; - use crate::RestCatalogConfig; + use crate::catalog::RestCatalogTestFixture; fn test_client() -> HttpClient { - HttpClient::new( - &RestCatalogConfig::builder() - .uri("http://localhost".to_string()) - .build(), - ) - .unwrap() + RestCatalogTestFixture::builder() + .uri("http://localhost".to_string()) + .build() + .http_client() + .unwrap() } #[tokio::test] diff --git a/crates/catalog/rest/src/catalog.rs b/crates/catalog/rest/src/catalog.rs index 8b8ed983de..602ea477a2 100644 --- a/crates/catalog/rest/src/catalog.rs +++ b/crates/catalog/rest/src/catalog.rs @@ -30,18 +30,19 @@ use iceberg::encryption::kms::{KeyManagementClient, KmsClientFactory}; use iceberg::io::{ ADLS_SAS_TOKEN, FileIO, FileIOBuilder, GCS_TOKEN, S3_SESSION_TOKEN, StorageFactory, }; +use iceberg::sensitive::SensitiveString; use iceberg::table::Table; use iceberg::{ Catalog, CatalogBuilder, Error, ErrorKind, Namespace, NamespaceIdent, Result, Runtime, SessionCatalog, SessionContext, TableCommit, TableCreation, TableIdent, }; +use iceberg_property_macro::Properties; use itertools::Itertools; use reqwest::header::{ HeaderMap, HeaderName, HeaderValue, {self}, }; use reqwest::{Client, Method, StatusCode, Url}; use tokio::sync::OnceCell; -use typed_builder::TypedBuilder; use crate::auth::{ AUTH_TYPE_NONE, AUTH_TYPE_OAUTH2, AuthManager, AuthSession, NoopAuthManager, OAuth2Manager, @@ -62,6 +63,20 @@ use crate::types::{ pub const REST_CATALOG_PROP_URI: &str = "uri"; /// REST catalog warehouse location pub const REST_CATALOG_PROP_WAREHOUSE: &str = "warehouse"; +/// REST catalog endpoint prefix. +pub const REST_CATALOG_PROP_PREFIX: &str = "prefix"; +/// Bearer token used to authenticate REST catalog requests. +pub const REST_CATALOG_PROP_TOKEN: &str = "token"; +/// OAuth2 client secret, optionally prefixed by a client ID and colon. +pub const REST_CATALOG_PROP_CREDENTIAL: &str = "credential"; +/// OAuth2 token endpoint URI. +pub const REST_CATALOG_PROP_OAUTH2_SERVER_URI: &str = "oauth2-server-uri"; +/// OAuth2 scope requested when exchanging credentials. +pub const REST_CATALOG_PROP_SCOPE: &str = "scope"; +/// OAuth2 audience requested when exchanging credentials. +pub const REST_CATALOG_PROP_AUDIENCE: &str = "audience"; +/// OAuth2 resource requested when exchanging credentials. +pub const REST_CATALOG_PROP_RESOURCE: &str = "resource"; /// Requested page size for namespace and table listings; see /// [`RestSessionCatalogBuilder::with_page_size`]. pub const REST_CATALOG_PROP_PAGE_SIZE: &str = "rest-page-size"; @@ -83,6 +98,8 @@ pub const REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES: &str = "rest-table-cache.ma pub const REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS: &str = "rest-table-cache.expire-after-write-ms"; +const REST_CATALOG_PROP_HEADER_PREFIX: &str = "header."; + const TABLE_CACHE_MAX_ENTRIES_DEFAULT: u64 = 100; const TABLE_CACHE_EXPIRE_AFTER_WRITE_MS_DEFAULT: u64 = 5 * 60 * 1000; /// moka panics on a time-to-live beyond 1000 years. @@ -181,72 +198,164 @@ impl RestCatalogBuilder { } } -/// Rest catalog configuration. -#[derive(Clone, TypedBuilder)] -pub(crate) struct RestCatalogConfig { - #[builder(default, setter(strip_option))] - name: Option, - +/// Typed REST catalog properties. +#[derive(Clone, Properties)] +#[cfg_attr(test, derive(PartialEq, Eq))] +pub(crate) struct RestCatalogProperties { + /// Catalog URI. + #[property(key = REST_CATALOG_PROP_URI, default = "", getter)] uri: String, - - #[builder(default, setter(strip_option(fallback = warehouse_opt)))] + /// Warehouse location or identifier. + #[property(key = REST_CATALOG_PROP_WAREHOUSE, default = None, getter)] warehouse: Option, + /// Catalog endpoint prefix. + #[property(key = REST_CATALOG_PROP_PREFIX, default = None, getter)] + prefix: Option, + /// Explicit authentication scheme. + #[property(key = REST_CATALOG_PROP_AUTH_TYPE, default = None, getter)] + auth_type: Option, + /// Configured bearer token. + #[property(key = REST_CATALOG_PROP_TOKEN, default = None, getter)] + token: Option, + /// Configured OAuth2 client ID and secret. + #[property(key = REST_CATALOG_PROP_CREDENTIAL, default = None, getter)] + credential: Option, + /// Explicit OAuth2 token endpoint. + #[property(key = REST_CATALOG_PROP_OAUTH2_SERVER_URI, default = None, getter)] + oauth2_server_uri: Option, + /// Explicit OAuth2 scope. + #[property(key = REST_CATALOG_PROP_SCOPE, default = None, getter)] + scope: Option, + /// OAuth2 audience. + #[property(key = REST_CATALOG_PROP_AUDIENCE, default = None, getter)] + audience: Option, + /// OAuth2 resource. + #[property(key = REST_CATALOG_PROP_RESOURCE, default = None, getter)] + resource: Option, + /// Explicit headers added to requests. + #[property(nested, getter)] + headers: RestHeaders, + /// Whether header redaction is disabled. + #[property( + key = REST_CATALOG_PROP_DISABLE_HEADER_REDACTION, + default = false, + parse_with = parse_header_redaction, + getter + )] + disable_header_redaction: bool, + /// Requested page size for namespace and table listings. + #[property(key = REST_CATALOG_PROP_PAGE_SIZE, default = None, parse_with = parse_page_size, getter)] + page_size: Option, + /// Maximum number of cached tables. + #[property(key = REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES, default = TABLE_CACHE_MAX_ENTRIES_DEFAULT, getter)] + table_cache_max_entries: u64, + /// How long a loaded table stays cached. + #[property( + key = REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS, + default = Duration::from_millis(TABLE_CACHE_EXPIRE_AFTER_WRITE_MS_DEFAULT), + parse_with = parse_cache_expiry, + getter + )] + table_cache_expire_after_write: Duration, +} - #[builder(default)] - props: HashMap, +/// OAuth2 client credentials. +#[derive(Clone, Debug)] +#[cfg_attr(test, derive(PartialEq, Eq))] +pub(crate) struct OAuth2Credential { + client_id: Option, + client_secret: SensitiveString, +} - #[builder(default)] - client: Option, +impl OAuth2Credential { + /// Creates a credential from a client ID and secret. + pub(crate) fn new(client_id: Option, client_secret: SensitiveString) -> Self { + Self { + client_id, + client_secret, + } + } + + /// Returns the optional client ID. + pub(crate) fn client_id(&self) -> Option<&str> { + self.client_id.as_deref() + } - /// Lazily-created default HTTP client, shared through clones of this - /// config so OAuth and catalog traffic reuse one connection pool - /// (matching the single-client behavior before the AuthManager refactor). - #[builder(default)] - default_client: Arc>, + /// Returns the client secret. + pub(crate) fn client_secret(&self) -> &SensitiveString { + &self.client_secret + } } -/// Property keys whose values are secrets, or may embed them (headers, -/// connection strings, keys like `adls.account-key` or `s3.sse.key`). -fn is_sensitive_prop(key: &str) -> bool { - key.contains("token") - || key.contains("credential") - || key.contains("secret") - || key.contains("password") - || key.contains("key") - || key.contains("connection-string") - || key.starts_with("header.") +impl FromStr for OAuth2Credential { + type Err = std::convert::Infallible; + + fn from_str(value: &str) -> std::result::Result { + Ok(match value.split_once(':') { + Some((id, secret)) => Self::new(Some(id.to_string()), secret.to_string().into()), + None => Self::new(None, value.to_string().into()), + }) + } } -/// Redacts secret property values: this config is printed by the derived -/// [`Debug`] implementations of [`RestSessionCatalog`] and [`RestCatalog`]. -impl Debug for RestCatalogConfig { +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) enum AuthType { + None, + OAuth2, + Custom(String), +} + +impl FromStr for AuthType { + type Err = std::convert::Infallible; + + fn from_str(value: &str) -> std::result::Result { + Ok(match value.to_ascii_lowercase().as_str() { + AUTH_TYPE_NONE => Self::None, + AUTH_TYPE_OAUTH2 => Self::OAuth2, + other => Self::Custom(other.to_string()), + }) + } +} + +impl Debug for RestCatalogProperties { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { - let props: HashMap<&str, &str> = self - .props - .iter() - .map(|(key, value)| { - let value = if is_sensitive_prop(key) { - "[REDACTED]" - } else { - value.as_str() - }; - (key.as_str(), value) - }) - .collect(); - f.debug_struct("RestCatalogConfig") - .field("name", &self.name) + f.debug_struct("RestCatalogProperties") .field("uri", &self.uri) .field("warehouse", &self.warehouse) - .field("props", &props) + .field("prefix", &self.prefix) + .field("auth_type", &self.auth_type) + .field("token", &self.token) + .field("credential", &self.credential) + .field("headers", &"[REDACTED]") + .field("page_size", &self.page_size) + .field("table_cache_max_entries", &self.table_cache_max_entries) + .field( + "table_cache_expire_after_write", + &self.table_cache_expire_after_write, + ) .finish_non_exhaustive() } } -impl RestCatalogConfig { +impl RestCatalogProperties { + pub(crate) fn for_handshake(props: &HashMap) -> Result { + // These settings are used only after the handshake. Server overrides may + // replace invalid client values before they are parsed. + let mut props = props.clone(); + for key in [ + REST_CATALOG_PROP_PAGE_SIZE, + REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES, + REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS, + ] { + props.remove(key); + } + Self::from_properties(&props) + } + fn url_prefixed(&self, parts: &[&str]) -> String { [&self.uri, PATH_V1] .into_iter() - .chain(self.props.get("prefix").map(|s| &**s)) + .chain(self.prefix().as_deref()) .chain(parts.iter().cloned()) .join("/") } @@ -256,37 +365,11 @@ impl RestCatalogConfig { } pub(crate) fn get_token_endpoint(&self) -> String { - self.explicit_oauth2_server_uri() + self.oauth2_server_uri + .clone() .unwrap_or_else(|| default_token_endpoint(&self.uri)) } - /// The `oauth2-server-uri` property, only when explicitly configured. - pub(crate) fn explicit_oauth2_server_uri(&self) -> Option { - self.props.get("oauth2-server-uri").cloned() - } - - /// Parse only the effective value, after server defaults and overrides are merged. - fn page_size(&self) -> Result> { - self.props - .get(REST_CATALOG_PROP_PAGE_SIZE) - .map(|value| { - let invalid_page_size = || { - Error::new( - ErrorKind::DataInvalid, - format!("Invalid {REST_CATALOG_PROP_PAGE_SIZE}: {value}"), - ) - }; - let page_size = value - .parse::() - .map_err(|err| invalid_page_size().with_source(err))?; - if page_size.get() > i32::MAX as u32 { - return Err(invalid_page_size()); - } - Ok(page_size) - }) - .transpose() - } - fn namespaces_endpoint(&self) -> String { self.url_prefixed(&["namespaces"]) } @@ -316,204 +399,140 @@ impl RestCatalogConfig { ]) } - /// The HTTP client: the configured one, or a lazily-created default that - /// is shared across every user of this config (and its clones), so token - /// and catalog requests keep sharing one connection pool. - pub(crate) fn client(&self) -> Client { - self.client - .clone() - .unwrap_or_else(|| self.default_client.get_or_init(Client::default).clone()) - } - - /// Get the token from the config. - /// - /// The client can use this token to send requests. - pub(crate) fn token(&self) -> Option { - self.props.get("token").cloned() - } - - /// Get the credentials from the config. The client can use these credentials to fetch a new - /// token. - pub(crate) fn credential(&self) -> Option<(Option, String)> { - credential_from_props(&self.props) - } - - /// Get the extra headers from config, see [`extra_headers_from_props`]. - pub(crate) fn extra_headers(&self) -> Result { - extra_headers_from_props(&self.props) + pub(crate) fn extra_headers(&self) -> HeaderMap { + let mut headers = HeaderMap::from_iter([ + ( + header::CONTENT_TYPE, + HeaderValue::from_static("application/json"), + ), + ( + HeaderName::from_static("x-client-version"), + HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION), + ), + ( + header::USER_AGENT, + HeaderValue::from_str(&format!("iceberg-rs/{CARGO_PKG_VERSION}")).unwrap(), + ), + ]); + headers.extend(self.headers.0.clone()); + headers } - /// Get the optional OAuth headers from the config. pub(crate) fn extra_oauth_params(&self) -> HashMap { - oauth_params_from_props(&self.props) - } - - /// Check if header redaction is disabled in error logs. - /// - /// Returns true if the `disable-header-redaction` property is set to "true". - /// Defaults to false for security (headers are redacted by default). - pub(crate) fn disable_header_redaction(&self) -> bool { - self.props - .get(REST_CATALOG_PROP_DISABLE_HEADER_REDACTION) - .map(|v| v.eq_ignore_ascii_case("true")) - .unwrap_or(false) - } - - /// Builds the table cache for freshness-aware loading, or `None` when - /// `rest-table-cache.max-entries` is 0. - fn table_cache(&self) -> Result> { - let parse = |key: &str, default: u64| -> Result { - self.props.get(key).map_or(Ok(default), |v| { - v.parse().map_err(|e| { - Error::new(ErrorKind::DataInvalid, format!("Invalid {key}: {v}")).with_source(e) - }) - }) - }; - let expire_after_write_ms = parse( - REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS, - TABLE_CACHE_EXPIRE_AFTER_WRITE_MS_DEFAULT, - )?; - let expire_after_write = Duration::from_millis(expire_after_write_ms); - // Zero is ambiguous: it could mean "disable" or "never expire". - if expire_after_write.is_zero() || expire_after_write > TABLE_CACHE_EXPIRE_AFTER_WRITE_MAX { - return Err(Error::new( - ErrorKind::DataInvalid, - format!( - "Invalid {REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS}: {expire_after_write_ms} (must be between 1 ms and 1000 years)" - ), - )); - } - let max_entries = parse( - REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES, - TABLE_CACHE_MAX_ENTRIES_DEFAULT, - )?; - Ok((max_entries > 0).then(|| { + let mut params = self.explicit_oauth_params(); + params + .entry(REST_CATALOG_PROP_SCOPE.to_string()) + .or_insert_with(|| "catalog".to_string()); + params + } + + pub(crate) fn explicit_oauth_params(&self) -> HashMap { + [ + (REST_CATALOG_PROP_SCOPE, self.scope()), + (REST_CATALOG_PROP_AUDIENCE, self.audience()), + (REST_CATALOG_PROP_RESOURCE, self.resource()), + ] + .into_iter() + .filter_map(|(key, value)| value.as_ref().map(|value| (key.to_string(), value.clone()))) + .collect() + } + + fn table_cache(&self) -> Option { + (self.table_cache_max_entries() > 0).then(|| { TableCache::builder() - .max_capacity(max_entries) - .time_to_live(expire_after_write) + .max_capacity(self.table_cache_max_entries()) + .time_to_live(*self.table_cache_expire_after_write()) .build() - })) - } - - /// Merge the `RestCatalogConfig` with the a [`CatalogConfig`] (fetched from the REST server). - pub(crate) fn merge_with_config(mut self, mut config: CatalogConfig) -> Self { - if let Some(uri) = config.overrides.remove(REST_CATALOG_PROP_URI) { - self.uri = uri; - } - - let mut props = config.defaults; - props.extend(self.props); - // The builder moved the client warehouse off the props; restore it - // between defaults and overrides (default < client < override). - if let Some(warehouse) = &self.warehouse { - props.insert(REST_CATALOG_PROP_WAREHOUSE.to_string(), warehouse.clone()); - } - props.extend(config.overrides); - - self.props = props; - self + }) } -} -/// Parses the `credential` property. -/// -/// ## Output -/// -/// - `None`: No credential is set. -/// - `Some(None, client_secret)`: No client_id is set, use client_secret directly. -/// - `Some(Some(client_id), client_secret)`: Both client_id and client_secret are set. -pub(crate) fn credential_from_props( - props: &HashMap, -) -> Option<(Option, String)> { - let cred = props.get("credential")?; - - match cred.split_once(':') { - Some((client_id, client_secret)) => { - Some((Some(client_id.to_string()), client_secret.to_string())) - } - None => Some((None, cred.to_string())), + fn resolved_auth_type(&self) -> &AuthType { + self.auth_type().as_ref().unwrap_or_else(|| { + if self.token.is_some() || self.credential.is_some() || self.oauth2_server_uri.is_some() + { + &AuthType::OAuth2 + } else { + &AuthType::None + } + }) } } -/// The extra headers added to each request, which include: -/// -/// - `content-type` -/// - `x-client-version` -/// - `user-agent` -/// - All headers specified by `header.xxx` in props. -pub(crate) fn extra_headers_from_props(props: &HashMap) -> Result { - let mut headers = HeaderMap::from_iter([ - ( - header::CONTENT_TYPE, - HeaderValue::from_static("application/json"), - ), - ( - HeaderName::from_static("x-client-version"), - HeaderValue::from_static(ICEBERG_REST_SPEC_VERSION), - ), - ( - header::USER_AGENT, - HeaderValue::from_str(&format!("iceberg-rs/{CARGO_PKG_VERSION}")).unwrap(), - ), - ]); - - headers.extend(explicit_headers_from_props(props)?); - - Ok(headers) +fn parse_header_redaction(value: &str) -> Result { + Ok(value.eq_ignore_ascii_case("true")) } -/// The default OAuth2 token endpoint for a catalog `uri`. -pub(crate) fn default_token_endpoint(uri: &str) -> String { - [uri, PATH_V1, "oauth", "tokens"].join("/") -} - -/// Only the headers explicitly configured via `header.xxx` props (no defaults). -pub(crate) fn explicit_headers_from_props(props: &HashMap) -> Result { - let mut headers = HeaderMap::new(); - for (key, value) in props - .iter() - .filter_map(|(k, v)| k.strip_prefix("header.").map(|k| (k, v))) - { - headers.insert( - HeaderName::from_str(key).map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - format!("Invalid header name: {key}"), - ) - .with_source(e) - })?, - HeaderValue::from_str(value).map_err(|e| { - Error::new( - ErrorKind::DataInvalid, - // The value itself is omitted: it may be a secret. - format!("Invalid value for header: {key}"), - ) - .with_source(e) - })?, - ); +fn parse_page_size(value: &str) -> Result { + let invalid = || { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid {REST_CATALOG_PROP_PAGE_SIZE}: {value}"), + ) + }; + let page_size = value + .parse::() + .map_err(|error| invalid().with_source(error))?; + if page_size.get() > i32::MAX as u32 { + return Err(invalid()); } - - Ok(headers) + Ok(page_size) } -/// The optional OAuth parameters added to each authentication request. -pub(crate) fn oauth_params_from_props(props: &HashMap) -> HashMap { - let mut params = HashMap::new(); - - if let Some(scope) = props.get("scope") { - params.insert("scope".to_string(), scope.to_string()); - } else { - params.insert("scope".to_string(), "catalog".to_string()); +fn parse_cache_expiry(value: &str) -> Result { + let millis = value.parse::().map_err(|error| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid {REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS}: {value}"), + ) + .with_source(error) + })?; + let duration = Duration::from_millis(millis); + if duration.is_zero() || duration > TABLE_CACHE_EXPIRE_AFTER_WRITE_MAX { + return Err(Error::new( + ErrorKind::DataInvalid, + format!( + "Invalid {REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS}: {millis} (must be between 1 ms and 1000 years)" + ), + )); } + Ok(duration) +} - let optional_params = ["audience", "resource"]; - for param_name in optional_params { - if let Some(value) = props.get(param_name) { - params.insert(param_name.to_string(), value.to_string()); +#[derive(Clone)] +#[cfg_attr(test, derive(PartialEq, Eq))] +pub(crate) struct RestHeaders(pub(crate) HeaderMap); + +impl RestHeaders { + fn from_properties(props: &HashMap) -> Result { + let mut headers = HeaderMap::new(); + for (key, value) in props.iter().filter_map(|(key, value)| { + key.strip_prefix(REST_CATALOG_PROP_HEADER_PREFIX) + .map(|key| (key, value)) + }) { + headers.insert( + HeaderName::from_str(key).map_err(|error| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid header name: {key}"), + ) + .with_source(error) + })?, + HeaderValue::from_str(value).map_err(|error| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid value for header: {key}"), + ) + .with_source(error) + })?, + ); } + Ok(Self(headers)) } +} - params +/// The default OAuth2 token endpoint for a catalog `uri`. +pub(crate) fn default_token_endpoint(uri: &str) -> String { + [uri, PATH_V1, "oauth", "tokens"].join("/") } struct RestCatalogClient { @@ -528,11 +547,10 @@ struct RestCatalogClient { /// Runtime config is fetched from rest server and stored here. /// /// It could be different from the user config. - config: RestCatalogConfig, + config: RestCatalogProperties, + props: HashMap, /// Capabilities the server advertises (see [`RestSessionCatalog::supports_endpoint`]). endpoints: HashSet, - /// Validated page size from the merged runtime configuration. - page_size: Option, /// Tables reused when the server answers a load with `304 Not Modified`. table_cache: Option, } @@ -553,18 +571,18 @@ 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, + user_props: &HashMap, + client: Client, auth_manager: Arc, ) -> Result { - let http_client = HttpClient::new(user_config)?; + let user_config = RestCatalogProperties::for_handshake(user_props)?; + let http_client = HttpClient::new(client, &user_config); // The init session lives only for the config handshake, so a // manager whose session guards a one-shot resource can release // it before deriving the catalog session. let catalog_config = { - let init_session = auth_manager - .init_session(&http_client, &Self::auth_props(user_config)) - .await?; - Self::load_config(&http_client, init_session.as_ref(), user_config).await? + let init_session = auth_manager.init_session(&http_client, user_props).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. @@ -572,23 +590,23 @@ impl RestCatalogClient { Some(advertised) if !advertised.is_empty() => advertised.iter().cloned().collect(), _ => crate::endpoint::DEFAULT_ENDPOINTS.clone(), }; - let config = user_config.clone().merge_with_config(catalog_config); - let page_size = config.page_size()?; - let http_client = http_client.update_with(&config)?; + let mut props = catalog_config.defaults; + props.extend(user_props.clone()); + props.extend(catalog_config.overrides); + let config = RestCatalogProperties::from_properties(&props)?; + 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 catalog_session = auth_manager - .catalog_session(&http_client, &Self::auth_props(&config)) - .await?; + let catalog_session = auth_manager.catalog_session(&http_client, &props).await?; Ok(Self { auth_manager, catalog_session, - table_cache: config.table_cache()?, + table_cache: config.table_cache(), config, + props, http_client, endpoints, - page_size, }) } @@ -619,36 +637,17 @@ impl RestCatalogClient { .await } - /// The properties handed to the [`AuthManager`], with the catalog `uri` - /// and `warehouse` made explicit. - fn auth_props(config: &RestCatalogConfig) -> HashMap { - // `oauth2-server-uri` stays absent unless explicitly configured, so an - // injected manager keeps its own endpoint. The resolved `uri` and - // `warehouse` ARE passed: the builder moved them off the props, and - // the built-in manager recomputes its token endpoint from the URI. - let mut props = config.props.clone(); - props.insert(REST_CATALOG_PROP_URI.to_string(), config.uri.clone()); - if let Some(warehouse) = &config.warehouse { - // A fallback only: after the handshake the merged props hold - // the resolved warehouse, server override included. - props - .entry(REST_CATALOG_PROP_WAREHOUSE.to_string()) - .or_insert_with(|| warehouse.clone()); - } - props - } - /// Loads the runtime config from the server using `user_config`. /// /// 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, + user_config: &RestCatalogProperties, ) -> Result { let mut request_builder = http_client.request(Method::GET, user_config.config_endpoint()); - if let Some(warehouse_location) = &user_config.warehouse { + if let Some(warehouse_location) = user_config.warehouse() { request_builder = request_builder.query(&[("warehouse", warehouse_location)]); } @@ -678,11 +677,10 @@ pub struct RestCatalog { } impl RestCatalog { - /// Creates a `RestCatalog` from a [`RestCatalogConfig`]. #[cfg(test)] fn new( context: SessionContext, - config: RestCatalogConfig, + config: RestCatalogTestFixture, auth_manager: Option>, storage_factory: Option>, runtime: Runtime, @@ -822,16 +820,14 @@ impl Catalog for RestCatalog { /// 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 { + name: String, + props: HashMap, + http_client: OnceLock, /// 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>, - /// 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, /// Storage factory for creating FileIO instances. storage_factory: Option>, @@ -840,18 +836,29 @@ pub struct RestSessionCatalog { kms_client: Option>, } +impl Debug for RestSessionCatalog { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RestSessionCatalog") + .field("name", &self.name) + .field("client", &self.client) + .finish_non_exhaustive() + } +} + impl RestSessionCatalog { - /// Creates a `RestSessionCatalog` from a [`RestCatalogConfig`]. + #[cfg(test)] fn new( - config: RestCatalogConfig, + config: RestCatalogTestFixture, auth_manager_override: Option>, storage_factory: Option>, runtime: Runtime, kms_client: Option>, ) -> Self { Self { + name: "test".to_string(), + props: config.into_props(), + http_client: OnceLock::new(), auth_manager_override: auth_manager_override.map(Arc::from), - user_config: config, client: OnceCell::new(), storage_factory, runtime, @@ -893,49 +900,26 @@ impl RestSessionCatalog { } } - /// The configured auth scheme: explicit `rest.auth.type` (matched - /// case-insensitively) when set; otherwise `oauth2` when a `token`, - /// `credential` or `oauth2-server-uri` is configured (preserving - /// pre-`rest.auth.type` setups), `none` when none is. - fn auth_type(config: &RestCatalogConfig) -> String { - config - .props - .get(REST_CATALOG_PROP_AUTH_TYPE) - // Matched case-insensitively, as the other flag properties are. - .map(|auth_type| auth_type.to_ascii_lowercase()) - .unwrap_or_else(|| { - if config.token().is_some() - || config.credential().is_some() - || config.explicit_oauth2_server_uri().is_some() - { - AUTH_TYPE_OAUTH2.to_string() - } else { - AUTH_TYPE_NONE.to_string() - } - }) - } - /// 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> { if let Some(auth_manager) = &self.auth_manager_override { return Ok(auth_manager.clone()); } - let config = &self.user_config; - let auth_type = Self::auth_type(config); + let config = RestCatalogProperties::for_handshake(&self.props)?; + let auth_type = config.resolved_auth_type(); // Java parity (`AuthManagers`): make the inference visible so users // configure the type explicitly. - if auth_type == AUTH_TYPE_OAUTH2 && !config.props.contains_key(REST_CATALOG_PROP_AUTH_TYPE) - { + if *auth_type == AuthType::OAuth2 && config.auth_type().is_none() { tracing::warn!( "Inferring {REST_CATALOG_PROP_AUTH_TYPE}={AUTH_TYPE_OAUTH2} from the configured \ OAuth properties; set it explicitly to avoid this warning" ); } - match auth_type.as_str() { - AUTH_TYPE_NONE => Ok(Arc::new(NoopAuthManager)), - AUTH_TYPE_OAUTH2 => Ok(Arc::new(OAuth2Manager::from_config(config)?)), - other => Err(Error::new( + match auth_type { + AuthType::None => Ok(Arc::new(NoopAuthManager)), + AuthType::OAuth2 => Ok(Arc::new(OAuth2Manager::from_properties(&config))), + AuthType::Custom(other) => Err(Error::new( ErrorKind::DataInvalid, format!( "unknown '{REST_CATALOG_PROP_AUTH_TYPE}': {other}; use \ @@ -950,7 +934,12 @@ impl RestSessionCatalog { async fn client(&self) -> Result<&RestCatalogClient> { self.client .get_or_try_init(|| async { - RestCatalogClient::init(&self.user_config, self.resolve_auth_manager()?).await + RestCatalogClient::init( + &self.props, + self.http_client.get_or_init(Client::default).clone(), + self.resolve_auth_manager()?, + ) + .await }) .await } @@ -988,14 +977,14 @@ impl RestSessionCatalog { metadata_location: Option<&str>, extra_config: Option>, ) -> Result { - let mut props = self.client().await?.config.props.clone(); + let mut props = self.client().await?.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.client().await?.config.warehouse.as_deref() { + let warehouse_path = match self.client().await?.config.warehouse().as_deref() { Some(url) if Url::parse(url).is_ok() => Some(url), Some(_) => None, None => None, @@ -1038,12 +1027,12 @@ impl SessionCatalog for RestSessionCatalog { let endpoint = client.config.namespaces_endpoint(); let mut namespaces = Vec::new(); // An empty initial token explicitly opts in to pagination. - let mut next_token = client.page_size.map(|_| String::new()); + let mut next_token = client.config.page_size().map(|_| String::new()); loop { let mut request = client.http_client.request(Method::GET, endpoint.clone()); - if let Some(page_size) = client.page_size { + if let Some(page_size) = client.config.page_size() { request = request.query(&[("pageSize", page_size)]); } @@ -1228,12 +1217,12 @@ impl SessionCatalog for RestSessionCatalog { let endpoint = client.config.tables_endpoint(namespace); let mut identifiers = Vec::new(); // An empty initial token explicitly opts in to pagination. - let mut next_token = client.page_size.map(|_| String::new()); + let mut next_token = client.config.page_size().map(|_| String::new()); loop { let mut request = client.http_client.request(Method::GET, endpoint.clone()); - if let Some(page_size) = client.page_size { + if let Some(page_size) = client.config.page_size() { request = request.query(&[("pageSize", page_size)]); } @@ -1338,7 +1327,7 @@ impl SessionCatalog for RestSessionCatalog { let config = response .config .into_iter() - .chain(self.user_config.props.clone()) + .chain(self.props.clone()) .collect(); let file_io = self @@ -1435,7 +1424,7 @@ impl SessionCatalog for RestSessionCatalog { let config = response .config .into_iter() - .chain(self.user_config.props.clone()) + .chain(self.props.clone()) .collect(); let file_io = self @@ -1692,24 +1681,20 @@ impl SessionCatalog for RestSessionCatalog { /// [`SessionContext`] with each [`SessionCatalog`] operation. #[derive(Debug)] pub struct RestSessionCatalogBuilder { - config: RestCatalogConfig, + page_size: Option, + client: Option, auth_manager: Option>, storage_factory: Option>, kms_client_factory: Option>, runtime: Option, } +#[allow(clippy::derivable_impls)] impl Default for RestSessionCatalogBuilder { fn default() -> Self { Self { - config: RestCatalogConfig { - name: None, - uri: "".to_string(), - warehouse: None, - props: HashMap::new(), - client: None, - default_client: Arc::new(OnceLock::new()), - }, + page_size: None, + client: None, auth_manager: None, storage_factory: None, kms_client_factory: None, @@ -1727,16 +1712,13 @@ impl RestSessionCatalogBuilder { /// server overrides. Catalog operations fail with [`ErrorKind::DataInvalid`] /// if the effective value is not in `1..=i32::MAX`. pub fn with_page_size(mut self, page_size: NonZeroU32) -> Self { - self.config.props.insert( - REST_CATALOG_PROP_PAGE_SIZE.to_string(), - page_size.to_string(), - ); + self.page_size = Some(page_size); self } /// Configures the catalog with a custom HTTP client. pub fn with_client(mut self, client: Client) -> Self { - self.config.client = Some(client); + self.client = Some(client); self } @@ -1822,58 +1804,60 @@ impl RestSessionCatalogBuilder { /// The server configuration handshake, endpoint negotiation, and /// authentication sessions are initialized lazily on the first operation. pub fn load( - mut self, + self, name: impl Into, - props: HashMap, + mut props: HashMap, ) -> impl Future> + Send { - self.config.name = Some(name.into()); - - if props.contains_key(REST_CATALOG_PROP_URI) { - self.config.uri = props - .get(REST_CATALOG_PROP_URI) - .cloned() - .unwrap_or_default(); - } - - if props.contains_key(REST_CATALOG_PROP_WAREHOUSE) { - self.config.warehouse = props.get(REST_CATALOG_PROP_WAREHOUSE).cloned() + let name = name.into(); + if let Some(page_size) = self.page_size { + props + .entry(REST_CATALOG_PROP_PAGE_SIZE.to_string()) + .or_insert_with(|| page_size.to_string()); } - // Collect other remaining properties - self.config.props.extend(props); - self.config.props.remove(REST_CATALOG_PROP_URI); - self.config.props.remove(REST_CATALOG_PROP_WAREHOUSE); - async move { - if self.config.name.is_none() { - Err(Error::new( + if name.trim().is_empty() { + return Err(Error::new( ErrorKind::DataInvalid, "Catalog name is required", - )) - } else if self.config.uri.is_empty() { - Err(Error::new( + )); + } + if props + .get(REST_CATALOG_PROP_URI) + .is_none_or(String::is_empty) + { + return Err(Error::new( ErrorKind::DataInvalid, "Catalog uri is required", - )) - } else { - let runtime = self.runtime.unwrap_or_else(Runtime::current); - let kms_client = match self.kms_client_factory { - Some(factory) => Some(factory.create_kms_client(&self.config.props).await?), - None => None, - }; - - Ok(RestSessionCatalog::new( - self.config, - self.auth_manager, - self.storage_factory, - runtime, - kms_client, - )) + )); + } + + let runtime = self.runtime.unwrap_or_else(Runtime::current); + let kms_client = match self.kms_client_factory { + Some(factory) => Some(factory.create_kms_client(&props).await?), + None => None, + }; + let http_client = OnceLock::new(); + if let Some(client) = self.client { + let _ = http_client.set(client); } + Ok(RestSessionCatalog { + name, + props, + http_client, + auth_manager_override: self.auth_manager.map(Arc::from), + client: OnceCell::new(), + storage_factory: self.storage_factory, + runtime, + kms_client, + }) } } } +#[cfg(test)] +pub(crate) use tests::RestCatalogTestFixture; + #[cfg(test)] mod tests { use std::fs::File; @@ -1891,12 +1875,336 @@ mod tests { use iceberg_test_utils::test_runtime; use mockito::{Matcher, Mock, Server, ServerGuard}; use serde_json::json; + use typed_builder::TypedBuilder; use uuid::uuid; use super::*; use crate::auth::AuthSession; use crate::request::HttpRequest; + #[derive(TypedBuilder)] + pub(crate) struct RestCatalogTestFixture { + uri: String, + #[builder(default, setter(strip_option))] + warehouse: Option, + #[builder(default)] + props: HashMap, + } + + impl RestCatalogTestFixture { + pub(super) fn into_props(mut self) -> HashMap { + self.props + .insert(REST_CATALOG_PROP_URI.to_string(), self.uri); + if let Some(warehouse) = self.warehouse { + self.props + .insert(REST_CATALOG_PROP_WAREHOUSE.to_string(), warehouse); + } + self.props + } + + pub(crate) fn properties(self) -> Result { + RestCatalogProperties::from_properties(&self.into_props()) + } + + pub(crate) fn http_client(self) -> Result { + Ok(HttpClient::new(Client::default(), &self.properties()?)) + } + } + + #[test] + fn test_catalog_properties() { + let props = HashMap::from([ + (REST_CATALOG_PROP_URI, "http://localhost:8181"), + (REST_CATALOG_PROP_WAREHOUSE, "s3://warehouse"), + ("prefix", "catalog"), + (REST_CATALOG_PROP_AUTH_TYPE, "OAuth2"), + ("token", "secret"), + ("credential", "client:secret"), + ("oauth2-server-uri", "http://localhost:8181/token"), + ("scope", "custom-scope"), + ("audience", "custom-audience"), + ("resource", "custom-resource"), + ("header.x-custom", "value"), + (REST_CATALOG_PROP_DISABLE_HEADER_REDACTION, "TRUE"), + (REST_CATALOG_PROP_PAGE_SIZE, "25"), + (REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES, "10"), + (REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS, "500"), + ("custom.property", "retained-by-catalog"), + ]) + .into_iter() + .map(|(key, value)| (key.to_string(), value.to_string())) + .collect(); + let expected = RestCatalogProperties { + uri: "http://localhost:8181".to_string(), + warehouse: Some("s3://warehouse".to_string()), + prefix: Some("catalog".to_string()), + auth_type: Some(AuthType::OAuth2), + token: Some("secret".to_string().into()), + credential: Some(OAuth2Credential { + client_id: Some("client".to_string()), + client_secret: "secret".to_string().into(), + }), + oauth2_server_uri: Some("http://localhost:8181/token".to_string()), + scope: Some("custom-scope".to_string()), + audience: Some("custom-audience".to_string()), + resource: Some("custom-resource".to_string()), + headers: RestHeaders(HeaderMap::from_iter([( + HeaderName::from_static("x-custom"), + HeaderValue::from_static("value"), + )])), + disable_header_redaction: true, + page_size: NonZeroU32::new(25), + table_cache_max_entries: 10, + table_cache_expire_after_write: Duration::from_millis(500), + }; + + assert_eq!( + RestCatalogProperties::from_properties(&props).unwrap(), + expected + ); + } + + #[test] + fn test_catalog_properties_defaults() { + let expected = RestCatalogProperties { + uri: String::new(), + warehouse: None, + prefix: None, + auth_type: None, + token: None, + credential: None, + oauth2_server_uri: None, + scope: None, + audience: None, + resource: None, + headers: RestHeaders(HeaderMap::new()), + disable_header_redaction: false, + page_size: None, + table_cache_max_entries: 100, + table_cache_expire_after_write: Duration::from_secs(300), + }; + + assert_eq!( + RestCatalogProperties::from_properties(&HashMap::new()).unwrap(), + expected + ); + } + + #[test] + fn test_catalog_properties_invalid_values() { + for (key, value) in [ + (REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES, "-1"), + ( + REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES, + "18446744073709551616", + ), + ( + REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS, + "invalid", + ), + (REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS, "-1"), + ("header.invalid name", "value"), + ("header.authorization", "secret\nvalue"), + ] { + let error = RestCatalogProperties::from_properties(&HashMap::from([( + key.to_string(), + value.to_string(), + )])) + .unwrap_err(); + assert_eq!(error.kind(), ErrorKind::DataInvalid); + assert!(format!("{error}").contains(key.strip_prefix("header.").unwrap_or(key))); + assert!(!format!("{error:?}").contains("secret\nvalue")); + } + } + + #[test] + fn test_credential_parsing_and_redaction_fallback() { + for (value, client_id, secret) in [ + ("secret", None, "secret"), + ( + "client:secret:with:colons", + Some("client"), + "secret:with:colons", + ), + (":secret", Some(""), "secret"), + ("client:", Some("client"), ""), + ("", None, ""), + ] { + let properties = RestCatalogProperties::from_properties(&HashMap::from([ + ("credential".to_string(), value.to_string()), + ( + REST_CATALOG_PROP_DISABLE_HEADER_REDACTION.to_string(), + "invalid".to_string(), + ), + ])) + .unwrap(); + let expected = ( + Some(OAuth2Credential { + client_id: client_id.map(str::to_string), + client_secret: secret.to_string().into(), + }), + false, + ); + assert_eq!( + ( + properties.credential().clone(), + properties.disable_header_redaction() + ), + expected, + ); + } + } + + #[tokio::test] + async fn test_cache_property_precedence() { + for (default, client, override_value, expected) in [ + ("10", None, None, 10), + ("invalid", Some("20"), None, 20), + ("10", Some("invalid"), Some("30"), 30), + ] { + let mut server = Server::new_async().await; + let cache_props = |value: &str| { + HashMap::from([ + ( + REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES.to_string(), + value.to_string(), + ), + ( + REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS.to_string(), + value.to_string(), + ), + ]) + }; + let config_mock = server + .mock("GET", "/v1/config") + .with_status(200) + .with_body( + json!({ + "defaults": cache_props(default), + "overrides": override_value.map(cache_props).unwrap_or_default(), + }) + .to_string(), + ) + .create_async() + .await; + let mut props = client.map(cache_props).unwrap_or_default(); + props.insert(REST_CATALOG_PROP_URI.to_string(), server.url()); + let catalog = RestSessionCatalogBuilder::default() + .load("test", props) + .await + .unwrap(); + let config = &catalog.client().await.unwrap().config; + assert_eq!( + ( + config.table_cache_max_entries, + config.table_cache_expire_after_write + ), + (expected, Duration::from_millis(expected)), + ); + config_mock.assert_async().await; + } + } + + #[tokio::test] + async fn test_load_forwards_all_properties_to_kms_factory() { + #[derive(Debug, Default)] + struct RecordingKmsClientFactory(Mutex>); + + #[async_trait] + impl KmsClientFactory for RecordingKmsClientFactory { + async fn create_kms_client( + &self, + properties: &HashMap, + ) -> Result> { + *self.0.lock().unwrap() = properties.clone(); + Err(Error::new( + ErrorKind::Unexpected, + "KMS initialization failed", + )) + } + } + + let properties = HashMap::from([ + ( + REST_CATALOG_PROP_URI.to_string(), + "http://localhost".to_string(), + ), + ( + REST_CATALOG_PROP_WAREHOUSE.to_string(), + "s3://warehouse".to_string(), + ), + ("token".to_string(), "secret".to_string()), + ("custom.property".to_string(), "value".to_string()), + (REST_CATALOG_PROP_PAGE_SIZE.to_string(), "20".to_string()), + ]); + let factory = Arc::new(RecordingKmsClientFactory::default()); + let error = RestSessionCatalogBuilder::default() + .with_kms_client_factory(factory.clone()) + .with_page_size(NonZeroU32::MIN) + .load("rest", properties.clone()) + .await + .unwrap_err(); + + assert_eq!(*factory.0.lock().unwrap(), properties); + assert_eq!( + (error.kind(), error.message()), + (ErrorKind::Unexpected, "KMS initialization failed") + ); + } + + #[tokio::test] + async fn test_file_io_receives_resolved_properties() { + let mut server = Server::new_async().await; + let config_mock = server + .mock("GET", "/v1/config") + .match_query(Matcher::Any) + .with_status(200) + .with_body( + json!({ + "defaults": {"warehouse": "file:///default", "custom.property": "default"}, + "overrides": {"warehouse": "file:///override", "s3.region": "us-east-2"}, + }) + .to_string(), + ) + .create_async() + .await; + let props = HashMap::from([ + (REST_CATALOG_PROP_URI.to_string(), server.url()), + ( + REST_CATALOG_PROP_WAREHOUSE.to_string(), + "file:///client".to_string(), + ), + ("custom.property".to_string(), "client".to_string()), + ("s3.region".to_string(), "us-east-1".to_string()), + ]); + let catalog = RestSessionCatalogBuilder::default() + .with_storage_factory(Arc::new(LocalFsStorageFactory)) + .load("rest", props) + .await + .unwrap(); + let file_io = catalog + .load_file_io( + None, + Some(HashMap::from([( + "custom.property".to_string(), + "table".to_string(), + )])), + ) + .await + .unwrap(); + let expected = HashMap::from([ + (REST_CATALOG_PROP_URI.to_string(), server.url()), + ( + REST_CATALOG_PROP_WAREHOUSE.to_string(), + "file:///override".to_string(), + ), + ("custom.property".to_string(), "table".to_string()), + ("s3.region".to_string(), "us-east-2".to_string()), + ]); + assert_eq!(file_io.config().props(), &expected); + config_mock.assert_async().await; + } + #[derive(Debug)] struct PlainContextSession; @@ -1966,11 +2274,11 @@ mod tests { } } - fn test_catalog(config: RestCatalogConfig) -> RestSessionCatalog { + fn test_catalog(config: RestCatalogTestFixture) -> RestSessionCatalog { RestSessionCatalog::new(config, None, None, Runtime::current(), None) } - fn test_catalog_with(config: RestCatalogConfig, auth_manager: M) -> RestSessionCatalog + fn test_catalog_with(config: RestCatalogTestFixture, auth_manager: M) -> RestSessionCatalog where M: AuthManager + 'static { RestSessionCatalog::new( config, @@ -1982,7 +2290,7 @@ mod tests { } fn context_catalog( - config: RestCatalogConfig, + config: RestCatalogTestFixture, fail_contextual_session: bool, ) -> (RestSessionCatalog, Arc>>) { let seen_session_ids = Arc::new(Mutex::new(Vec::new())); @@ -1996,16 +2304,15 @@ mod tests { } fn test_client() -> HttpClient { - HttpClient::new( - &RestCatalogConfig::builder() - .uri("http://localhost".to_string()) - .build(), - ) - .unwrap() + RestCatalogTestFixture::builder() + .uri("http://localhost".to_string()) + .build() + .http_client() + .unwrap() } /// Builds a [`RestSessionCatalog`] with the default test storage factory and runtime. - fn session_catalog(config: RestCatalogConfig) -> RestSessionCatalog { + fn session_catalog(config: RestCatalogTestFixture) -> RestSessionCatalog { RestSessionCatalog::new( config, None, @@ -2313,16 +2620,10 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); assert_eq!( - catalog - .client() - .await - .unwrap() - .config - .props - .get("warehouse"), + catalog.client().await.unwrap().props.get("warehouse"), Some(&"s3://iceberg-catalog".to_string()) ); @@ -2387,7 +2688,7 @@ mod tests { let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), @@ -2420,7 +2721,7 @@ mod tests { let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), @@ -2457,7 +2758,7 @@ mod tests { let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), @@ -2509,7 +2810,7 @@ mod tests { props.insert("credential".to_string(), "client1:secret1".to_string()); let catalog = session_catalog( - RestCatalogConfig::builder() + RestCatalogTestFixture::builder() .uri(server.url()) .props(props) .build(), @@ -2551,7 +2852,7 @@ mod tests { let config_mock = create_config_mock(&mut server).await; let catalog = session_catalog( - RestCatalogConfig::builder() + RestCatalogTestFixture::builder() .uri(server.url()) .props(props) .build(), @@ -2570,11 +2871,11 @@ mod tests { let mut props = HashMap::new(); props.insert("credential".to_string(), "client1:secret1".to_string()); - let config = RestCatalogConfig::builder() + let config = RestCatalogTestFixture::builder() .uri(server.url()) .props(props) .build(); - let headers: HeaderMap = config.extra_headers().unwrap(); + let headers: HeaderMap = config.properties().unwrap().extra_headers(); let expected_headers = HeaderMap::from_iter([ ( @@ -2607,11 +2908,11 @@ mod tests { "some/value".to_string(), ); - let config = RestCatalogConfig::builder() + let config = RestCatalogTestFixture::builder() .uri(server.url()) .props(props) .build(); - let headers: HeaderMap = config.extra_headers().unwrap(); + let headers: HeaderMap = config.properties().unwrap().extra_headers(); let expected_headers = HeaderMap::from_iter([ ( @@ -2653,7 +2954,7 @@ mod tests { ); let catalog = session_catalog( - RestCatalogConfig::builder() + RestCatalogTestFixture::builder() .uri(server.url()) .props(props) .build(), @@ -2701,7 +3002,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let _namespaces = catalog .list_namespaces(&SessionContext::empty(), None) @@ -2731,7 +3032,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let namespaces = catalog .list_namespaces(&SessionContext::empty(), None) @@ -2756,7 +3057,7 @@ mod tests { (REST_CATALOG_PROP_AUTH_TYPE.to_string(), "none".to_string()), ("token".to_string(), "some-oauth-token".to_string()), ]); - let config = RestCatalogConfig::builder() + let config = RestCatalogTestFixture::builder() .uri("http://localhost".to_string()) .props(props) .build(); @@ -2797,7 +3098,7 @@ mod tests { ]); let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder() + RestCatalogTestFixture::builder() .uri(server.url()) .props(props) .build(), @@ -2836,7 +3137,7 @@ mod tests { let props = HashMap::from([("credential".to_string(), "client1:secret1".to_string())]); let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder() + RestCatalogTestFixture::builder() .uri(bootstrap.url()) .props(props) .build(), @@ -2994,7 +3295,7 @@ mod tests { )])); let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), Some(Box::new(manager)), Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), @@ -3060,16 +3361,15 @@ mod tests { .create_async() .await; - let catalog_client = HttpClient::new( - &RestCatalogConfig::builder() - .uri(server.url()) - .props(HashMap::from([( - "header.x-catalog-only".to_string(), - "not-on-token-requests".to_string(), - )])) - .build(), - ) - .unwrap(); + let catalog_client = RestCatalogTestFixture::builder() + .uri(server.url()) + .props(HashMap::from([( + "header.x-catalog-only".to_string(), + "not-on-token-requests".to_string(), + )])) + .build() + .http_client() + .unwrap(); let manager = OAuth2Manager::new(format!("{}/v1/oauth/tokens", server.url())) .with_credential(Some("client1".to_string()), "secret1".to_string()) @@ -3107,7 +3407,7 @@ mod tests { let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder() + RestCatalogTestFixture::builder() .uri(server.url()) .props(HashMap::from([( "token".to_string(), @@ -3165,7 +3465,7 @@ mod tests { let captured = Arc::new(AsyncMutex::new(None)); let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder() + RestCatalogTestFixture::builder() .uri(server.url()) .props(HashMap::from([( "token".to_string(), @@ -3240,7 +3540,7 @@ mod tests { let captured = Arc::new(AsyncMutex::new(None)); let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder() + RestCatalogTestFixture::builder() .uri(server.url()) .warehouse("client-wh".to_string()) .build(), @@ -3249,7 +3549,8 @@ mod tests { Runtime::current(), None, ); - catalog.client().await.unwrap(); + let client = catalog.client().await.unwrap(); + assert_eq!(client.config.warehouse.as_deref(), Some("client-wh")); config_mock.assert_async().await; let props = captured.lock().await.clone().unwrap(); assert_eq!( @@ -3272,7 +3573,7 @@ mod tests { let captured = Arc::new(AsyncMutex::new(None)); let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder() + RestCatalogTestFixture::builder() .uri(server.url()) .warehouse("client-wh".to_string()) .build(), @@ -3281,7 +3582,8 @@ mod tests { Runtime::current(), None, ); - catalog.client().await.unwrap(); + let client = catalog.client().await.unwrap(); + assert_eq!(client.config.warehouse.as_deref(), Some("override-wh")); config_mock.assert_async().await; let props = captured.lock().await.clone().unwrap(); assert_eq!( @@ -3351,7 +3653,7 @@ mod tests { let dropped = Arc::new(AtomicBool::new(false)); let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), Some(Box::new(GuardManager(dropped.clone()))), Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), @@ -3381,7 +3683,7 @@ mod tests { .await; let (catalog, seen_session_ids) = context_catalog( - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), false, ); let context = SessionContext::builder() @@ -3414,7 +3716,7 @@ mod tests { .create_async() .await; let (catalog, seen_session_ids) = context_catalog( - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), false, ); let context = SessionContext::builder() @@ -3443,7 +3745,7 @@ mod tests { .create_async() .await; let (catalog, seen_session_ids) = context_catalog( - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), false, ); let context = SessionContext::builder() @@ -3475,7 +3777,7 @@ mod tests { .create_async() .await; let (catalog, seen_session_ids) = context_catalog( - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), false, ); let context = SessionContext::builder() @@ -3500,8 +3802,10 @@ mod tests { .expect(0) .create_async() .await; - let (catalog, seen_session_ids) = - context_catalog(RestCatalogConfig::builder().uri(server.url()).build(), true); + let (catalog, seen_session_ids) = context_catalog( + RestCatalogTestFixture::builder().uri(server.url()).build(), + true, + ); let context = SessionContext::builder() .session_id("failing-session".to_string()) .build(); @@ -3514,10 +3818,18 @@ mod tests { list_mock.assert_async().await; } - #[test] - fn test_config_debug_redacts_secrets() { - let config = RestCatalogConfig::builder() - .uri("http://localhost".to_string()) + #[tokio::test] + async fn test_config_debug_redacts_secrets() { + let mut server = Server::new_async().await; + let config_mock = server + .mock("GET", "/v1/config") + .match_query(Matcher::Any) + .with_status(200) + .with_body(r#"{"defaults": {}, "overrides": {}}"#) + .create_async() + .await; + let config = RestCatalogTestFixture::builder() + .uri(server.url()) .props(HashMap::from([ ("token".to_string(), "tok-secret".to_string()), ("credential".to_string(), "id:cred-secret".to_string()), @@ -3532,22 +3844,38 @@ mod tests { ])) .build(); - let out = format!("{config:?}"); - assert!(!out.contains("tok-secret")); - assert!(!out.contains("cred-secret")); - assert!(!out.contains("Basic xyz")); - assert!(!out.contains("adls-secret")); - assert!(!out.contains("sse-secret")); - assert!(!out.contains("cs-secret")); + let props = config.into_props(); + let typed = RestCatalogProperties::from_properties(&props).unwrap(); + let out = format!("{typed:?}"); assert!(out.contains("[REDACTED]")); assert!(out.contains("wh1")); + let catalog = RestSessionCatalogBuilder::default() + .with_auth_manager(NoopAuthManager) + .load("rest", props) + .await + .unwrap(); + let before_handshake = format!("{catalog:?}"); + catalog.client().await.unwrap(); + for output in [out, before_handshake, format!("{catalog:?}")] { + for secret in [ + "tok-secret", + "cred-secret", + "Basic xyz", + "adls-secret", + "sse-secret", + "cs-secret", + ] { + assert!(!output.contains(secret)); + } + } + config_mock.assert_async().await; } #[tokio::test] async fn test_auth_type_defaults() { // Unset `rest.auth.type`: `oauth2` when any OAuth material is // configured (existing setups keep working), `none` otherwise. - let bare = RestCatalogConfig::builder() + let bare = RestCatalogTestFixture::builder() .uri("http://localhost".to_string()) .build(); assert!( @@ -3555,7 +3883,7 @@ mod tests { .contains("NoopAuthManager") ); - let with_token = RestCatalogConfig::builder() + let with_token = RestCatalogTestFixture::builder() .uri("http://localhost".to_string()) .props(HashMap::from([("token".to_string(), "tok".to_string())])) .build(); @@ -3568,7 +3896,7 @@ mod tests { ); // An explicit type is matched case-insensitively. - let mixed_case = RestCatalogConfig::builder() + let mixed_case = RestCatalogTestFixture::builder() .uri("http://localhost".to_string()) .props(HashMap::from([( REST_CATALOG_PROP_AUTH_TYPE.to_string(), @@ -3585,7 +3913,7 @@ mod tests { // An explicit OAuth endpoint is oauth2 intent too: the manager can // still pick up a server-supplied token from `/v1/config`. - let with_endpoint = RestCatalogConfig::builder() + let with_endpoint = RestCatalogTestFixture::builder() .uri("http://localhost".to_string()) .props(HashMap::from([( "oauth2-server-uri".to_string(), @@ -3607,7 +3935,7 @@ mod tests { REST_CATALOG_PROP_AUTH_TYPE.to_string(), "kerberos".to_string(), )]); - let config = RestCatalogConfig::builder() + let config = RestCatalogTestFixture::builder() .uri("http://localhost".to_string()) .props(props) .build(); @@ -3639,7 +3967,7 @@ mod tests { } } - let config = RestCatalogConfig::builder() + let config = RestCatalogTestFixture::builder() .uri("http://localhost".to_string()) .props(HashMap::from([( REST_CATALOG_PROP_AUTH_TYPE.to_string(), @@ -3685,7 +4013,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let namespaces = catalog .list_namespaces(&SessionContext::empty(), None) @@ -3784,7 +4112,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let namespaces = catalog .list_namespaces(&SessionContext::empty(), None) @@ -3837,7 +4165,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let namespaces = catalog .create_namespace( @@ -3878,7 +4206,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let namespaces = catalog .get_namespace( @@ -3911,7 +4239,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); assert!( catalog @@ -3948,7 +4276,7 @@ mod tests { let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), @@ -3978,7 +4306,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); catalog .drop_namespace( @@ -4018,7 +4346,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let tables = catalog .list_tables( @@ -4086,7 +4414,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let tables = catalog .list_tables( @@ -4217,7 +4545,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let tables = catalog .list_tables( @@ -4261,7 +4589,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); catalog .drop_table( @@ -4287,7 +4615,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); assert!( catalog @@ -4323,7 +4651,7 @@ mod tests { let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), @@ -4356,7 +4684,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); catalog .rename_table( @@ -4388,7 +4716,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let table = catalog .load_table( @@ -4524,7 +4852,7 @@ mod tests { let _config = create_config_mock(&mut server).await; let _v1 = load_table_mock(&mut server, Matcher::Missing, 200, Some("\"v1\"")).await; let reuse_v1 = load_table_mock(&mut server, "\"v1\"".into(), 304, None).await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let context = SessionContext::empty(); let location = |table: Table| table.metadata_location().unwrap().to_string(); @@ -4548,7 +4876,7 @@ mod tests { let mut server = Server::new_async().await; let _config = create_config_mock(&mut server).await; let _load = load_table_mock(&mut server, Matcher::Missing, 304, None).await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let err = catalog .load_table(&SessionContext::empty(), &test_table_ident()) @@ -4566,7 +4894,7 @@ mod tests { .await .expect(2); let hit = load_table_mock(&mut server, "\"v1\"".into(), 304, None).await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let session = |id: &str| SessionContext::builder().session_id(id.to_string()).build(); for id in ["a", "b", "a"] { @@ -4588,7 +4916,7 @@ mod tests { .await .expect(2); let catalog = session_catalog( - RestCatalogConfig::builder() + RestCatalogTestFixture::builder() .uri(server.url()) .props(HashMap::from([( REST_CATALOG_PROP_TABLE_CACHE_MAX_ENTRIES.to_string(), @@ -4645,7 +4973,7 @@ mod tests { .await; let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), @@ -4686,7 +5014,7 @@ mod tests { .await .expect(2); let _gone = load_table_mock(&mut server, "\"v1\"".into(), 404, None).await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let context = SessionContext::empty(); let ident = test_table_ident(); @@ -4718,7 +5046,7 @@ mod tests { .with_body(body.to_string()) .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let context = SessionContext::empty(); for _ in 0..3 { @@ -4746,7 +5074,7 @@ mod tests { .expect(2) .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let context = SessionContext::empty(); catalog @@ -4764,7 +5092,7 @@ mod tests { #[test] fn test_table_cache_rejects_out_of_range_expiry() { for value in ["0", "31536000000001"] { - let config = RestCatalogConfig::builder() + let config = RestCatalogTestFixture::builder() .uri("http://localhost".to_string()) .props(HashMap::from([( REST_CATALOG_PROP_TABLE_CACHE_EXPIRE_AFTER_WRITE_MS.to_string(), @@ -4773,7 +5101,7 @@ mod tests { .build(); assert_eq!( - config.table_cache().unwrap_err().kind(), + config.properties().unwrap_err().kind(), ErrorKind::DataInvalid ); } @@ -4806,7 +5134,7 @@ mod tests { .with_chunked_body(|_| Err(std::io::Error::other("connection reset"))) .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let context = SessionContext::empty(); let ident = test_table_ident(); @@ -4838,7 +5166,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let table = catalog .load_table( @@ -4871,7 +5199,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let table_creation = TableCreation::builder() .name("test1".to_string()) @@ -5022,7 +5350,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let table_creation = TableCreation::builder() .name("test1".to_string()) @@ -5094,7 +5422,7 @@ mod tests { let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), @@ -5242,7 +5570,7 @@ mod tests { let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), @@ -5309,7 +5637,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let table_ident = TableIdent::new(NamespaceIdent::new("ns1".to_string()), "test1".to_string()); let metadata_location = String::from( @@ -5357,7 +5685,7 @@ mod tests { .create_async() .await; - let catalog = session_catalog(RestCatalogConfig::builder().uri(server.url()).build()); + let catalog = session_catalog(RestCatalogTestFixture::builder().uri(server.url()).build()); let table_ident = TableIdent::new(NamespaceIdent::new("ns1".to_string()), "test1".to_string()); @@ -5377,15 +5705,32 @@ mod tests { #[tokio::test] async fn test_create_rest_catalog() { - let builder = RestCatalogBuilder::default().with_client(Client::new()); + let mut server = Server::new_async().await; + let config_mock = server + .mock("GET", "/v1/config") + .match_query(Matcher::Any) + .match_header("x-injected", "shared-client") + .with_status(200) + .with_body(r#"{"defaults": {}, "overrides": {}}"#) + .create_async() + .await; + let client = Client::builder() + .default_headers(HeaderMap::from_iter([( + HeaderName::from_static("x-injected"), + HeaderValue::from_static("shared-client"), + )])) + .build() + .unwrap(); + let builder = RestCatalogBuilder::default().with_client(client); let catalog = builder .load( "test", HashMap::from([ + (REST_CATALOG_PROP_URI.to_string(), server.url()), ( - REST_CATALOG_PROP_URI.to_string(), - "http://localhost:8080".to_string(), + REST_CATALOG_PROP_WAREHOUSE.to_string(), + "s3://warehouse".to_string(), ), ("a".to_string(), "b".to_string()), ]), @@ -5395,14 +5740,19 @@ mod tests { assert!(catalog.is_ok()); let catalog = catalog.unwrap(); - let catalog_config = &catalog.inner.user_config; - assert_eq!(catalog_config.name.as_deref(), Some("test")); - assert_eq!(catalog_config.uri, "http://localhost:8080"); - assert_eq!(catalog_config.warehouse, None); - assert!(catalog_config.client.is_some()); - - assert_eq!(catalog_config.props.get("a"), Some(&"b".to_string())); - assert!(!catalog_config.props.contains_key(REST_CATALOG_PROP_URI)); + let expected_props = HashMap::from([ + (REST_CATALOG_PROP_URI.to_string(), server.url()), + ( + REST_CATALOG_PROP_WAREHOUSE.to_string(), + "s3://warehouse".to_string(), + ), + ("a".to_string(), "b".to_string()), + ]); + assert_eq!(catalog.inner.name, "test"); + assert_eq!(catalog.inner.props, expected_props); + assert!(catalog.inner.http_client.get().is_some()); + catalog.client().await.unwrap(); + config_mock.assert_async().await; } #[tokio::test] @@ -5426,6 +5776,23 @@ mod tests { } } + #[tokio::test] + async fn test_create_rest_catalog_empty_name() { + let error = RestCatalogBuilder::default() + .load( + "", + HashMap::from([( + REST_CATALOG_PROP_URI.to_string(), + "http://localhost:8080".to_string(), + )]), + ) + .await + .unwrap_err(); + + assert_eq!(error.kind(), ErrorKind::DataInvalid); + assert_eq!(error.message(), "Catalog name is required"); + } + #[tokio::test] async fn test_create_session_catalog() { let builder = RestSessionCatalogBuilder::default(); @@ -5447,16 +5814,16 @@ mod tests { let catalog = result.unwrap(); - let catalog_config = catalog.user_config; - assert_eq!(catalog_config.name.as_deref(), Some("test")); - assert_eq!(catalog_config.uri, "http://localhost:8080"); - assert_eq!(catalog_config.warehouse, None); - // The default builder sets no client (only `with_client` does). - assert!(catalog_config.client.is_none()); - - // `uri` is consumed into its own field; other props are retained. - assert_eq!(catalog_config.props.get("a"), Some(&"b".to_string())); - assert!(!catalog_config.props.contains_key(REST_CATALOG_PROP_URI)); + let expected_props = HashMap::from([ + ( + REST_CATALOG_PROP_URI.to_string(), + "http://localhost:8080".to_string(), + ), + ("a".to_string(), "b".to_string()), + ]); + assert_eq!(catalog.name, "test"); + assert_eq!(catalog.props, expected_props); + assert!(catalog.http_client.get().is_none()); } #[tokio::test] @@ -5520,7 +5887,7 @@ mod tests { let catalog = RestCatalog::new( SessionContext::empty(), - RestCatalogConfig::builder().uri(server.url()).build(), + RestCatalogTestFixture::builder().uri(server.url()).build(), None, Some(Arc::new(LocalFsStorageFactory)), Runtime::current(), diff --git a/crates/catalog/rest/src/client.rs b/crates/catalog/rest/src/client.rs index f81f86574b..013cecee87 100644 --- a/crates/catalog/rest/src/client.rs +++ b/crates/catalog/rest/src/client.rs @@ -23,7 +23,7 @@ use reqwest::header::HeaderMap; use reqwest::{Client, IntoUrl, Method, RequestBuilder}; use serde::de::DeserializeOwned; -use crate::RestCatalogConfig; +use crate::RestCatalogProperties; use crate::auth::AuthSession; use crate::request::HttpRequest; use crate::response::HttpResponse; @@ -89,30 +89,22 @@ impl HttpClient { HttpResponse::read(response).await } - /// Create a new http client. - pub(crate) fn new(cfg: &RestCatalogConfig) -> Result { - Ok(HttpClient { - client: cfg.client(), - extra_headers: cfg.extra_headers()?, + /// Creates an HTTP client with typed catalog settings. + pub(crate) fn new(client: Client, cfg: &RestCatalogProperties) -> Self { + HttpClient { + client, + extra_headers: cfg.extra_headers(), disable_header_redaction: cfg.disable_header_redaction(), - }) + } } - /// Update the http client with new configuration. - /// - /// If cfg carries new value, we will use cfg instead. - /// Otherwise, we will keep the old value. - pub(crate) fn update_with(self, cfg: &RestCatalogConfig) -> Result { - let extra_headers = (!cfg.extra_headers()?.is_empty()) - .then(|| cfg.extra_headers()) - .transpose()? - .unwrap_or(self.extra_headers); - Ok(HttpClient { - // `cfg.client()` returns the same shared client. - client: cfg.client(), - extra_headers, + /// Applies resolved catalog settings to the shared HTTP transport. + pub(crate) fn update_with(self, cfg: &RestCatalogProperties) -> Self { + HttpClient { + client: self.client, + extra_headers: cfg.extra_headers(), disable_header_redaction: cfg.disable_header_redaction(), - }) + } } /// Testing only: the bearer token `session` would attach. @@ -254,6 +246,7 @@ pub(crate) fn deserialize_unexpected_catalog_error( #[cfg(test)] mod tests { use super::*; + use crate::catalog::RestCatalogTestFixture; #[tokio::test] async fn test_a_truncated_body_error_names_the_url() { @@ -274,7 +267,10 @@ mod tests { }); let url = format!("http://{addr}/token"); - let err = HttpClient::new(&RestCatalogConfig::builder().uri(url.clone()).build()) + let err = RestCatalogTestFixture::builder() + .uri(url.clone()) + .build() + .http_client() .unwrap() .post_form(&url, &HeaderMap::new(), &HashMap::new()) .await @@ -294,7 +290,10 @@ mod tests { .create_async() .await; - let response = HttpClient::new(&RestCatalogConfig::builder().uri(server.url()).build()) + let response = RestCatalogTestFixture::builder() + .uri(server.url()) + .build() + .http_client() .unwrap() .post_form( &format!("{}/token", server.url()), @@ -347,23 +346,22 @@ mod tests { .create_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(); + RestCatalogTestFixture::builder() + .uri(server.url()) + .props(HashMap::from([( + "header.authorization".to_string(), + "Basic xyz".to_string(), + )])) + .build() + .http_client() + .unwrap() + .post_form( + &format!("{}/token", server.url()), + &HeaderMap::new(), + &HashMap::new(), + ) + .await + .unwrap(); mock.assert_async().await; } @@ -391,7 +389,7 @@ mod tests { #[tokio::test] async fn test_http_client_debug_redacts_headers() { - let config = RestCatalogConfig::builder() + let config = RestCatalogTestFixture::builder() .uri("http://localhost".to_string()) .props(HashMap::from([ ("header.authorization".to_string(), "Basic xyz".to_string()), @@ -405,7 +403,7 @@ mod tests { ), ])) .build(); - let client = HttpClient::new(&config).unwrap(); + let client = config.http_client().unwrap(); let out = format!("{client:?}"); assert!(!out.contains("Basic xyz")); diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 701cb7f95a..26e501840f 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1493,6 +1493,9 @@ pub fn iceberg::sensitive::SensitiveString::from(value: alloc::string::String) - impl core::fmt::Debug for iceberg::sensitive::SensitiveString pub fn iceberg::sensitive::SensitiveString::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result impl core::marker::StructuralPartialEq for iceberg::sensitive::SensitiveString +impl core::str::traits::FromStr for iceberg::sensitive::SensitiveString +pub type iceberg::sensitive::SensitiveString::Err = core::convert::Infallible +pub fn iceberg::sensitive::SensitiveString::from_str(value: &str) -> core::result::Result pub mod iceberg::spec pub use iceberg::spec::ByteBuf pub mod iceberg::spec::geospatial diff --git a/crates/iceberg/src/sensitive.rs b/crates/iceberg/src/sensitive.rs index f7f92efd62..028f01358b 100644 --- a/crates/iceberg/src/sensitive.rs +++ b/crates/iceberg/src/sensitive.rs @@ -18,6 +18,7 @@ //! This module contains types to keep sensitive data in memory. use std::fmt; +use std::str::FromStr; use zeroize::Zeroizing; @@ -81,6 +82,14 @@ impl From for SensitiveString { } } +impl FromStr for SensitiveString { + type Err = std::convert::Infallible; + + fn from_str(value: &str) -> Result { + Ok(Self::from(value.to_string())) + } +} + /// Wrapper for sensitive byte data (encryption keys, DEKs, etc.) that: /// - Zeroizes memory on drop /// - Redacts content in [`Debug`] and [`std::fmt::Display`] output @@ -136,7 +145,9 @@ mod tests { fn test_sensitive_string_redacts_debug_value() { let sensitive_value = "my-pw-12346"; - let logged = format!("{:?}", SensitiveString::from(sensitive_value.to_string())); + let parsed = sensitive_value.parse::().unwrap(); + assert_eq!(parsed, SensitiveString::from(sensitive_value.to_string())); + let logged = format!("{parsed:?}"); assert!(!logged.contains(sensitive_value)); }