From c447c1ecf2a7117c22a613dbd1fbcade0787d632 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Tue, 20 Oct 2026 15:01:24 +0700 Subject: [PATCH 1/7] fix: keep URL passwords and SAS queries out of Azure store errors and logs The native Azure store printed whole URLs in its errors and debug lines, and handed them to object_store, whose own errors quote the URL. A password in the userinfo or a SAS token in the query could end up in task errors and logs. object_store now gets a copy without the password, query and fragment, keeping the container it reads from the username. Messages name the URL by scheme, container and host only. --- native/core/src/parquet/objectstore/azure.rs | 132 +++++++++++++++++-- 1 file changed, 123 insertions(+), 9 deletions(-) diff --git a/native/core/src/parquet/objectstore/azure.rs b/native/core/src/parquet/objectstore/azure.rs index 4b49cc07eb3..d51ef1f9ed5 100644 --- a/native/core/src/parquet/objectstore/azure.rs +++ b/native/core/src/parquet/objectstore/azure.rs @@ -120,9 +120,12 @@ pub fn create_store( url: &Url, configs: &HashMap, ) -> Result<(Box, Path), object_store::Error> { - let (scheme, path) = ObjectStoreScheme::parse(url)?; + let (scheme, path) = ObjectStoreScheme::parse(&without_credentials(url))?; if scheme != ObjectStoreScheme::MicrosoftAzure { - return Err(config_error(format!("Scheme of URL is not Azure: {url}"))); + return Err(config_error(format!( + "Scheme of URL is not Azure: {}", + redacted(url) + ))); } let path = Path::parse(path)?; let store = builder_for(url, configs)?.build()?; @@ -137,17 +140,21 @@ fn builder_for( let marker = |key: &str| configs.get(key).map(String::as_str); match marker(MARKER_RESOLUTION) { Some("none") => { - debug!("Azure authentication for {url}: none configured in Hadoop, reading AZURE_*"); - Ok(MicrosoftAzureBuilder::from_env().with_url(url.to_string())) + debug!( + "Azure authentication for {}: none configured in Hadoop, reading AZURE_*", + display_authority(url) + ); + Ok(MicrosoftAzureBuilder::from_env().with_url(without_secrets(url).to_string())) } Some("resolved") => { let settings = resolved_settings(configs)?; debug!( - "Azure authentication for {url}: {} with keys {:?}", + "Azure authentication for {}: {} with keys {:?}", + display_authority(url), marker(MARKER_AUTH_TYPE).unwrap_or_default(), settings.iter().map(|(k, _)| k.as_ref()).collect::>() ); - let builder = MicrosoftAzureBuilder::new().with_url(url.to_string()); + let builder = MicrosoftAzureBuilder::new().with_url(without_secrets(url).to_string()); Ok(settings .into_iter() .fold(builder, |b, (key, value)| b.with_config(key, value))) @@ -162,9 +169,8 @@ fn builder_for( )) } Some("error") => Err(config_error(format!( - "Hadoop ABFS authentication failed on the driver for {}://{}: {}: {}", - url.scheme(), - &url[Position::BeforeUsername..Position::AfterPort], + "Hadoop ABFS authentication failed on the driver for {}: {}: {}", + display_authority(url), marker(MARKER_ERROR_CLASS).unwrap_or(""), marker(MARKER_ERROR_MESSAGE).unwrap_or(""), ))), @@ -322,6 +328,50 @@ fn authority_host(authority: &str) -> String { authority.trim_end_matches('/').to_string() } +// A URL's password and query (where a SAS token sits) can carry a credential, so no message +// or URL handed to object_store, which quotes URLs in its errors, keeps them. The path is not +// treated as secret. + +/// `url` without its password, query and fragment. The username stays: in an ABFS URL it is the +/// container, which `MicrosoftAzureBuilder` reads. +fn without_secrets(url: &Url) -> Url { + let mut stripped = url.clone(); + stripped.set_query(None); + stripped.set_fragment(None); + // Fails only for a URL that cannot have userinfo, so there is nothing to clear. + let _ = stripped.set_password(None); + stripped +} + +/// [`without_secrets`] without the username too, for a URL that may not be Azure, where the +/// username can be an access key ID. `ObjectStoreScheme::parse` reads only the scheme, host and +/// path. +fn without_credentials(url: &Url) -> Url { + let mut stripped = without_secrets(url); + // Fails only for a URL that cannot have userinfo, so there is nothing to clear. + let _ = stripped.set_username(""); + stripped +} + +/// `scheme://[container@]host[:port]` of an ABFS URL, for messages. +fn display_authority(url: &Url) -> String { + let safe = without_secrets(url); + format!( + "{}://{}", + safe.scheme(), + &safe[Position::BeforeUsername..Position::AfterPort] + ) +} + +/// `scheme://host[:port]` of a URL that may not be Azure, for messages. +fn redacted(url: &Url) -> String { + format!( + "{}://{}", + url.scheme(), + &url[Position::BeforeHost..Position::AfterPort] + ) +} + fn config_error(message: String) -> object_store::Error { object_store::Error::Generic { store: "MicrosoftAzure", @@ -529,6 +579,57 @@ mod tests { assert!(err.contains("Scheme of URL is not Azure"), "{err}"); } + #[test] + fn create_store_errors_do_not_echo_userinfo_or_query() { + let configs = with(shared_key(), HADOOP_KEY, KEY); + let error_for = |raw: String| { + let url = Url::parse(&raw).unwrap(); + let err = create_store(&url, &configs).unwrap_err().to_string(); + for leaked in ["alice", SECRET, "SAS-MARKER"] { + assert!(!err.contains(leaked), "{leaked} echoed: {err}"); + } + err + }; + let not_azure = error_for(format!( + "s3://alice:{SECRET}@bucket/dir/file.parquet?sig=SAS-MARKER" + )); + assert!( + not_azure.contains("Scheme of URL is not Azure: s3://bucket"), + "{not_azure}" + ); + assert!( + !not_azure.contains("file.parquet"), + "path echoed: {not_azure}" + ); + error_for(format!( + "foo://alice:{SECRET}@host/dir/file.parquet?sig=SAS-MARKER" + )); + } + + #[test] + fn builder_errors_do_not_echo_the_password_or_query() { + let configs = with(shared_key(), HADOOP_KEY, KEY); + // object_store recognises only the public cloud and Fabric hosts, and quotes the URL. + let url = Url::parse(&format!( + "abfss://data:{SECRET}@myacct.dfs.core.chinacloudapi.cn/path/file.parquet?sig=SAS-MARKER" + )) + .unwrap(); + let err = create_store(&url, &configs).unwrap_err().to_string(); + for leaked in [SECRET, "SAS-MARKER"] { + assert!(!err.contains(leaked), "{leaked} echoed: {err}"); + } + } + + #[test] + fn create_store_keeps_the_container_of_a_url_with_a_query() { + let configs = with(shared_key(), HADOOP_KEY, KEY); + let url = + Url::parse("abfss://data@myacct.dfs.core.windows.net/path/file.parquet?sig=SAS-MARKER") + .unwrap(); + let (_store, path) = create_store(&url, &configs).expect("store builds"); + assert_eq!(path.as_ref(), "path/file.parquet"); + } + #[test] fn client_creds_take_tenant_and_authority_host_from_the_endpoint_alone() { assert_exactly( @@ -844,6 +945,19 @@ mod tests { org.apache.hadoop.fs.azurebfs.contracts.exceptions.KeyProviderException: \ Failure to initialize configuration" ); + let with_password = Url::parse(&format!( + "abfss://data:{SECRET}@myacct.dfs.core.windows.net/path/file.parquet" + )) + .unwrap(); + let err = builder_for(&with_password, &configs) + .err() + .map(|e| e.to_string()) + .expect("an error"); + assert!(!err.contains(SECRET), "password echoed: {err}"); + assert!( + err.contains("for abfss://data@myacct.dfs.core.windows.net: "), + "{err}" + ); } #[test] From 65b5ad234ea99cffced7aca49c3d970cc1de84fb Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Tue, 20 Oct 2026 19:46:18 +0700 Subject: [PATCH 2/7] fix: keep ABFS URL passwords out of store keys, resolution errors and driver messages object_store_authority kept an ABFS URL's password, so it reached the store cache key, the registration URL, debug lines and the deregister_store error. object_store_backend parsed the raw URL and normalize_object_store_url echoed input it could not parse. On the driver, AbfsAuthResolver put the URI into forwarded errors and a warning, and the mixed-store decline reason and objectStoreKey kept the raw authority. All of these now name a URL by scheme, container and host only. The URL helpers move to objectstore/url_redaction.rs, NativeConfig gains a matching displayAuthority, objectStoreKey matches the native key again, and the resolver scrubs the URI's password and query from Hadoop messages it forwards. Credentials only come from configuration, so URLs that differ only by password share a store. --- native/core/src/parquet/objectstore/azure.rs | 48 +---- native/core/src/parquet/objectstore/mod.rs | 1 + .../parquet/objectstore/s3_blob_fs_support.rs | 43 +++- .../src/parquet/objectstore/url_redaction.rs | 66 ++++++ native/core/src/parquet/parquet_support.rs | 203 +++++++++++++++++- .../comet/objectstore/AbfsAuthResolver.scala | 28 ++- .../comet/objectstore/NativeConfig.scala | 35 ++- .../objectstore/AbfsAuthResolverSuite.scala | 43 +++- .../comet/objectstore/NativeConfigSuite.scala | 47 ++++ 9 files changed, 434 insertions(+), 80 deletions(-) create mode 100644 native/core/src/parquet/objectstore/url_redaction.rs diff --git a/native/core/src/parquet/objectstore/azure.rs b/native/core/src/parquet/objectstore/azure.rs index d51ef1f9ed5..57762f09c97 100644 --- a/native/core/src/parquet/objectstore/azure.rs +++ b/native/core/src/parquet/objectstore/azure.rs @@ -71,7 +71,7 @@ use log::debug; use std::collections::HashMap; -use url::{Position, Url}; +use url::Url; use object_store::{ azure::{AzureConfigKey, MicrosoftAzureBuilder}, @@ -79,6 +79,8 @@ use object_store::{ ObjectStore, ObjectStoreScheme, }; +use super::url_redaction::{display_authority, redacted, without_credentials, without_secrets}; + const MARKER_RESOLUTION: &str = "comet.azure.resolution"; const MARKER_AUTH_TYPE: &str = "comet.azure.auth.type"; const MARKER_OAUTH_PROVIDER_CLASS: &str = "comet.azure.oauth.provider.class"; @@ -328,50 +330,6 @@ fn authority_host(authority: &str) -> String { authority.trim_end_matches('/').to_string() } -// A URL's password and query (where a SAS token sits) can carry a credential, so no message -// or URL handed to object_store, which quotes URLs in its errors, keeps them. The path is not -// treated as secret. - -/// `url` without its password, query and fragment. The username stays: in an ABFS URL it is the -/// container, which `MicrosoftAzureBuilder` reads. -fn without_secrets(url: &Url) -> Url { - let mut stripped = url.clone(); - stripped.set_query(None); - stripped.set_fragment(None); - // Fails only for a URL that cannot have userinfo, so there is nothing to clear. - let _ = stripped.set_password(None); - stripped -} - -/// [`without_secrets`] without the username too, for a URL that may not be Azure, where the -/// username can be an access key ID. `ObjectStoreScheme::parse` reads only the scheme, host and -/// path. -fn without_credentials(url: &Url) -> Url { - let mut stripped = without_secrets(url); - // Fails only for a URL that cannot have userinfo, so there is nothing to clear. - let _ = stripped.set_username(""); - stripped -} - -/// `scheme://[container@]host[:port]` of an ABFS URL, for messages. -fn display_authority(url: &Url) -> String { - let safe = without_secrets(url); - format!( - "{}://{}", - safe.scheme(), - &safe[Position::BeforeUsername..Position::AfterPort] - ) -} - -/// `scheme://host[:port]` of a URL that may not be Azure, for messages. -fn redacted(url: &Url) -> String { - format!( - "{}://{}", - url.scheme(), - &url[Position::BeforeHost..Position::AfterPort] - ) -} - fn config_error(message: String) -> object_store::Error { object_store::Error::Generic { store: "MicrosoftAzure", diff --git a/native/core/src/parquet/objectstore/mod.rs b/native/core/src/parquet/objectstore/mod.rs index fe7f38040e9..e1b71610139 100644 --- a/native/core/src/parquet/objectstore/mod.rs +++ b/native/core/src/parquet/objectstore/mod.rs @@ -20,3 +20,4 @@ pub(crate) mod http_metrics; pub mod location_scoped; pub mod s3; pub mod s3_blob_fs_support; +pub(crate) mod url_redaction; 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 ab09e7d24c3..59100494932 100644 --- a/native/core/src/parquet/objectstore/s3_blob_fs_support.rs +++ b/native/core/src/parquet/objectstore/s3_blob_fs_support.rs @@ -77,8 +77,15 @@ pub(crate) fn normalize_object_store_url( url_str: &str, object_store_configs: &HashMap, ) -> Result { - let url = Url::parse(url_str) - .map_err(|e| ExecutionError::GeneralError(format!("Error parsing URL {url_str}: {e}")))?; + let url = Url::parse(url_str).map_err(|e| { + // `url_str` may carry credentials, so only a valid scheme is named. `ParseError` names + // the problem without quoting the input. + 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) + })?; if is_hdfs_scheme(&url, object_store_configs) { return Ok(NormalizedObjectStoreUrl { url, is_hdfs: true }); } @@ -95,6 +102,15 @@ pub(crate) fn normalize_object_store_url( }) } +/// 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("://")?; + let mut chars = scheme.chars(); + let is_scheme = chars.next().is_some_and(|c| c.is_ascii_alphabetic()) + && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '-' | '.')); + is_scheme.then_some(scheme) +} + /// True if `scheme` is a configured s3-compliant alias (`fs.comet.s3Compliant.schemes`; empty or /// unset means none). These route to `AmazonS3` but are not recognized by /// `ObjectStoreScheme::parse`. `s3a` is excluded -- object_store knows it and callers special-case @@ -303,6 +319,29 @@ mod tests { configs } + #[test] + fn parse_errors_name_only_a_valid_scheme() { + let empty = HashMap::new(); + let error = |url: &str| match normalize_object_store_url(url, &empty) { + Err(e) => e.to_string(), + Ok(_) => panic!("{url} parsed"), + }; + let err = error("s3a://alice:SECRET@bucket:notaport/x?sig=SAS-MARKER"); + assert!( + err.contains("Error parsing s3a URL: invalid port number"), + "{err}" + ); + assert!(!err.contains("alice") && !err.contains("SECRET"), "{err}"); + // The text before `://` is not a scheme, so none of the input is echoed. + let err = error("1alice:SECRET@host://x"); + assert!(err.contains("Error parsing URL: "), "{err}"); + assert!(!err.contains("alice") && !err.contains("SECRET"), "{err}"); + // Starts like a scheme, but '@' is not a scheme character. + let err = error("alice@host://x"); + assert!(err.contains("Error parsing URL: "), "{err}"); + assert!(!err.contains("alice") && !err.contains("host"), "{err}"); + } + #[test] fn test_normalize_object_store_url() { let blob = configs_with("blob"); diff --git a/native/core/src/parquet/objectstore/url_redaction.rs b/native/core/src/parquet/objectstore/url_redaction.rs new file mode 100644 index 00000000000..f126343f546 --- /dev/null +++ b/native/core/src/parquet/objectstore/url_redaction.rs @@ -0,0 +1,66 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! URL forms safe to put in messages, store keys and URLs handed to object_store, which quotes +//! URLs in its errors. A URL's password and query (where a SAS token sits) can carry a +//! credential. The path is not treated as secret. + +use std::borrow::Cow; +use url::{Position, Url}; + +/// `url` without its password, query and fragment. The username stays: in an ABFS URL it is the +/// container, which `MicrosoftAzureBuilder` reads. +pub(crate) fn without_secrets(url: &Url) -> Url { + let mut stripped = url.clone(); + stripped.set_query(None); + stripped.set_fragment(None); + // Fails only for a URL that cannot have userinfo, so there is nothing to clear. + let _ = stripped.set_password(None); + stripped +} + +/// [`without_secrets`] without the username too, for a URL that may not be Azure, where the +/// username can be an access key ID. `ObjectStoreScheme::parse` reads only the scheme, host and +/// path. +pub(crate) fn without_credentials(url: &Url) -> Url { + let mut stripped = without_secrets(url); + // Fails only for a URL that cannot have userinfo, so there is nothing to clear. + let _ = stripped.set_username(""); + stripped +} + +/// `[username@]host[:port]` of `url`, without the password. +pub(crate) fn authority_without_password(url: &Url) -> Cow<'_, str> { + if url.password().is_none() { + return Cow::Borrowed(&url[Position::BeforeUsername..Position::AfterPort]); + } + Cow::Owned(without_secrets(url)[Position::BeforeUsername..Position::AfterPort].to_string()) +} + +/// `scheme://[container@]host[:port]` of an ABFS URL, for messages. +pub(crate) fn display_authority(url: &Url) -> String { + format!("{}://{}", url.scheme(), authority_without_password(url)) +} + +/// `scheme://host[:port]` of a URL that may not be Azure, for messages. +pub(crate) fn redacted(url: &Url) -> String { + format!( + "{}://{}", + url.scheme(), + &url[Position::BeforeHost..Position::AfterPort] + ) +} diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 997bd72bbb8..1e292053eb0 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -45,6 +45,7 @@ use log::debug; use object_store::path::Path; use object_store::{parse_url, ObjectStore, ObjectStoreScheme}; use parquet::arrow::PARQUET_FIELD_ID_META_KEY; +use std::borrow::Cow; use std::collections::HashMap; use std::sync::OnceLock; use std::time::Duration; @@ -60,6 +61,7 @@ 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}; pub(crate) fn duplicate_parquet_field_error(name: &str) -> DataFusionError { DataFusionError::Execution(format!("Found duplicate Parquet field name '{name}'")) @@ -687,14 +689,14 @@ fn uses_azure_container_key(scheme: &str) -> bool { || scheme.starts_with("abfss+comet-") } -pub(crate) fn object_store_authority(url: &Url) -> &str { - let start = if uses_azure_container_key(url.scheme()) { - // ABFS URLs encode the container in the userinfo. - url::Position::BeforeUsername +/// `host[:port]` of `url`, prefixed with `container@` for ABFS, which encodes the container in +/// the username. Never includes the password. +pub(crate) fn object_store_authority(url: &Url) -> Cow<'_, str> { + if uses_azure_container_key(url.scheme()) { + authority_without_password(url) } else { - url::Position::BeforeHost - }; - &url[start..url::Position::AfterPort] + Cow::Borrowed(&url[url::Position::BeforeHost..url::Position::AfterPort]) + } } fn registry_url_key(url: &Url) -> String { @@ -733,7 +735,10 @@ impl ObjectStoreRegistry for CometObjectStoreRegistry { .unwrap_or_else(PoisonError::into_inner) .remove(®istry_url_key(url)) .ok_or_else(|| { - DataFusionError::Internal(format!("No suitable object store found for {url}")) + DataFusionError::Internal(format!( + "No suitable object store found for {}", + registry_url_key(url) + )) }) } else { self.default.deregister_store(url) @@ -909,8 +914,9 @@ fn object_store_backend(url: &Url, is_hdfs: bool) -> Result ObjectStoreBackend::Local, ObjectStoreScheme::AmazonS3 @@ -1138,6 +1144,27 @@ mod tests { ("gs://bucket/k.parquet", "", "", "gs://bucket", false), (abfss_path.as_str(), "", "", abfss, false), (abfs_path.as_str(), "", "", abfs, false), + ( + "abfss://container:secret@account.dfs.core.windows.net/k.parquet", + "", + "", + abfss, + false, + ), + ( + "abfss://container:secret@account.dfs.core.windows.net:8443/k.parquet", + "", + "", + "abfss://container@account.dfs.core.windows.net:8443", + false, + ), + ( + "abfss://:secret@account.dfs.core.windows.net/k.parquet", + "", + "", + "abfss://account.dfs.core.windows.net", + false, + ), ( "wasbs://container@account.blob.core.windows.net/k.parquet", "", @@ -1179,6 +1206,10 @@ mod tests { "http://example.com/path", "https://example.com/path", "https://account.blob.core.windows.net/container/path", + // Userinfo and query are stripped before parsing and do not change the class. + "s3://alice:SECRET@bucket/path?sig=SAS-MARKER", + "abfss://data:SECRET@acct.dfs.core.windows.net/path?sig=SAS-MARKER", + "https://alice:SECRET@account.blob.core.windows.net/container/path?sig=SAS-MARKER", ] { let url = url::Url::parse(address).unwrap(); assert_eq!( @@ -1793,6 +1824,158 @@ mod tests { )); } + #[test] + fn object_store_authority_drops_only_the_abfs_password() { + for (url, expected) in [ + ( + "abfss://data:SECRET@acct.dfs.core.windows.net/p?sig=SAS-MARKER", + "data@acct.dfs.core.windows.net", + ), + ( + "abfs://data@acct.dfs.core.windows.net:8443/p", + "data@acct.dfs.core.windows.net:8443", + ), + ( + "abfss://:SECRET@acct.dfs.core.windows.net/p", + "acct.dfs.core.windows.net", + ), + ( + "abfss+comet-0123456789abcdef-native://data:SECRET@acct.dfs.core.windows.net/", + "data@acct.dfs.core.windows.net", + ), + ( + "s3://alice:SECRET@bucket:9000/k?sig=SAS-MARKER", + "bucket:9000", + ), + ("hdfs://nn:8020/t", "nn:8020"), + ] { + let parsed = url::Url::parse(url).unwrap(); + assert_eq!(super::object_store_authority(&parsed), expected, "{url}"); + } + } + + #[test] + fn deregister_error_does_not_echo_the_password_or_query() { + use datafusion::execution::object_store::ObjectStoreRegistry; + + let url = url::Url::parse("abfss://data:SECRET@acct.dfs.core.windows.net/p?sig=SAS-MARKER") + .unwrap(); + let err = super::CometObjectStoreRegistry::default() + .deregister_store(&url) + .expect_err("nothing is registered") + .to_string(); + for leaked in ["SECRET", "SAS-MARKER"] { + assert!(!err.contains(leaked), "{leaked} echoed: {err}"); + } + assert!( + err.contains("abfss://data@acct.dfs.core.windows.net"), + "{err}" + ); + } + + #[test] + fn prepare_errors_do_not_echo_userinfo_or_query() { + let runtime = Arc::new(RuntimeEnv::default()); + let options = HashMap::from([ + ("comet.azure.resolution".into(), "resolved".into()), + ("comet.azure.auth.type".into(), "SharedKey".into()), + ("fs.azure.account.key".into(), "c2VjcmV0".into()), + ]); + let errors: Vec<(String, &str)> = [ + // Rejected by the Azure builder, which recognises only public cloud and Fabric hosts. + // The username is the container here, so it may appear. + ( + "abfss://data:SECRET@acct.dfs.core.chinacloudapi.cn/x?sig=SAS-MARKER", + "URL did not match any known pattern", + ), + // Rejected by `ObjectStoreScheme::parse`. + ( + "foo://alice:SECRET@host/x?sig=SAS-MARKER", + "Unable to recognise URL", + ), + // Rejected by `Url::parse`. + ( + "s3a://alice:SECRET@bucket:notaport/x?sig=SAS-MARKER", + "invalid port number", + ), + ] + .into_iter() + .map(|(url, reason)| { + let err = prepare_object_store_with_configs(Arc::clone(&runtime), url.into(), &options) + .expect_err("the URL is rejected") + .to_string(); + (err, reason) + }) + .collect(); + for (err, reason) in &errors { + assert!(err.contains(reason), "{err}"); + } + assert!( + errors[2].0.contains("Error parsing s3a URL"), + "{}", + errors[2].0 + ); + let leaks: Vec<&String> = errors + .iter() + .map(|(err, _)| err) + .filter(|err| { + ["alice", "SECRET", "SAS-MARKER"] + .iter() + .any(|s| err.contains(s)) + }) + .collect(); + assert!(leaks.is_empty(), "credentials echoed: {leaks:#?}"); + } + + /// The ABFS password stays out of the cache key and registration URL, so two URLs that + /// differ only by password share one store. The container stays in both. + #[test] + fn abfs_password_stays_out_of_store_identity() { + use datafusion::execution::runtime_env::RuntimeEnvBuilder; + + let host = "comet-password-identity.dfs.core.windows.net"; + let options = HashMap::from([ + ("comet.azure.resolution".into(), "resolved".into()), + ("comet.azure.auth.type".into(), "SharedKey".into()), + ("fs.azure.account.key".into(), "c2VjcmV0".into()), + ]); + let runtime = RuntimeEnvBuilder::new() + .with_object_store_registry(Arc::new(super::CometObjectStoreRegistry::default())) + .build_arc() + .unwrap(); + let register = |password: &str| { + let (url, _, _) = prepare_object_store_with_configs( + Arc::clone(&runtime), + format!("abfss://data:{password}@{host}/p/file.parquet?sig=SAS-MARKER"), + &options, + ) + .unwrap(); + (url.clone(), runtime.object_store(&url).unwrap()) + }; + + let (url, store) = register("SECRET"); + for leaked in ["SECRET", "SAS-MARKER"] { + assert!(!url.as_str().contains(leaked), "{leaked} in {url}"); + } + assert!(url.as_str().contains(&format!("://data@{host}")), "{url}"); + let (other_url, other_store) = register("OTHER"); + assert_eq!(other_url, url); + assert!(Arc::ptr_eq(&other_store, &store)); + + let keys: Vec = object_store_cache() + .read() + .unwrap() + .keys() + .filter(|(key, _, _)| key.contains(host)) + .map(|(key, _, _)| key.clone()) + .collect(); + object_store_cache() + .write() + .unwrap() + .retain(|(key, _, _), _| !key.contains(host)); + assert_eq!(keys, vec![format!("abfss://data@{host}")]); + } + /// Parses the url, registers the object store, and returns a tuple of the object store url and object store path #[cfg(not(feature = "hdfs-opendal"))] pub(crate) fn prepare_object_store( diff --git a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala index a7938ce79a8..cdad2dda3ef 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala @@ -208,7 +208,7 @@ private[comet] object AbfsAuthResolver extends Logging { hadoopConf: Configuration, uri: URI, handles: Try[Handles]): Outcome = { - val secrets = ArrayBuffer[String]() + val secrets = ArrayBuffer[String](uriSecrets(uri): _*) // Set once the AbfsConfiguration exists, so a failure can read the credentials it skipped. var openSession: Option[Session] = None try { @@ -257,17 +257,17 @@ private[comet] object AbfsAuthResolver extends Logging { private[objectstore] def authorityParts(uri: URI): Either[Failed, (String, String)] = { Option(uri.getRawAuthority) match { case None => - Left(Failed(Exceptions.INVALID_URI_AUTHORITY, s"$uri has invalid authority.")) + Left(Failed(Exceptions.INVALID_URI_AUTHORITY, s"${display(uri)} has invalid authority.")) case Some(authority) if !authority.contains("@") => - Left(Failed(Exceptions.INVALID_URI_AUTHORITY, s"$uri has invalid authority.")) + Left(Failed(Exceptions.INVALID_URI_AUTHORITY, s"${display(uri)} has invalid authority.")) case Some(authority) => val parts = authority.split("@", 2) if (parts.length < 2 || parts(0).isEmpty) { Left( Failed( Exceptions.INVALID_URI, - s"Invalid URI '$uri' has a malformed authority, expected container name. " + - "Authority takes the form abfs://[@]")) + s"Invalid URI '${display(uri)}' has a malformed authority, expected container " + + "name. Authority takes the form abfs://[@]")) } else { Right((parts(0), parts(1))) } @@ -351,7 +351,9 @@ private[comet] object AbfsAuthResolver extends Logging { private def sharedKey(session: Session, uri: URI, account: String): Outcome = { // Store.initializeClient checks the account name before fetching the key. if (account.indexOf('.') <= 0) { - Failed(Exceptions.INVALID_URI, s"Invalid URI $uri - account name is not fully qualified.") + Failed( + Exceptions.INVALID_URI, + s"Invalid URI ${display(uri)} - account name is not fully qualified.") } else { val key = session.storageAccountKey() if (key.isEmpty) { @@ -481,6 +483,16 @@ private[comet] object AbfsAuthResolver extends Logging { .foldLeft(withoutQuotedKey)((text, secret) => text.replace(secret, REDACTED)) } + /** The parts of `uri` that can carry a credential: the password and the raw query. */ + private[objectstore] def uriSecrets(uri: URI): Seq[String] = { + val password = Option(uri.getRawUserInfo).map(_.dropWhile(_ != ':').drop(1)) + (password ++ Option(uri.getRawQuery)).filter(_.nonEmpty).toSeq + } + + // Hadoop's messages quote the whole URI. These reach native errors and logs, so they leave out + // the password, path and query, which can carry a credential. + private def display(uri: URI): String = NativeConfig.displayAuthority(uri) + private def failed(uri: URI, e: Throwable, secrets: Iterable[String]): Failed = { val chain = causeChain(e) val thrown = chain.head @@ -489,8 +501,8 @@ private[comet] object AbfsAuthResolver extends Logging { if (verbatim) s"$name: ${redact(Option(t.getMessage).getOrElse(""), secrets)}" else name } logWarning( - s"Hadoop ABFS authentication for ${uri.getScheme}://${uri.getRawAuthority} failed on the " + - s"driver: ${chain.map(describe(_, verbatim = true)).mkString(", caused by ")}") + s"Hadoop ABFS authentication for ${display(uri)} failed on the driver: " + + s"${chain.map(describe(_, verbatim = true)).mkString(", caused by ")}") val forwarded = chain.map(t => describe(t, isVerbatim(t))) val hidden = chain.exists(t => !isVerbatim(t)) diff --git a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala index 0cdf8183bfe..05f4ee07a6b 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala @@ -132,6 +132,25 @@ object NativeConfig { rawAuthority.substring(rawAuthority.lastIndexOf('@') + 1) } + /** + * The raw authority as `[user@]host[:port]`, without the password. For ABFS the user is the + * container. As in native, the user ends at the first `:` and the host follows the last `@`. + */ + private def authorityWithoutPassword(uri: URI): String = { + val rawAuthority = Option(uri.getRawAuthority).getOrElse("") + val at = rawAuthority.lastIndexOf('@') + val user = rawAuthority.take(at).takeWhile(_ != ':') + val host = rawAuthority.substring(at + 1) + if (user.isEmpty) host else s"$user@$host" + } + + /** + * `scheme://[user@]host[:port]` of `uri` for messages: no password, path or query. A path + * without a scheme is local, as in [[objectStoreKey]]. + */ + private[comet] def displayAuthority(uri: URI): String = + s"${lowerScheme(uri).getOrElse("file")}://${authorityWithoutPassword(uri)}" + /** * The schemes native reads through libhdfs: the `fs.comet.libhdfs.schemes` list, or only `hdfs` * when it is unset or blank, as native `is_hdfs_scheme` decides. @@ -186,10 +205,10 @@ object NativeConfig { * The object store native planning resolves for `uri`. Mirrors native * `normalize_object_store_url` followed by `object_store_url_key`: a libhdfs scheme (decided on * the scheme as written) keeps its spelling, `s3a` and opted-in aliases become `s3` with a - * hostless bucket promoted from the path, user info is dropped except for ABFS containers, and - * a path without a scheme is local. The one divergence is a default port of a special scheme - * (`https://host:443`), which native drops; that only splits files across more partitions. - * Files with different keys cannot share a native partition. + * hostless bucket promoted from the path, user info is dropped except for ABFS containers + * (never their password), and a path without a scheme is local. The one divergence is a default + * port of a special scheme (`https://host:443`), which native drops; that only splits files + * across more partitions. Files with different keys cannot share a native partition. */ def objectStoreKey( uri: URI, @@ -198,7 +217,7 @@ object NativeConfig { val scheme = lowerScheme(uri).getOrElse("file") val isLibhdfs = libhdfsSchemes.contains(scheme) val authority = - if (scheme == "abfs" || scheme == "abfss") Option(uri.getRawAuthority).getOrElse("") + if (scheme == "abfs" || scheme == "abfss") authorityWithoutPassword(uri) else hostAndPort(uri) val isS3Alias = !isLibhdfs && (scheme == "s3a" || @@ -424,9 +443,13 @@ object NativeConfig { s"(${first.getScheme} and ${uri.getScheme})") } if (firstIsAbfs && uri.getRawAuthority != first.getRawAuthority) { + // Hadoop reads the whole user info as the container, so a different password is + // another container to it. + val (a, b) = (displayAuthority(first), displayAuthority(uri)) + val stores = if (a == b) s"$a, with different user info" else s"$a and $b" return Left( "Native scan does not support ABFS files from more than one container or account " + - s"(${first.getRawAuthority} and ${uri.getRawAuthority})") + s"($stores)") } } if (firstIsAbfs) { diff --git a/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala index 4d69c2e01a3..5a984f65df7 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala @@ -96,20 +96,23 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers { } test("authorityParts - no '@', empty container and null authority fail as Hadoop does") { - val noAt = "abfss://myacct.dfs.core.windows.net/p" - authorityParts(new URI(noAt)) shouldBe - Left(Failed(invalidUriAuthorityException, s"$noAt has invalid authority.")) + // Messages name the URI by scheme and authority only, so a SAS query never reaches them. + val noAt = "abfss://myacct.dfs.core.windows.net/p?sig=SAS-MARKER" + authorityParts(new URI(noAt)) shouldBe Left( + Failed( + invalidUriAuthorityException, + "abfss://myacct.dfs.core.windows.net has invalid authority.")) - val emptyContainer = "abfss://@myacct.dfs.core.windows.net/p" + val emptyContainer = "abfss://@myacct.dfs.core.windows.net/p?sig=SAS-MARKER" authorityParts(new URI(emptyContainer)) shouldBe Left( Failed( invalidUriException, - s"Invalid URI '$emptyContainer' has a malformed authority, expected container name. " + - "Authority takes the form abfs://[@]")) + "Invalid URI 'abfss://myacct.dfs.core.windows.net' has a malformed authority, expected " + + "container name. Authority takes the form abfs://[@]")) - val nullAuthority = "abfss:///p" + val nullAuthority = "abfss:///p?sig=SAS-MARKER" authorityParts(new URI(nullAuthority)) shouldBe - Left(Failed(invalidUriAuthorityException, s"$nullAuthority has invalid authority.")) + Left(Failed(invalidUriAuthorityException, "abfss:// has invalid authority.")) } // ---- reflection ---- @@ -335,7 +338,29 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers { val shortUri = new URI("abfss://data@myacct/path") val outcome = resolve(conf("fs.azure.account.key.myacct" -> "c2VjcmV0"), shortUri) outcome shouldBe - Failed(invalidUriException, s"Invalid URI $shortUri - account name is not fully qualified.") + Failed( + invalidUriException, + "Invalid URI abfss://data@myacct - account name is not fully qualified.") + } + + test("Failed - the forwarded message drops the URI's password, path and query") { + // Hadoop reads the container as everything before the first '@', password included. + val shortUri = new URI("abfss://data:SECRET@myacct/path?sig=SAS-MARKER") + val outcome = resolve(conf("fs.azure.account.key.myacct" -> "c2VjcmV0"), shortUri) + outcome shouldBe a[Failed] + val message = toOptions(outcome)(Markers.ERROR_MESSAGE) + Seq("SECRET", "SAS-MARKER", "/path").foreach(leaked => message should not include leaked) + message should include("abfss://data@myacct") + } + + test("Failed - a forwarded message is scrubbed of the URI's password and query") { + // ClassNotFoundException is forwarded verbatim, so its message stands in for a Hadoop one + // that quotes the URI. + val secretUri = new URI(s"abfss://data:the-uri-password@$account/p?sig=the-sas-signature") + val thrown = new ClassNotFoundException(s"no class for $secretUri") + val message = toOptions(resolve(conf(), secretUri, Failure(thrown)))(Markers.ERROR_MESSAGE) + message should include(s"no class for abfss://data:@$account/p?") + Seq("the-uri-password", "the-sas-signature").foreach(s => message should not include s) } test("Failed - a LinkageError from the reflection layer is an error marker, not a crash") { diff --git a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala index 3863f66105a..69cf4856b44 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala @@ -553,6 +553,20 @@ class NativeConfigSuite extends AnyFunSuite with Matchers { // ABFS keeps the container from the user info; WASB does not. Case(s"abfss://container@$account/k.parquet", "", "", s"abfss://container@$account", false), Case(s"abfs://container@$account/k.parquet", "", "", s"abfs://container@$account", false), + // The password never joins the key; credentials come from the configuration. + Case( + s"abfss://container:secret@$account/k.parquet", + "", + "", + s"abfss://container@$account", + hdfs = false), + Case(s"abfss://:secret@$account/k.parquet", "", "", s"abfss://$account", hdfs = false), + Case( + s"abfss://container:secret@$account:8443/k.parquet", + "", + "", + s"abfss://container@$account:8443", + hdfs = false), Case( "wasbs://container@account.blob.core.windows.net/k.parquet", "", @@ -715,6 +729,39 @@ class NativeConfigSuite extends AnyFunSuite with Matchers { Seq(abfsFile("data", "accta", "a.parquet"), abfsFile("data", "acctb", "b.parquet"))) } + test("extractScanObjectStoreOptions - a decline names ABFS stores without password or query") { + val hadoopConf = new Configuration() + hadoopConf.set("fs.azure.account.key.accta.dfs.core.windows.net", "a2V5LWE=") + val uris = Seq("SECRET", "OTHER").map { password => + new URI(s"abfss://data:$password@accta.dfs.core.windows.net/path/a.parquet?sig=SAS-MARKER") + } + + def declineReason(uris: Seq[URI]): String = + NativeConfig.extractScanObjectStoreOptions(hadoopConf, uris) match { + case Left(reason) => + Seq("SECRET", "OTHER", "SAS-MARKER", "/path").foreach { leaked => + reason should not include leaked + } + reason + case Right(options) => fail(s"expected a decline, got $options") + } + + declineReason(uris) should include( + "(abfss://data@accta.dfs.core.windows.net, with different user info)") + val otherContainer = + new URI("abfss://logs:OTHER@accta.dfs.core.windows.net/path/b.parquet?sig=SAS-MARKER") + declineReason(Seq(uris.head, otherContainer)) should include( + "(abfss://data@accta.dfs.core.windows.net and abfss://logs@accta.dfs.core.windows.net)") + } + + test("displayAuthority - lowercase scheme and no password, path or query") { + NativeConfig.displayAuthority( + new URI("ABFSS://data:SECRET@acct.dfs.core.windows.net:8443/p?sig=SAS-MARKER")) shouldBe + "abfss://data@acct.dfs.core.windows.net:8443" + // A path without a scheme is local, as in objectStoreKey. + NativeConfig.displayAuthority(new URI("/tmp/t/k.parquet")) shouldBe "file://" + } + test("extractScanObjectStoreOptions - files in one container keep the first file's options") { val hadoopConf = new Configuration() hadoopConf.set("fs.azure.account.key.accta.dfs.core.windows.net", "a2V5LWE=") From a3275e6689d3d66d5c550ec7effc570f1a5fc26c Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Wed, 21 Oct 2026 10:43:56 +0700 Subject: [PATCH 3/7] fix: name only the scheme when an Azure store gets a non-Azure URL The message no longer needs the host, so the redacted helper that built it is removed. --- native/core/src/parquet/objectstore/azure.rs | 11 ++++------- native/core/src/parquet/objectstore/url_redaction.rs | 9 --------- 2 files changed, 4 insertions(+), 16 deletions(-) diff --git a/native/core/src/parquet/objectstore/azure.rs b/native/core/src/parquet/objectstore/azure.rs index 57762f09c97..ab56a96073a 100644 --- a/native/core/src/parquet/objectstore/azure.rs +++ b/native/core/src/parquet/objectstore/azure.rs @@ -79,7 +79,7 @@ use object_store::{ ObjectStore, ObjectStoreScheme, }; -use super::url_redaction::{display_authority, redacted, without_credentials, without_secrets}; +use super::url_redaction::{display_authority, without_credentials, without_secrets}; const MARKER_RESOLUTION: &str = "comet.azure.resolution"; const MARKER_AUTH_TYPE: &str = "comet.azure.auth.type"; @@ -126,7 +126,7 @@ pub fn create_store( if scheme != ObjectStoreScheme::MicrosoftAzure { return Err(config_error(format!( "Scheme of URL is not Azure: {}", - redacted(url) + url.scheme() ))); } let path = Path::parse(path)?; @@ -552,13 +552,10 @@ mod tests { "s3://alice:{SECRET}@bucket/dir/file.parquet?sig=SAS-MARKER" )); assert!( - not_azure.contains("Scheme of URL is not Azure: s3://bucket"), + not_azure.ends_with("Scheme of URL is not Azure: s3"), "{not_azure}" ); - assert!( - !not_azure.contains("file.parquet"), - "path echoed: {not_azure}" - ); + assert!(!not_azure.contains("bucket"), "host echoed: {not_azure}"); error_for(format!( "foo://alice:{SECRET}@host/dir/file.parquet?sig=SAS-MARKER" )); diff --git a/native/core/src/parquet/objectstore/url_redaction.rs b/native/core/src/parquet/objectstore/url_redaction.rs index f126343f546..c50f2f05c1e 100644 --- a/native/core/src/parquet/objectstore/url_redaction.rs +++ b/native/core/src/parquet/objectstore/url_redaction.rs @@ -55,12 +55,3 @@ pub(crate) fn authority_without_password(url: &Url) -> Cow<'_, str> { pub(crate) fn display_authority(url: &Url) -> String { format!("{}://{}", url.scheme(), authority_without_password(url)) } - -/// `scheme://host[:port]` of a URL that may not be Azure, for messages. -pub(crate) fn redacted(url: &Url) -> String { - format!( - "{}://{}", - url.scheme(), - &url[Position::BeforeHost..Position::AfterPort] - ) -} From 8acc4a2a54d04b532d240866a4b7d49b54249b0d Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Thu, 22 Oct 2026 09:14:26 +0700 Subject: [PATCH 4/7] docs: keep only the comments that explain the URL redaction rules Drop comments that restate the code or a test's name, and shorten the rest to why each part of a URL is kept or removed. --- .../parquet/objectstore/s3_blob_fs_support.rs | 6 ++---- .../src/parquet/objectstore/url_redaction.rs | 14 ++++--------- native/core/src/parquet/parquet_support.rs | 13 ++++-------- .../comet/objectstore/AbfsAuthResolver.scala | 3 +-- .../comet/objectstore/NativeConfig.scala | 21 +++++++------------ .../objectstore/AbfsAuthResolverSuite.scala | 4 +--- .../comet/objectstore/NativeConfigSuite.scala | 1 - 7 files changed, 19 insertions(+), 43 deletions(-) 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 59100494932..4c55ec8c4b1 100644 --- a/native/core/src/parquet/objectstore/s3_blob_fs_support.rs +++ b/native/core/src/parquet/objectstore/s3_blob_fs_support.rs @@ -78,8 +78,7 @@ pub(crate) fn normalize_object_store_url( object_store_configs: &HashMap, ) -> Result { let url = Url::parse(url_str).map_err(|e| { - // `url_str` may carry credentials, so only a valid scheme is named. `ParseError` names - // the problem without quoting the input. + // `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}"), @@ -332,11 +331,10 @@ mod tests { "{err}" ); assert!(!err.contains("alice") && !err.contains("SECRET"), "{err}"); - // The text before `://` is not a scheme, so none of the input is echoed. + // A leading digit, then an '@': neither is a scheme, so none of the input is echoed. let err = error("1alice:SECRET@host://x"); assert!(err.contains("Error parsing URL: "), "{err}"); assert!(!err.contains("alice") && !err.contains("SECRET"), "{err}"); - // Starts like a scheme, but '@' is not a scheme character. let err = error("alice@host://x"); assert!(err.contains("Error parsing URL: "), "{err}"); assert!(!err.contains("alice") && !err.contains("host"), "{err}"); diff --git a/native/core/src/parquet/objectstore/url_redaction.rs b/native/core/src/parquet/objectstore/url_redaction.rs index c50f2f05c1e..c94bb06e5ee 100644 --- a/native/core/src/parquet/objectstore/url_redaction.rs +++ b/native/core/src/parquet/objectstore/url_redaction.rs @@ -15,15 +15,13 @@ // specific language governing permissions and limitations // under the License. -//! URL forms safe to put in messages, store keys and URLs handed to object_store, which quotes -//! URLs in its errors. A URL's password and query (where a SAS token sits) can carry a -//! credential. The path is not treated as secret. +//! URL forms for messages, store keys and object_store, which quotes URLs in its errors. The +//! password and query (a SAS token) can carry a credential; the path is not treated as secret. use std::borrow::Cow; use url::{Position, Url}; -/// `url` without its password, query and fragment. The username stays: in an ABFS URL it is the -/// container, which `MicrosoftAzureBuilder` reads. +/// Keeps the username: in an ABFS URL it is the container, which `MicrosoftAzureBuilder` reads. pub(crate) fn without_secrets(url: &Url) -> Url { let mut stripped = url.clone(); stripped.set_query(None); @@ -33,9 +31,7 @@ pub(crate) fn without_secrets(url: &Url) -> Url { stripped } -/// [`without_secrets`] without the username too, for a URL that may not be Azure, where the -/// username can be an access key ID. `ObjectStoreScheme::parse` reads only the scheme, host and -/// path. +/// [`without_secrets`] minus the username, which outside Azure can be an access key ID. pub(crate) fn without_credentials(url: &Url) -> Url { let mut stripped = without_secrets(url); // Fails only for a URL that cannot have userinfo, so there is nothing to clear. @@ -43,7 +39,6 @@ pub(crate) fn without_credentials(url: &Url) -> Url { stripped } -/// `[username@]host[:port]` of `url`, without the password. pub(crate) fn authority_without_password(url: &Url) -> Cow<'_, str> { if url.password().is_none() { return Cow::Borrowed(&url[Position::BeforeUsername..Position::AfterPort]); @@ -51,7 +46,6 @@ pub(crate) fn authority_without_password(url: &Url) -> Cow<'_, str> { Cow::Owned(without_secrets(url)[Position::BeforeUsername..Position::AfterPort].to_string()) } -/// `scheme://[container@]host[:port]` of an ABFS URL, for messages. pub(crate) fn display_authority(url: &Url) -> String { format!("{}://{}", url.scheme(), authority_without_password(url)) } diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 1e292053eb0..dd273ac94dd 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -689,10 +689,9 @@ fn uses_azure_container_key(scheme: &str) -> bool { || scheme.starts_with("abfss+comet-") } -/// `host[:port]` of `url`, prefixed with `container@` for ABFS, which encodes the container in -/// the username. Never includes the password. pub(crate) fn object_store_authority(url: &Url) -> Cow<'_, str> { if uses_azure_container_key(url.scheme()) { + // ABFS URLs encode the container in the userinfo. authority_without_password(url) } else { Cow::Borrowed(&url[url::Position::BeforeHost..url::Position::AfterPort]) @@ -1206,7 +1205,7 @@ mod tests { "http://example.com/path", "https://example.com/path", "https://account.blob.core.windows.net/container/path", - // Userinfo and query are stripped before parsing and do not change the class. + // Stripping the credentials before parsing must not change the class. "s3://alice:SECRET@bucket/path?sig=SAS-MARKER", "abfss://data:SECRET@acct.dfs.core.windows.net/path?sig=SAS-MARKER", "https://alice:SECRET@account.blob.core.windows.net/container/path?sig=SAS-MARKER", @@ -1882,18 +1881,15 @@ mod tests { ("fs.azure.account.key".into(), "c2VjcmV0".into()), ]); let errors: Vec<(String, &str)> = [ - // Rejected by the Azure builder, which recognises only public cloud and Fabric hosts. - // The username is the container here, so it may appear. + // The Azure builder recognises only public cloud and Fabric hosts. ( "abfss://data:SECRET@acct.dfs.core.chinacloudapi.cn/x?sig=SAS-MARKER", "URL did not match any known pattern", ), - // Rejected by `ObjectStoreScheme::parse`. ( "foo://alice:SECRET@host/x?sig=SAS-MARKER", "Unable to recognise URL", ), - // Rejected by `Url::parse`. ( "s3a://alice:SECRET@bucket:notaport/x?sig=SAS-MARKER", "invalid port number", @@ -1927,8 +1923,7 @@ mod tests { assert!(leaks.is_empty(), "credentials echoed: {leaks:#?}"); } - /// The ABFS password stays out of the cache key and registration URL, so two URLs that - /// differ only by password share one store. The container stays in both. + /// Credentials come from the configuration, so the key keeps the container but no password. #[test] fn abfs_password_stays_out_of_store_identity() { use datafusion::execution::runtime_env::RuntimeEnvBuilder; diff --git a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala index cdad2dda3ef..9e8cf133fa3 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala @@ -489,8 +489,7 @@ private[comet] object AbfsAuthResolver extends Logging { (password ++ Option(uri.getRawQuery)).filter(_.nonEmpty).toSeq } - // Hadoop's messages quote the whole URI. These reach native errors and logs, so they leave out - // the password, path and query, which can carry a credential. + // Messages reach native errors and logs, so they omit the password, path and query. private def display(uri: URI): String = NativeConfig.displayAuthority(uri) private def failed(uri: URI, e: Throwable, secrets: Iterable[String]): Failed = { diff --git a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala index 05f4ee07a6b..e27fb2cc3af 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala @@ -132,10 +132,7 @@ object NativeConfig { rawAuthority.substring(rawAuthority.lastIndexOf('@') + 1) } - /** - * The raw authority as `[user@]host[:port]`, without the password. For ABFS the user is the - * container. As in native, the user ends at the first `:` and the host follows the last `@`. - */ + /** As in native, the user ends at the first `:` and the host follows the last `@`. */ private def authorityWithoutPassword(uri: URI): String = { val rawAuthority = Option(uri.getRawAuthority).getOrElse("") val at = rawAuthority.lastIndexOf('@') @@ -144,10 +141,7 @@ object NativeConfig { if (user.isEmpty) host else s"$user@$host" } - /** - * `scheme://[user@]host[:port]` of `uri` for messages: no password, path or query. A path - * without a scheme is local, as in [[objectStoreKey]]. - */ + /** `scheme://[user@]host[:port]` of `uri`, for messages: no password, path or query. */ private[comet] def displayAuthority(uri: URI): String = s"${lowerScheme(uri).getOrElse("file")}://${authorityWithoutPassword(uri)}" @@ -205,10 +199,10 @@ object NativeConfig { * The object store native planning resolves for `uri`. Mirrors native * `normalize_object_store_url` followed by `object_store_url_key`: a libhdfs scheme (decided on * the scheme as written) keeps its spelling, `s3a` and opted-in aliases become `s3` with a - * hostless bucket promoted from the path, user info is dropped except for ABFS containers - * (never their password), and a path without a scheme is local. The one divergence is a default - * port of a special scheme (`https://host:443`), which native drops; that only splits files - * across more partitions. Files with different keys cannot share a native partition. + * hostless bucket promoted from the path, user info is dropped except for ABFS containers, and + * a path without a scheme is local. The one divergence is a default port of a special scheme + * (`https://host:443`), which native drops; that only splits files across more partitions. + * Files with different keys cannot share a native partition. */ def objectStoreKey( uri: URI, @@ -443,8 +437,7 @@ object NativeConfig { s"(${first.getScheme} and ${uri.getScheme})") } if (firstIsAbfs && uri.getRawAuthority != first.getRawAuthority) { - // Hadoop reads the whole user info as the container, so a different password is - // another container to it. + // Hadoop reads the whole user info, password too, as the container. val (a, b) = (displayAuthority(first), displayAuthority(uri)) val stores = if (a == b) s"$a, with different user info" else s"$a and $b" return Left( diff --git a/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala index 5a984f65df7..25f32bc49f2 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala @@ -344,7 +344,6 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers { } test("Failed - the forwarded message drops the URI's password, path and query") { - // Hadoop reads the container as everything before the first '@', password included. val shortUri = new URI("abfss://data:SECRET@myacct/path?sig=SAS-MARKER") val outcome = resolve(conf("fs.azure.account.key.myacct" -> "c2VjcmV0"), shortUri) outcome shouldBe a[Failed] @@ -354,8 +353,7 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers { } test("Failed - a forwarded message is scrubbed of the URI's password and query") { - // ClassNotFoundException is forwarded verbatim, so its message stands in for a Hadoop one - // that quotes the URI. + // Forwarded verbatim, so it stands in for a Hadoop message that quotes the URI. val secretUri = new URI(s"abfss://data:the-uri-password@$account/p?sig=the-sas-signature") val thrown = new ClassNotFoundException(s"no class for $secretUri") val message = toOptions(resolve(conf(), secretUri, Failure(thrown)))(Markers.ERROR_MESSAGE) diff --git a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala index 69cf4856b44..7ffa749f49b 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala @@ -758,7 +758,6 @@ class NativeConfigSuite extends AnyFunSuite with Matchers { NativeConfig.displayAuthority( new URI("ABFSS://data:SECRET@acct.dfs.core.windows.net:8443/p?sig=SAS-MARKER")) shouldBe "abfss://data@acct.dfs.core.windows.net:8443" - // A path without a scheme is local, as in objectStoreKey. NativeConfig.displayAuthority(new URI("/tmp/t/k.parquet")) shouldBe "file://" } From 2532764b70bda12c039415a78f63852c02d28042 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Mon, 26 Oct 2026 23:18:29 +0700 Subject: [PATCH 5/7] fix: key encrypted ABFS reads by the password-free authority Native now builds the key callback path from the authority without its password, so the retriever registered under the listed path was never found. Register and look up retrievers by one key: S3 aliases folded to s3a, and for abfs/abfss the scheme lowercased and the password dropped by the same rule as native. The missing-retriever error prints that key. --- .../comet/parquet/CometFileKeyUnwrapper.java | 46 +++++++++-- .../parquet/TestCometFileKeyUnwrapper.java | 80 +++++++++++++++++++ 2 files changed, 121 insertions(+), 5 deletions(-) diff --git a/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java b/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java index 79e32400bad..433b8b4b0ea 100644 --- a/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java +++ b/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java @@ -154,6 +154,43 @@ static String normalizeS3Scheme(final String filePath, final Set s3Compl return "s3a://" + StringUtils.stripStart(filePath.substring(schemeEnd + 1), "/"); } + /** + * The ABFS path native looks keys up by: the scheme lowercased and the password dropped from the + * authority, where the user ends at the first {@code :} and the host follows the last {@code @}, + * as native {@code authority_without_password} reads it. Non-ABFS paths are returned unchanged. + */ + static String withoutAbfsPassword(final String filePath) { + final int schemeEnd = filePath.indexOf(':'); + if (schemeEnd <= 0) { + return filePath; + } + final String scheme = filePath.substring(0, schemeEnd).toLowerCase(Locale.ROOT); + if (!scheme.equals("abfs") && !scheme.equals("abfss")) { + return filePath; + } + if (!filePath.startsWith("//", schemeEnd + 1)) { + return scheme + filePath.substring(schemeEnd); + } + final int start = schemeEnd + 3; + int end = start; + while (end < filePath.length() && "/?#".indexOf(filePath.charAt(end)) < 0) { + end++; + } + final String authority = filePath.substring(start, end); + final int at = authority.lastIndexOf('@'); + final String userInfo = at < 0 ? "" : authority.substring(0, at); + final int colon = userInfo.indexOf(':'); + final String user = colon < 0 ? userInfo : userInfo.substring(0, colon); + final String host = authority.substring(at + 1); + final String key = user.isEmpty() ? host : user + "@" + host; + return scheme + "://" + key + filePath.substring(end); + } + + /** The retriever cache key: S3 aliases folded to s3a and the ABFS password dropped. */ + private String cacheKey(final String filePath) { + return withoutAbfsPassword(normalizeS3Scheme(filePath)); + } + /** * Creates and stores a DecryptionKeyRetriever instance for the given file path. * @@ -175,13 +212,13 @@ public void storeDecryptionKeyRetriever(final String filePath, final Configurati // Check the assumption that all files have the same hadoopConf and thus same Factory assert (conf == hadoopConf); } - final String normalizedPath = normalizeS3Scheme(filePath); + final String key = cacheKey(filePath); Path path = new Path(filePath); FileDecryptionProperties decryptionProperties = factory.getFileDecryptionProperties(hadoopConf, path); DecryptionKeyRetriever keyRetriever = decryptionProperties.getKeyRetriever(); - retrieverCache.put(normalizedPath, keyRetriever); + retrieverCache.put(key, keyRetriever); } /** @@ -195,11 +232,10 @@ public void storeDecryptionKeyRetriever(final String filePath, final Configurati */ public byte[] getKey(final String filePath, final byte[] keyMetadata) throws ParquetCryptoRuntimeException { - final String normalizedPath = normalizeS3Scheme(filePath); - DecryptionKeyRetriever keyRetriever = retrieverCache.get(normalizedPath); + DecryptionKeyRetriever keyRetriever = retrieverCache.get(cacheKey(filePath)); if (keyRetriever == null) { throw new ParquetCryptoRuntimeException( - "Failed to find DecryptionKeyRetriever for path: " + filePath); + "Failed to find DecryptionKeyRetriever for path: " + cacheKey(filePath)); } return keyRetriever.getKey(keyMetadata); } diff --git a/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java b/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java index 621e68ff98d..c1fcaae1a3d 100644 --- a/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java +++ b/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java @@ -19,14 +19,23 @@ package org.apache.comet.parquet; +import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.HashSet; import java.util.Set; import org.junit.Test; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.crypto.DecryptionPropertiesFactory; +import org.apache.parquet.crypto.FileDecryptionProperties; +import org.apache.parquet.crypto.ParquetCryptoRuntimeException; + import static java.util.Arrays.asList; +import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; /** * Regression guard for {@link CometFileKeyUnwrapper#normalizeS3Scheme(String, Set)}. The put side @@ -95,4 +104,75 @@ public void aliasSingleAndTripleSlashPromoteBucketToAuthority() { assertEquals(canonical, norm("blob:/bucket/foo/part-0.parquet", BLOB)); assertEquals(canonical, norm("blob:///bucket/foo/part-0.parquet", BLOB)); } + + @Test + public void abfsKeyDropsThePasswordAsNativeDoes() { + // Native looks keys up by the URL it read the file through, whose ABFS authority has no + // password and whose scheme the url crate lowercased. + String host = "acct.dfs.core.windows.net"; + String[][] cases = { + {"abfss://data:secret@" + host + "/t/r.parquet", "abfss://data@" + host + "/t/r.parquet"}, + { + "abfs://data:secret@" + host + ":8443/t/r.parquet", + "abfs://data@" + host + ":8443/t/r.parquet" + }, + {"abfss://:secret@" + host + "/t/r.parquet", "abfss://" + host + "/t/r.parquet"}, + {"abfss://data@" + host + "/t/r.parquet", "abfss://data@" + host + "/t/r.parquet"}, + {"ABFSS://data:secret@" + host + "/t/r.parquet", "abfss://data@" + host + "/t/r.parquet"}, + {"abfss://data@" + host + "/a@b:c/r.parquet", "abfss://data@" + host + "/a@b:c/r.parquet"}, + {"abfss:/t/r.parquet", "abfss:/t/r.parquet"}, + {"abfss://" + host + "/p", "abfss://" + host + "/p"}, + {"abfss://data:se:cret@" + host + "/p", "abfss://data@" + host + "/p"}, + // The url crate drops the `:` of an empty password. + {"abfss://data:@" + host + "/p", "abfss://data@" + host + "/p"}, + {"abfss://data:secret@" + host + "?q", "abfss://data@" + host + "?q"}, + {"abfss://data:secret@" + host + "#f", "abfss://data@" + host + "#f"}, + // Only ABFS is rewritten here; S3 user info is left as it is. + {"s3a://user:secret@bucket/t/r.parquet", "s3a://user:secret@bucket/t/r.parquet"}, + {"hdfs://nn:8020/t/r.parquet", "hdfs://nn:8020/t/r.parquet"}, + }; + for (String[] c : cases) { + assertEquals(c[0], c[1], CometFileKeyUnwrapper.withoutAbfsPassword(c[0])); + } + } + + @Test + public void abfsKeyStoredWithPasswordIsFoundWithout() { + Configuration conf = new Configuration(false); + conf.set( + DecryptionPropertiesFactory.CRYPTO_FACTORY_CLASS_PROPERTY_NAME, + FixedKeyFactory.class.getName()); + CometFileKeyUnwrapper unwrapper = new CometFileKeyUnwrapper(); + String stored = "abfss://data:fixture@acct.dfs.core.windows.net/t/r.parquet"; + unwrapper.storeDecryptionKeyRetriever(stored, conf); + byte[] metadata = new byte[0]; + assertArrayEquals( + FixedKeyFactory.KEY, + unwrapper.getKey("abfss://data@acct.dfs.core.windows.net/t/r.parquet", metadata)); + assertArrayEquals(FixedKeyFactory.KEY, unwrapper.getKey(stored, metadata)); + } + + @Test + public void missingAbfsKeyErrorOmitsThePassword() { + CometFileKeyUnwrapper unwrapper = new CometFileKeyUnwrapper(); + try { + unwrapper.getKey("abfss://data:fixture@acct.dfs.core.windows.net/t/r.parquet", new byte[0]); + fail("expected a missing retriever"); + } catch (ParquetCryptoRuntimeException e) { + assertEquals( + "Failed to find DecryptionKeyRetriever for path: " + + "abfss://data@acct.dfs.core.windows.net/t/r.parquet", + e.getMessage()); + } + } + + /** Hands every file a retriever that returns {@link #KEY}. */ + public static class FixedKeyFactory implements DecryptionPropertiesFactory { + static final byte[] KEY = "0123456789abcdef".getBytes(StandardCharsets.US_ASCII); + + @Override + public FileDecryptionProperties getFileDecryptionProperties(Configuration conf, Path path) { + return FileDecryptionProperties.builder().withKeyRetriever(metadata -> KEY).build(); + } + } } From c487ef848771317d709b6a50f1ea834f78c18858 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Mon, 26 Oct 2026 23:18:36 +0700 Subject: [PATCH 6/7] fix: mask overlapping secrets as one span in ABFS diagnostics Secrets were replaced one after another, so a URI password inside a configured credential was masked first and the rest of the credential stayed in the message. Match every secret in the same text and replace each run of masked characters with one marker. --- .../comet/objectstore/AbfsAuthResolver.scala | 24 +++++++++++++---- .../objectstore/AbfsAuthResolverSuite.scala | 26 +++++++++++++++++++ 2 files changed, 45 insertions(+), 5 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala index 9e8cf133fa3..1898c7feae0 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/AbfsAuthResolver.scala @@ -20,6 +20,7 @@ package org.apache.comet.objectstore import java.net.URI +import java.util.Arrays import scala.collection.mutable.ArrayBuffer import scala.util.{Failure, Success, Try} @@ -474,13 +475,26 @@ private[comet] object AbfsAuthResolver extends Logging { /** * Hides every `secret` of at least [[MIN_REDACTED_LENGTH]] characters in `message`, and the raw - * key a `SimpleKeyProvider` message quotes. + * key a `SimpleKeyProvider` message quotes. Secrets are matched in the same text, so one that + * overlaps or contains another is hidden whole, and each run of hidden characters becomes one + * marker. */ private[objectstore] def redact(message: String, secrets: Iterable[String]): String = { - val withoutQuotedKey = QUOTED_KEY_PATTERN.replaceAllIn(message, "key =\"" + REDACTED + "\"") - secrets - .filter(_.length >= MIN_REDACTED_LENGTH) - .foldLeft(withoutQuotedKey)((text, secret) => text.replace(secret, REDACTED)) + val text = QUOTED_KEY_PATTERN.replaceAllIn(message, "key =\"" + REDACTED + "\"") + val hidden = new Array[Boolean](text.length) + secrets.filter(_.length >= MIN_REDACTED_LENGTH).foreach { secret => + var i = text.indexOf(secret) + while (i >= 0) { + Arrays.fill(hidden, i, i + secret.length, true) + i = text.indexOf(secret, i + 1) + } + } + val out = new StringBuilder(text.length) + for (i <- 0 until text.length) { + if (!hidden(i)) out += text(i) + else if (i == 0 || !hidden(i - 1)) out ++= REDACTED + } + out.toString } /** The parts of `uri` that can carry a credential: the password and the raw query. */ diff --git a/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala index 25f32bc49f2..f7e1e0e495f 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/AbfsAuthResolverSuite.scala @@ -298,6 +298,18 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers { failed.message should not include secret } + test("Failed - a credential that contains the URI's password is redacted whole") { + val secret = "oauth-client-secret-4f9a2c" + val secretUri = new URI(s"abfss://data:client-secret@$account/path/file.parquet") + val outcome = + resolve(conf(acct(Keys.AUTH_TYPE) -> secret, acct(Keys.CLIENT_SECRET) -> secret), secretUri) + outcome shouldBe a[Failed] + val failed = outcome.asInstanceOf[Failed] + failed.message should include("No enum constant") + Seq("oauth-", "-4f9a2c", "client-secret").foreach(part => + failed.message should not include part) + } + test("Failed - an unknown auth type value is Hadoop's enum error") { val outcome = resolveWith(acct(Keys.AUTH_TYPE) -> "oauth") outcome shouldBe a[Failed] @@ -384,6 +396,20 @@ class AbfsAuthResolverSuite extends AnyFunSuite with Matchers { "Failure for acct key =\"\": token= port=8080" } + test("redact - overlapping, contained and adjacent secrets become one marker") { + // Each secret is matched in the original text, so the order they arrive in does not matter. + val cases = Seq( + ("x oauth-client-secret-4f9a2c y", Seq("client-secret", "oauth-client-secret-4f9a2c")), + ("x first-secret-tail y", Seq("first-secret", "secret-tail")), + ("x first-secretsecond-secret y", Seq("first-secret", "second-secret")), + ("x abababababab y", Seq("abababab"))) + for ((message, secrets) <- cases; ordered <- Seq(secrets, secrets.reverse)) { + withClue(s"$message with $ordered: ") { + redact(message, ordered) shouldBe "x y" + } + } + } + test("redact - a padded secret is hidden in its trimmed form too") { val padded = " the-padded-secret " redact("provider said the-padded-secret was rejected", Seq(padded, padded.trim)) shouldBe From 5c61d9ce5e5311650745b89df84f9801a6b34684 Mon Sep 17 00:00:00 2001 From: Dustin Smith Date: Mon, 26 Oct 2026 23:18:42 +0700 Subject: [PATCH 7/7] perf: reuse the raw ABFS authority when it has no password The object store key is built per file during fallback checks and again per split in partition packing. Return the raw authority as is when it has no `@`, or no `:` before the last `@`, instead of rebuilding the same string. --- .../apache/comet/objectstore/NativeConfig.scala | 17 ++++++++++++----- .../comet/objectstore/NativeConfigSuite.scala | 8 ++++++++ 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala index e27fb2cc3af..731e2cc6dc4 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala @@ -132,13 +132,20 @@ object NativeConfig { rawAuthority.substring(rawAuthority.lastIndexOf('@') + 1) } - /** As in native, the user ends at the first `:` and the host follows the last `@`. */ - private def authorityWithoutPassword(uri: URI): String = { + /** + * As in native, the user ends at the first `:` and the host follows the last `@`. An authority + * with no password is returned as is. + */ + private[objectstore] def authorityWithoutPassword(uri: URI): String = { val rawAuthority = Option(uri.getRawAuthority).getOrElse("") val at = rawAuthority.lastIndexOf('@') - val user = rawAuthority.take(at).takeWhile(_ != ':') - val host = rawAuthority.substring(at + 1) - if (user.isEmpty) host else s"$user@$host" + if (at < 0 || (at > 0 && rawAuthority.lastIndexOf(':', at - 1) < 0)) { + rawAuthority + } else { + val user = rawAuthority.take(at).takeWhile(_ != ':') + val host = rawAuthority.substring(at + 1) + if (user.isEmpty) host else s"$user@$host" + } } /** `scheme://[user@]host[:port]` of `uri`, for messages: no password, path or query. */ diff --git a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala index 7ffa749f49b..9a44d20a1c2 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala @@ -754,6 +754,14 @@ class NativeConfigSuite extends AnyFunSuite with Matchers { "(abfss://data@accta.dfs.core.windows.net and abfss://logs@accta.dfs.core.windows.net)") } + test("authorityWithoutPassword - an authority without a password is returned as is") { + for (authority <- Seq("container@acct.dfs.core.windows.net", "container@acct:8443")) { + val uri = new URI(s"abfss://$authority/k.parquet") + NativeConfig.authorityWithoutPassword(uri) should be theSameInstanceAs uri.getRawAuthority + } + NativeConfig.authorityWithoutPassword(new URI("/tmp/t/k.parquet")) shouldBe "" + } + test("displayAuthority - lowercase scheme and no password, path or query") { NativeConfig.displayAuthority( new URI("ABFSS://data:SECRET@acct.dfs.core.windows.net:8443/p?sig=SAS-MARKER")) shouldBe