opendal_service_gcs_grpc/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::fmt::Debug;
19use std::sync::Arc;
20
21use log::debug;
22use reqsign_core::Env as _;
23use reqsign_core::{Context, OsEnv, ProvideCredential, ProvideCredentialChain, Signer, StaticEnv};
24use reqsign_file_read_tokio::TokioFileRead;
25use reqsign_google::{
26    Credential, DefaultCredentialProvider, FileCredentialProvider, RequestSigner,
27    StaticCredentialProvider, TokenCredentialProvider, VmMetadataCredentialProvider,
28};
29use tonic::transport::{ClientTlsConfig, Endpoint};
30
31use opendal_core::raw::*;
32use opendal_core::*;
33
34use crate::GCS_GRPC_SCHEME;
35use crate::config::GcsGrpcConfig;
36use crate::copier::new_gcs_grpc_copier;
37use crate::core::{GcsGrpcCore, parse_generation, parse_object, parse_status};
38use crate::deleter::GcsGrpcDeleter;
39use crate::generated::google::storage::v2::GetObjectRequest;
40use crate::lister::GcsGrpcLister;
41use crate::reader::GcsGrpcReader;
42use crate::writer::GcsGrpcWriter;
43
44const DEFAULT_GCS_GRPC_ENDPOINT: &str = "https://storage.googleapis.com";
45const DEFAULT_GCS_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read_write";
46
47/// Builder for the Google Cloud Storage gRPC service.
48#[doc = include_str!("docs.md")]
49#[derive(Default)]
50pub struct GcsGrpcBuilder {
51    pub(super) config: GcsGrpcConfig,
52    pub(super) credential_provider_chain: Option<ProvideCredentialChain<Credential>>,
53}
54
55impl Debug for GcsGrpcBuilder {
56    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
57        f.debug_struct("GcsGrpcBuilder")
58            .field("config", &self.config)
59            .finish_non_exhaustive()
60    }
61}
62
63impl GcsGrpcBuilder {
64    /// Set the working directory root.
65    pub fn root(mut self, root: &str) -> Self {
66        self.config.root = (!root.is_empty()).then(|| root.to_string());
67        self
68    }
69
70    /// Set the bucket name.
71    pub fn bucket(mut self, bucket: &str) -> Self {
72        self.config.bucket = bucket.to_string();
73        self
74    }
75
76    /// Set the gRPC endpoint.
77    pub fn endpoint(mut self, endpoint: &str) -> Self {
78        self.config.endpoint = (!endpoint.is_empty()).then(|| endpoint.to_string());
79        self
80    }
81
82    /// Set the Google OAuth 2.0 scope.
83    pub fn scope(mut self, scope: &str) -> Self {
84        self.config.scope = (!scope.is_empty()).then(|| scope.to_string());
85        self
86    }
87
88    /// Set the service account used by the GCE metadata server.
89    pub fn service_account(mut self, service_account: &str) -> Self {
90        self.config.service_account =
91            (!service_account.is_empty()).then(|| service_account.to_string());
92        self
93    }
94
95    /// Set a base64-encoded service account credential.
96    pub fn credential(mut self, credential: &str) -> Self {
97        self.config.credential = (!credential.is_empty()).then(|| credential.to_string());
98        self
99    }
100
101    /// Set the path to a service account credential file.
102    pub fn credential_path(mut self, path: &str) -> Self {
103        self.config.credential_path = (!path.is_empty()).then(|| path.to_string());
104        self
105    }
106
107    /// Set a custom Google credential provider.
108    pub fn credential_provider(
109        mut self,
110        provider: impl ProvideCredential<Credential = Credential> + 'static,
111    ) -> Self {
112        self.credential_provider_chain = Some(
113            self.credential_provider_chain
114                .unwrap_or_default()
115                .push_front(provider),
116        );
117        self
118    }
119
120    /// Set a custom Google credential provider chain.
121    pub fn credential_provider_chain(mut self, chain: ProvideCredentialChain<Credential>) -> Self {
122        self.credential_provider_chain = Some(chain);
123        self
124    }
125
126    /// Set an OAuth 2.0 access token.
127    pub fn token(mut self, token: String) -> Self {
128        self.config.token = Some(token);
129        self
130    }
131
132    /// Disable the GCE metadata credential provider.
133    pub fn disable_vm_metadata(mut self) -> Self {
134        self.config.disable_vm_metadata = true;
135        self
136    }
137
138    /// Disable environment and well-known credential loading.
139    pub fn disable_config_load(mut self) -> Self {
140        self.config.disable_config_load = true;
141        self
142    }
143
144    /// Send requests without authentication.
145    pub fn skip_signature(mut self) -> Self {
146        self.config.skip_signature = true;
147        self
148    }
149}
150
151impl Builder for GcsGrpcBuilder {
152    type Config = GcsGrpcConfig;
153
154    fn build(self) -> Result<impl Service> {
155        debug!("backend build started: {self:?}");
156        let root = normalize_root(&self.config.root.unwrap_or_default());
157        if self.config.bucket.is_empty() {
158            return Err(
159                Error::new(ErrorKind::ConfigInvalid, "The bucket is misconfigured")
160                    .with_operation("Builder::build")
161                    .with_context("service", GCS_GRPC_SCHEME),
162            );
163        }
164
165        let endpoint = self
166            .config
167            .endpoint
168            .clone()
169            .unwrap_or_else(|| DEFAULT_GCS_GRPC_ENDPOINT.to_string());
170        let channel_endpoint = build_endpoint(&endpoint)?;
171        let scope = self
172            .config
173            .scope
174            .clone()
175            .unwrap_or_else(|| DEFAULT_GCS_SCOPE.to_string());
176
177        let os_env = OsEnv;
178        let mut envs = os_env.vars();
179        envs.insert("GOOGLE_SCOPE".to_string(), scope.clone());
180        let ctx = Context::new()
181            .with_file_read(TokioFileRead)
182            .with_env(StaticEnv {
183                home_dir: os_env.home_dir(),
184                envs,
185            });
186
187        let mut default_credential = DefaultCredentialProvider::builder();
188        if self.config.disable_config_load {
189            default_credential = default_credential.no_env().no_well_known();
190        }
191        if self.config.disable_vm_metadata || self.config.service_account.is_some() {
192            default_credential = default_credential.no_vm_metadata();
193        }
194        let mut credential_chain = ProvideCredentialChain::new().push(default_credential.build());
195        if !self.config.disable_vm_metadata
196            && let Some(service_account) = self.config.service_account.as_deref()
197        {
198            credential_chain = credential_chain.push(
199                VmMetadataCredentialProvider::new()
200                    .with_scope(&scope)
201                    .with_service_account(service_account),
202            );
203        }
204        if let Some(path) = self.config.credential_path.as_deref() {
205            credential_chain =
206                credential_chain.push_front(FileCredentialProvider::new(path).with_scope(&scope));
207        }
208        if let Some(content) = self.config.credential.as_deref()
209            && let Ok(provider) = StaticCredentialProvider::from_base64(content)
210        {
211            credential_chain = credential_chain.push_front(provider.with_scope(&scope));
212        }
213        if let Some(token) = self.config.token.as_deref() {
214            credential_chain = credential_chain.push_front(TokenCredentialProvider::new(token));
215        }
216        if let Some(custom) = self.credential_provider_chain {
217            credential_chain = credential_chain.push_front(custom);
218        }
219
220        let signer = Signer::new(
221            ctx.clone(),
222            credential_chain,
223            RequestSigner::new("storage").with_scope(&scope),
224        );
225        let capability = Capability {
226            stat: true,
227            stat_with_version: true,
228            read: true,
229            read_with_version: true,
230            read_with_suffix: true,
231            write: true,
232            write_can_empty: true,
233            write_can_multi: true,
234            write_with_content_type: true,
235            write_with_content_disposition: true,
236            write_with_content_encoding: true,
237            write_with_cache_control: true,
238            write_with_if_not_exists: true,
239            write_with_user_metadata: true,
240            delete: true,
241            delete_with_version: true,
242            copy: true,
243            copy_with_if_not_exists: true,
244            copy_with_source_version: true,
245            list: true,
246            list_with_limit: true,
247            list_with_start_after: true,
248            list_with_recursive: true,
249            shared: true,
250            ..Default::default()
251        };
252        let bucket = self.config.bucket;
253        Ok(GcsGrpcBackend {
254            core: Arc::new(GcsGrpcCore {
255                info: ServiceInfo::new(GCS_GRPC_SCHEME, &root, &bucket),
256                capability,
257                endpoint,
258                bucket,
259                root,
260                channel_endpoint,
261                channel: Default::default(),
262                signer,
263                sign_ctx: ctx,
264                skip_signature: self.config.skip_signature,
265            }),
266        })
267    }
268}
269
270fn build_endpoint(endpoint: &str) -> Result<Endpoint> {
271    let mut endpoint_builder = Endpoint::from_shared(endpoint.to_string()).map_err(|err| {
272        Error::new(ErrorKind::ConfigInvalid, "invalid GCS gRPC endpoint").set_source(err)
273    })?;
274    match endpoint_builder.uri().scheme() {
275        Some(scheme) if scheme == &http::uri::Scheme::HTTPS => {
276            endpoint_builder = endpoint_builder
277                .tls_config(ClientTlsConfig::new().with_webpki_roots())
278                .map_err(|err| {
279                    Error::new(
280                        ErrorKind::ConfigInvalid,
281                        "invalid GCS gRPC TLS configuration",
282                    )
283                    .set_source(err)
284                })?;
285        }
286        Some(scheme) if scheme == &http::uri::Scheme::HTTP => {}
287        _ => {
288            return Err(Error::new(
289                ErrorKind::ConfigInvalid,
290                "GCS gRPC endpoint must use http or https",
291            ));
292        }
293    }
294    Ok(endpoint_builder)
295}
296
297/// Google Cloud Storage gRPC backend.
298#[derive(Clone, Debug)]
299pub struct GcsGrpcBackend {
300    core: Arc<GcsGrpcCore>,
301}
302
303impl Service for GcsGrpcBackend {
304    type Reader = oio::Reader;
305    type Writer = oio::Writer;
306    type Lister = oio::Lister;
307    type Deleter = oio::Deleter;
308    type Copier = oio::Copier;
309
310    fn info(&self) -> ServiceInfo {
311        self.core.info.clone()
312    }
313
314    fn capability(&self) -> Capability {
315        self.core.capability
316    }
317
318    async fn create_dir(
319        &self,
320        _ctx: &OperationContext,
321        _path: &str,
322        _args: OpCreateDir,
323    ) -> Result<RpCreateDir> {
324        Err(Error::new(
325            ErrorKind::Unsupported,
326            "operation is not supported",
327        ))
328    }
329
330    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
331        if path == "/" {
332            return Ok(RpStat::new(Metadata::new(EntryMode::DIR)));
333        }
334        let object = self.core.object_name(path);
335        let request = GetObjectRequest {
336            bucket: self.core.bucket_resource(),
337            object,
338            generation: parse_generation(args.version())?,
339        };
340        let request = self
341            .core
342            .request(ctx, request, &[("bucket", &self.core.bucket_resource())])
343            .await?;
344        let response = self
345            .core
346            .client()
347            .get_object(request)
348            .await
349            .map_err(parse_status)?;
350        Ok(RpStat::new(parse_object(response.get_ref())))
351    }
352
353    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
354        Ok(Box::new(GcsGrpcReader::new(
355            self.core.clone(),
356            ctx.clone(),
357            path,
358            args,
359        )))
360    }
361
362    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
363        Ok(Box::new(GcsGrpcWriter::new(
364            self.core.clone(),
365            ctx.clone(),
366            path,
367            args,
368        )))
369    }
370
371    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
372        Ok(Box::new(GcsGrpcDeleter::new(
373            self.core.clone(),
374            ctx.clone(),
375        )))
376    }
377
378    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
379        Ok(Box::new(GcsGrpcLister::new(
380            self.core.clone(),
381            ctx.clone(),
382            path,
383            args,
384        )))
385    }
386
387    fn copy(
388        &self,
389        ctx: &OperationContext,
390        from: &str,
391        to: &str,
392        args: OpCopy,
393    ) -> Result<Self::Copier> {
394        Ok(Box::new(new_gcs_grpc_copier(
395            self.core.clone(),
396            ctx.clone(),
397            from,
398            to,
399            args,
400        )))
401    }
402
403    async fn rename(
404        &self,
405        _ctx: &OperationContext,
406        _from: &str,
407        _to: &str,
408        _args: OpRename,
409    ) -> Result<RpRename> {
410        Err(Error::new(
411            ErrorKind::Unsupported,
412            "operation is not supported",
413        ))
414    }
415
416    async fn presign(
417        &self,
418        _ctx: &OperationContext,
419        _path: &str,
420        _args: OpPresign,
421    ) -> Result<RpPresign> {
422        Err(Error::new(
423            ErrorKind::Unsupported,
424            "operation is not supported",
425        ))
426    }
427}
428#[cfg(test)]
429mod tests {
430    use super::*;
431
432    #[test]
433    fn builder_does_not_require_a_tokio_runtime() {
434        GcsGrpcBuilder::default()
435            .bucket("example-bucket")
436            .skip_signature()
437            .build()
438            .unwrap();
439    }
440
441    #[test]
442    fn endpoint_accepts_http_schemes_case_insensitively() {
443        assert!(build_endpoint("HTTPS://storage.googleapis.com").is_ok());
444        assert!(build_endpoint("ftp://storage.googleapis.com").is_err());
445    }
446}
opendal_service_gcs_grpc/backend.rs

opendal_service_gcs_grpc/
backend.rs
