opendal_service_cos/
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 http::StatusCode;
22use http::Uri;
23use log::debug;
24use reqsign_core::Context;
25use reqsign_core::OsEnv;
26use reqsign_core::Signer;
27use reqsign_file_read_tokio::TokioFileRead;
28use reqsign_tencent_cos::DefaultCredentialProvider;
29use reqsign_tencent_cos::RequestSigner;
30use reqsign_tencent_cos::StaticCredentialProvider;
31
32use super::COS_SCHEME;
33use super::config::CosConfig;
34use super::core::parse_error;
35use super::core::*;
36use super::deleter::CosDeleter;
37use super::lister::CosLister;
38use super::lister::CosListers;
39use super::lister::CosObjectVersionsLister;
40use super::reader::*;
41use super::writer::CosWriter;
42use super::writer::CosWriters;
43use opendal_core::raw::*;
44use opendal_core::*;
45
46/// Tencent-Cloud COS services support.
47#[doc = include_str!("docs.md")]
48#[derive(Default)]
49pub struct CosBuilder {
50    pub(super) config: CosConfig,
51}
52
53impl Debug for CosBuilder {
54    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
55        f.debug_struct("CosBuilder")
56            .field("config", &self.config)
57            .finish_non_exhaustive()
58    }
59}
60
61impl CosBuilder {
62    /// Set root of this backend.
63    ///
64    /// All operations will happen under this root.
65    pub fn root(mut self, root: &str) -> Self {
66        self.config.root = if root.is_empty() {
67            None
68        } else {
69            Some(root.to_string())
70        };
71
72        self
73    }
74
75    /// Set endpoint of this backend.
76    ///
77    /// NOTE: no bucket or account id in endpoint, we will trim them if exists.
78    ///
79    /// # Examples
80    ///
81    /// - `https://cos.ap-singapore.myqcloud.com`
82    pub fn endpoint(mut self, endpoint: &str) -> Self {
83        if !endpoint.is_empty() {
84            self.config.endpoint = Some(endpoint.trim_end_matches('/').to_string());
85        }
86
87        self
88    }
89
90    /// Set secret_id of this backend.
91    /// - If it is set, we will take user's input first.
92    /// - If not, we will try to load it from environment.
93    pub fn secret_id(mut self, secret_id: &str) -> Self {
94        if !secret_id.is_empty() {
95            self.config.secret_id = Some(secret_id.to_string());
96        }
97
98        self
99    }
100
101    /// Set secret_key of this backend.
102    /// - If it is set, we will take user's input first.
103    /// - If not, we will try to load it from environment.
104    pub fn secret_key(mut self, secret_key: &str) -> Self {
105        if !secret_key.is_empty() {
106            self.config.secret_key = Some(secret_key.to_string());
107        }
108
109        self
110    }
111
112    /// Set security_token (a.k.a. session token) of this backend.
113    ///
114    /// This is used when authenticating via Tencent Cloud STS temporary
115    /// credentials (e.g. obtained from `GetFederationToken` or
116    /// `AssumeRole`). When provided, it will be combined with `secret_id`
117    /// and `secret_key` to sign requests, and the `x-cos-security-token`
118    /// header will be attached automatically.
119    ///
120    /// - If this is set along with `secret_id` and `secret_key`, a static
121    ///   credential provider with the token will be used.
122    /// - If this is not set, the default credential chain in reqsign will
123    ///   try to load credentials (including the token) from environment
124    ///   variables such as `TENCENTCLOUD_TOKEN`,
125    ///   `TENCENTCLOUD_SECURITY_TOKEN`, and `QCLOUD_SECRET_TOKEN`
126    ///   (unless `disable_config_load` is enabled).
127    pub fn security_token(mut self, security_token: &str) -> Self {
128        if !security_token.is_empty() {
129            self.config.security_token = Some(security_token.to_string());
130        }
131
132        self
133    }
134
135    /// Set bucket of this backend.
136    /// The param is required.
137    pub fn bucket(mut self, bucket: &str) -> Self {
138        if !bucket.is_empty() {
139            self.config.bucket = Some(bucket.to_string());
140        }
141
142        self
143    }
144
145    /// Deprecated: COS versioning capability is enabled by default.
146    #[deprecated(
147        since = "0.57.0",
148        note = "COS versioning capability is enabled by default and this option is no longer needed."
149    )]
150    pub fn enable_versioning(self, _enabled: bool) -> Self {
151        self
152    }
153
154    /// Disable config load so that opendal will not load config from
155    /// environment.
156    ///
157    /// For examples:
158    ///
159    /// - envs like `TENCENTCLOUD_SECRET_ID`
160    pub fn disable_config_load(mut self) -> Self {
161        self.config.disable_config_load = true;
162        self
163    }
164}
165
166impl Builder for CosBuilder {
167    type Config = CosConfig;
168
169    fn build(self) -> Result<impl Service> {
170        debug!("backend build started: {:?}", self);
171
172        let root = normalize_root(&self.config.root.unwrap_or_default());
173        debug!("backend use root {root}");
174
175        let bucket = match &self.config.bucket {
176            Some(bucket) => Ok(bucket.to_string()),
177            None => Err(
178                Error::new(ErrorKind::ConfigInvalid, "The bucket is misconfigured")
179                    .with_context("service", COS_SCHEME),
180            ),
181        }?;
182        debug!("backend use bucket {}", bucket);
183
184        let uri = match &self.config.endpoint {
185            Some(endpoint) => endpoint.parse::<Uri>().map_err(|err| {
186                Error::new(ErrorKind::ConfigInvalid, "endpoint is invalid")
187                    .with_context("service", COS_SCHEME)
188                    .with_context("endpoint", endpoint)
189                    .set_source(err)
190            }),
191            None => Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
192                .with_context("service", COS_SCHEME)),
193        }?;
194
195        let scheme = match uri.scheme_str() {
196            Some(scheme) => scheme.to_string(),
197            None => "https".to_string(),
198        };
199
200        // If endpoint contains bucket name, we should trim them.
201        let endpoint = uri.host().unwrap().replace(&format!("//{bucket}."), "//");
202        debug!("backend use endpoint {}", endpoint);
203
204        let os_env = OsEnv;
205        let ctx = Context::new()
206            .with_file_read(TokioFileRead)
207            .with_env(os_env);
208
209        let mut credential = if self.config.disable_config_load {
210            DefaultCredentialProvider::builder()
211                .no_env()
212                .no_web_identity()
213                .build()
214        } else {
215            DefaultCredentialProvider::new()
216        };
217
218        if let (Some(secret_id), Some(secret_key)) = (
219            self.config.secret_id.as_deref(),
220            self.config.secret_key.as_deref(),
221        ) {
222            let static_provider = if let Some(token) = self.config.security_token.as_deref() {
223                StaticCredentialProvider::with_security_token(secret_id, secret_key, token)
224            } else {
225                StaticCredentialProvider::new(secret_id, secret_key)
226            };
227
228            credential = credential.push_front(static_provider);
229        }
230
231        let signer = Signer::new(ctx, credential, RequestSigner::new());
232
233        let info = ServiceInfo::new(COS_SCHEME, &root, &bucket);
234        let capability = Capability {
235            stat: true,
236            stat_with_if_match: true,
237            stat_with_if_none_match: true,
238            stat_with_version: true,
239
240            read: true,
241            read_with_suffix: true,
242
243            read_with_if_match: true,
244            read_with_if_none_match: true,
245            read_with_if_modified_since: true,
246            read_with_if_unmodified_since: true,
247            read_with_version: true,
248
249            write: true,
250            write_can_empty: true,
251            write_can_append: true,
252            write_can_multi: true,
253            write_with_content_type: true,
254            write_with_cache_control: true,
255            write_with_content_disposition: true,
256            write_with_if_not_exists: true,
257            copy_with_if_not_exists: true,
258            // The min multipart size of COS is 1 MiB.
259            //
260            // ref: <https://www.tencentcloud.com/document/product/436/14112>
261            write_multi_min_size: Some(1024 * 1024),
262            // The max multipart size of COS is 5 GiB.
263            //
264            // ref: <https://www.tencentcloud.com/document/product/436/14112>
265            write_multi_max_size: if cfg!(target_pointer_width = "64") {
266                Some(5 * 1024 * 1024 * 1024)
267            } else {
268                Some(usize::MAX)
269            },
270            write_with_user_metadata: true,
271
272            delete: true,
273            delete_with_version: true,
274            copy: true,
275
276            list: true,
277            list_with_recursive: true,
278            list_with_versions: true,
279            list_with_deleted: true,
280
281            presign: true,
282            presign_stat: true,
283            presign_read: true,
284            presign_write: true,
285
286            shared: true,
287
288            ..Default::default()
289        };
290
291        Ok(CosBackend {
292            core: Arc::new(CosCore {
293                info,
294                capability,
295                bucket: bucket.clone(),
296                root,
297                endpoint: format!("{}://{}.{}", scheme, bucket, endpoint),
298                signer,
299            }),
300        })
301    }
302}
303
304/// Backend for Tencent-Cloud COS services.
305#[derive(Debug, Clone)]
306pub struct CosBackend {
307    pub(crate) core: Arc<CosCore>,
308}
309
310impl Service for CosBackend {
311    type Reader = oio::StreamReader<CosReader>;
312    type Writer = CosWriters;
313    type Lister = CosListers;
314    type Deleter = oio::OneShotDeleter<CosDeleter>;
315    type Copier = oio::OneShotCopier;
316
317    fn info(&self) -> ServiceInfo {
318        self.core.info.clone()
319    }
320
321    fn capability(&self) -> Capability {
322        self.core.capability
323    }
324
325    async fn create_dir(
326        &self,
327        _ctx: &OperationContext,
328        _path: &str,
329        _args: OpCreateDir,
330    ) -> Result<RpCreateDir> {
331        Err(Error::new(
332            ErrorKind::Unsupported,
333            "operation is not supported",
334        ))
335    }
336
337    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
338        let resp = self.core.cos_head_object(ctx, path, &args).await?;
339
340        let status = resp.status();
341
342        match status {
343            StatusCode::OK => {
344                let headers = resp.headers();
345                let mut meta = parse_into_metadata(path, headers)?;
346
347                let user_meta = parse_prefixed_headers(headers, "x-cos-meta-");
348                if !user_meta.is_empty() {
349                    meta = meta.with_user_metadata(user_meta);
350                }
351
352                if let Some(v) = parse_header_to_str(headers, constants::X_COS_VERSION_ID)?
353                    && v != "null"
354                {
355                    meta.set_version(v);
356                }
357
358                Ok(RpStat::new(meta))
359            }
360            _ => Err(parse_error(resp)),
361        }
362    }
363    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
364        let output: oio::StreamReader<CosReader> = {
365            Ok(oio::StreamReader::new(CosReader::new(
366                self.clone(),
367                ctx.clone(),
368                path,
369                args,
370            )))
371        }?;
372
373        Ok(output)
374    }
375
376    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
377        let output: CosWriters = {
378            let writer = CosWriter::new(self.core.clone(), ctx.clone(), path, args.clone());
379
380            let w = if args.append() {
381                CosWriters::Two(oio::AppendWriter::new(writer))
382            } else {
383                CosWriters::One(oio::MultipartWriter::new(
384                    ctx.executor().clone(),
385                    writer,
386                    args.concurrent(),
387                ))
388            };
389
390            Ok(w)
391        }?;
392
393        Ok(output)
394    }
395
396    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
397        let output: oio::OneShotDeleter<CosDeleter> = {
398            Ok(oio::OneShotDeleter::new(CosDeleter::new(
399                self.core.clone(),
400                ctx.clone(),
401            )))
402        }?;
403
404        Ok(output)
405    }
406
407    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
408        let output: CosListers = {
409            let l = if args.versions() || args.deleted() {
410                TwoWays::Two(oio::PageLister::new(CosObjectVersionsLister::new(
411                    self.core.clone(),
412                    ctx.clone(),
413                    path,
414                    args,
415                )))
416            } else {
417                TwoWays::One(oio::PageLister::new(CosLister::new(
418                    self.core.clone(),
419                    ctx.clone(),
420                    path,
421                    args.recursive(),
422                    args.limit(),
423                )))
424            };
425
426            Ok(l)
427        }?;
428
429        Ok(output)
430    }
431
432    fn copy(
433        &self,
434        ctx: &OperationContext,
435        from: &str,
436        to: &str,
437        args: OpCopy,
438        _opts: OpCopier,
439    ) -> Result<Self::Copier> {
440        let core = self.core.clone();
441        let ctx = ctx.clone();
442        let from = from.to_string();
443        let to = to.to_string();
444        Ok(oio::OneShotCopier::new(async move {
445            let resp = core.cos_copy_object(&ctx, &from, &to, &args).await?;
446
447            let status = resp.status();
448
449            match status {
450                StatusCode::OK => Ok(Metadata::default()),
451                _ => Err(parse_error(resp)),
452            }
453        }))
454    }
455
456    async fn rename(
457        &self,
458        _ctx: &OperationContext,
459        _from: &str,
460        _to: &str,
461        _args: OpRename,
462    ) -> Result<RpRename> {
463        Err(Error::new(
464            ErrorKind::Unsupported,
465            "operation is not supported",
466        ))
467    }
468
469    async fn presign(
470        &self,
471        ctx: &OperationContext,
472        path: &str,
473        args: OpPresign,
474    ) -> Result<RpPresign> {
475        let req = match args.operation() {
476            PresignOperation::Stat(v) => self.core.cos_head_object_request(path, v),
477            PresignOperation::Read(range, v) => self.core.cos_get_object_request(path, *range, v),
478            PresignOperation::Write(v) => {
479                self.core
480                    .cos_put_object_request(path, None, v, Buffer::new())
481            }
482            PresignOperation::Delete(_) => Err(Error::new(
483                ErrorKind::Unsupported,
484                "operation is not supported",
485            )),
486            _ => Err(Error::new(
487                ErrorKind::Unsupported,
488                "operation is not supported",
489            )),
490        };
491        let req = req?;
492        let req = self.core.sign_query(ctx, req, args.expire()).await?;
493
494        // We don't need this request anymore, consume it directly.
495        let (parts, _) = req.into_parts();
496
497        Ok(RpPresign::new(PresignedRequest::new(
498            parts.method,
499            parts.uri,
500            parts.headers,
501        )))
502    }
503}
opendal_service_cos/backend.rs

opendal_service_cos/
backend.rs
