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

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,14 +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| {
-        // `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<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()
             )));
         }
@@ -182,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));
     }
@@ -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}");
+    }
 }

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<String, object_store::Error> {
         }
         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() {