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::{ErrorContext, 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    type Composer = ();
310
311    fn info(&self) -> ServiceInfo {
312        self.core.info.clone()
313    }
314
315    fn capability(&self) -> Capability {
316        self.core.capability
317    }
318
319    async fn create_dir(
320        &self,
321        _ctx: &OperationContext,
322        _path: &str,
323        _args: OpCreateDir,
324    ) -> Result<RpCreateDir> {
325        Err(Error::new(
326            ErrorKind::Unsupported,
327            "operation is not supported",
328        ))
329    }
330
331    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
332        if path == "/" {
333            return Ok(RpStat::new(MetadataBuilder::dir().build()));
334        }
335        let object = self.core.object_name(path);
336        let request = GetObjectRequest {
337            bucket: self.core.bucket_resource(),
338            object,
339            generation: parse_generation(args.version())?,
340        };
341        let request = self
342            .core
343            .request(ctx, request, &[("bucket", &self.core.bucket_resource())])
344            .await?;
345        let response = self
346            .core
347            .client()
348            .get_object(request)
349            .await
350            .map_err(|status| {
351                parse_status(ErrorContext::new(ServiceOperation("GetObject")), status)
352            })?;
353        Ok(RpStat::new(parse_object(response.get_ref())))
354    }
355
356    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
357        Ok(Box::new(GcsGrpcReader::new(
358            self.core.clone(),
359            ctx.clone(),
360            path,
361            args,
362        )))
363    }
364
365    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
366        Ok(Box::new(GcsGrpcWriter::new(
367            self.core.clone(),
368            ctx.clone(),
369            path,
370            args,
371        )))
372    }
373
374    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
375        Ok(Box::new(GcsGrpcDeleter::new(
376            self.core.clone(),
377            ctx.clone(),
378        )))
379    }
380
381    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
382        Ok(Box::new(GcsGrpcLister::new(
383            self.core.clone(),
384            ctx.clone(),
385            path,
386            args,
387        )))
388    }
389
390    fn copy(
391        &self,
392        ctx: &OperationContext,
393        from: &str,
394        to: &str,
395        args: OpCopy,
396    ) -> Result<Self::Copier> {
397        Ok(Box::new(new_gcs_grpc_copier(
398            self.core.clone(),
399            ctx.clone(),
400            from,
401            to,
402            args,
403        )))
404    }
405
406    async fn rename(
407        &self,
408        _ctx: &OperationContext,
409        _from: &str,
410        _to: &str,
411        _args: OpRename,
412    ) -> Result<RpRename> {
413        Err(Error::new(
414            ErrorKind::Unsupported,
415            "operation is not supported",
416        ))
417    }
418
419    async fn presign(
420        &self,
421        _ctx: &OperationContext,
422        _path: &str,
423        _args: OpPresign,
424    ) -> Result<RpPresign> {
425        Err(Error::new(
426            ErrorKind::Unsupported,
427            "operation is not supported",
428        ))
429    }
430}
431#[cfg(test)]
432mod tests {
433    use super::*;
434
435    #[test]
436    fn builder_does_not_require_a_tokio_runtime() {
437        GcsGrpcBuilder::default()
438            .bucket("example-bucket")
439            .skip_signature()
440            .build()
441            .unwrap();
442    }
443
444    #[test]
445    fn endpoint_accepts_http_schemes_case_insensitively() {
446        assert!(build_endpoint("HTTPS://storage.googleapis.com").is_ok());
447        assert!(build_endpoint("ftp://storage.googleapis.com").is_err());
448    }
449}
opendal_service_gcs_grpc/backend.rs

opendal_service_gcs_grpc/
backend.rs
