fix: keep URL passwords and query strings out of object store errors

native/core/src/parquet/objectstore/azure.rs

@@ -78,14 +78,16 @@
 
 use log::debug;
 use std::collections::HashMap;
-use url::{Position, Url};
+use url::Url;
 
 use object_store::{
     azure::{AzureConfigKey, MicrosoftAzureBuilder},
     path::Path,
     ObjectStore, ObjectStoreScheme,
 };
 
+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";
 const MARKER_OAUTH_PROVIDER_CLASS: &str = "comet.azure.oauth.provider.class";
@@ -127,9 +129,12 @@ pub fn create_store(
     url: &Url,
     configs: &HashMap<String, String>,
 ) -> Result<(Box<dyn ObjectStore>, 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: {}",
+            url.scheme()
+        )));
     }
     let path = Path::parse(path)?;
     let store = builder_for(url, configs)?.build()?;
@@ -159,17 +164,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::<Vec<_>>()
             );
-            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)))
@@ -184,9 +193,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("<unknown>"),
             marker(MARKER_ERROR_MESSAGE).unwrap_or("<unknown>"),
         ))),
@@ -570,6 +578,54 @@ 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.ends_with("Scheme of URL is not Azure: s3"),
+            "{not_azure}"
+        );
+        assert!(!not_azure.contains("bucket"), "host 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 store_error_answers_whether_create_store_builds() {
         let shared_key = with(shared_key(), HADOOP_KEY, KEY);
@@ -923,6 +979,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]

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;

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<String, String>,
     min_ttl: Duration,
 ) -> Result<(Box<dyn ObjectStore>, 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()

native/core/src/parquet/objectstore/s3_blob_fs_support.rs

@@ -77,8 +77,7 @@ pub(crate) fn normalize_object_store_url(
     url_str: &str,
     object_store_configs: &HashMap<String, String>,
 ) -> Result<NormalizedObjectStoreUrl, ExecutionError> {
-    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| parse_error(url_str, e))?;
     if is_hdfs_scheme(&url, object_store_configs) {
         return Ok(NormalizedObjectStoreUrl { url, is_hdfs: true });
     }
@@ -95,6 +94,25 @@ 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("://")?;
+    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
@@ -135,7 +153,7 @@ fn rewrite_alias_to_s3(mut url: Url) -> Result<Url, ExecutionError> {
         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()
             )));
         }
@@ -167,8 +185,7 @@ fn rewrite_alias_to_s3(mut url: Url) -> Result<Url, ExecutionError> {
 /// 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<Cow<'_, str>, 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));
     }
@@ -303,6 +320,28 @@ 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}");
+        // 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}");
+        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");
@@ -486,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}");
+    }
 }