diff --git a/native/core/src/parquet/objectstore/s3.rs b/native/core/src/parquet/objectstore/s3.rs index a0a514885a3..0ffaff25033 100644 --- a/native/core/src/parquet/objectstore/s3.rs +++ b/native/core/src/parquet/objectstore/s3.rs @@ -25,6 +25,7 @@ use crate::execution::jni_api::get_runtime; use crate::parquet::objectstore::location_scoped::{ LocationScopedObjectStore, LocationSource, LocationStoreFactory, }; +use crate::parquet::objectstore::url_redaction::without_credentials; use async_trait::async_trait; use aws_config::{ ecs::EcsCredentialsProvider, environment::EnvironmentVariableCredentialsProvider, @@ -69,11 +70,12 @@ pub fn create_store( configs: &HashMap, min_ttl: Duration, ) -> Result<(Box, Path), object_store::Error> { - let (scheme, path) = ObjectStoreScheme::parse(url)?; + // The parser quotes the URL it rejects. + let (scheme, path) = ObjectStoreScheme::parse(&without_credentials(url))?; if scheme != ObjectStoreScheme::AmazonS3 { return Err(object_store::Error::Generic { store: "S3", - source: format!("Scheme of URL is not S3: {url}").into(), + source: format!("Scheme of URL is not S3: {}", url.scheme()).into(), }); } let path = Path::parse(path)?; @@ -180,7 +182,8 @@ impl S3StoreTemplate { }; Ok(Self { - url: url.to_string(), + // The builder reads only the bucket from the URL and quotes the URL in its errors. + url: without_credentials(url).to_string(), region, s3_configs, }) @@ -1322,6 +1325,51 @@ mod tests { assert!(store.is_ok(), "{:?}", store.err()); } + #[test] + fn create_store_errors_do_not_echo_userinfo_or_query() { + let error_for = |raw: &str| { + let url = Url::parse(raw).unwrap(); + let err = match create_store(&url, &HashMap::new(), Duration::from_secs(300)) { + Err(e) => e.to_string(), + Ok(_) => panic!("{raw} built a store"), + }; + for leaked in ["alice", "SECRET", "SAS-MARKER"] { + assert!(!err.contains(leaked), "{leaked} echoed: {err}"); + } + err + }; + let not_s3 = error_for("abfss://alice:SECRET@acct.dfs.core.windows.net/p?sig=SAS-MARKER"); + assert!( + not_s3.ends_with("Scheme of URL is not S3: abfss"), + "{not_s3}" + ); + assert!(!not_s3.contains("acct"), "host echoed: {not_s3}"); + let unknown = error_for("foo://alice:SECRET@host/dir/f.parquet?sig=SAS-MARKER"); + assert!(unknown.contains("Unable to recognise URL"), "{unknown}"); + assert!(unknown.contains("foo://host"), "{unknown}"); + } + + /// The builder reads only the bucket from the template URL and quotes the URL in its errors. + #[test] + fn template_url_carries_no_credentials() { + let bucket = "comet-template-credentials-test-bucket"; + // A configured region skips the region lookup. + let configs = HashMap::from([( + "fs.s3a.endpoint.region".to_string(), + "us-west-2".to_string(), + )]); + let url = Url::parse(&format!("s3a://alice:SECRET@{bucket}/k?sig=SAS-MARKER")).unwrap(); + let template = S3StoreTemplate::new(&url, &configs, bucket).unwrap(); + for leaked in ["alice", "SECRET", "SAS-MARKER"] { + assert!( + !template.url.contains(leaked), + "{leaked} echoed: {}", + template.url + ); + } + assert_eq!(template.url, format!("s3a://{bucket}/k")); + } + #[test] fn test_get_config_trimmed() { let configs = TestConfigBuilder::new() diff --git a/native/core/src/parquet/objectstore/s3_blob_fs_support.rs b/native/core/src/parquet/objectstore/s3_blob_fs_support.rs index 4c55ec8c4b1..17afb0c7c0d 100644 --- a/native/core/src/parquet/objectstore/s3_blob_fs_support.rs +++ b/native/core/src/parquet/objectstore/s3_blob_fs_support.rs @@ -77,14 +77,7 @@ pub(crate) fn normalize_object_store_url( url_str: &str, object_store_configs: &HashMap, ) -> Result { - let url = Url::parse(url_str).map_err(|e| { - // `url_str` may carry credentials; `ParseError` does not quote it. - let message = match scheme_of_unparsed(url_str) { - Some(scheme) => format!("Error parsing {scheme} URL: {e}"), - None => format!("Error parsing URL: {e}"), - }; - ExecutionError::GeneralError(message) - })?; + let url = Url::parse(url_str).map_err(|e| parse_error(url_str, e))?; if is_hdfs_scheme(&url, object_store_configs) { return Ok(NormalizedObjectStoreUrl { url, is_hdfs: true }); } @@ -101,6 +94,16 @@ pub(crate) fn normalize_object_store_url( }) } +/// `url_str` may carry credentials, so the message names only its scheme. `ParseError` does not +/// quote the input. +fn parse_error(url_str: &str, e: url::ParseError) -> ExecutionError { + let message = match scheme_of_unparsed(url_str) { + Some(scheme) => format!("Error parsing {scheme} URL: {e}"), + None => format!("Error parsing URL: {e}"), + }; + ExecutionError::GeneralError(message) +} + /// The text before `://` in a URL that failed to parse, if it is a valid scheme. fn scheme_of_unparsed(url_str: &str) -> Option<&str> { let (scheme, _) = url_str.split_once("://")?; @@ -150,7 +153,7 @@ fn rewrite_alias_to_s3(mut url: Url) -> Result { let (bucket, key) = trimmed.split_once('/').unwrap_or((trimmed, "")); if bucket.is_empty() { return Err(ExecutionError::GeneralError(format!( - "{}:// URL is missing bucket name: {url}", + "{}:// URL is missing bucket name", url.scheme() ))); } @@ -182,8 +185,7 @@ fn rewrite_alias_to_s3(mut url: Url) -> Result { /// raw recorded string iceberg-rust matches positional/equality deletes against, so promotion /// cannot desync the two and silently drop deletes. pub(crate) fn promote_hostless_alias_url(path: &str) -> Result, ExecutionError> { - let url = Url::parse(path) - .map_err(|e| ExecutionError::GeneralError(format!("Error parsing URL {path}: {e}")))?; + let url = Url::parse(path).map_err(|e| parse_error(path, e))?; if url.host_str().is_some() { return Ok(Cow::Borrowed(path)); } @@ -523,4 +525,30 @@ mod tests { // than opening a bucketless URL. (The JVM gate already declines these; defense in depth.) assert!(promote_hostless_alias_url("blob:///").is_err()); } + + #[test] + fn promote_errors_do_not_echo_the_url() { + let err = + promote_hostless_alias_url("blob://alice:SECRET@bucket:notaport/x?sig=SAS-MARKER") + .unwrap_err() + .to_string(); + for leaked in ["alice", "SECRET", "SAS-MARKER"] { + assert!(!err.contains(leaked), "{leaked} echoed: {err}"); + } + assert!( + err.contains("Error parsing blob URL: invalid port number"), + "{err}" + ); + } + + #[test] + fn missing_bucket_error_does_not_echo_the_query() { + let err = match normalize_object_store_url("blob:///?sig=SAS-MARKER", &configs_with("blob")) + { + Err(e) => e.to_string(), + Ok(_) => panic!("a URL with no bucket was normalized"), + }; + assert!(!err.contains("SAS-MARKER"), "SAS-MARKER echoed: {err}"); + assert!(err.contains("blob:// URL is missing bucket name"), "{err}"); + } } diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 1a4601a7503..50971806057 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -63,7 +63,9 @@ use super::objectstore; use super::objectstore::s3_blob_fs_support::{ normalize_object_store_url, NormalizedObjectStoreUrl, }; -use super::objectstore::url_redaction::{authority_without_password, without_credentials}; +use super::objectstore::url_redaction::{ + authority_without_password, without_credentials, without_secrets, +}; pub(crate) fn duplicate_parquet_field_error(name: &str) -> DataFusionError { DataFusionError::Execution(format!("Found duplicate Parquet field name '{name}'")) @@ -855,10 +857,13 @@ fn get_name_node_uri(url: &Url) -> Result { } Ok(uri_builder) } else { - Err(object_store::Error::InvalidPath { - source: object_store::path::Error::InvalidPath { - path: std::path::PathBuf::from(url.as_str()), - }, + Err(object_store::Error::Generic { + store: "hdfs-opendal", + source: format!( + "{}:// URL has no host, so no name node can be derived", + url.scheme() + ) + .into(), }) } } @@ -1078,7 +1083,14 @@ pub(crate) fn prepare_object_store_with_config_hash( } else if is_azure_scheme(url.scheme()) { objectstore::azure::create_store(url, &backend_configs) } else { - parse_url(url) + // The builders and the HTTP store quote the URL in their errors. None of them + // reads the password or query; an `az` URL's username is read as its container. + let url = if url.scheme() == "az" { + without_secrets(url) + } else { + without_credentials(url) + }; + parse_url(&url) } .map_err(|e| ExecutionError::GeneralError(e.to_string()))?; @@ -2172,6 +2184,104 @@ mod tests { assert!(leaks.is_empty(), "credentials echoed: {leaks:#?}"); } + /// object_store sends these hosts to the S3 and Azure builders by suffix, and the builders + /// reject the ones they cannot split, quoting the URL. + #[test] + fn store_builder_errors_do_not_echo_userinfo_or_query() { + let runtime = Arc::new(RuntimeEnv::default()); + for (url, kept) in [ + ( + "https://alice:SECRET@x.foo.amazonaws.com/p?sig=SAS-MARKER", + "https://x.foo.amazonaws.com/p", + ), + ( + "https://alice:SECRET@a.b.dfs.core.windows.net/p?sig=SAS-MARKER", + "https://a.b.dfs.core.windows.net/p", + ), + ( + "az://alice:SECRET@acct.dfs.core.chinacloudapi.cn/p?sig=SAS-MARKER", + "az://alice@acct.dfs.core.chinacloudapi.cn/p", + ), + ] { + let err = prepare_object_store_with_configs( + Arc::clone(&runtime), + url.into(), + &HashMap::new(), + ) + .expect_err("the builder rejects the host") + .to_string(); + for leaked in ["SECRET", "SAS-MARKER"] { + assert!(!err.contains(leaked), "{leaked} echoed: {err}"); + } + if !url.starts_with("az:") { + assert!(!err.contains("alice"), "alice echoed: {err}"); + } + assert!(err.contains("URL did not match any known pattern"), "{err}"); + assert!(err.contains(kept), "{err}"); + } + } + + /// The HTTP store quotes its URL in request errors. Plain http is refused before any request + /// is sent, so this needs no server. + #[tokio::test] + async fn http_store_errors_do_not_echo_userinfo_or_query() { + use object_store::ObjectStoreExt; + + let host = "127.0.0.1:9"; + let runtime = Arc::new(RuntimeEnv::default()); + let (url, path, _) = prepare_object_store_with_configs( + Arc::clone(&runtime), + format!("http://alice:SECRET@{host}/f.parquet?sig=SAS-MARKER"), + &HashMap::new(), + ) + .unwrap(); + let err = runtime + .object_store(&url) + .unwrap() + .head(&path) + .await + .expect_err("plain http is not allowed") + .to_string(); + object_store_cache() + .write() + .unwrap() + .retain(|(key, _, _), _| !key.contains(host)); + for leaked in ["alice", "SECRET", "SAS-MARKER"] { + assert!(!err.contains(leaked), "{leaked} echoed: {err}"); + } + assert!(err.contains("127.0.0.1"), "{err}"); + } + + /// The Azure builder reads the container of an `az` URL from its username. + #[test] + fn az_url_keeps_its_container() { + let host = "comet-az-container.dfs.core.windows.net"; + let result = prepare_object_store_with_configs( + Arc::new(RuntimeEnv::default()), + format!("az://data:SECRET@{host}/p/f.parquet?sig=SAS-MARKER"), + &HashMap::new(), + ); + object_store_cache() + .write() + .unwrap() + .retain(|(key, _, _), _| !key.contains(host)); + assert!(result.is_ok(), "{:?}", result.err()); + } + + #[cfg(feature = "hdfs-opendal")] + #[test] + fn hostless_hdfs_url_error_names_only_the_scheme() { + let url = url::Url::parse("hdfs:///tmp/x?sig=SAS-MARKER").unwrap(); + assert_eq!(url.host(), None); + let err = super::get_name_node_uri(&url).unwrap_err().to_string(); + for leaked in ["SAS-MARKER", "/tmp/x"] { + assert!(!err.contains(leaked), "{leaked} echoed: {err}"); + } + assert!(err.contains("hdfs://"), "{err}"); + let url = url::Url::parse("hdfs://nn:8020/t").unwrap(); + assert_eq!(super::get_name_node_uri(&url).unwrap(), "hdfs://nn:8020"); + } + /// Credentials come from the configuration, so the key keeps the container but no password. #[test] fn abfs_password_stays_out_of_store_identity() {