feat: robustify pre-signed URL checks (#760)

## What changes are proposed in this pull request?

Current checks for a pre-signed url in our readers may mis-classify
object store URLs that do not use store specific schemes (`s3://`m
`az://` ...) as pre-signed urls. We extend the current checks, to also
consider the presence of a query string and define a helper trait to
ensure we can maintain the check in one place.

## How was this change tested?

additional test to validate url matches.

Author: roeap

kernel/src/engine/default/filesystem.rs

@@ -7,6 +7,7 @@ use object_store::path::Path;
 use object_store::{DynObjectStore, ObjectStore};
 use url::Url;
 
+use super::UrlExt;
 use crate::engine::default::executor::TaskExecutor;
 use crate::{DeltaResult, Error, FileMeta, FileSlice, FileSystemClient};
 
@@ -131,19 +132,14 @@ impl<E: TaskExecutor> FileSystemClient for ObjectStoreFileSystemClient<E> {
                     };
                     let store = store.clone();
                     async move {
-                        match url.scheme() {
-                            "http" | "https" => {
-                                // have to annotate type here or rustc can't figure it out
-                                Ok::<bytes::Bytes, Error>(reqwest::get(url).await?.bytes().await?)
-                            }
-                            _ => {
-                                if let Some(rng) = range {
-                                    Ok(store.get_range(&path, rng).await?)
-                                } else {
-                                    let result = store.get(&path).await?;
-                                    Ok(result.bytes().await?)
-                                }
-                            }
+                        if url.is_presigned() {
+                            // have to annotate type here or rustc can't figure it out
+                            Ok::<bytes::Bytes, Error>(reqwest::get(url).await?.bytes().await?)
+                        } else if let Some(rng) = range {
+                            Ok(store.get_range(&path, rng).await?)
+                        } else {
+                            let result = store.get(&path).await?;
+                            Ok(result.bytes().await?)
                         }
                     }
                 })

kernel/src/engine/default/mod.rs

@@ -158,6 +158,35 @@ impl<E: TaskExecutor> Engine for DefaultEngine<E> {
     }
 }
 
+trait UrlExt {
+    // Check if a given url is a presigned url and can be used
+    // to access the object store via simple http requests
+    fn is_presigned(&self) -> bool;
+}
+
+impl UrlExt for Url {
+    fn is_presigned(&self) -> bool {
+        matches!(self.scheme(), "http" | "https")
+            && (
+                // https://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-query-string-auth.html
+                // https://developers.cloudflare.com/r2/api/s3/presigned-urls/
+                self
+                .query_pairs()
+                .any(|(k, _)| k.eq_ignore_ascii_case("X-Amz-Signature")) ||
+                // https://learn.microsoft.com/en-us/rest/api/storageservices/create-user-delegation-sas#version-2020-12-06-and-later
+                // note signed permission (sp) must always be present
+                self
+                .query_pairs().any(|(k, _)| k.eq_ignore_ascii_case("sp")) ||
+                // https://cloud.google.com/storage/docs/authentication/signatures
+                self
+                .query_pairs().any(|(k, _)| k.eq_ignore_ascii_case("X-Goog-Credential")) ||
+                // https://www.alibabacloud.com/help/en/oss/user-guide/upload-files-using-presigned-urls
+                self
+                .query_pairs().any(|(k, _)| k.eq_ignore_ascii_case("X-OSS-Credential"))
+            )
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::executor::tokio::TokioBackgroundExecutor;
@@ -173,4 +202,29 @@ mod tests {
         let engine = DefaultEngine::new(store, Arc::new(TokioBackgroundExecutor::new()));
         test_arrow_engine(&engine, &url);
     }
+
+    #[test]
+    fn test_pre_signed_url() {
+        let url = Url::parse("https://example.com?X-Amz-Signature=foo").unwrap();
+        assert!(url.is_presigned());
+
+        let url = Url::parse("https://example.com?sp=foo").unwrap();
+        assert!(url.is_presigned());
+
+        let url = Url::parse("https://example.com?X-Goog-Credential=foo").unwrap();
+        assert!(url.is_presigned());
+
+        let url = Url::parse("https://example.com?X-OSS-Credential=foo").unwrap();
+        assert!(url.is_presigned());
+
+        // assert that query keys are case insensitive
+        let url = Url::parse("https://example.com?x-gooG-credenTIAL=foo").unwrap();
+        assert!(url.is_presigned());
+
+        let url = Url::parse("https://example.com?x-oss-CREDENTIAL=foo").unwrap();
+        assert!(url.is_presigned());
+
+        let url = Url::parse("https://example.com").unwrap();
+        assert!(!url.is_presigned());
+    }
 }

kernel/src/engine/default/parquet.rs

@@ -17,6 +17,7 @@ use object_store::DynObjectStore;
 use uuid::Uuid;
 
 use super::file_stream::{FileOpenFuture, FileOpener, FileStream};
+use super::UrlExt;
 use crate::engine::arrow_data::ArrowEngineData;
 use crate::engine::arrow_utils::{fixup_parquet_read, generate_mask, get_requested_indices};
 use crate::engine::default::executor::TaskExecutor;
@@ -191,18 +192,19 @@ impl<E: TaskExecutor> ParquetHandler for DefaultParquetHandler<E> {
         //   -> reqwest to get data
         //   -> parse to parquet
         // SAFETY: we did is_empty check above, this is ok.
-        let file_opener: Box<dyn FileOpener> = match files[0].location.scheme() {
-            "http" | "https" => Box::new(PresignedUrlOpener::new(
+        let file_opener: Box<dyn FileOpener> = if files[0].location.is_presigned() {
+            Box::new(PresignedUrlOpener::new(
                 1024,
                 physical_schema.clone(),
                 predicate,
-            )),
-            _ => Box::new(ParquetOpener::new(
+            ))
+        } else {
+            Box::new(ParquetOpener::new(
                 1024,
                 physical_schema.clone(),
                 predicate,
                 self.store.clone(),
-            )),
+            ))
         };
         FileStream::new_async_read_iterator(
             self.task_executor.clone(),