Enable anonymous access to GCS buckets (#322)

* Gcp skip signature

* Skip signing if empty bearer token

Author: kylebarron

src/gcp/builder.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use crate::client::{http_connector, HttpConnector, TokenCredentialProvider};
+use crate::config::ConfigValue;
 use crate::gcp::client::{GoogleCloudStorageClient, GoogleCloudStorageConfig};
 use crate::gcp::credential::{
     ApplicationDefaultCredentials, InstanceCredentialProvider, ServiceAccountCredentials,
@@ -109,6 +110,8 @@ pub struct GoogleCloudStorageBuilder {
     client_options: ClientOptions,
     /// Credentials
     credentials: Option<GcpCredentialProvider>,
+    /// Skip signing requests
+    skip_signature: ConfigValue<bool>,
     /// Credentials for sign url
     signing_credentials: Option<GcpSigningCredentialProvider>,
     /// The [`HttpConnector`] to use
@@ -161,6 +164,9 @@ pub enum GoogleConfigKey {
     /// See [`GoogleCloudStorageBuilder::with_application_credentials`].
     ApplicationCredentials,
 
+    /// Skip signing request
+    SkipSignature,
+
     /// Client options
     Client(ClientConfigKey),
 }
@@ -172,6 +178,7 @@ impl AsRef<str> for GoogleConfigKey {
             Self::ServiceAccountKey => "google_service_account_key",
             Self::Bucket => "google_bucket",
             Self::ApplicationCredentials => "google_application_credentials",
+            Self::SkipSignature => "google_skip_signature",
             Self::Client(key) => key.as_ref(),
         }
     }
@@ -189,6 +196,7 @@ impl FromStr for GoogleConfigKey {
             "google_service_account_key" | "service_account_key" => Ok(Self::ServiceAccountKey),
             "google_bucket" | "google_bucket_name" | "bucket" | "bucket_name" => Ok(Self::Bucket),
             "google_application_credentials" => Ok(Self::ApplicationCredentials),
+            "google_skip_signature" | "skip_signature" => Ok(Self::SkipSignature),
             _ => match s.strip_prefix("google_").unwrap_or(s).parse() {
                 Ok(key) => Ok(Self::Client(key)),
                 Err(_) => Err(Error::UnknownConfigurationKey { key: s.into() }.into()),
@@ -208,6 +216,7 @@ impl Default for GoogleCloudStorageBuilder {
             client_options: ClientOptions::new().with_allow_http(true),
             url: None,
             credentials: None,
+            skip_signature: Default::default(),
             signing_credentials: None,
             http_connector: None,
         }
@@ -288,6 +297,7 @@ impl GoogleCloudStorageBuilder {
             GoogleConfigKey::ApplicationCredentials => {
                 self.application_credentials_path = Some(value.into())
             }
+            GoogleConfigKey::SkipSignature => self.skip_signature.parse(value),
             GoogleConfigKey::Client(key) => {
                 self.client_options = self.client_options.with_config(key, value)
             }
@@ -312,6 +322,7 @@ impl GoogleCloudStorageBuilder {
             GoogleConfigKey::ServiceAccountKey => self.service_account_key.clone(),
             GoogleConfigKey::Bucket => self.bucket_name.clone(),
             GoogleConfigKey::ApplicationCredentials => self.application_credentials_path.clone(),
+            GoogleConfigKey::SkipSignature => Some(self.skip_signature.to_string()),
             GoogleConfigKey::Client(key) => self.client_options.get_config_value(key),
         }
     }
@@ -389,6 +400,14 @@ impl GoogleCloudStorageBuilder {
         self
     }
 
+    /// If enabled, [`GoogleCloudStorage`] will not fetch credentials and will not sign requests.
+    ///
+    /// This can be useful when interacting with public GCS buckets that deny authorized requests.
+    pub fn with_skip_signature(mut self, skip_signature: bool) -> Self {
+        self.skip_signature = skip_signature.into();
+        self
+    }
+
     /// Set the credential provider overriding any other options
     pub fn with_credentials(mut self, credentials: GcpCredentialProvider) -> Self {
         self.credentials = Some(credentials);
@@ -546,14 +565,15 @@ impl GoogleCloudStorageBuilder {
             )) as _
         };
 
-        let config = GoogleCloudStorageConfig::new(
-            gcs_base_url,
+        let config = GoogleCloudStorageConfig {
+            base_url: gcs_base_url,
             credentials,
             signing_credentials,
             bucket_name,
-            self.retry_config,
-            self.client_options,
-        );
+            retry_config: self.retry_config,
+            client_options: self.client_options,
+            skip_signature: self.skip_signature.get()?,
+        };
 
         let http_client = http.connect(&config.client_options)?;
         Ok(GoogleCloudStorage {

src/gcp/client.rs

@@ -25,6 +25,7 @@ use crate::client::s3::{
     ListResponse,
 };
 use crate::client::{GetOptionsExt, HttpClient, HttpError, HttpResponse};
+use crate::gcp::credential::CredentialExt;
 use crate::gcp::{GcpCredential, GcpCredentialProvider, GcpSigningCredentialProvider, STORE};
 use crate::multipart::PartId;
 use crate::path::{Path, DELIMITER};
@@ -144,30 +145,21 @@ pub(crate) struct GoogleCloudStorageConfig {
     pub retry_config: RetryConfig,
 
     pub client_options: ClientOptions,
+
+    pub skip_signature: bool,
 }
 
 impl GoogleCloudStorageConfig {
-    pub(crate) fn new(
-        base_url: String,
-        credentials: GcpCredentialProvider,
-        signing_credentials: GcpSigningCredentialProvider,
-        bucket_name: String,
-        retry_config: RetryConfig,
-        client_options: ClientOptions,
-    ) -> Self {
-        Self {
-            base_url,
-            credentials,
-            signing_credentials,
-            bucket_name,
-            retry_config,
-            client_options,
-        }
-    }
-
     pub(crate) fn path_url(&self, path: &Path) -> String {
         format!("{}/{}/{}", self.base_url, self.bucket_name, path)
     }
+
+    pub(crate) async fn get_credential(&self) -> Result<Option<Arc<GcpCredential>>> {
+        Ok(match self.skip_signature {
+            false => Some(self.credentials.get_credential().await?),
+            true => None,
+        })
+    }
 }
 
 /// A builder for a put request allowing customisation of the headers and query string
@@ -304,8 +296,8 @@ impl GoogleCloudStorageClient {
         &self.config
     }
 
-    async fn get_credential(&self) -> Result<Arc<GcpCredential>> {
-        self.config.credentials.get_credential().await
+    async fn get_credential(&self) -> Result<Option<Arc<GcpCredential>>> {
+        self.config.get_credential().await
     }
 
     /// Create a signature from a string-to-sign using Google Cloud signBlob method.
@@ -341,7 +333,7 @@ impl GoogleCloudStorageClient {
         let response = self
             .client
             .post(&url)
-            .bearer_auth(&credential.bearer)
+            .with_bearer_auth(credential.as_deref())
             .json(&body)
             .retryable(&self.config.retry_config)
             .idempotent(true)
@@ -493,7 +485,7 @@ impl GoogleCloudStorageClient {
 
         self.client
             .request(Method::DELETE, &url)
-            .bearer_auth(&credential.bearer)
+            .with_bearer_auth(credential.as_deref())
             .header(CONTENT_TYPE, "application/octet-stream")
             .header(CONTENT_LENGTH, "0")
             .query(&[("uploadId", multipart_id)])
@@ -538,7 +530,7 @@ impl GoogleCloudStorageClient {
         let response = self
             .client
             .request(Method::POST, &url)
-            .bearer_auth(&credential.bearer)
+            .with_bearer_auth(credential.as_deref())
             .query(&[("uploadId", upload_id)])
             .body(data)
             .retryable(&self.config.retry_config)
@@ -594,7 +586,7 @@ impl GoogleCloudStorageClient {
         }
 
         builder
-            .bearer_auth(&credential.bearer)
+            .with_bearer_auth(credential.as_deref())
             // Needed if reqwest is compiled with native-tls instead of rustls-tls
             // See https://github.com/apache/arrow-rs/pull/3921
             .header(CONTENT_LENGTH, 0)
@@ -640,11 +632,8 @@ impl GetClient for GoogleCloudStorageClient {
             request = request.query(&[("generation", version)]);
         }
 
-        if !credential.bearer.is_empty() {
-            request = request.bearer_auth(&credential.bearer);
-        }
-
         let response = request
+            .with_bearer_auth(credential.as_deref())
             .with_get_options(options)
             .send_retry(&self.config.retry_config)
             .await
@@ -696,7 +685,7 @@ impl ListClient for Arc<GoogleCloudStorageClient> {
             .client
             .request(Method::GET, url)
             .query(&query)
-            .bearer_auth(&credential.bearer)
+            .with_bearer_auth(credential.as_deref())
             .send_retry(&self.config.retry_config)
             .await
             .map_err(|source| Error::ListRequest { source })?

src/gcp/credential.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use super::client::GoogleCloudStorageClient;
+use crate::client::builder::HttpRequestBuilder;
 use crate::client::retry::RetryExt;
 use crate::client::token::TemporaryToken;
 use crate::client::{HttpClient, HttpError, TokenProvider};
@@ -424,7 +425,7 @@ impl TokenProvider for InstanceCredentialProvider {
     /// Since the connection is local we need to enable http access and don't actually use the client object passed in.
     /// Respects the `GCE_METADATA_HOST`, `GCE_METADATA_ROOT`, and `GCE_METADATA_IP`
     /// environment variables.
-    ///  
+    ///
     /// References: <https://googleapis.dev/python/google-auth/latest/reference/google.auth.environment_vars.html>
     async fn fetch_token(
         &self,
@@ -494,7 +495,7 @@ impl TokenProvider for InstanceSigningCredentialProvider {
     /// Since the connection is local we need to enable http access and don't actually use the client object passed in.
     /// Respects the `GCE_METADATA_HOST`, `GCE_METADATA_ROOT`, and `GCE_METADATA_IP`
     /// environment variables.
-    ///  
+    ///
     /// References: <https://googleapis.dev/python/google-auth/latest/reference/google.auth.environment_vars.html>
     async fn fetch_token(
         &self,
@@ -854,6 +855,26 @@ impl GCSAuthorizer {
     }
 }
 
+pub(crate) trait CredentialExt {
+    /// Apply bearer authentication to the request if the credential is not None
+    fn with_bearer_auth(self, credential: Option<&GcpCredential>) -> Self;
+}
+
+impl CredentialExt for HttpRequestBuilder {
+    fn with_bearer_auth(self, credential: Option<&GcpCredential>) -> Self {
+        match credential {
+            Some(credential) => {
+                if credential.bearer.is_empty() {
+                    self
+                } else {
+                    self.bearer_auth(&credential.bearer)
+                }
+            }
+            None => self,
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;