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 endpoint = build_endpoint(&uri, &bucket)?;
196        debug!("backend use endpoint {}", endpoint);
197
198        let os_env = OsEnv;
199        let ctx = Context::new()
200            .with_file_read(TokioFileRead)
201            .with_env(os_env);
202
203        let mut credential = if self.config.disable_config_load {
204            DefaultCredentialProvider::builder()
205                .no_env()
206                .no_web_identity()
207                .build()
208        } else {
209            DefaultCredentialProvider::new()
210        };
211
212        if let (Some(secret_id), Some(secret_key)) = (
213            self.config.secret_id.as_deref(),
214            self.config.secret_key.as_deref(),
215        ) {
216            let static_provider = if let Some(token) = self.config.security_token.as_deref() {
217                StaticCredentialProvider::with_security_token(secret_id, secret_key, token)
218            } else {
219                StaticCredentialProvider::new(secret_id, secret_key)
220            };
221
222            credential = credential.push_front(static_provider);
223        }
224
225        let signer = Signer::new(ctx, credential, RequestSigner::new());
226
227        let info = ServiceInfo::new(COS_SCHEME, &root, &bucket);
228        let capability = Capability {
229            stat: true,
230            stat_with_if_match: true,
231            stat_with_if_none_match: true,
232            stat_with_version: true,
233
234            read: true,
235            read_with_suffix: true,
236
237            read_with_if_match: true,
238            read_with_if_none_match: true,
239            read_with_if_modified_since: true,
240            read_with_if_unmodified_since: true,
241            read_with_version: true,
242
243            write: true,
244            write_can_empty: true,
245            write_can_append: true,
246            write_can_multi: true,
247            write_with_content_type: true,
248            write_with_cache_control: true,
249            write_with_content_disposition: true,
250            write_with_if_not_exists: true,
251            copy_with_if_not_exists: true,
252            // The min multipart size of COS is 1 MiB.
253            //
254            // ref: <https://www.tencentcloud.com/document/product/436/14112>
255            write_multi_min_size: Some(1024 * 1024),
256            // The max multipart size of COS is 5 GiB.
257            //
258            // ref: <https://www.tencentcloud.com/document/product/436/14112>
259            write_multi_max_size: if cfg!(target_pointer_width = "64") {
260                Some(5 * 1024 * 1024 * 1024)
261            } else {
262                Some(usize::MAX)
263            },
264            write_with_user_metadata: true,
265
266            delete: true,
267            delete_with_version: true,
268            copy: true,
269
270            list: true,
271            list_with_recursive: true,
272            list_with_versions: true,
273            list_with_deleted: true,
274
275            presign: true,
276            presign_stat: true,
277            presign_read: true,
278            presign_write: true,
279
280            shared: true,
281
282            ..Default::default()
283        };
284
285        Ok(CosBackend {
286            core: Arc::new(CosCore {
287                info,
288                capability,
289                bucket: bucket.clone(),
290                root,
291                endpoint,
292                signer,
293            }),
294        })
295    }
296}
297
298/// Backend for Tencent-Cloud COS services.
299#[derive(Debug, Clone)]
300pub struct CosBackend {
301    pub(crate) core: Arc<CosCore>,
302}
303
304impl Service for CosBackend {
305    type Reader = oio::StreamReader<CosReader>;
306    type Writer = CosWriters;
307    type Lister = CosListers;
308    type Deleter = oio::OneShotDeleter<CosDeleter>;
309    type Copier = oio::OneShotCopier;
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        let resp = self.core.cos_head_object(ctx, path, &args).await?;
333
334        let status = resp.status();
335
336        match status {
337            StatusCode::OK => {
338                let headers = resp.headers();
339                let mut meta = parse_into_metadata(path, headers)?;
340
341                let user_meta = parse_prefixed_headers(headers, "x-cos-meta-");
342                if !user_meta.is_empty() {
343                    meta = meta.with_user_metadata(user_meta);
344                }
345
346                if let Some(v) = parse_header_to_str(headers, constants::X_COS_VERSION_ID)?
347                    && v != "null"
348                {
349                    meta.set_version(v);
350                }
351
352                Ok(RpStat::new(meta))
353            }
354            _ => Err(parse_error(resp)),
355        }
356    }
357    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
358        let output: oio::StreamReader<CosReader> = {
359            Ok(oio::StreamReader::new(CosReader::new(
360                self.clone(),
361                ctx.clone(),
362                path,
363                args,
364            )))
365        }?;
366
367        Ok(output)
368    }
369
370    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
371        let output: CosWriters = {
372            let writer = CosWriter::new(self.core.clone(), ctx.clone(), path, args.clone());
373
374            let w = if args.append() {
375                CosWriters::Two(oio::AppendWriter::new(writer))
376            } else {
377                CosWriters::One(oio::MultipartWriter::new(
378                    ctx.executor().clone(),
379                    writer,
380                    args.concurrent(),
381                ))
382            };
383
384            Ok(w)
385        }?;
386
387        Ok(output)
388    }
389
390    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
391        let output: oio::OneShotDeleter<CosDeleter> = {
392            Ok(oio::OneShotDeleter::new(CosDeleter::new(
393                self.core.clone(),
394                ctx.clone(),
395            )))
396        }?;
397
398        Ok(output)
399    }
400
401    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
402        let output: CosListers = {
403            let l = if args.versions() || args.deleted() {
404                TwoWays::Two(oio::PageLister::new(CosObjectVersionsLister::new(
405                    self.core.clone(),
406                    ctx.clone(),
407                    path,
408                    args,
409                )))
410            } else {
411                TwoWays::One(oio::PageLister::new(CosLister::new(
412                    self.core.clone(),
413                    ctx.clone(),
414                    path,
415                    args.recursive(),
416                    args.limit(),
417                )))
418            };
419
420            Ok(l)
421        }?;
422
423        Ok(output)
424    }
425
426    fn copy(
427        &self,
428        ctx: &OperationContext,
429        from: &str,
430        to: &str,
431        args: OpCopy,
432        _opts: OpCopier,
433    ) -> Result<Self::Copier> {
434        let core = self.core.clone();
435        let ctx = ctx.clone();
436        let from = from.to_string();
437        let to = to.to_string();
438        Ok(oio::OneShotCopier::new(async move {
439            let resp = core.cos_copy_object(&ctx, &from, &to, &args).await?;
440
441            let status = resp.status();
442
443            match status {
444                StatusCode::OK => Ok(Metadata::default()),
445                _ => Err(parse_error(resp)),
446            }
447        }))
448    }
449
450    async fn rename(
451        &self,
452        _ctx: &OperationContext,
453        _from: &str,
454        _to: &str,
455        _args: OpRename,
456    ) -> Result<RpRename> {
457        Err(Error::new(
458            ErrorKind::Unsupported,
459            "operation is not supported",
460        ))
461    }
462
463    async fn presign(
464        &self,
465        ctx: &OperationContext,
466        path: &str,
467        args: OpPresign,
468    ) -> Result<RpPresign> {
469        let req = match args.operation() {
470            PresignOperation::Stat(v) => self.core.cos_head_object_request(path, v),
471            PresignOperation::Read(range, v) => self.core.cos_get_object_request(path, *range, v),
472            PresignOperation::Write(v) => {
473                self.core
474                    .cos_put_object_request(path, None, v, Buffer::new())
475            }
476            PresignOperation::Delete(_) => Err(Error::new(
477                ErrorKind::Unsupported,
478                "operation is not supported",
479            )),
480            _ => Err(Error::new(
481                ErrorKind::Unsupported,
482                "operation is not supported",
483            )),
484        };
485        let req = req?;
486        let req = self.core.sign_query(ctx, req, args.expire()).await?;
487
488        // We don't need this request anymore, consume it directly.
489        let (parts, _) = req.into_parts();
490
491        Ok(RpPresign::new(PresignedRequest::new(
492            parts.method,
493            parts.uri,
494            parts.headers,
495        )))
496    }
497}
498
499/// Compose the request endpoint for a bucket, as `scheme://bucket.host[:port]`.
500///
501/// Extracted so it can be unit tested, mirroring `S3Builder::build_endpoint`.
502fn build_endpoint(uri: &Uri, bucket: &str) -> Result<String> {
503    let scheme = uri.scheme_str().unwrap_or("https");
504
505    let host = uri.host().ok_or_else(|| {
506        Error::new(ErrorKind::ConfigInvalid, "endpoint host is empty")
507            .with_context("service", COS_SCHEME)
508            .with_context("endpoint", uri.to_string())
509    })?;
510
511    // If the endpoint already carries the bucket as its leftmost label, don't add it twice.
512    let host = host.strip_prefix(&format!("{bucket}.")).unwrap_or(host);
513
514    // Keep the port. `Uri::host` omits it, so composing from the host alone silently sent every
515    // request to the scheme default.
516    Ok(match uri.port_u16() {
517        Some(port) => format!("{scheme}://{bucket}.{host}:{port}"),
518        None => format!("{scheme}://{bucket}.{host}"),
519    })
520}
521
522#[cfg(test)]
523mod tests {
524    use super::*;
525
526    fn endpoint_of(raw: &str, bucket: &str) -> String {
527        build_endpoint(&raw.parse::<Uri>().unwrap(), bucket).unwrap()
528    }
529
530    #[test]
531    fn build_endpoint_keeps_a_custom_port() {
532        assert_eq!(
533            endpoint_of(
534                "https://cos.internal.example.com:8443",
535                "examplebucket-1250000000"
536            ),
537            "https://examplebucket-1250000000.cos.internal.example.com:8443"
538        );
539    }
540
541    #[test]
542    fn build_endpoint_prefixes_the_bucket() {
543        assert_eq!(
544            endpoint_of(
545                "https://cos.ap-guangzhou.myqcloud.com",
546                "examplebucket-1250000000"
547            ),
548            "https://examplebucket-1250000000.cos.ap-guangzhou.myqcloud.com"
549        );
550    }
551
552    #[test]
553    fn build_endpoint_does_not_repeat_a_bucket_already_in_the_host() {
554        // The previous `replace("//{bucket}.", "//")` could never match, because Uri::host never
555        // contains "//", so this doubled the bucket label.
556        assert_eq!(
557            endpoint_of(
558                "https://examplebucket-1250000000.cos.ap-guangzhou.myqcloud.com",
559                "examplebucket-1250000000"
560            ),
561            "https://examplebucket-1250000000.cos.ap-guangzhou.myqcloud.com"
562        );
563    }
564
565    #[test]
566    fn build_endpoint_defaults_the_scheme_to_https() {
567        // A bare host parses with no scheme; "//host" parses with no host at all, which is why
568        // the missing-host case below is a real input rather than a contrived one.
569        assert_eq!(
570            endpoint_of("cos.ap-guangzhou.myqcloud.com", "b"),
571            "https://b.cos.ap-guangzhou.myqcloud.com"
572        );
573    }
574
575    #[test]
576    fn build_endpoint_reports_a_missing_host_instead_of_panicking() {
577        // Both of these parse to host = None. The previous code called .unwrap() on that.
578        for raw in ["/just/a/path", "//cos.ap-guangzhou.myqcloud.com"] {
579            let uri = raw.parse::<Uri>().unwrap();
580            assert!(
581                build_endpoint(&uri, "b").is_err(),
582                "expected an error for {raw}"
583            );
584        }
585    }
586}
opendal_service_cos/backend.rs

opendal_service_cos/
backend.rs
