opendal/services/oss/
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::fmt::Formatter;
20use std::sync::Arc;
21
22use http::Response;
23use http::StatusCode;
24use http::Uri;
25use log::debug;
26use reqsign::AliyunConfig;
27use reqsign::AliyunLoader;
28use reqsign::AliyunOssSigner;
29
30use super::core::*;
31use super::delete::OssDeleter;
32use super::error::parse_error;
33use super::lister::OssLister;
34use super::lister::OssListers;
35use super::lister::OssObjectVersionsLister;
36use super::writer::OssWriter;
37use super::writer::OssWriters;
38use super::DEFAULT_SCHEME;
39use crate::raw::*;
40use crate::services::OssConfig;
41use crate::*;
42const DEFAULT_BATCH_MAX_OPERATIONS: usize = 1000;
43
44impl Configurator for OssConfig {
45    type Builder = OssBuilder;
46
47    #[allow(deprecated)]
48    fn into_builder(self) -> Self::Builder {
49        OssBuilder {
50            config: self,
51
52            http_client: None,
53        }
54    }
55}
56
57/// Aliyun Object Storage Service (OSS) support
58#[doc = include_str!("docs.md")]
59#[derive(Default)]
60pub struct OssBuilder {
61    config: OssConfig,
62
63    #[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
64    http_client: Option<HttpClient>,
65}
66
67impl Debug for OssBuilder {
68    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
69        let mut d = f.debug_struct("OssBuilder");
70
71        d.field("config", &self.config);
72        d.finish_non_exhaustive()
73    }
74}
75
76impl OssBuilder {
77    /// Set root of this backend.
78    ///
79    /// All operations will happen under this root.
80    pub fn root(mut self, root: &str) -> Self {
81        self.config.root = if root.is_empty() {
82            None
83        } else {
84            Some(root.to_string())
85        };
86
87        self
88    }
89
90    /// Set bucket name of this backend.
91    pub fn bucket(mut self, bucket: &str) -> Self {
92        self.config.bucket = bucket.to_string();
93
94        self
95    }
96
97    /// Set endpoint of this backend.
98    pub fn endpoint(mut self, endpoint: &str) -> Self {
99        if !endpoint.is_empty() {
100            // Trim trailing `/` so that we can accept `http://127.0.0.1:9000/`
101            self.config.endpoint = Some(endpoint.trim_end_matches('/').to_string())
102        }
103
104        self
105    }
106
107    /// Set bucket versioning status for this backend
108    pub fn enable_versioning(mut self, enabled: bool) -> Self {
109        self.config.enable_versioning = enabled;
110
111        self
112    }
113
114    /// Set an endpoint for generating presigned urls.
115    ///
116    /// You can offer a public endpoint like <https://oss-cn-beijing.aliyuncs.com> to return a presinged url for
117    /// public accessors, along with an internal endpoint like <https://oss-cn-beijing-internal.aliyuncs.com>
118    /// to access objects in a faster path.
119    ///
120    /// - If presign_endpoint is set, we will use presign_endpoint on generating presigned urls.
121    /// - if not, we will use endpoint as default.
122    pub fn presign_endpoint(mut self, endpoint: &str) -> Self {
123        if !endpoint.is_empty() {
124            // Trim trailing `/` so that we can accept `http://127.0.0.1:9000/`
125            self.config.presign_endpoint = Some(endpoint.trim_end_matches('/').to_string())
126        }
127
128        self
129    }
130
131    /// Set access_key_id of this backend.
132    ///
133    /// - If access_key_id is set, we will take user's input first.
134    /// - If not, we will try to load it from environment.
135    pub fn access_key_id(mut self, v: &str) -> Self {
136        if !v.is_empty() {
137            self.config.access_key_id = Some(v.to_string())
138        }
139
140        self
141    }
142
143    /// Set access_key_secret of this backend.
144    ///
145    /// - If access_key_secret is set, we will take user's input first.
146    /// - If not, we will try to load it from environment.
147    pub fn access_key_secret(mut self, v: &str) -> Self {
148        if !v.is_empty() {
149            self.config.access_key_secret = Some(v.to_string())
150        }
151
152        self
153    }
154
155    /// Set security_token for this backend.
156    ///
157    /// - If security_token is set, we will take user's input first.
158    /// - If not, we will try to load it from environment.
159    pub fn security_token(mut self, security_token: &str) -> Self {
160        if !security_token.is_empty() {
161            self.config.security_token = Some(security_token.to_string())
162        }
163
164        self
165    }
166
167    /// Specify the http client that used by this service.
168    ///
169    /// # Notes
170    ///
171    /// This API is part of OpenDAL's Raw API. `HttpClient` could be changed
172    /// during minor updates.
173    #[deprecated(since = "0.53.0", note = "Use `Operator::update_http_client` instead")]
174    #[allow(deprecated)]
175    pub fn http_client(mut self, client: HttpClient) -> Self {
176        self.http_client = Some(client);
177        self
178    }
179
180    /// preprocess the endpoint option
181    fn parse_endpoint(&self, endpoint: &Option<String>, bucket: &str) -> Result<(String, String)> {
182        let (endpoint, host) = match endpoint.clone() {
183            Some(ep) => {
184                let uri = ep.parse::<Uri>().map_err(|err| {
185                    Error::new(ErrorKind::ConfigInvalid, "endpoint is invalid")
186                        .with_context("service", Scheme::Oss)
187                        .with_context("endpoint", &ep)
188                        .set_source(err)
189                })?;
190                let host = uri.host().ok_or_else(|| {
191                    Error::new(ErrorKind::ConfigInvalid, "endpoint host is empty")
192                        .with_context("service", Scheme::Oss)
193                        .with_context("endpoint", &ep)
194                })?;
195                let full_host = if let Some(port) = uri.port_u16() {
196                    format!("{bucket}.{host}:{port}")
197                } else {
198                    format!("{bucket}.{host}")
199                };
200                let endpoint = match uri.scheme_str() {
201                    Some(scheme_str) => match scheme_str {
202                        "http" | "https" => format!("{scheme_str}://{full_host}"),
203                        _ => {
204                            return Err(Error::new(
205                                ErrorKind::ConfigInvalid,
206                                "endpoint protocol is invalid",
207                            )
208                            .with_context("service", Scheme::Oss));
209                        }
210                    },
211                    None => format!("https://{full_host}"),
212                };
213                (endpoint, full_host)
214            }
215            None => {
216                return Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
217                    .with_context("service", Scheme::Oss));
218            }
219        };
220        Ok((endpoint, host))
221    }
222
223    /// Set server_side_encryption for this backend.
224    ///
225    /// Available values: `AES256`, `KMS`.
226    ///
227    /// Reference: <https://www.alibabacloud.com/help/en/object-storage-service/latest/server-side-encryption-5>
228    /// Brief explanation:
229    /// There are two server-side encryption methods available:
230    /// SSE-AES256:
231    ///     1. Configure the bucket encryption mode as OSS-managed and specify the encryption algorithm as AES256.
232    ///     2. Include the `x-oss-server-side-encryption` parameter in the request and set its value to AES256.
233    /// SSE-KMS:
234    ///     1. To use this service, you need to first enable KMS.
235    ///     2. Configure the bucket encryption mode as KMS, and specify the specific CMK ID for BYOK (Bring Your Own Key)
236    ///        or not specify the specific CMK ID for OSS-managed KMS key.
237    ///     3. Include the `x-oss-server-side-encryption` parameter in the request and set its value to KMS.
238    ///     4. If a specific CMK ID is specified, include the `x-oss-server-side-encryption-key-id` parameter in the request, and set its value to the specified CMK ID.
239    pub fn server_side_encryption(mut self, v: &str) -> Self {
240        if !v.is_empty() {
241            self.config.server_side_encryption = Some(v.to_string())
242        }
243        self
244    }
245
246    /// Set server_side_encryption_key_id for this backend.
247    ///
248    /// # Notes
249    ///
250    /// This option only takes effect when server_side_encryption equals to KMS.
251    pub fn server_side_encryption_key_id(mut self, v: &str) -> Self {
252        if !v.is_empty() {
253            self.config.server_side_encryption_key_id = Some(v.to_string())
254        }
255        self
256    }
257
258    /// Set maximum batch operations of this backend.
259    #[deprecated(
260        since = "0.52.0",
261        note = "Please use `delete_max_size` instead of `batch_max_operations`"
262    )]
263    pub fn batch_max_operations(mut self, delete_max_size: usize) -> Self {
264        self.config.delete_max_size = Some(delete_max_size);
265
266        self
267    }
268
269    /// Set maximum delete operations of this backend.
270    pub fn delete_max_size(mut self, delete_max_size: usize) -> Self {
271        self.config.delete_max_size = Some(delete_max_size);
272
273        self
274    }
275
276    /// Allow anonymous will allow opendal to send request without signing
277    /// when credential is not loaded.
278    pub fn allow_anonymous(mut self) -> Self {
279        self.config.allow_anonymous = true;
280        self
281    }
282
283    /// Set role_arn for this backend.
284    ///
285    /// If `role_arn` is set, we will use already known config as source
286    /// credential to assume role with `role_arn`.
287    pub fn role_arn(mut self, role_arn: &str) -> Self {
288        if !role_arn.is_empty() {
289            self.config.role_arn = Some(role_arn.to_string())
290        }
291
292        self
293    }
294
295    /// Set role_session_name for this backend.
296    pub fn role_session_name(mut self, role_session_name: &str) -> Self {
297        if !role_session_name.is_empty() {
298            self.config.role_session_name = Some(role_session_name.to_string())
299        }
300
301        self
302    }
303
304    /// Set oidc_provider_arn for this backend.
305    pub fn oidc_provider_arn(mut self, oidc_provider_arn: &str) -> Self {
306        if !oidc_provider_arn.is_empty() {
307            self.config.oidc_provider_arn = Some(oidc_provider_arn.to_string())
308        }
309
310        self
311    }
312
313    /// Set oidc_token_file for this backend.
314    pub fn oidc_token_file(mut self, oidc_token_file: &str) -> Self {
315        if !oidc_token_file.is_empty() {
316            self.config.oidc_token_file = Some(oidc_token_file.to_string())
317        }
318
319        self
320    }
321
322    /// Set sts_endpoint for this backend.
323    pub fn sts_endpoint(mut self, sts_endpoint: &str) -> Self {
324        if !sts_endpoint.is_empty() {
325            self.config.sts_endpoint = Some(sts_endpoint.to_string())
326        }
327
328        self
329    }
330}
331
332impl Builder for OssBuilder {
333    type Config = OssConfig;
334
335    fn build(self) -> Result<impl Access> {
336        debug!("backend build started: {:?}", &self);
337
338        let root = normalize_root(&self.config.root.clone().unwrap_or_default());
339        debug!("backend use root {}", &root);
340
341        // Handle endpoint, region and bucket name.
342        let bucket = match self.config.bucket.is_empty() {
343            false => Ok(&self.config.bucket),
344            true => Err(
345                Error::new(ErrorKind::ConfigInvalid, "The bucket is misconfigured")
346                    .with_context("service", Scheme::Oss),
347            ),
348        }?;
349
350        // Retrieve endpoint and host by parsing the endpoint option and bucket. If presign_endpoint is not
351        // set, take endpoint as default presign_endpoint.
352        let (endpoint, host) = self.parse_endpoint(&self.config.endpoint, bucket)?;
353        debug!("backend use bucket {}, endpoint: {}", &bucket, &endpoint);
354
355        let presign_endpoint = if self.config.presign_endpoint.is_some() {
356            self.parse_endpoint(&self.config.presign_endpoint, bucket)?
357                .0
358        } else {
359            endpoint.clone()
360        };
361        debug!("backend use presign_endpoint: {}", &presign_endpoint);
362
363        let server_side_encryption = match &self.config.server_side_encryption {
364            None => None,
365            Some(v) => Some(
366                build_header_value(v)
367                    .map_err(|err| err.with_context("key", "server_side_encryption"))?,
368            ),
369        };
370
371        let server_side_encryption_key_id = match &self.config.server_side_encryption_key_id {
372            None => None,
373            Some(v) => Some(
374                build_header_value(v)
375                    .map_err(|err| err.with_context("key", "server_side_encryption_key_id"))?,
376            ),
377        };
378
379        let mut cfg = AliyunConfig::default();
380        // Load cfg from env first.
381        cfg = cfg.from_env();
382
383        if let Some(v) = self.config.access_key_id {
384            cfg.access_key_id = Some(v);
385        }
386
387        if let Some(v) = self.config.access_key_secret {
388            cfg.access_key_secret = Some(v);
389        }
390
391        if let Some(v) = self.config.security_token {
392            cfg.security_token = Some(v);
393        }
394
395        if let Some(v) = self.config.role_arn {
396            cfg.role_arn = Some(v);
397        }
398
399        // override default role_session_name if set
400        if let Some(v) = self.config.role_session_name {
401            cfg.role_session_name = v;
402        }
403
404        if let Some(v) = self.config.oidc_provider_arn {
405            cfg.oidc_provider_arn = Some(v);
406        }
407
408        if let Some(v) = self.config.oidc_token_file {
409            cfg.oidc_token_file = Some(v);
410        }
411
412        if let Some(v) = self.config.sts_endpoint {
413            cfg.sts_endpoint = Some(v);
414        }
415
416        let loader = AliyunLoader::new(GLOBAL_REQWEST_CLIENT.clone(), cfg);
417
418        let signer = AliyunOssSigner::new(bucket);
419
420        let delete_max_size = self
421            .config
422            .delete_max_size
423            .unwrap_or(DEFAULT_BATCH_MAX_OPERATIONS);
424
425        Ok(OssBackend {
426            core: Arc::new(OssCore {
427                info: {
428                    let am = AccessorInfo::default();
429                    am.set_scheme(DEFAULT_SCHEME)
430                        .set_root(&root)
431                        .set_name(bucket)
432                        .set_native_capability(Capability {
433                            stat: true,
434                            stat_with_if_match: true,
435                            stat_with_if_none_match: true,
436                            stat_with_version: self.config.enable_versioning,
437
438                            read: true,
439
440                            read_with_if_match: true,
441                            read_with_if_none_match: true,
442                            read_with_version: self.config.enable_versioning,
443                            read_with_if_modified_since: true,
444                            read_with_if_unmodified_since: true,
445
446                            write: true,
447                            write_can_empty: true,
448                            write_can_append: true,
449                            write_can_multi: true,
450                            write_with_cache_control: true,
451                            write_with_content_type: true,
452                            write_with_content_disposition: true,
453                            // TODO: set this to false while version has been enabled.
454                            write_with_if_not_exists: !self.config.enable_versioning,
455
456                            // The min multipart size of OSS is 100 KiB.
457                            //
458                            // ref: <https://www.alibabacloud.com/help/en/oss/user-guide/multipart-upload-12>
459                            write_multi_min_size: Some(100 * 1024),
460                            // The max multipart size of OSS is 5 GiB.
461                            //
462                            // ref: <https://www.alibabacloud.com/help/en/oss/user-guide/multipart-upload-12>
463                            write_multi_max_size: if cfg!(target_pointer_width = "64") {
464                                Some(5 * 1024 * 1024 * 1024)
465                            } else {
466                                Some(usize::MAX)
467                            },
468                            write_with_user_metadata: true,
469
470                            delete: true,
471                            delete_with_version: self.config.enable_versioning,
472                            delete_max_size: Some(delete_max_size),
473
474                            copy: true,
475
476                            list: true,
477                            list_with_limit: true,
478                            list_with_start_after: true,
479                            list_with_recursive: true,
480                            list_with_versions: self.config.enable_versioning,
481                            list_with_deleted: self.config.enable_versioning,
482
483                            presign: true,
484                            presign_stat: true,
485                            presign_read: true,
486                            presign_write: true,
487
488                            shared: true,
489
490                            ..Default::default()
491                        });
492
493                    // allow deprecated api here for compatibility
494                    #[allow(deprecated)]
495                    if let Some(client) = self.http_client {
496                        am.update_http_client(|_| client);
497                    }
498
499                    am.into()
500                },
501                root,
502                bucket: bucket.to_owned(),
503                endpoint,
504                host,
505                presign_endpoint,
506                allow_anonymous: self.config.allow_anonymous,
507                signer,
508                loader,
509                server_side_encryption,
510                server_side_encryption_key_id,
511            }),
512        })
513    }
514}
515
516#[derive(Debug, Clone)]
517/// Aliyun Object Storage Service backend
518pub struct OssBackend {
519    core: Arc<OssCore>,
520}
521
522impl Access for OssBackend {
523    type Reader = HttpBody;
524    type Writer = OssWriters;
525    type Lister = OssListers;
526    type Deleter = oio::BatchDeleter<OssDeleter>;
527
528    fn info(&self) -> Arc<AccessorInfo> {
529        self.core.info.clone()
530    }
531
532    async fn stat(&self, path: &str, args: OpStat) -> Result<RpStat> {
533        let resp = self.core.oss_head_object(path, &args).await?;
534
535        let status = resp.status();
536
537        match status {
538            StatusCode::OK => {
539                let headers = resp.headers();
540                let mut meta = self.core.parse_metadata(path, resp.headers())?;
541
542                if let Some(v) = parse_header_to_str(headers, constants::X_OSS_VERSION_ID)? {
543                    meta.set_version(v);
544                }
545
546                Ok(RpStat::new(meta))
547            }
548            _ => Err(parse_error(resp)),
549        }
550    }
551
552    async fn read(&self, path: &str, args: OpRead) -> Result<(RpRead, Self::Reader)> {
553        let resp = self.core.oss_get_object(path, &args).await?;
554
555        let status = resp.status();
556
557        match status {
558            StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
559                Ok((RpRead::default(), resp.into_body()))
560            }
561            _ => {
562                let (part, mut body) = resp.into_parts();
563                let buf = body.to_buffer().await?;
564                Err(parse_error(Response::from_parts(part, buf)))
565            }
566        }
567    }
568
569    async fn write(&self, path: &str, args: OpWrite) -> Result<(RpWrite, Self::Writer)> {
570        let writer = OssWriter::new(self.core.clone(), path, args.clone());
571
572        let w = if args.append() {
573            OssWriters::Two(oio::AppendWriter::new(writer))
574        } else {
575            OssWriters::One(oio::MultipartWriter::new(
576                self.core.info.clone(),
577                writer,
578                args.concurrent(),
579            ))
580        };
581
582        Ok((RpWrite::default(), w))
583    }
584
585    async fn delete(&self) -> Result<(RpDelete, Self::Deleter)> {
586        Ok((
587            RpDelete::default(),
588            oio::BatchDeleter::new(OssDeleter::new(self.core.clone())),
589        ))
590    }
591
592    async fn list(&self, path: &str, args: OpList) -> Result<(RpList, Self::Lister)> {
593        let l = if args.versions() || args.deleted() {
594            TwoWays::Two(oio::PageLister::new(OssObjectVersionsLister::new(
595                self.core.clone(),
596                path,
597                args,
598            )))
599        } else {
600            TwoWays::One(oio::PageLister::new(OssLister::new(
601                self.core.clone(),
602                path,
603                args.recursive(),
604                args.limit(),
605                args.start_after(),
606            )))
607        };
608
609        Ok((RpList::default(), l))
610    }
611
612    async fn copy(&self, from: &str, to: &str, _args: OpCopy) -> Result<RpCopy> {
613        let resp = self.core.oss_copy_object(from, to).await?;
614        let status = resp.status();
615
616        match status {
617            StatusCode::OK => Ok(RpCopy::default()),
618            _ => Err(parse_error(resp)),
619        }
620    }
621
622    async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
623        // We will not send this request out, just for signing.
624        let req = match args.operation() {
625            PresignOperation::Stat(v) => self.core.oss_head_object_request(path, true, v),
626            PresignOperation::Read(v) => self.core.oss_get_object_request(path, true, v),
627            PresignOperation::Write(v) => {
628                self.core
629                    .oss_put_object_request(path, None, v, Buffer::new(), true)
630            }
631            PresignOperation::Delete(_) => Err(Error::new(
632                ErrorKind::Unsupported,
633                "operation is not supported",
634            )),
635        };
636        let mut req = req?;
637
638        self.core.sign_query(&mut req, args.expire()).await?;
639
640        // We don't need this request anymore, consume it directly.
641        let (parts, _) = req.into_parts();
642
643        Ok(RpPresign::new(PresignedRequest::new(
644            parts.method,
645            parts.uri,
646            parts.headers,
647        )))
648    }
649}
opendal/services/oss/backend.rs

opendal/services/oss/
backend.rs
