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

opendal_service_s3/
backend.rs
