object_store/aws/
mod.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
18//! An object store implementation for S3
19//!
20//! See the [Feature Flags](crate#feature-flags) section for the difference
21//! between the `aws` and `aws-base` features.
22//!
23//! ## Multipart uploads
24//!
25//! Multipart uploads can be initiated with the [`ObjectStore::put_multipart_opts`] method.
26//!
27//! If the writer fails for any reason, you may have parts uploaded to AWS but not
28//! used that you will be charged for. [`MultipartUpload::abort`] may be invoked to drop
29//! these unneeded parts, however, it is recommended that you consider implementing
30//! [automatic cleanup] of unused parts that are older than some threshold.
31//!
32//! [automatic cleanup]: https://aws.amazon.com/blogs/aws/s3-lifecycle-management-update-support-for-multipart-uploads-and-delete-markers/
33
34use async_trait::async_trait;
35use futures_util::stream::BoxStream;
36use futures_util::{StreamExt, TryStreamExt};
37use http::header::{HeaderName, IF_MATCH, IF_NONE_MATCH};
38use http::{Method, StatusCode};
39use std::{sync::Arc, time::Duration};
40use url::Url;
41
42use crate::aws::client::{CompleteMultipartMode, PutPartPayload, RequestError, S3Client};
43use crate::client::CredentialProvider;
44use crate::client::get::GetClientExt;
45use crate::client::list::{ListClient, ListClientExt};
46use crate::multipart::{MultipartStore, PartId};
47use crate::signer::Signer;
48use crate::util::STRICT_ENCODE_SET;
49use crate::{
50    CopyMode, CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartId, MultipartUpload,
51    ObjectMeta, ObjectStore, Path, PutMode, PutMultipartOptions, PutOptions, PutPayload, PutResult,
52    Result, UploadPart,
53};
54
55static TAGS_HEADER: HeaderName = HeaderName::from_static("x-amz-tagging");
56static COPY_SOURCE_HEADER: HeaderName = HeaderName::from_static("x-amz-copy-source");
57
58mod builder;
59mod checksum;
60mod client;
61mod credential;
62mod precondition;
63
64#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
65mod resolve;
66
67pub use builder::{AmazonS3Builder, AmazonS3ConfigKey};
68pub use checksum::Checksum;
69pub use precondition::{S3ConditionalPut, S3CopyIfNotExists};
70
71#[cfg(all(feature = "reqwest", not(target_arch = "wasm32")))]
72pub use resolve::resolve_bucket_region;
73
74/// This struct is used to maintain the URI path encoding
75const STRICT_PATH_ENCODE_SET: percent_encoding::AsciiSet = STRICT_ENCODE_SET.remove(b'/');
76
77const STORE: &str = "S3";
78
79/// [`CredentialProvider`] for [`AmazonS3`]
80pub type AwsCredentialProvider = Arc<dyn CredentialProvider<Credential = AwsCredential>>;
81use crate::client::parts::Parts;
82use crate::list::{PaginatedListOptions, PaginatedListResult, PaginatedListStore};
83pub use credential::{AwsAuthorizer, AwsCredential};
84
85/// Interface for [Amazon S3](https://aws.amazon.com/s3/).
86#[derive(Debug, Clone)]
87pub struct AmazonS3 {
88    client: Arc<S3Client>,
89}
90
91impl std::fmt::Display for AmazonS3 {
92    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
93        write!(f, "AmazonS3({})", self.client.config.bucket)
94    }
95}
96
97impl AmazonS3 {
98    /// Returns the [`AwsCredentialProvider`] used by [`AmazonS3`]
99    pub fn credentials(&self) -> &AwsCredentialProvider {
100        &self.client.config.credentials
101    }
102
103    /// Create a full URL to the resource specified by `path` with this instance's configuration.
104    fn path_url(&self, path: &Path) -> String {
105        self.client.config.path_url(path)
106    }
107}
108
109#[async_trait]
110impl Signer for AmazonS3 {
111    /// Create a URL containing the relevant [AWS SigV4] query parameters that authorize a request
112    /// via `method` to the resource at `path` valid for the duration specified in `expires_in`.
113    ///
114    /// [AWS SigV4]: https://docs.aws.amazon.com/IAM/latest/UserGuide/create-signed-request.html
115    ///
116    /// # Example
117    ///
118    /// This example returns a URL that will enable a user to upload a file to
119    /// "some-folder/some-file.txt" in the next hour.
120    ///
121    /// ```
122    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
123    /// # use object_store::{aws::AmazonS3Builder, path::Path, signer::Signer};
124    /// # use http::Method;
125    /// # use std::time::Duration;
126    /// #
127    /// let region = "us-east-1";
128    /// let s3 = AmazonS3Builder::new()
129    ///     .with_region(region)
130    ///     .with_bucket_name("my-bucket")
131    ///     .with_access_key_id("my-access-key-id")
132    ///     .with_secret_access_key("my-secret-access-key")
133    ///     .build()?;
134    ///
135    /// let url = s3.signed_url(
136    ///     Method::PUT,
137    ///     &Path::from("some-folder/some-file.txt"),
138    ///     Duration::from_secs(60 * 60)
139    /// ).await?;
140    /// #     Ok(())
141    /// # }
142    /// ```
143    async fn signed_url(&self, method: Method, path: &Path, expires_in: Duration) -> Result<Url> {
144        let crypto = self.client.config.crypto()?;
145        let credential = self.credentials().get_credential().await?;
146        let authorizer = AwsAuthorizer::new(&credential, "s3", &self.client.config.region)
147            .with_request_payer(self.client.config.request_payer)
148            .with_crypto(crypto);
149
150        let path_url = self.path_url(path);
151        let mut url = path_url.parse().map_err(|e| Error::Generic {
152            store: STORE,
153            source: format!("Unable to parse url {path_url}: {e}").into(),
154        })?;
155
156        authorizer.sign(method, &mut url, expires_in)?;
157
158        Ok(url)
159    }
160}
161
162#[async_trait]
163impl ObjectStore for AmazonS3 {
164    async fn put_opts(
165        &self,
166        location: &Path,
167        payload: PutPayload,
168        opts: PutOptions,
169    ) -> Result<PutResult> {
170        let PutOptions {
171            mode,
172            tags,
173            attributes,
174            extensions,
175        } = opts;
176
177        let request = self
178            .client
179            .request(Method::PUT, location)
180            .with_payload(payload)?
181            .with_attributes(attributes)
182            .with_tags(tags)
183            .with_extensions(extensions)
184            .with_encryption_headers();
185
186        match (mode, &self.client.config.conditional_put) {
187            (PutMode::Overwrite, _) => request.idempotent(true).do_put().await,
188            (PutMode::Create, S3ConditionalPut::Disabled) => Err(Error::NotImplemented {
189                operation:
190                    "`put_opts` with mode `PutMode::Create` when conditional put is disabled".into(),
191                implementer: self.to_string(),
192            }),
193            (PutMode::Create, S3ConditionalPut::ETagMatch) => {
194                match request.header(&IF_NONE_MATCH, "*").do_put().await {
195                    // Technically If-None-Match should return NotModified but some stores,
196                    // such as R2, instead return PreconditionFailed
197                    // https://developers.cloudflare.com/r2/api/s3/extensions/#conditional-operations-in-putobject
198                    Err(e @ Error::NotModified { .. } | e @ Error::Precondition { .. }) => {
199                        Err(Error::AlreadyExists {
200                            path: location.to_string(),
201                            source: Box::new(e),
202                        })
203                    }
204                    r => r,
205                }
206            }
207            (PutMode::Update(v), put) => {
208                let etag = v.e_tag.ok_or_else(|| Error::Generic {
209                    store: STORE,
210                    source: "ETag required for conditional put".to_string().into(),
211                })?;
212                match put {
213                    S3ConditionalPut::ETagMatch => {
214                        match request
215                            .header(&IF_MATCH, etag.as_str())
216                            // Real S3 will occasionally report 409 Conflict
217                            // if there are concurrent `If-Match` requests
218                            // in flight, so we need to be prepared to retry
219                            // 409 responses.
220                            .retry_on_conflict(true)
221                            .do_put()
222                            .await
223                        {
224                            // Real S3 reports NotFound rather than PreconditionFailed when the
225                            // object doesn't exist. Convert to PreconditionFailed for
226                            // consistency with R2. This also matches what the HTTP spec
227                            // says the behavior should be.
228                            Err(Error::NotFound { path, source }) => {
229                                Err(Error::Precondition { path, source })
230                            }
231                            r => r,
232                        }
233                    }
234                    S3ConditionalPut::Disabled => Err(Error::NotImplemented {
235                        operation:
236                            "`put_opts` with mode `PutMode::Update` when conditional put is disabled"
237                                .into(),
238                        implementer: self.to_string(),
239                    }),
240                }
241            }
242        }
243    }
244
245    async fn put_multipart_opts(
246        &self,
247        location: &Path,
248        opts: PutMultipartOptions,
249    ) -> Result<Box<dyn MultipartUpload>> {
250        let upload_id = self.client.create_multipart(location, opts).await?;
251
252        Ok(Box::new(S3MultiPartUpload {
253            part_idx: 0,
254            state: Arc::new(UploadState {
255                client: Arc::clone(&self.client),
256                location: location.clone(),
257                upload_id: upload_id.clone(),
258                parts: Default::default(),
259            }),
260        }))
261    }
262
263    async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
264        self.client.get_opts(location, options).await
265    }
266
267    fn delete_stream(
268        &self,
269        locations: BoxStream<'static, Result<Path>>,
270    ) -> BoxStream<'static, Result<Path>> {
271        let client = Arc::clone(&self.client);
272
273        // Some S3-compatible providers do not implement
274        // the bulk `DeleteObjects` API (`POST /?delete`). When bulk delete is
275        // disabled, fall back to parallel single-object `DELETE /key` requests,
276        // which are part of the core S3 API supported by every provider.
277        if client.config.disable_bulk_delete {
278            return locations
279                .map(move |location| {
280                    let client = Arc::clone(&client);
281                    async move {
282                        let location = location?;
283                        client.delete_request(&location).await?;
284                        Ok(location)
285                    }
286                })
287                .buffered(20)
288                .boxed();
289        }
290
291        locations
292            .try_chunks(1_000)
293            .map(move |locations| {
294                let client = Arc::clone(&client);
295                async move {
296                    // Early return the error. We ignore the paths that have already been
297                    // collected into the chunk.
298                    let locations = locations.map_err(|e| e.1)?;
299                    client
300                        .bulk_delete_request(locations)
301                        .await
302                        .map(futures_util::stream::iter)
303                }
304            })
305            .buffered(20)
306            .try_flatten()
307            .boxed()
308    }
309
310    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, Result<ObjectMeta>> {
311        self.client.list(prefix)
312    }
313
314    fn list_with_offset(
315        &self,
316        prefix: Option<&Path>,
317        offset: &Path,
318    ) -> BoxStream<'static, Result<ObjectMeta>> {
319        if self.client.config.is_s3_express() {
320            let offset = offset.clone();
321            // S3 Express does not support start-after
322            return self
323                .client
324                .list(prefix)
325                .try_filter(move |f| futures_util::future::ready(f.location > offset))
326                .boxed();
327        }
328
329        self.client.list_with_offset(prefix, offset)
330    }
331
332    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
333        self.client.list_with_delimiter(prefix).await
334    }
335
336    async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> Result<()> {
337        let CopyOptions {
338            mode,
339            extensions: _,
340        } = options;
341
342        match mode {
343            CopyMode::Overwrite => {
344                self.client
345                    .copy_request(from, to)
346                    .idempotent(true)
347                    .send()
348                    .await?;
349                Ok(())
350            }
351            CopyMode::Create => {
352                let (k, v, status) = match &self.client.config.copy_if_not_exists {
353                    Some(S3CopyIfNotExists::Header(k, v)) => {
354                        (k, v, StatusCode::PRECONDITION_FAILED)
355                    }
356                    Some(S3CopyIfNotExists::HeaderWithStatus(k, v, status)) => (k, v, *status),
357                    Some(S3CopyIfNotExists::Multipart) => {
358                        let upload_id = self
359                            .client
360                            .create_multipart(to, PutMultipartOptions::default())
361                            .await?;
362
363                        let res = async {
364                            let part_id = self
365                                .client
366                                .put_part(to, &upload_id, 0, PutPartPayload::Copy(from))
367                                .await?;
368                            match self
369                                .client
370                                .complete_multipart(
371                                    to,
372                                    &upload_id,
373                                    vec![part_id],
374                                    CompleteMultipartMode::Create,
375                                )
376                                .await
377                            {
378                                Err(e @ Error::Precondition { .. }) => Err(Error::AlreadyExists {
379                                    path: to.to_string(),
380                                    source: Box::new(e),
381                                }),
382                                Ok(_) => Ok(()),
383                                Err(e) => Err(e),
384                            }
385                        }
386                        .await;
387
388                        // If the multipart upload failed, make a best effort attempt to
389                        // clean it up. It's the caller's responsibility to add a
390                        // lifecycle rule if guaranteed cleanup is required, as we
391                        // cannot protect against an ill-timed process crash.
392                        if res.is_err() {
393                            let _ = self.client.abort_multipart(to, &upload_id).await;
394                        }
395
396                        return res;
397                    }
398                    None => {
399                        return Err(Error::NotSupported {
400                            source: "S3 does not support copy-if-not-exists".to_string().into(),
401                        });
402                    }
403                };
404
405                let req = self.client.copy_request(from, to);
406                match req.header(k, v).send().await {
407                    Err(RequestError::Retry { source, path })
408                        if source.status() == Some(status) =>
409                    {
410                        Err(Error::AlreadyExists {
411                            source: Box::new(source),
412                            path,
413                        })
414                    }
415                    Err(e) => Err(e.into()),
416                    Ok(_) => Ok(()),
417                }
418            }
419        }
420    }
421}
422
423#[derive(Debug)]
424struct S3MultiPartUpload {
425    part_idx: usize,
426    state: Arc<UploadState>,
427}
428
429#[derive(Debug)]
430struct UploadState {
431    parts: Parts,
432    location: Path,
433    upload_id: String,
434    client: Arc<S3Client>,
435}
436
437#[async_trait]
438impl MultipartUpload for S3MultiPartUpload {
439    fn put_part(&mut self, data: PutPayload) -> UploadPart {
440        let idx = self.part_idx;
441        self.part_idx += 1;
442        let state = Arc::clone(&self.state);
443        Box::pin(async move {
444            let part = state
445                .client
446                .put_part(
447                    &state.location,
448                    &state.upload_id,
449                    idx,
450                    PutPartPayload::Part(data),
451                )
452                .await?;
453            state.parts.put(idx, part);
454            Ok(())
455        })
456    }
457
458    async fn complete(&mut self) -> Result<PutResult> {
459        let parts = self.state.parts.finish(self.part_idx)?;
460
461        self.state
462            .client
463            .complete_multipart(
464                &self.state.location,
465                &self.state.upload_id,
466                parts,
467                CompleteMultipartMode::Overwrite,
468            )
469            .await
470    }
471
472    async fn abort(&mut self) -> Result<()> {
473        self.state
474            .client
475            .request(Method::DELETE, &self.state.location)
476            .query(&[("uploadId", &self.state.upload_id)])
477            .idempotent(true)
478            .send()
479            .await?;
480
481        Ok(())
482    }
483}
484
485#[async_trait]
486impl MultipartStore for AmazonS3 {
487    async fn create_multipart(&self, path: &Path) -> Result<MultipartId> {
488        self.client
489            .create_multipart(path, PutMultipartOptions::default())
490            .await
491    }
492
493    async fn create_multipart_opts(
494        &self,
495        path: &Path,
496        opts: PutMultipartOptions,
497    ) -> Result<MultipartId> {
498        self.client.create_multipart(path, opts).await
499    }
500
501    async fn put_part(
502        &self,
503        path: &Path,
504        id: &MultipartId,
505        part_idx: usize,
506        data: PutPayload,
507    ) -> Result<PartId> {
508        self.client
509            .put_part(path, id, part_idx, PutPartPayload::Part(data))
510            .await
511    }
512
513    async fn complete_multipart(
514        &self,
515        path: &Path,
516        id: &MultipartId,
517        parts: Vec<PartId>,
518    ) -> Result<PutResult> {
519        self.client
520            .complete_multipart(path, id, parts, CompleteMultipartMode::Overwrite)
521            .await
522    }
523
524    async fn abort_multipart(&self, path: &Path, id: &MultipartId) -> Result<()> {
525        self.client
526            .request(Method::DELETE, path)
527            .query(&[("uploadId", id)])
528            .send()
529            .await?;
530        Ok(())
531    }
532}
533
534#[async_trait]
535impl PaginatedListStore for AmazonS3 {
536    async fn list_paginated(
537        &self,
538        prefix: Option<&str>,
539        opts: PaginatedListOptions,
540    ) -> Result<PaginatedListResult> {
541        self.client.list_request(prefix, opts).await
542    }
543}
544
545#[cfg(test)]
546mod tests {
547    use super::*;
548    use crate::ClientOptions;
549    use crate::ObjectStoreExt;
550    #[cfg(feature = "reqwest")]
551    use crate::client::SpawnedReqwestConnector;
552    use crate::client::get::GetClient;
553    use crate::client::retry::RetryContext;
554    use crate::integration::*;
555    use crate::tests::*;
556    use base64::Engine;
557    use base64::prelude::BASE64_STANDARD;
558    use http::HeaderMap;
559
560    const NON_EXISTENT_NAME: &str = "nonexistentname";
561
562    #[tokio::test]
563    async fn write_multipart_file_with_signature() {
564        maybe_skip_integration!();
565
566        let bucket = "test-bucket-for-checksum";
567        for checksum in [Checksum::SHA256, Checksum::CRC64NVME] {
568            let store = AmazonS3Builder::from_env()
569                .with_bucket_name(bucket)
570                .with_checksum_algorithm(checksum)
571                .build()
572                .unwrap();
573
574            let str = "test.bin";
575            let path = Path::parse(str).unwrap();
576            let opts = PutMultipartOptions::default();
577            let mut upload = store.put_multipart_opts(&path, opts).await.unwrap();
578
579            upload
580                .put_part(PutPayload::from(vec![0u8; 10_000_000]))
581                .await
582                .unwrap();
583            upload
584                .put_part(PutPayload::from(vec![0u8; 5_000_000]))
585                .await
586                .unwrap();
587
588            let res = upload.complete().await.unwrap();
589            assert!(res.e_tag.is_some(), "Should have valid etag");
590
591            store.delete(&path).await.unwrap();
592        }
593    }
594
595    #[tokio::test]
596    async fn copy_multipart_file_with_signature() {
597        maybe_skip_integration!();
598
599        let bucket = "test-bucket-for-copy-if-not-exists";
600        for checksum in [Checksum::SHA256, Checksum::CRC64NVME] {
601            let store = AmazonS3Builder::from_env()
602                .with_bucket_name(bucket)
603                .with_checksum_algorithm(checksum)
604                .with_copy_if_not_exists(S3CopyIfNotExists::Multipart)
605                .build()
606                .unwrap();
607
608            let src = Path::parse("src.bin").unwrap();
609            let dst = Path::parse("dst.bin").unwrap();
610            store
611                .put(&src, PutPayload::from(vec![0u8; 100_000]))
612                .await
613                .unwrap();
614            if store.head(&dst).await.is_ok() {
615                store.delete(&dst).await.unwrap();
616            }
617            store.copy_if_not_exists(&src, &dst).await.unwrap();
618            store.delete(&src).await.unwrap();
619            store.delete(&dst).await.unwrap();
620        }
621    }
622
623    #[tokio::test]
624    async fn copy_multipart_file_with_signature_change_checksum() {
625        maybe_skip_integration!();
626
627        let bucket = "test-bucket-for-copy-if-not-exists";
628        let checksum_src = Checksum::SHA256;
629        let checksum_dst = Checksum::CRC64NVME;
630
631        let src = Path::parse("change_checksum_src.bin").unwrap();
632        let dst = Path::parse("change_checksum_dst.bin").unwrap();
633
634        let store = AmazonS3Builder::from_env()
635            .with_bucket_name(bucket)
636            .with_checksum_algorithm(checksum_src)
637            .build()
638            .unwrap();
639
640        store
641            .put(&src, PutPayload::from(vec![0u8; 100_000]))
642            .await
643            .unwrap();
644        if store.head(&dst).await.is_ok() {
645            store.delete(&dst).await.unwrap();
646        }
647
648        let store = AmazonS3Builder::from_env()
649            .with_bucket_name(bucket)
650            .with_checksum_algorithm(checksum_dst)
651            .with_copy_if_not_exists(S3CopyIfNotExists::Multipart)
652            .build()
653            .unwrap();
654
655        store.copy_if_not_exists(&src, &dst).await.unwrap();
656        store.delete(&src).await.unwrap();
657        store.delete(&dst).await.unwrap();
658    }
659
660    #[tokio::test]
661    async fn write_multipart_file_with_signature_object_lock() {
662        maybe_skip_integration!();
663
664        for checksum in [Checksum::SHA256, Checksum::CRC64NVME] {
665            let bucket = "test-object-lock";
666            let store = AmazonS3Builder::from_env()
667                .with_bucket_name(bucket)
668                .with_checksum_algorithm(checksum)
669                .build()
670                .unwrap();
671
672            let str = "test.bin";
673            let path = Path::parse(str).unwrap();
674            let opts = PutMultipartOptions::default();
675            let mut upload = store.put_multipart_opts(&path, opts).await.unwrap();
676
677            upload
678                .put_part(PutPayload::from(vec![0u8; 10_000_000]))
679                .await
680                .unwrap();
681            upload
682                .put_part(PutPayload::from(vec![0u8; 5_000_000]))
683                .await
684                .unwrap();
685
686            let res = upload.complete().await.unwrap();
687            assert!(res.e_tag.is_some(), "Should have valid etag");
688
689            store.delete(&path).await.unwrap();
690        }
691    }
692
693    #[tokio::test]
694    async fn s3_test() {
695        maybe_skip_integration!();
696        // tag the extensions of every HTTP response with a marker,
697        // allowing response_extensions to verify their propagation
698        let config =
699            AmazonS3Builder::from_env().with_http_connector(MarkerHttpConnector::default());
700
701        let integration = config.build().unwrap();
702        let config = &integration.client.config;
703        let test_not_exists = config.copy_if_not_exists.is_some();
704        let test_conditional_put = config.conditional_put != S3ConditionalPut::Disabled;
705
706        put_get_delete_list(&integration).await;
707        list_with_offset_exclusivity(&integration).await;
708        get_opts(&integration).await;
709        list_uses_directories_correctly(&integration).await;
710        list_with_delimiter(&integration).await;
711        rename_and_copy(&integration).await;
712        stream_get(&integration).await;
713        multipart(&integration, &integration).await;
714        multipart_with_opts(&integration, &integration).await;
715        multipart_put_part_out_of_order(&integration, &integration).await;
716        multipart_race_condition(&integration, true).await;
717        multipart_out_of_order(&integration).await;
718        signing(&integration).await;
719        s3_encryption(&integration).await;
720        put_get_attributes(&integration).await;
721        list_paginated(&integration, &integration).await;
722        response_extensions(&integration, true).await;
723
724        // Object tagging is not supported by S3 Express One Zone
725        if config.session_provider.is_none() {
726            tagging(
727                Arc::new(AmazonS3 {
728                    client: Arc::clone(&integration.client),
729                }),
730                !config.disable_tagging,
731                |p| {
732                    let client = Arc::clone(&integration.client);
733                    async move { client.get_object_tagging(&p).await }
734                },
735            )
736            .await;
737        }
738
739        if test_not_exists {
740            copy_if_not_exists(&integration).await;
741        }
742        if test_conditional_put {
743            put_opts(&integration, true).await;
744        }
745
746        // run integration test with unsigned payload enabled
747        let builder = AmazonS3Builder::from_env().with_unsigned_payload(true);
748        let integration = builder.build().unwrap();
749        put_get_delete_list(&integration).await;
750
751        // run integration test with checksum set to sha256
752        let builder = AmazonS3Builder::from_env().with_checksum_algorithm(Checksum::SHA256);
753        let integration = builder.build().unwrap();
754        put_get_delete_list(&integration).await;
755
756        // run integration test with checksum set to crc64nvme
757        let builder = AmazonS3Builder::from_env().with_checksum_algorithm(Checksum::CRC64NVME);
758        let integration = builder.build().unwrap();
759        put_get_delete_list(&integration).await;
760    }
761
762    #[tokio::test]
763    async fn s3_test_get_nonexistent_location() {
764        maybe_skip_integration!();
765        let integration = AmazonS3Builder::from_env().build().unwrap();
766
767        let location = Path::from_iter([NON_EXISTENT_NAME]);
768
769        let err = get_nonexistent_object(&integration, Some(location))
770            .await
771            .unwrap_err();
772        assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
773    }
774
775    #[tokio::test]
776    async fn s3_test_get_nonexistent_bucket() {
777        maybe_skip_integration!();
778        let config = AmazonS3Builder::from_env().with_bucket_name(NON_EXISTENT_NAME);
779        let integration = config.build().unwrap();
780
781        let location = Path::from_iter([NON_EXISTENT_NAME]);
782
783        let err = integration.get(&location).await.unwrap_err();
784        assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
785    }
786
787    #[tokio::test]
788    async fn s3_test_put_nonexistent_bucket() {
789        maybe_skip_integration!();
790        let config = AmazonS3Builder::from_env().with_bucket_name(NON_EXISTENT_NAME);
791        let integration = config.build().unwrap();
792
793        let location = Path::from_iter([NON_EXISTENT_NAME]);
794        let data = PutPayload::from("arbitrary data");
795
796        let err = integration.put(&location, data).await.unwrap_err();
797        assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
798    }
799
800    #[tokio::test]
801    async fn s3_test_delete_nonexistent_location() {
802        maybe_skip_integration!();
803        let integration = AmazonS3Builder::from_env().build().unwrap();
804
805        let location = Path::from_iter([NON_EXISTENT_NAME]);
806
807        integration.delete(&location).await.unwrap();
808    }
809
810    #[tokio::test]
811    async fn s3_test_delete_nonexistent_bucket() {
812        maybe_skip_integration!();
813        let config = AmazonS3Builder::from_env().with_bucket_name(NON_EXISTENT_NAME);
814        let integration = config.build().unwrap();
815
816        let location = Path::from_iter([NON_EXISTENT_NAME]);
817
818        let err = integration.delete(&location).await.unwrap_err();
819        assert!(matches!(err, crate::Error::NotFound { .. }), "{}", err);
820    }
821
822    #[tokio::test]
823    #[ignore = "Tests shouldn't call use remote services by default"]
824    async fn test_disable_creds() {
825        // https://registry.opendata.aws/daylight-osm/
826        let v1 = AmazonS3Builder::new()
827            .with_bucket_name("daylight-map-distribution")
828            .with_region("us-west-1")
829            .with_access_key_id("local")
830            .with_secret_access_key("development")
831            .build()
832            .unwrap();
833
834        let prefix = Path::from("release");
835
836        v1.list_with_delimiter(Some(&prefix)).await.unwrap_err();
837
838        let v2 = AmazonS3Builder::new()
839            .with_bucket_name("daylight-map-distribution")
840            .with_region("us-west-1")
841            .with_skip_signature(true)
842            .build()
843            .unwrap();
844
845        v2.list_with_delimiter(Some(&prefix)).await.unwrap();
846    }
847
848    async fn s3_encryption(store: &AmazonS3) {
849        maybe_skip_integration!();
850
851        let data = PutPayload::from(vec![3u8; 1024]);
852
853        let encryption_headers: HeaderMap = store.client.config.encryption_headers.clone().into();
854        let expected_encryption =
855            if let Some(encryption_type) = encryption_headers.get("x-amz-server-side-encryption") {
856                encryption_type
857            } else {
858                eprintln!("Skipping S3 encryption test - encryption not configured");
859                return;
860            };
861
862        let locations = [
863            Path::from("test-encryption-1"),
864            Path::from("test-encryption-2"),
865            Path::from("test-encryption-3"),
866        ];
867
868        store.put(&locations[0], data.clone()).await.unwrap();
869        store.copy(&locations[0], &locations[1]).await.unwrap();
870
871        let mut upload = store.put_multipart(&locations[2]).await.unwrap();
872        upload.put_part(data.clone()).await.unwrap();
873        upload.complete().await.unwrap();
874
875        for location in &locations {
876            let mut context = RetryContext::new(&store.client.config.retry_config);
877
878            let res = store
879                .client
880                .get_request(&mut context, location, GetOptions::default())
881                .await
882                .unwrap();
883
884            let headers = res.headers();
885            assert_eq!(
886                headers
887                    .get("x-amz-server-side-encryption")
888                    .expect("object is not encrypted"),
889                expected_encryption
890            );
891
892            store.delete(location).await.unwrap();
893        }
894    }
895
896    /// See CONTRIBUTING.md for the MinIO setup for this test.
897    #[tokio::test]
898    async fn test_s3_ssec_encryption_with_minio() {
899        if std::env::var("TEST_S3_SSEC_ENCRYPTION").is_err() {
900            eprintln!("Skipping S3 SSE-C encryption test");
901            return;
902        }
903        eprintln!("Running S3 SSE-C encryption test");
904
905        let customer_key = "1234567890abcdef1234567890abcdef";
906        let expected_md5 = "JMwgiexXqwuPqIPjYFmIZQ==";
907
908        let store = AmazonS3Builder::from_env()
909            .with_ssec_encryption(BASE64_STANDARD.encode(customer_key))
910            .with_client_options(ClientOptions::default().with_allow_invalid_certificates(true))
911            .build()
912            .unwrap();
913
914        let data = PutPayload::from(vec![3u8; 1024]);
915
916        let locations = [
917            Path::from("test-encryption-1"),
918            Path::from("test-encryption-2"),
919            Path::from("test-encryption-3"),
920        ];
921
922        // Test put with sse-c.
923        store.put(&locations[0], data.clone()).await.unwrap();
924
925        // Test copy with sse-c.
926        store.copy(&locations[0], &locations[1]).await.unwrap();
927
928        // Test multipart upload with sse-c.
929        let mut upload = store.put_multipart(&locations[2]).await.unwrap();
930        upload.put_part(data.clone()).await.unwrap();
931        upload.complete().await.unwrap();
932
933        // Test get with sse-c.
934        for location in &locations {
935            let mut context = RetryContext::new(&store.client.config.retry_config);
936
937            let res = store
938                .client
939                .get_request(&mut context, location, GetOptions::default())
940                .await
941                .unwrap();
942
943            let headers = res.headers();
944            assert_eq!(
945                headers
946                    .get("x-amz-server-side-encryption-customer-algorithm")
947                    .expect("object is not encrypted with SSE-C"),
948                "AES256"
949            );
950
951            assert_eq!(
952                headers
953                    .get("x-amz-server-side-encryption-customer-key-MD5")
954                    .expect("object is not encrypted with SSE-C"),
955                expected_md5
956            );
957
958            store.delete(location).await.unwrap();
959        }
960    }
961
962    /// Integration test that ensures I/O is done on an alternate threadpool
963    /// when using the `SpawnedReqwestConnector`.
964    #[cfg(feature = "reqwest")]
965    #[test]
966    fn s3_alternate_threadpool_spawned_request_connector() {
967        maybe_skip_integration!();
968        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
969
970        // Runtime with I/O enabled
971        let io_runtime = tokio::runtime::Builder::new_current_thread()
972            .enable_all() // <-- turns on IO
973            .build()
974            .unwrap();
975
976        // Runtime without I/O enabled
977        let non_io_runtime = tokio::runtime::Builder::new_current_thread()
978            // note: no call to enable_all
979            .build()
980            .unwrap();
981
982        // run the io runtime in a different thread
983        let io_handle = io_runtime.handle().clone();
984        let thread_handle = std::thread::spawn(move || {
985            io_runtime.block_on(async move {
986                shutdown_rx.await.unwrap();
987            });
988        });
989
990        let store = AmazonS3Builder::from_env()
991            // use different bucket to avoid collisions with other tests
992            .with_bucket_name("test-bucket-for-spawn")
993            .with_http_connector(SpawnedReqwestConnector::new(io_handle))
994            .build()
995            .unwrap();
996
997        // run a request on the non io runtime -- will fail if the connector
998        // does not spawn the request to the io runtime
999        non_io_runtime
1000            .block_on(async move {
1001                let path = Path::from("alternate_threadpool/test.txt");
1002                store.delete(&path).await.ok(); // remove the file if it exists from prior runs
1003                store.put(&path, "foo".into()).await?;
1004                let res = store.get(&path).await?.bytes().await?;
1005                assert_eq!(res.as_ref(), b"foo");
1006                store.delete(&path).await?; // cleanup
1007                Ok(()) as Result<()>
1008            })
1009            .expect("failed to run request on non io runtime");
1010
1011        // shutdown the io runtime and thread
1012        shutdown_tx.send(()).ok();
1013        thread_handle.join().expect("runtime thread panicked");
1014    }
1015}
object_store/aws/mod.rs

object_store/aws/
mod.rs
