opendal_service_s3/
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::collections::HashMap;
19use std::fmt::Debug;
20use std::fmt::Write;
21use std::sync::Arc;
22use std::sync::LazyLock;
23
24use base64::Engine;
25use base64::prelude::BASE64_STANDARD;
26use constants::X_AMZ_META_PREFIX;
27use constants::X_AMZ_VERSION_ID;
28use http::StatusCode;
29use log::debug;
30use log::warn;
31use md5::Digest;
32use md5::Md5;
33use reqsign_aws_v4::AssumeRoleCredentialProvider;
34use reqsign_aws_v4::Credential;
35use reqsign_aws_v4::DefaultCredentialProvider;
36use reqsign_aws_v4::RequestSigner as AwsV4Signer;
37use reqsign_aws_v4::StaticCredentialProvider;
38use reqsign_core::Context;
39use reqsign_core::OsEnv;
40use reqsign_core::ProvideCredentialChain;
41use reqsign_core::Signer;
42use reqsign_file_read_tokio::TokioFileRead;
43use url::Url;
44
45use crate::S3_SCHEME;
46use crate::config::S3Config;
47use crate::copier::S3Copiers;
48use crate::copier::new_s3_copier;
49use crate::core::parse_error;
50use crate::core::*;
51use crate::deleter::S3Deleter;
52use crate::lister::S3ListerV1;
53use crate::lister::S3ListerV2;
54use crate::lister::S3Listers;
55use crate::lister::S3ObjectVersionsLister;
56use crate::reader::*;
57use crate::writer::S3Writer;
58use crate::writer::S3Writers;
59use opendal_core::raw::*;
60use opendal_core::*;
61
62/// Allow constructing correct region endpoint if user gives a global endpoint.
63static ENDPOINT_TEMPLATES: LazyLock<HashMap<&'static str, &'static str>> = LazyLock::new(|| {
64    let mut m = HashMap::new();
65    // AWS S3 Service.
66    m.insert(
67        "https://s3.amazonaws.com",
68        "https://s3.{region}.amazonaws.com",
69    );
70    m
71});
72
73const DEFAULT_BATCH_MAX_OPERATIONS: usize = 1000;
74
75/// Aws S3 and compatible services (including minio, digitalocean space, Tencent Cloud Object Storage(COS) and so on) support.
76/// For more information about s3-compatible services, refer to [Compatible Services](#compatible-services).
77#[doc = include_str!("docs.md")]
78#[doc = include_str!("compatible_services.md")]
79#[derive(Default)]
80pub struct S3Builder {
81    pub(super) config: S3Config,
82    pub(super) credential_providers: Option<ProvideCredentialChain<Credential>>,
83}
84
85impl Debug for S3Builder {
86    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
87        f.debug_struct("S3Builder")
88            .field("config", &self.config)
89            .finish_non_exhaustive()
90    }
91}
92
93impl S3Builder {
94    /// Set root of this backend.
95    ///
96    /// All operations will happen under this root.
97    pub fn root(mut self, root: &str) -> Self {
98        self.config.root = if root.is_empty() {
99            None
100        } else {
101            Some(root.to_string())
102        };
103
104        self
105    }
106
107    /// Set bucket name of this backend.
108    pub fn bucket(mut self, bucket: &str) -> Self {
109        self.config.bucket = bucket.to_string();
110
111        self
112    }
113
114    /// Set endpoint of this backend.
115    ///
116    /// Endpoint must be full uri, e.g.
117    ///
118    /// - AWS S3: `https://s3.amazonaws.com` or `https://s3.{region}.amazonaws.com`
119    /// - Cloudflare R2: `https://<ACCOUNT_ID>.r2.cloudflarestorage.com`
120    /// - Aliyun OSS: `https://{region}.aliyuncs.com`
121    /// - Tencent COS: `https://cos.{region}.myqcloud.com`
122    /// - Minio: `http://127.0.0.1:9000`
123    ///
124    /// If user inputs endpoint without scheme like "s3.amazonaws.com", we
125    /// will prepend "https://" before it.
126    pub fn endpoint(mut self, endpoint: &str) -> Self {
127        if !endpoint.is_empty() {
128            // Trim trailing `/` so that we can accept `http://127.0.0.1:9000/`
129            self.config.endpoint = Some(endpoint.trim_end_matches('/').to_string())
130        }
131
132        self
133    }
134
135    /// Region represent the signing region of this endpoint. This is required
136    /// if you are using the default AWS S3 endpoint.
137    ///
138    /// If using a custom endpoint,
139    /// - If region is set, we will take user's input first.
140    /// - If not, we will try to load it from environment.
141    pub fn region(mut self, region: &str) -> Self {
142        if !region.is_empty() {
143            self.config.region = Some(region.to_string())
144        }
145
146        self
147    }
148
149    /// Set access_key_id of this backend.
150    ///
151    /// - If access_key_id is set, we will take user's input first.
152    /// - If not, we will try to load it from environment.
153    pub fn access_key_id(mut self, v: &str) -> Self {
154        if !v.is_empty() {
155            self.config.access_key_id = Some(v.to_string())
156        }
157
158        self
159    }
160
161    /// Set secret_access_key of this backend.
162    ///
163    /// - If secret_access_key is set, we will take user's input first.
164    /// - If not, we will try to load it from environment.
165    pub fn secret_access_key(mut self, v: &str) -> Self {
166        if !v.is_empty() {
167            self.config.secret_access_key = Some(v.to_string())
168        }
169
170        self
171    }
172
173    /// Set role_arn for this backend.
174    ///
175    /// If `role_arn` is set, we will use already known config as source
176    /// credential to assume role with `role_arn`.
177    pub fn role_arn(mut self, v: &str) -> Self {
178        if !v.is_empty() {
179            self.config.role_arn = Some(v.to_string())
180        }
181
182        self
183    }
184
185    /// Set external_id for this backend.
186    pub fn external_id(mut self, v: &str) -> Self {
187        if !v.is_empty() {
188            self.config.external_id = Some(v.to_string())
189        }
190
191        self
192    }
193
194    /// Set role_session_name for this backend.
195    pub fn role_session_name(mut self, v: &str) -> Self {
196        if !v.is_empty() {
197            self.config.role_session_name = Some(v.to_string())
198        }
199
200        self
201    }
202
203    /// Set assume_role_duration_seconds for this backend.
204    pub fn assume_role_duration_seconds(mut self, v: u32) -> Self {
205        self.config.assume_role_duration_seconds = Some(v);
206        self
207    }
208
209    /// Set assume_role_session_tags for this backend.
210    pub fn assume_role_session_tags(mut self, tags: HashMap<String, String>) -> Self {
211        self.config.assume_role_session_tags = Some(tags);
212        self
213    }
214
215    /// Set default storage_class for this backend.
216    ///
217    /// Available values:
218    /// - `DEEP_ARCHIVE`
219    /// - `GLACIER`
220    /// - `GLACIER_IR`
221    /// - `INTELLIGENT_TIERING`
222    /// - `ONEZONE_IA`
223    /// - `OUTPOSTS`
224    /// - `REDUCED_REDUNDANCY`
225    /// - `STANDARD`
226    /// - `STANDARD_IA`
227    pub fn default_storage_class(mut self, v: &str) -> Self {
228        if !v.is_empty() {
229            self.config.default_storage_class = Some(v.to_string())
230        }
231
232        self
233    }
234
235    /// Set server_side_encryption for this backend.
236    ///
237    /// Available values: `AES256`, `aws:kms`.
238    ///
239    /// # Note
240    ///
241    /// This function is the low-level setting for SSE related features.
242    ///
243    /// SSE related options should be set carefully to make them works.
244    /// Please use `server_side_encryption_with_*` helpers if even possible.
245    pub fn server_side_encryption(mut self, v: &str) -> Self {
246        if !v.is_empty() {
247            self.config.server_side_encryption = Some(v.to_string())
248        }
249
250        self
251    }
252
253    /// Set server_side_encryption_aws_kms_key_id for this backend
254    ///
255    /// - If `server_side_encryption` set to `aws:kms`, and `server_side_encryption_aws_kms_key_id`
256    ///   is not set, S3 will use aws managed kms key to encrypt data.
257    /// - If `server_side_encryption` set to `aws:kms`, and `server_side_encryption_aws_kms_key_id`
258    ///   is a valid kms key id, S3 will use the provided kms key to encrypt data.
259    /// - If the `server_side_encryption_aws_kms_key_id` is invalid or not found, an error will be
260    ///   returned.
261    /// - If `server_side_encryption` is not `aws:kms`, setting `server_side_encryption_aws_kms_key_id` is a noop.
262    ///
263    /// # Note
264    ///
265    /// This function is the low-level setting for SSE related features.
266    ///
267    /// SSE related options should be set carefully to make them works.
268    /// Please use `server_side_encryption_with_*` helpers if even possible.
269    pub fn server_side_encryption_aws_kms_key_id(mut self, v: &str) -> Self {
270        if !v.is_empty() {
271            self.config.server_side_encryption_aws_kms_key_id = Some(v.to_string())
272        }
273
274        self
275    }
276
277    /// Set server_side_encryption_customer_algorithm for this backend.
278    ///
279    /// Available values: `AES256`.
280    ///
281    /// # Note
282    ///
283    /// This function is the low-level setting for SSE related features.
284    ///
285    /// SSE related options should be set carefully to make them works.
286    /// Please use `server_side_encryption_with_*` helpers if even possible.
287    pub fn server_side_encryption_customer_algorithm(mut self, v: &str) -> Self {
288        if !v.is_empty() {
289            self.config.server_side_encryption_customer_algorithm = Some(v.to_string())
290        }
291
292        self
293    }
294
295    /// Set server_side_encryption_customer_key for this backend.
296    ///
297    /// # Args
298    ///
299    /// `v`: base64 encoded key that matches algorithm specified in
300    /// `server_side_encryption_customer_algorithm`.
301    ///
302    /// # Note
303    ///
304    /// This function is the low-level setting for SSE related features.
305    ///
306    /// SSE related options should be set carefully to make them works.
307    /// Please use `server_side_encryption_with_*` helpers if even possible.
308    pub fn server_side_encryption_customer_key(mut self, v: &str) -> Self {
309        if !v.is_empty() {
310            self.config.server_side_encryption_customer_key = Some(v.to_string())
311        }
312
313        self
314    }
315
316    /// Set server_side_encryption_customer_key_md5 for this backend.
317    ///
318    /// # Args
319    ///
320    /// `v`: MD5 digest of key specified in `server_side_encryption_customer_key`.
321    ///
322    /// # Note
323    ///
324    /// This function is the low-level setting for SSE related features.
325    ///
326    /// SSE related options should be set carefully to make them works.
327    /// Please use `server_side_encryption_with_*` helpers if even possible.
328    pub fn server_side_encryption_customer_key_md5(mut self, v: &str) -> Self {
329        if !v.is_empty() {
330            self.config.server_side_encryption_customer_key_md5 = Some(v.to_string())
331        }
332
333        self
334    }
335
336    /// Enable server side encryption with aws managed kms key
337    ///
338    /// As known as: SSE-KMS
339    ///
340    /// NOTE: This function should not be used along with other `server_side_encryption_with_` functions.
341    pub fn server_side_encryption_with_aws_managed_kms_key(mut self) -> Self {
342        self.config.server_side_encryption = Some("aws:kms".to_string());
343        self
344    }
345
346    /// Enable server side encryption with customer managed kms key
347    ///
348    /// As known as: SSE-KMS
349    ///
350    /// NOTE: This function should not be used along with other `server_side_encryption_with_` functions.
351    pub fn server_side_encryption_with_customer_managed_kms_key(
352        mut self,
353        aws_kms_key_id: &str,
354    ) -> Self {
355        self.config.server_side_encryption = Some("aws:kms".to_string());
356        self.config.server_side_encryption_aws_kms_key_id = Some(aws_kms_key_id.to_string());
357        self
358    }
359
360    /// Enable server side encryption with s3 managed key
361    ///
362    /// As known as: SSE-S3
363    ///
364    /// NOTE: This function should not be used along with other `server_side_encryption_with_` functions.
365    pub fn server_side_encryption_with_s3_key(mut self) -> Self {
366        self.config.server_side_encryption = Some("AES256".to_string());
367        self
368    }
369
370    /// Enable server side encryption with customer key.
371    ///
372    /// As known as: SSE-C
373    ///
374    /// NOTE: This function should not be used along with other `server_side_encryption_with_` functions.
375    pub fn server_side_encryption_with_customer_key(mut self, algorithm: &str, key: &[u8]) -> Self {
376        self.config.server_side_encryption_customer_algorithm = Some(algorithm.to_string());
377        self.config.server_side_encryption_customer_key = Some(BASE64_STANDARD.encode(key));
378        let key_md5 = Md5::digest(key);
379        self.config.server_side_encryption_customer_key_md5 = Some(BASE64_STANDARD.encode(key_md5));
380        self
381    }
382
383    /// Set temporary credential used in AWS S3 connections
384    ///
385    /// # Warning
386    ///
387    /// session token's lifetime is short and requires users to refresh in time.
388    pub fn session_token(mut self, token: &str) -> Self {
389        if !token.is_empty() {
390            self.config.session_token = Some(token.to_string());
391        }
392        self
393    }
394
395    /// Disable config load so that opendal will not load config from
396    /// environment.
397    ///
398    /// For examples:
399    ///
400    /// - envs like `AWS_ACCESS_KEY_ID`
401    /// - files like `~/.aws/config`
402    pub fn disable_config_load(mut self) -> Self {
403        self.config.disable_config_load = true;
404        self
405    }
406
407    /// Disable list objects v2 so that opendal will fall back to the older
408    /// List Objects V1 to list objects.
409    ///
410    /// By default, OpenDAL uses List Objects V2 to list objects. However,
411    /// some legacy services do not yet support V2.
412    pub fn disable_list_objects_v2(mut self) -> Self {
413        self.config.disable_list_objects_v2 = true;
414        self
415    }
416
417    /// Enable request payer so that OpenDAL will send requests with `x-amz-request-payer` header.
418    ///
419    /// With this option the client accepts to pay for the request and data transfer costs.
420    pub fn enable_request_payer(mut self) -> Self {
421        self.config.enable_request_payer = true;
422        self
423    }
424
425    /// Disable load credential from ec2 metadata.
426    ///
427    /// This option is used to disable the default behavior of opendal
428    /// to load credential from ec2 metadata, a.k.a, IMDSv2
429    pub fn disable_ec2_metadata(mut self) -> Self {
430        self.config.disable_ec2_metadata = true;
431        self
432    }
433
434    /// Skip signature will skip loading credentials and signing requests.
435    pub fn skip_signature(mut self) -> Self {
436        self.config.skip_signature = true;
437        self
438    }
439
440    /// Allow anonymous will allow opendal to send request without signing
441    /// when credential is not loaded.
442    #[deprecated(
443        since = "0.57.0",
444        note = "Please use `skip_signature` instead of `allow_anonymous`"
445    )]
446    pub fn allow_anonymous(self) -> Self {
447        self.skip_signature()
448    }
449
450    /// Enable virtual host style so that opendal will send API requests
451    /// in virtual host style instead of path style.
452    ///
453    /// - By default, opendal will send API to `https://s3.us-east-1.amazonaws.com/bucket_name`
454    /// - Enabled, opendal will send API to `https://bucket_name.s3.us-east-1.amazonaws.com`
455    pub fn enable_virtual_host_style(mut self) -> Self {
456        self.config.enable_virtual_host_style = true;
457        self
458    }
459
460    /// Deprecated: S3 stat override capabilities are enabled by default.
461    #[deprecated(
462        since = "0.57.0",
463        note = "S3 stat override capabilities are enabled by default and this option is no longer needed."
464    )]
465    pub fn disable_stat_with_override(self) -> Self {
466        self
467    }
468
469    /// Deprecated: S3 versioning capability is enabled by default.
470    #[deprecated(
471        since = "0.57.0",
472        note = "S3 versioning capability is enabled by default and this option is no longer needed."
473    )]
474    pub fn enable_versioning(self, _enabled: bool) -> Self {
475        self
476    }
477
478    /// Replace the credential providers with a custom chain.
479    pub fn credential_provider_chain(mut self, chain: ProvideCredentialChain<Credential>) -> Self {
480        self.credential_providers = Some(chain);
481        self
482    }
483
484    /// Check if `bucket` is valid.
485    /// `bucket` must be not empty and if `enable_virtual_host_style` is true
486    /// it could not contain dot (.) character.
487    fn is_bucket_valid(config: &S3Config) -> bool {
488        if config.bucket.is_empty() {
489            return false;
490        }
491        // If enable virtual host style, `bucket` will reside in domain part,
492        // for example `https://bucket_name.s3.us-east-1.amazonaws.com`,
493        // so `bucket` with dot can't be recognized correctly for this format.
494        if config.enable_virtual_host_style && config.bucket.contains('.') {
495            return false;
496        }
497        true
498    }
499
500    /// Build endpoint with given region.
501    fn build_endpoint(config: &S3Config, region: &str) -> String {
502        let bucket = {
503            debug_assert!(Self::is_bucket_valid(config), "bucket must be valid");
504
505            config.bucket.as_str()
506        };
507
508        let mut endpoint = match &config.endpoint {
509            Some(endpoint) => {
510                if endpoint.starts_with("http") {
511                    endpoint.to_string()
512                } else {
513                    // Prefix https if endpoint doesn't start with scheme.
514                    format!("https://{endpoint}")
515                }
516            }
517            None => "https://s3.amazonaws.com".to_string(),
518        };
519
520        // If endpoint contains bucket name, we should trim them.
521        endpoint = endpoint.replace(&format!("//{bucket}."), "//");
522
523        // Omit default ports if specified.
524        if let Ok(url) = Url::parse(&endpoint) {
525            // Remove the trailing `/` of root path.
526            endpoint = url.to_string().trim_end_matches('/').to_string();
527        }
528
529        // Update with endpoint templates.
530        endpoint = if let Some(template) = ENDPOINT_TEMPLATES.get(endpoint.as_str()) {
531            template.replace("{region}", region)
532        } else {
533            // If we don't know where about this endpoint, just leave
534            // them as it.
535            endpoint.to_string()
536        };
537
538        // Apply virtual host style.
539        if config.enable_virtual_host_style {
540            endpoint = endpoint.replace("//", &format!("//{bucket}."))
541        } else {
542            write!(endpoint, "/{bucket}").expect("write into string must succeed");
543        };
544
545        endpoint
546    }
547
548    /// Deprecated: S3 delete batch capability is enabled by default.
549    #[deprecated(
550        since = "0.57.0",
551        note = "S3 delete batch capability is enabled by default and this option is no longer needed."
552    )]
553    pub fn batch_max_operations(self, _batch_max_operations: usize) -> Self {
554        self
555    }
556
557    /// Deprecated: S3 delete batch capability is enabled by default.
558    #[deprecated(
559        since = "0.57.0",
560        note = "S3 delete batch capability is enabled by default and this option is no longer needed."
561    )]
562    pub fn delete_max_size(self, _delete_max_size: usize) -> Self {
563        self
564    }
565
566    /// Set checksum algorithm of this backend.
567    /// This is necessary when writing to AWS S3 Buckets with Object Lock enabled for example.
568    ///
569    /// Available options:
570    /// - "crc32c"
571    /// - "md5"
572    pub fn checksum_algorithm(mut self, checksum_algorithm: &str) -> Self {
573        self.config.checksum_algorithm = Some(checksum_algorithm.to_string());
574
575        self
576    }
577
578    /// Deprecated: S3 write with If-Match capability is enabled by default.
579    #[deprecated(
580        since = "0.57.0",
581        note = "S3 write with If-Match capability is enabled by default and this option is no longer needed."
582    )]
583    pub fn disable_write_with_if_match(self) -> Self {
584        self
585    }
586
587    /// Deprecated: S3 append capability is enabled by default.
588    #[deprecated(
589        since = "0.57.0",
590        note = "S3 append capability is enabled by default and this option is no longer needed."
591    )]
592    pub fn enable_write_with_append(self) -> Self {
593        self
594    }
595
596    /// Detect region of S3 bucket.
597    ///
598    /// # Args
599    ///
600    /// - endpoint: the endpoint of S3 service
601    /// - bucket: the bucket of S3 service
602    ///
603    /// # Return
604    ///
605    /// - `Some(region)` means we detect the region successfully
606    /// - `None` means we can't detect the region or meeting errors.
607    ///
608    /// # Notes
609    ///
610    /// We will try to detect region by the following methods.
611    ///
612    /// - Match endpoint with given rules to get region
613    ///   - Cloudflare R2
614    ///   - AWS S3
615    ///   - Aliyun OSS
616    /// - Send a `HEAD` request to endpoint with bucket name to get `x-amz-bucket-region`.
617    ///
618    /// # Examples
619    ///
620    /// ```no_run
621    /// use opendal_service_s3::S3;
622    ///
623    /// # async fn example() {
624    /// let region: Option<String> = S3::detect_region("https://s3.amazonaws.com", "example").await;
625    /// # }
626    /// ```
627    ///
628    /// # Reference
629    ///
630    /// - [Amazon S3 HeadBucket API](https://docs.aws.amazon.com/zh_cn/AmazonS3/latest/API/API_HeadBucket.html)
631    pub async fn detect_region(endpoint: &str, bucket: &str) -> Option<String> {
632        // Remove the possible trailing `/` in endpoint.
633        let endpoint = endpoint.trim_end_matches('/');
634
635        // Make sure the endpoint contains the scheme.
636        let mut endpoint = if endpoint.starts_with("http") {
637            endpoint.to_string()
638        } else {
639            // Prefix https if endpoint doesn't start with scheme.
640            format!("https://{endpoint}")
641        };
642
643        // Remove bucket name from endpoint.
644        endpoint = endpoint.replace(&format!("//{bucket}."), "//");
645        let url = format!("{endpoint}/{bucket}");
646
647        debug!("detect region with url: {url}");
648
649        // Try to detect region by endpoint.
650
651        // If this bucket is R2, we can return auto directly.
652        //
653        // Reference: <https://developers.cloudflare.com/r2/api/s3/api/>
654        if endpoint.ends_with("r2.cloudflarestorage.com") {
655            return Some("auto".to_string());
656        }
657
658        // If this bucket is AWS, we can try to match the endpoint.
659        if endpoint == "https://s3.amazonaws.com" {
660            return Some("us-east-1".to_string());
661        }
662
663        if let Some(region) = endpoint
664            .strip_prefix("https://s3.")
665            .and_then(|v| v.strip_suffix(".amazonaws.com"))
666        {
667            return Some(region.to_string());
668        }
669
670        // If this bucket is OSS, we can try to match the endpoint.
671        //
672        // - `oss-ap-southeast-1.aliyuncs.com` => `oss-ap-southeast-1`
673        // - `oss-cn-hangzhou-internal.aliyuncs.com` => `oss-cn-hangzhou`
674        if let Some(v) = endpoint.strip_prefix("https://") {
675            if let Some(region) = v.strip_suffix("-internal.aliyuncs.com") {
676                return Some(region.to_string());
677            }
678            if let Some(region) = v.strip_suffix(".aliyuncs.com") {
679                return Some(region.to_string());
680            }
681        }
682
683        // Try to detect region by HeadBucket.
684        let req = http::Request::head(&url).body(Buffer::new()).ok()?;
685
686        let client = HttpTransporter::default();
687        let res = client
688            .send(req)
689            .await
690            .map_err(|err| warn!("detect region failed for: {err:?}"))
691            .ok()?;
692
693        debug!(
694            "auto detect region got response: status {:?}, header: {:?}",
695            res.status(),
696            res.headers()
697        );
698
699        // Get region from response header no matter status code.
700        if let Some(region) = res
701            .headers()
702            .get("x-amz-bucket-region")
703            .and_then(|header| header.to_str().ok())
704        {
705            return Some(region.to_string());
706        }
707
708        // Status code is 403 or 200 means we already visit the correct
709        // region, we can use the default region directly.
710        if res.status() == StatusCode::FORBIDDEN || res.status() == StatusCode::OK {
711            return Some("us-east-1".to_string());
712        }
713
714        None
715    }
716
717    /// Set default ACL for new objects.
718    pub fn default_acl(mut self, acl: &str) -> Self {
719        self.config.default_acl = Some(acl.to_string());
720        self
721    }
722}
723
724impl Builder for S3Builder {
725    type Config = S3Config;
726
727    fn build(self) -> Result<impl Service> {
728        debug!("backend build started: {:?}", self);
729
730        let S3Builder {
731            mut config,
732            credential_providers,
733        } = self;
734
735        #[allow(deprecated)]
736        if config.allow_anonymous {
737            config.skip_signature = true;
738        }
739
740        let root = normalize_root(&config.root.clone().unwrap_or_default());
741        debug!("backend use root {}", root);
742
743        // Handle bucket name.
744        let bucket = if Self::is_bucket_valid(&config) {
745            Ok(&config.bucket)
746        } else {
747            Err(
748                Error::new(ErrorKind::ConfigInvalid, "The bucket is misconfigured")
749                    .with_context("service", S3_SCHEME),
750            )
751        }?;
752        debug!("backend use bucket {}", bucket);
753
754        let default_storage_class = match &config.default_storage_class {
755            None => None,
756            Some(v) => Some(
757                build_header_value(v).map_err(|err| err.with_context("key", "storage_class"))?,
758            ),
759        };
760
761        let server_side_encryption = match &config.server_side_encryption {
762            None => None,
763            Some(v) => Some(
764                build_header_value(v)
765                    .map_err(|err| err.with_context("key", "server_side_encryption"))?,
766            ),
767        };
768
769        let server_side_encryption_aws_kms_key_id =
770            match &config.server_side_encryption_aws_kms_key_id {
771                None => None,
772                Some(v) => Some(build_header_value(v).map_err(|err| {
773                    err.with_context("key", "server_side_encryption_aws_kms_key_id")
774                })?),
775            };
776
777        let server_side_encryption_customer_algorithm =
778            match &config.server_side_encryption_customer_algorithm {
779                None => None,
780                Some(v) => Some(build_header_value(v).map_err(|err| {
781                    err.with_context("key", "server_side_encryption_customer_algorithm")
782                })?),
783            };
784
785        let server_side_encryption_customer_key =
786            match &config.server_side_encryption_customer_key {
787                None => None,
788                Some(v) => Some(build_header_value(v).map_err(|err| {
789                    err.with_context("key", "server_side_encryption_customer_key")
790                })?),
791            };
792
793        let server_side_encryption_customer_key_md5 =
794            match &config.server_side_encryption_customer_key_md5 {
795                None => None,
796                Some(v) => Some(build_header_value(v).map_err(|err| {
797                    err.with_context("key", "server_side_encryption_customer_key_md5")
798                })?),
799            };
800
801        let checksum_algorithm = match config.checksum_algorithm.as_deref() {
802            Some("crc32c") => Some(ChecksumAlgorithm::Crc32c),
803            Some("md5") => Some(ChecksumAlgorithm::Md5),
804            None => None,
805            v => {
806                return Err(Error::new(
807                    ErrorKind::ConfigInvalid,
808                    format!("{v:?} is not a supported checksum_algorithm."),
809                ));
810            }
811        };
812
813        // Determine the region
814        let region = if let Some(ref v) = config.region {
815            v.to_string()
816        } else {
817            std::env::var("AWS_REGION")
818                .or_else(|_| std::env::var("AWS_DEFAULT_REGION"))
819                .map_err(|_| {
820                    Error::new(
821                        ErrorKind::ConfigInvalid,
822                        "region is missing. Please find it by S3::detect_region() or set them in env.",
823                    )
824                    .with_operation("Builder::build")
825                    .with_context("service", S3_SCHEME)
826                })?
827        };
828        debug!("backend use region: {region}");
829
830        if config.endpoint.is_none() && !config.disable_config_load {
831            let endpoint_from_env = std::env::var("AWS_ENDPOINT_URL")
832                .or_else(|_| std::env::var("AWS_ENDPOINT"))
833                .or_else(|_| std::env::var("AWS_S3_ENDPOINT"))
834                .ok();
835            if let Some(endpoint) = endpoint_from_env {
836                let normalized = endpoint.trim_end_matches('/').to_string();
837                config.endpoint = Some(normalized);
838            }
839        }
840
841        // Building endpoint.
842        let endpoint = Self::build_endpoint(&config, &region);
843        debug!("backend use endpoint: {endpoint}");
844
845        // The base signer context only carries local config readers. HTTP
846        // sending is injected from OperationContext when S3Core signs each
847        // operation.
848        let ctx = Context::new().with_file_read(TokioFileRead).with_env(OsEnv);
849
850        let mut provider = {
851            let mut builder = DefaultCredentialProvider::builder();
852
853            if config.disable_config_load {
854                builder = builder.no_env().no_profile();
855            }
856
857            if config.disable_ec2_metadata {
858                builder = builder.no_imds();
859            }
860
861            ProvideCredentialChain::new().push(builder.build())
862        };
863
864        // Insert static key if user provided.
865        if let (Some(ak), Some(sk)) = (&config.access_key_id, &config.secret_access_key) {
866            let static_provider = if let Some(token) = config.session_token.as_deref() {
867                StaticCredentialProvider::new(ak, sk).with_session_token(token)
868            } else {
869                StaticCredentialProvider::new(ak, sk)
870            };
871            provider = provider.push_front(static_provider);
872        }
873
874        // Insert assume role provider if user provided.
875        if let Some(role_arn) = &config.role_arn {
876            // The assume-role provider owns its STS signer, so give it a
877            // concrete HTTP sender instead of relying on a future operation
878            // context.
879            let sts_ctx = ctx.clone().with_http_send(HttpTransporter::default());
880            let sts_request_signer = AwsV4Signer::new("sts", &region);
881            let sts_signer = Signer::new(sts_ctx, provider, sts_request_signer);
882            let mut assume_role_provider =
883                AssumeRoleCredentialProvider::new(role_arn.clone(), sts_signer)
884                    .with_region(region.clone())
885                    .with_regional_sts_endpoint();
886
887            if let Some(external_id) = &config.external_id {
888                assume_role_provider = assume_role_provider.with_external_id(external_id.clone());
889            }
890            if let Some(role_session_name) = &config.role_session_name {
891                assume_role_provider =
892                    assume_role_provider.with_role_session_name(role_session_name.clone());
893            }
894            if let Some(duration_seconds) = config.assume_role_duration_seconds {
895                assume_role_provider = assume_role_provider.with_duration_seconds(duration_seconds);
896            }
897            if let Some(tags) = &config.assume_role_session_tags {
898                assume_role_provider = assume_role_provider
899                    .with_tags(tags.iter().map(|(k, v)| (k.clone(), v.clone())).collect());
900            }
901            provider = ProvideCredentialChain::new().push(assume_role_provider);
902        }
903
904        // Replace provider if user provide their own.
905        let provider = if let Some(credential_providers) = credential_providers {
906            credential_providers
907        } else {
908            provider
909        };
910
911        // Create request signer for S3
912        let request_signer = AwsV4Signer::new("s3", &region);
913
914        // Create the signer
915        let signer = Signer::new(ctx, provider, request_signer);
916
917        Ok(S3Backend {
918            core: Arc::new(S3Core {
919                info: ServiceInfo::new(S3_SCHEME, &root, bucket),
920                capability: Capability {
921                    stat: true,
922                    stat_with_if_match: true,
923                    stat_with_if_none_match: true,
924                    stat_with_if_modified_since: true,
925                    stat_with_if_unmodified_since: true,
926                    stat_with_override_cache_control: true,
927                    stat_with_override_content_disposition: true,
928                    stat_with_override_content_type: true,
929                    stat_with_version: true,
930
931                    read: true,
932                    read_with_if_match: true,
933                    read_with_if_none_match: true,
934                    read_with_if_modified_since: true,
935                    read_with_if_unmodified_since: true,
936                    read_with_override_cache_control: true,
937                    read_with_override_content_disposition: true,
938                    read_with_override_content_type: true,
939                    read_with_version: true,
940                    read_with_suffix: true,
941
942                    write: true,
943                    write_can_empty: true,
944                    write_can_multi: true,
945                    write_can_append: true,
946
947                    write_with_cache_control: true,
948                    write_with_content_type: true,
949                    write_with_content_disposition: true,
950                    write_with_content_encoding: true,
951                    write_with_if_match: true,
952                    write_with_if_not_exists: true,
953                    write_with_user_metadata: true,
954
955                    // The min multipart size of S3 is 5 MiB.
956                    //
957                    // ref: <https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html>
958                    write_multi_min_size: Some(5 * 1024 * 1024),
959                    // The max multipart size of S3 is 5 GiB.
960                    //
961                    // ref: <https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html>
962                    write_multi_max_size: if cfg!(target_pointer_width = "64") {
963                        Some(5 * 1024 * 1024 * 1024)
964                    } else {
965                        Some(usize::MAX)
966                    },
967                    // S3 allows at most 10,000 parts and 5 GiB for each part.
968                    //
969                    // ref: <https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html>
970                    write_total_max_size: if cfg!(target_pointer_width = "64") {
971                        Some(10_000 * 5 * 1024 * 1024 * 1024)
972                    } else {
973                        None
974                    },
975
976                    delete: true,
977                    delete_max_size: Some(DEFAULT_BATCH_MAX_OPERATIONS),
978                    delete_with_version: true,
979
980                    copy: true,
981                    copy_can_multi: true,
982                    copy_with_if_not_exists: true,
983                    copy_with_if_match: true,
984                    copy_with_source_version: true,
985                    // The min multipart size of S3 is 5 MiB.
986                    //
987                    // ref: <https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html>
988                    copy_multi_min_size: Some(5 * 1024 * 1024),
989                    // The max multipart size of S3 is 5 GiB.
990                    //
991                    // ref: <https://docs.aws.amazon.com/AmazonS3/latest/userguide/qfacts.html>
992                    copy_multi_max_size: if cfg!(target_pointer_width = "64") {
993                        Some(5 * 1024 * 1024 * 1024)
994                    } else {
995                        Some(usize::MAX)
996                    },
997
998                    list: true,
999                    list_with_limit: true,
1000                    list_with_start_after: true,
1001                    list_with_recursive: true,
1002                    list_with_versions: true,
1003                    list_with_deleted: true,
1004
1005                    presign: true,
1006                    presign_stat: true,
1007                    presign_read: true,
1008                    presign_write: true,
1009                    presign_delete: true,
1010
1011                    shared: true,
1012
1013                    ..Default::default()
1014                },
1015                bucket: bucket.to_string(),
1016                endpoint,
1017                root,
1018                server_side_encryption,
1019                server_side_encryption_aws_kms_key_id,
1020                server_side_encryption_customer_algorithm,
1021                server_side_encryption_customer_key,
1022                server_side_encryption_customer_key_md5,
1023                default_storage_class,
1024                skip_signature: config.skip_signature,
1025                disable_list_objects_v2: config.disable_list_objects_v2,
1026                enable_request_payer: config.enable_request_payer,
1027                signer,
1028                checksum_algorithm,
1029                default_acl: config.default_acl,
1030            }),
1031        })
1032    }
1033}
1034
1035/// Backend for s3 services.
1036#[derive(Debug, Clone)]
1037pub struct S3Backend {
1038    pub(crate) core: Arc<S3Core>,
1039}
1040
1041impl Service for S3Backend {
1042    type Reader = oio::StreamReader<S3Reader>;
1043    type Writer = S3Writers;
1044    type Lister = S3Listers;
1045    type Deleter = oio::BatchDeleter<S3Deleter>;
1046    type Copier = S3Copiers;
1047
1048    fn info(&self) -> ServiceInfo {
1049        self.core.info.clone()
1050    }
1051
1052    fn capability(&self) -> Capability {
1053        self.core.capability
1054    }
1055
1056    async fn create_dir(
1057        &self,
1058        _ctx: &OperationContext,
1059        _path: &str,
1060        _args: OpCreateDir,
1061    ) -> Result<RpCreateDir> {
1062        Err(Error::new(
1063            ErrorKind::Unsupported,
1064            "operation is not supported",
1065        ))
1066    }
1067
1068    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
1069        let resp = self.core.s3_head_object(ctx, path, args).await?;
1070
1071        let status = resp.status();
1072
1073        match status {
1074            StatusCode::OK => {
1075                let headers = resp.headers();
1076                let mut meta = parse_into_metadata(path, headers)?;
1077
1078                let user_meta = parse_prefixed_headers(headers, X_AMZ_META_PREFIX);
1079                if !user_meta.is_empty() {
1080                    meta = meta.with_user_metadata(user_meta);
1081                }
1082
1083                if let Some(v) = parse_header_to_str(headers, X_AMZ_VERSION_ID)? {
1084                    meta.set_version(v);
1085                }
1086
1087                Ok(RpStat::new(meta))
1088            }
1089            _ => Err(parse_error(resp)),
1090        }
1091    }
1092    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
1093        let output: oio::StreamReader<S3Reader> = {
1094            Ok(oio::StreamReader::new(S3Reader::new(
1095                self.clone(),
1096                ctx.clone(),
1097                path,
1098                args,
1099            )))
1100        }?;
1101
1102        Ok(output)
1103    }
1104
1105    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
1106        let output: S3Writers = {
1107            let writer = S3Writer::new(self.core.clone(), ctx.clone(), path, args.clone());
1108
1109            let w = if args.append() {
1110                S3Writers::Two(oio::AppendWriter::new(writer))
1111            } else {
1112                // Multipart uploads schedule work through the operation
1113                // executor supplied by the caller.
1114                S3Writers::One(oio::MultipartWriter::new(
1115                    ctx.executor().clone(),
1116                    writer,
1117                    args.concurrent(),
1118                ))
1119            };
1120
1121            Ok(w)
1122        }?;
1123
1124        Ok(output)
1125    }
1126
1127    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
1128        let output: oio::BatchDeleter<S3Deleter> = {
1129            Ok(oio::BatchDeleter::new(
1130                S3Deleter::new(self.core.clone(), ctx.clone()),
1131                self.core.capability.delete_max_size,
1132            ))
1133        }?;
1134
1135        Ok(output)
1136    }
1137
1138    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
1139        let output: S3Listers = {
1140            let l = if args.versions() || args.deleted() {
1141                ThreeWays::Three(oio::PageLister::new(S3ObjectVersionsLister::new(
1142                    self.core.clone(),
1143                    ctx.clone(),
1144                    path,
1145                    args,
1146                )))
1147            } else if self.core.disable_list_objects_v2 {
1148                ThreeWays::One(oio::PageLister::new(S3ListerV1::new(
1149                    self.core.clone(),
1150                    ctx.clone(),
1151                    path,
1152                    args,
1153                )))
1154            } else {
1155                ThreeWays::Two(oio::PageLister::new(S3ListerV2::new(
1156                    self.core.clone(),
1157                    ctx.clone(),
1158                    path,
1159                    args,
1160                )))
1161            };
1162
1163            Ok(l)
1164        }?;
1165
1166        Ok(output)
1167    }
1168
1169    fn copy(
1170        &self,
1171        ctx: &OperationContext,
1172        from: &str,
1173        to: &str,
1174        args: OpCopy,
1175        opts: OpCopier,
1176    ) -> Result<Self::Copier> {
1177        let output: S3Copiers = {
1178            let copier = new_s3_copier(self.core.clone(), ctx, from, to, args, opts)?;
1179            Ok(copier)
1180        }?;
1181
1182        Ok(output)
1183    }
1184
1185    async fn rename(
1186        &self,
1187        _ctx: &OperationContext,
1188        _from: &str,
1189        _to: &str,
1190        _args: OpRename,
1191    ) -> Result<RpRename> {
1192        Err(Error::new(
1193            ErrorKind::Unsupported,
1194            "operation is not supported",
1195        ))
1196    }
1197
1198    async fn presign(
1199        &self,
1200        ctx: &OperationContext,
1201        path: &str,
1202        args: OpPresign,
1203    ) -> Result<RpPresign> {
1204        let (expire, op) = args.into_parts();
1205        // We will not send this request out, just for signing.
1206        let req = match op {
1207            PresignOperation::Stat(v) => self.core.s3_head_object_request(path, v),
1208            PresignOperation::Read(range, v) => self.core.s3_get_object_request(path, range, &v),
1209            PresignOperation::Write(v) => {
1210                self.core
1211                    .s3_put_object_request(path, None, &v, Buffer::new())
1212            }
1213            PresignOperation::Delete(v) => self.core.s3_delete_object_request(path, &v),
1214            _ => Err(Error::new(
1215                ErrorKind::Unsupported,
1216                "operation is not supported",
1217            )),
1218        };
1219        let req = req?;
1220
1221        let req = self.core.sign_query(ctx, req, expire).await?;
1222
1223        // We don't need this request anymore, consume it directly.
1224        let (parts, _) = req.into_parts();
1225
1226        Ok(RpPresign::new(PresignedRequest::new(
1227            parts.method,
1228            parts.uri,
1229            parts.headers,
1230        )))
1231    }
1232}
1233#[cfg(test)]
1234mod tests {
1235    use super::*;
1236
1237    #[test]
1238    fn test_is_valid_bucket() {
1239        let bucket_cases = vec![
1240            ("", false, false),
1241            ("test", false, true),
1242            ("test.xyz", false, true),
1243            ("", true, false),
1244            ("test", true, true),
1245            ("test.xyz", true, false),
1246        ];
1247
1248        for (bucket, enable_virtual_host_style, expected) in bucket_cases {
1249            let mut b = S3Builder::default();
1250            b = b.bucket(bucket);
1251            if enable_virtual_host_style {
1252                b = b.enable_virtual_host_style();
1253            }
1254            assert_eq!(S3Builder::is_bucket_valid(&b.config), expected)
1255        }
1256    }
1257
1258    #[test]
1259    fn test_build_endpoint() {
1260        let endpoint_cases = vec![
1261            Some("s3.amazonaws.com"),
1262            Some("https://s3.amazonaws.com"),
1263            Some("https://s3.us-east-2.amazonaws.com"),
1264            None,
1265        ];
1266
1267        for endpoint in &endpoint_cases {
1268            let mut b = S3Builder::default().bucket("test");
1269            if let Some(endpoint) = endpoint {
1270                b = b.endpoint(endpoint);
1271            }
1272
1273            let endpoint = S3Builder::build_endpoint(&b.config, "us-east-2");
1274            assert_eq!(endpoint, "https://s3.us-east-2.amazonaws.com/test");
1275        }
1276
1277        for endpoint in &endpoint_cases {
1278            let mut b = S3Builder::default()
1279                .bucket("test")
1280                .enable_virtual_host_style();
1281            if let Some(endpoint) = endpoint {
1282                b = b.endpoint(endpoint);
1283            }
1284
1285            let endpoint = S3Builder::build_endpoint(&b.config, "us-east-2");
1286            assert_eq!(endpoint, "https://test.s3.us-east-2.amazonaws.com");
1287        }
1288    }
1289
1290    #[tokio::test]
1291    async fn test_detect_region() {
1292        let cases = vec![
1293            (
1294                "aws s3 without region in endpoint",
1295                "https://s3.amazonaws.com",
1296                "example",
1297                Some("us-east-1"),
1298            ),
1299            (
1300                "aws s3 with region in endpoint",
1301                "https://s3.us-east-1.amazonaws.com",
1302                "example",
1303                Some("us-east-1"),
1304            ),
1305            (
1306                "oss with public endpoint",
1307                "https://oss-ap-southeast-1.aliyuncs.com",
1308                "example",
1309                Some("oss-ap-southeast-1"),
1310            ),
1311            (
1312                "oss with internal endpoint",
1313                "https://oss-cn-hangzhou-internal.aliyuncs.com",
1314                "example",
1315                Some("oss-cn-hangzhou"),
1316            ),
1317            (
1318                "r2",
1319                "https://abc.xxxxx.r2.cloudflarestorage.com",
1320                "example",
1321                Some("auto"),
1322            ),
1323            (
1324                "invalid service",
1325                "https://opendal.apache.org",
1326                "example",
1327                None,
1328            ),
1329        ];
1330
1331        for (name, endpoint, bucket, expected) in cases {
1332            let region = S3Builder::detect_region(endpoint, bucket).await;
1333            assert_eq!(region.as_deref(), expected, "{name}");
1334        }
1335    }
1336
1337    #[tokio::test]
1338    async fn test_presign_write_preserves_content_type() {
1339        let backend = S3Builder::default()
1340            .bucket("test")
1341            .region("us-east-1")
1342            .skip_signature()
1343            .disable_config_load()
1344            .disable_ec2_metadata()
1345            .build()
1346            .expect("build");
1347
1348        let op = OpWrite::default().with_content_type("application/json");
1349        let args = OpPresign::new(op, Duration::from_secs(3600));
1350        let ctx = OperationContext::new();
1351        let presigned = backend
1352            .presign(&ctx, "test.txt", args)
1353            .await
1354            .expect("presign")
1355            .into_presigned_request();
1356
1357        assert_eq!(
1358            presigned.header().get(http::header::CONTENT_TYPE).unwrap(),
1359            "application/json"
1360        );
1361    }
1362
1363    #[tokio::test]
1364    async fn test_presign_stat_encodes_version_id() {
1365        let backend = S3Builder::default()
1366            .bucket("test")
1367            .region("us-east-1")
1368            .skip_signature()
1369            .disable_config_load()
1370            .disable_ec2_metadata()
1371            .build()
1372            .expect("build");
1373
1374        let op = OpStat::default().with_version("a+b/c=d%25&e");
1375        let args = OpPresign::new(op, Duration::from_secs(3600));
1376        let ctx = OperationContext::new();
1377        let presigned = backend
1378            .presign(&ctx, "test.txt", args)
1379            .await
1380            .expect("presign")
1381            .into_presigned_request();
1382
1383        assert_eq!(
1384            presigned.uri().to_string(),
1385            "https://s3.us-east-1.amazonaws.com/test/test.txt?versionId=a%2Bb/c%3Dd%2525%26e"
1386        );
1387    }
1388
1389    #[tokio::test]
1390    async fn test_presign_read_encodes_version_id() {
1391        let backend = S3Builder::default()
1392            .bucket("test")
1393            .region("us-east-1")
1394            .skip_signature()
1395            .disable_config_load()
1396            .disable_ec2_metadata()
1397            .build()
1398            .expect("build");
1399
1400        let op = OpRead::default().with_version("a+b/c=d%25&e");
1401        let args = OpPresign::new(
1402            PresignOperation::Read(BytesRange::default(), op),
1403            Duration::from_secs(3600),
1404        );
1405        let ctx = OperationContext::new();
1406        let presigned = backend
1407            .presign(&ctx, "test.txt", args)
1408            .await
1409            .expect("presign")
1410            .into_presigned_request();
1411
1412        assert_eq!(
1413            presigned.uri().to_string(),
1414            "https://s3.us-east-1.amazonaws.com/test/test.txt?versionId=a%2Bb/c%3Dd%2525%26e"
1415        );
1416    }
1417}
opendal_service_s3/backend.rs

opendal_service_s3/
backend.rs
