feat(gcs): allow unauthenticated requests (#4965)

* feat: add no_authentication fields

* feat: add load_credentials skip from no_authentication

* fix: skip signing with no auth

* fix: use allow_anonymous similar to s3

* docs: clarify allow_anonymous

* feat: add disable vm/config values

* feat: wire up credential/token skip

* docs: allow_anonymous fn comment

* feat: disable well known locations for disable_config_load

* fix: move allow_anonymous to after cred load

* doc: clarify allow_anonymous

I've taken the similar description to the s3 backend here

core/src/services/gcs/backend.rs

@@ -66,6 +66,14 @@ pub struct GcsConfig {
     pub predefined_acl: Option<String>,
     /// The default storage class used by gcs.
     pub default_storage_class: Option<String>,
+    /// Allow opendal to send requests without signing when credentials are not
+    /// loaded.
+    pub allow_anonymous: bool,
+    /// Disable attempting to load credentials from the GCE metadata server when
+    /// running within Google Cloud.
+    pub disable_vm_metadata: bool,
+    /// Disable loading configuration from the environment.
+    pub disable_config_load: bool,
 }
 
 impl Debug for GcsConfig {
@@ -206,6 +214,18 @@ impl GcsBuilder {
         self
     }
 
+    /// Disable attempting to load credentials from the GCE metadata server.
+    pub fn disable_vm_metadata(mut self) -> Self {
+        self.config.disable_vm_metadata = true;
+        self
+    }
+
+    /// Disable loading configuration from the environment.
+    pub fn disable_config_load(mut self) -> Self {
+        self.config.disable_config_load = true;
+        self
+    }
+
     /// Set the predefined acl for GCS.
     ///
     /// Available values are:
@@ -235,6 +255,15 @@ impl GcsBuilder {
         };
         self
     }
+
+    /// Allow anonymous requests.
+    ///
+    /// This is typically used for buckets which are open to the public or GCS
+    /// storage emulators.
+    pub fn allow_anonymous(mut self) -> Self {
+        self.config.allow_anonymous = true;
+        self
+    }
 }
 
 impl Builder for GcsBuilder {
@@ -288,6 +317,12 @@ impl Builder for GcsBuilder {
             cred_loader = cred_loader.with_disable_well_known_location();
         }
 
+        if self.config.disable_config_load {
+            cred_loader = cred_loader
+                .with_disable_env()
+                .with_disable_well_known_location();
+        }
+
         let scope = if let Some(scope) = &self.config.scope {
             scope
         } else {
@@ -305,6 +340,10 @@ impl Builder for GcsBuilder {
             token_loader = token_loader.with_customized_token_loader(loader)
         }
 
+        if self.config.disable_vm_metadata {
+            token_loader = token_loader.with_disable_vm_metadata(true);
+        }
+
         let signer = GoogleSigner::new("storage");
 
         let backend = GcsBackend {
@@ -318,6 +357,7 @@ impl Builder for GcsBuilder {
                 credential_loader: cred_loader,
                 predefined_acl: self.config.predefined_acl.clone(),
                 default_storage_class: self.config.default_storage_class.clone(),
+                allow_anonymous: self.config.allow_anonymous,
             }),
         };
 

core/src/services/gcs/core.rs

@@ -57,6 +57,8 @@ pub struct GcsCore {
 
     pub predefined_acl: Option<String>,
     pub default_storage_class: Option<String>,
+
+    pub allow_anonymous: bool,
 }
 
 impl Debug for GcsCore {
@@ -89,23 +91,30 @@ impl GcsCore {
         }
     }
 
-    fn load_credential(&self) -> Result<GoogleCredential> {
+    fn load_credential(&self) -> Result<Option<GoogleCredential>> {
         let cred = self
             .credential_loader
             .load()
             .map_err(new_request_credential_error)?;
 
         if let Some(cred) = cred {
-            Ok(cred)
-        } else {
-            Err(Error::new(
-                ErrorKind::ConfigInvalid,
-                "no valid credential found",
-            ))
+            return Ok(Some(cred));
+        }
+
+        if self.allow_anonymous {
+            return Ok(None);
         }
+
+        Err(Error::new(
+            ErrorKind::ConfigInvalid,
+            "no valid credential found",
+        ))
     }
 
     pub async fn sign<T>(&self, req: &mut Request<T>) -> Result<()> {
+        if self.allow_anonymous {
+            return Ok(());
+        }
         let cred = self.load_token().await?;
 
         self.signer
@@ -124,11 +133,13 @@ impl GcsCore {
     }
 
     pub async fn sign_query<T>(&self, req: &mut Request<T>, duration: Duration) -> Result<()> {
-        let cred = self.load_credential()?;
-
-        self.signer
-            .sign_query(req, duration, &cred)
-            .map_err(new_request_sign_error)?;
+        if let Some(cred) = self.load_credential()? {
+            self.signer
+                .sign_query(req, duration, &cred)
+                .map_err(new_request_sign_error)?;
+        } else {
+            return Ok(());
+        }
 
         // Always remove host header, let users' client to set it based on HTTP
         // version.