| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| |
| use crate::aws::builder::S3EncryptionHeaders; |
| use crate::aws::checksum::Checksum; |
| use crate::aws::credential::{AwsCredential, CredentialExt}; |
| use crate::aws::{ |
| AwsAuthorizer, AwsCredentialProvider, COPY_SOURCE_HEADER, S3ConditionalPut, S3CopyIfNotExists, |
| STORE, STRICT_PATH_ENCODE_SET, TAGS_HEADER, |
| }; |
| use crate::client::builder::{HttpRequestBuilder, RequestBuilderError}; |
| use crate::client::get::GetClient; |
| use crate::client::header::{HeaderConfig, get_etag}; |
| use crate::client::header::{get_put_result, get_version}; |
| use crate::client::list::ListClient; |
| use crate::client::retry::{RetryContext, RetryExt}; |
| use crate::client::s3::{ |
| CompleteMultipartUpload, CompleteMultipartUploadResult, CopyPartResult, |
| InitiateMultipartUploadResult, ListResponse, PartMetadata, |
| }; |
| use crate::client::{ |
| CryptoProvider, DigestAlgorithm, GetOptionsExt, HttpClient, HttpError, HttpResponse, |
| crypto_provider, |
| }; |
| use crate::list::{PaginatedListOptions, PaginatedListResult}; |
| use crate::multipart::PartId; |
| use crate::{ |
| Attribute, Attributes, ClientOptions, GetOptions, ListResult, MultipartId, Path, |
| PutMultipartOptions, PutPayload, PutResult, Result, RetryConfig, TagSet, |
| }; |
| use async_trait::async_trait; |
| use base64::Engine; |
| use base64::prelude::BASE64_STANDARD; |
| use bytes::{Buf, Bytes}; |
| use http::header::{ |
| CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_LENGTH, |
| CONTENT_TYPE, |
| }; |
| use http::{HeaderMap, HeaderName, Method}; |
| use itertools::Itertools; |
| use md5::{Digest, Md5}; |
| use percent_encoding::{PercentEncode, utf8_percent_encode}; |
| use quick_xml::events::{self as xml_events}; |
| use serde::{Deserialize, Serialize}; |
| use std::sync::Arc; |
| |
| const VERSION_HEADER: &str = "x-amz-version-id"; |
| const SHA256_CHECKSUM: &str = "x-amz-checksum-sha256"; |
| const CRC64NVME_CHECKSUM: &str = "x-amz-checksum-crc64nvme"; |
| const USER_DEFINED_METADATA_HEADER_PREFIX: &str = "x-amz-meta-"; |
| const ALGORITHM: &str = "x-amz-checksum-algorithm"; |
| const STORAGE_CLASS: &str = "x-amz-storage-class"; |
| |
| /// A specialized `Error` for object store-related errors |
| #[derive(Debug, thiserror::Error)] |
| pub(crate) enum Error { |
| #[error("Error performing DeleteObjects request: {}", source)] |
| DeleteObjectsRequest { |
| source: crate::client::retry::RetryError, |
| paths: Vec<String>, |
| }, |
| |
| #[error( |
| "DeleteObjects request failed for key {}: {} (code: {})", |
| path, |
| message, |
| code |
| )] |
| DeleteFailed { |
| path: String, |
| code: String, |
| message: String, |
| }, |
| |
| #[error("Error getting DeleteObjects response body: {}", source)] |
| DeleteObjectsResponse { source: HttpError }, |
| |
| #[error("Got invalid DeleteObjects response: {}", source)] |
| InvalidDeleteObjectsResponse { |
| source: Box<dyn std::error::Error + Send + Sync + 'static>, |
| }, |
| |
| #[error("Error performing list request: {}", source)] |
| ListRequest { |
| source: crate::client::retry::RetryError, |
| }, |
| |
| #[error("Error getting list response body: {}", source)] |
| ListResponseBody { source: HttpError }, |
| |
| #[error("Error getting create multipart response body: {}", source)] |
| CreateMultipartResponseBody { source: HttpError }, |
| |
| #[error("Error performing complete multipart request: {}: {}", path, source)] |
| CompleteMultipartRequest { |
| source: crate::client::retry::RetryError, |
| path: String, |
| }, |
| |
| #[error("Error getting complete multipart response body: {}", source)] |
| CompleteMultipartResponseBody { source: HttpError }, |
| |
| #[error("Got invalid list response: {}", source)] |
| InvalidListResponse { source: quick_xml::de::DeError }, |
| |
| #[error("Got invalid multipart response: {}", source)] |
| InvalidMultipartResponse { source: quick_xml::de::DeError }, |
| |
| #[error("Unable to extract metadata from headers: {}", source)] |
| Metadata { |
| source: crate::client::header::Error, |
| }, |
| } |
| |
| impl From<Error> for crate::Error { |
| fn from(err: Error) -> Self { |
| match err { |
| Error::CompleteMultipartRequest { source, path } => source.error(STORE, path), |
| Error::DeleteObjectsRequest { source, paths } => source.error(STORE, paths.join(",")), |
| _ => Self::Generic { |
| store: STORE, |
| source: Box::new(err), |
| }, |
| } |
| } |
| } |
| |
| pub(crate) enum PutPartPayload<'a> { |
| Part(PutPayload), |
| Copy(&'a Path), |
| } |
| |
| impl Default for PutPartPayload<'_> { |
| fn default() -> Self { |
| Self::Part(PutPayload::default()) |
| } |
| } |
| |
| pub(crate) enum CompleteMultipartMode { |
| Overwrite, |
| Create, |
| } |
| |
| #[derive(Deserialize)] |
| #[serde(rename_all = "PascalCase", rename = "DeleteResult")] |
| struct BatchDeleteResponse { |
| #[serde(rename = "$value")] |
| content: Vec<DeleteObjectResult>, |
| } |
| |
| #[derive(Deserialize)] |
| enum DeleteObjectResult { |
| #[allow(unused)] |
| Deleted(DeletedObject), |
| Error(DeleteError), |
| } |
| |
| #[derive(Deserialize)] |
| #[serde(rename_all = "PascalCase", rename = "Deleted")] |
| struct DeletedObject { |
| #[allow(dead_code)] |
| key: String, |
| } |
| |
| #[derive(Deserialize)] |
| #[serde(rename_all = "PascalCase", rename = "Error")] |
| struct DeleteError { |
| key: String, |
| code: String, |
| message: String, |
| } |
| |
| impl From<DeleteError> for Error { |
| fn from(err: DeleteError) -> Self { |
| Self::DeleteFailed { |
| path: err.key, |
| code: err.code, |
| message: err.message, |
| } |
| } |
| } |
| |
| #[derive(Debug)] |
| pub(crate) struct S3Config { |
| pub region: String, |
| pub bucket: String, |
| pub bucket_endpoint: String, |
| pub credentials: AwsCredentialProvider, |
| pub crypto: Option<Arc<dyn CryptoProvider>>, |
| pub session_provider: Option<AwsCredentialProvider>, |
| pub retry_config: RetryConfig, |
| pub client_options: ClientOptions, |
| pub sign_payload: bool, |
| pub skip_signature: bool, |
| pub disable_tagging: bool, |
| pub disable_bulk_delete: bool, |
| pub checksum: Option<Checksum>, |
| pub copy_if_not_exists: Option<S3CopyIfNotExists>, |
| pub conditional_put: S3ConditionalPut, |
| pub request_payer: bool, |
| pub(super) encryption_headers: S3EncryptionHeaders, |
| } |
| |
| impl S3Config { |
| pub(crate) fn path_url(&self, path: &Path) -> String { |
| format!("{}/{}", self.bucket_endpoint, encode_path(path)) |
| } |
| |
| async fn get_session_credential(&self) -> Result<Option<SessionCredential<'_>>> { |
| Ok(match self.skip_signature { |
| false => { |
| let provider = self.session_provider.as_ref().unwrap_or(&self.credentials); |
| let credential = provider.get_credential().await?; |
| Some(SessionCredential { |
| credential, |
| session_token: self.session_provider.is_some(), |
| config: self, |
| }) |
| } |
| true => None, |
| }) |
| } |
| |
| pub(crate) async fn get_credential(&self) -> Result<Option<Arc<AwsCredential>>> { |
| Ok(match self.skip_signature { |
| false => Some(self.credentials.get_credential().await?), |
| true => None, |
| }) |
| } |
| |
| #[inline] |
| pub(crate) fn is_s3_express(&self) -> bool { |
| self.session_provider.is_some() |
| } |
| |
| pub(crate) fn crypto(&self) -> Result<&dyn CryptoProvider> { |
| crypto_provider(self.crypto.as_deref()) |
| } |
| } |
| |
| struct SessionCredential<'a> { |
| credential: Arc<AwsCredential>, |
| session_token: bool, |
| config: &'a S3Config, |
| } |
| |
| impl SessionCredential<'_> { |
| fn authorizer(&self) -> Result<AwsAuthorizer<'_>> { |
| let mut authorizer = |
| AwsAuthorizer::new(self.credential.as_ref(), "s3", &self.config.region) |
| .with_sign_payload(self.config.sign_payload) |
| .with_request_payer(self.config.request_payer) |
| .with_crypto(self.config.crypto()?); |
| |
| if self.session_token { |
| let token = HeaderName::from_static("x-amz-s3session-token"); |
| authorizer = authorizer.with_token_header(token) |
| } |
| |
| Ok(authorizer) |
| } |
| } |
| |
| #[derive(Debug, thiserror::Error)] |
| pub enum RequestError { |
| #[error(transparent)] |
| Generic { |
| #[from] |
| source: crate::Error, |
| }, |
| |
| #[error("Retry")] |
| Retry { |
| source: crate::client::retry::RetryError, |
| path: String, |
| }, |
| } |
| |
| impl From<RequestError> for crate::Error { |
| fn from(value: RequestError) -> Self { |
| match value { |
| RequestError::Generic { source } => source, |
| RequestError::Retry { source, path } => source.error(STORE, path), |
| } |
| } |
| } |
| |
| /// A builder for a request allowing customization of the headers and query string |
| pub(crate) struct Request<'a> { |
| path: &'a Path, |
| config: &'a S3Config, |
| builder: HttpRequestBuilder, |
| payload_sha256: Option<[u8; 32]>, |
| payload: Option<PutPayload>, |
| use_session_creds: bool, |
| idempotent: bool, |
| retry_on_conflict: bool, |
| retry_error_body: bool, |
| } |
| |
| impl Request<'_> { |
| pub(crate) fn query<T: Serialize + ?Sized + Sync>(self, query: &T) -> Self { |
| let builder = self.builder.query(query); |
| Self { builder, ..self } |
| } |
| |
| pub(crate) fn header<K>(self, k: K, v: &str) -> Self |
| where |
| K: TryInto<HeaderName>, |
| K::Error: Into<RequestBuilderError>, |
| { |
| let builder = self.builder.header(k, v); |
| Self { builder, ..self } |
| } |
| |
| pub(crate) fn headers(self, headers: HeaderMap) -> Self { |
| let builder = self.builder.headers(headers); |
| Self { builder, ..self } |
| } |
| |
| pub(crate) fn idempotent(self, idempotent: bool) -> Self { |
| Self { idempotent, ..self } |
| } |
| |
| pub(crate) fn retry_on_conflict(self, retry_on_conflict: bool) -> Self { |
| Self { |
| retry_on_conflict, |
| ..self |
| } |
| } |
| |
| pub(crate) fn retry_error_body(self, retry_error_body: bool) -> Self { |
| Self { |
| retry_error_body, |
| ..self |
| } |
| } |
| |
| pub(crate) fn with_encryption_headers(self) -> Self { |
| let headers = self.config.encryption_headers.clone().into(); |
| let builder = self.builder.headers(headers); |
| Self { builder, ..self } |
| } |
| |
| pub(crate) fn with_session_creds(self, use_session_creds: bool) -> Self { |
| Self { |
| use_session_creds, |
| ..self |
| } |
| } |
| |
| pub(crate) fn with_tags(mut self, tags: TagSet) -> Self { |
| let tags = tags.encoded(); |
| if !tags.is_empty() && !self.config.disable_tagging { |
| self.builder = self.builder.header(&TAGS_HEADER, tags); |
| } |
| self |
| } |
| |
| pub(crate) fn with_attributes(self, attributes: Attributes) -> Self { |
| let mut has_content_type = false; |
| let mut builder = self.builder; |
| for (k, v) in &attributes { |
| builder = match k { |
| Attribute::CacheControl => builder.header(CACHE_CONTROL, v.as_ref()), |
| Attribute::ContentDisposition => builder.header(CONTENT_DISPOSITION, v.as_ref()), |
| Attribute::ContentEncoding => builder.header(CONTENT_ENCODING, v.as_ref()), |
| Attribute::ContentLanguage => builder.header(CONTENT_LANGUAGE, v.as_ref()), |
| Attribute::ContentType => { |
| has_content_type = true; |
| builder.header(CONTENT_TYPE, v.as_ref()) |
| } |
| Attribute::StorageClass => builder.header(STORAGE_CLASS, v.as_ref()), |
| Attribute::Metadata(k_suffix) => builder.header( |
| &format!("{USER_DEFINED_METADATA_HEADER_PREFIX}{k_suffix}"), |
| v.as_ref(), |
| ), |
| }; |
| } |
| |
| if !has_content_type { |
| if let Some(value) = self.config.client_options.get_content_type(self.path) { |
| builder = builder.header(CONTENT_TYPE, value); |
| } |
| } |
| Self { builder, ..self } |
| } |
| |
| pub(crate) fn with_extensions(self, extensions: ::http::Extensions) -> Self { |
| let builder = self.builder.extensions(extensions); |
| Self { builder, ..self } |
| } |
| |
| pub(crate) fn with_payload(mut self, payload: PutPayload) -> Result<Self> { |
| let mut cached_digest: Option<[u8; 32]> = None; |
| let mut sha256_digest = || -> Result<[u8; 32]> { |
| if let Some(digest) = cached_digest { |
| return Ok(digest); |
| } |
| let mut ctx = self.config.crypto()?.digest(DigestAlgorithm::Sha256)?; |
| for part in &payload { |
| ctx.update(part); |
| } |
| let digest = ctx.finish()?.try_into().unwrap(); |
| cached_digest = Some(digest); |
| Ok(digest) |
| }; |
| |
| if !self.config.skip_signature && self.config.sign_payload { |
| self.payload_sha256 = Some(sha256_digest()?); |
| } |
| |
| match self.config.checksum { |
| Some(Checksum::SHA256) => { |
| self.builder = self |
| .builder |
| .header(SHA256_CHECKSUM, BASE64_STANDARD.encode(sha256_digest()?)); |
| } |
| Some(Checksum::CRC64NVME) => { |
| let crc_algo = crc_fast::CrcAlgorithm::Crc64Nvme; |
| let mut digest = crc_fast::Digest::new(crc_algo); |
| payload.iter().for_each(|x| digest.update(x)); |
| let checksum = digest.finalize(); |
| |
| self.builder = self.builder.header( |
| CRC64NVME_CHECKSUM, |
| BASE64_STANDARD.encode(checksum.to_be_bytes()), |
| ) |
| } |
| None => {} |
| } |
| |
| let content_length = payload.content_length(); |
| self.builder = self.builder.header(CONTENT_LENGTH, content_length); |
| self.payload = Some(payload); |
| Ok(self) |
| } |
| |
| pub(crate) async fn send(self) -> Result<HttpResponse, RequestError> { |
| let credential = match self.use_session_creds { |
| true => self.config.get_session_credential().await?, |
| false => { |
| let credential = self.config.get_credential().await?; |
| credential.map(|credential| SessionCredential { |
| credential, |
| session_token: false, |
| config: self.config, |
| }) |
| } |
| }; |
| let authorizer = credential.as_ref().map(|x| x.authorizer()).transpose()?; |
| |
| let sha = self.payload_sha256.as_ref().map(|x| x.as_ref()); |
| |
| let path = self.path.as_ref(); |
| self.builder |
| .with_aws_sigv4(authorizer, sha)? |
| .retryable(&self.config.retry_config) |
| .retry_on_conflict(self.retry_on_conflict) |
| .idempotent(self.idempotent) |
| .retry_error_body(self.retry_error_body) |
| .payload(self.payload) |
| .send() |
| .await |
| .map_err(|source| { |
| let path = path.into(); |
| RequestError::Retry { source, path } |
| }) |
| } |
| |
| pub(crate) async fn do_put(self) -> Result<PutResult> { |
| let response = self.send().await?; |
| Ok( |
| get_put_result(response, VERSION_HEADER) |
| .map_err(|source| Error::Metadata { source })?, |
| ) |
| } |
| } |
| |
| #[derive(Debug)] |
| pub(crate) struct S3Client { |
| pub config: S3Config, |
| pub client: HttpClient, |
| } |
| |
| impl S3Client { |
| pub(crate) fn new(config: S3Config, client: HttpClient) -> Self { |
| Self { config, client } |
| } |
| |
| pub(crate) fn request<'a>(&'a self, method: Method, path: &'a Path) -> Request<'a> { |
| let url = self.config.path_url(path); |
| let mut builder = self.client.request(method, url); |
| if let Some(headers) = self.config.client_options.get_default_headers() { |
| builder = builder.headers(headers.clone()); |
| } |
| Request { |
| path, |
| builder, |
| payload: None, |
| payload_sha256: None, |
| config: &self.config, |
| use_session_creds: true, |
| idempotent: false, |
| retry_on_conflict: false, |
| retry_error_body: false, |
| } |
| } |
| |
| /// Make an S3 Delete Objects request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjects.html> |
| /// |
| /// Produces a vector of results, one for each path in the input vector. If |
| /// the delete was successful, the path is returned in the `Ok` variant. If |
| /// there was an error for a certain path, the error will be returned in the |
| /// vector. If there was an issue with making the overall request, an error |
| /// will be returned at the top level. |
| pub(crate) async fn bulk_delete_request(&self, paths: Vec<Path>) -> Result<Vec<Result<Path>>> { |
| if paths.is_empty() { |
| return Ok(Vec::new()); |
| } |
| |
| let credential = self.config.get_session_credential().await?; |
| let authorizer = credential.as_ref().map(|x| x.authorizer()).transpose()?; |
| let url = format!("{}?delete", self.config.bucket_endpoint); |
| |
| let mut buffer = Vec::new(); |
| let mut writer = quick_xml::Writer::new(&mut buffer); |
| writer |
| .write_event(xml_events::Event::Start( |
| xml_events::BytesStart::new("Delete") |
| .with_attributes([("xmlns", "http://s3.amazonaws.com/doc/2006-03-01/")]), |
| )) |
| .unwrap(); |
| for path in &paths { |
| // <Object><Key>{path}</Key></Object> |
| writer |
| .write_event(xml_events::Event::Start(xml_events::BytesStart::new( |
| "Object", |
| ))) |
| .unwrap(); |
| writer |
| .write_event(xml_events::Event::Start(xml_events::BytesStart::new("Key"))) |
| .unwrap(); |
| writer |
| .write_event(xml_events::Event::Text(xml_events::BytesText::new( |
| path.as_ref(), |
| ))) |
| .map_err(|err| crate::Error::Generic { |
| store: STORE, |
| source: Box::new(err), |
| })?; |
| writer |
| .write_event(xml_events::Event::End(xml_events::BytesEnd::new("Key"))) |
| .unwrap(); |
| writer |
| .write_event(xml_events::Event::End(xml_events::BytesEnd::new("Object"))) |
| .unwrap(); |
| } |
| writer |
| .write_event(xml_events::Event::End(xml_events::BytesEnd::new("Delete"))) |
| .unwrap(); |
| |
| let body = Bytes::from(buffer); |
| |
| let mut builder = self.client.request(Method::POST, url); |
| if let Some(headers) = self.config.client_options.get_default_headers() { |
| builder = builder.headers(headers.clone()); |
| } |
| |
| let crypto = self.config.crypto()?; |
| let mut ctx = crypto.digest(DigestAlgorithm::Sha256)?; |
| ctx.update(body.as_ref()); |
| let digest = ctx.finish()?; |
| |
| builder = builder.header(SHA256_CHECKSUM, BASE64_STANDARD.encode(digest)); |
| |
| // S3 *requires* DeleteObjects to include a Content-MD5 header: |
| // https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObjects.html |
| // > "The Content-MD5 request header is required for all Multi-Object Delete requests" |
| // Some platforms, like MinIO, enforce this requirement and fail requests without the header. |
| let mut hasher = Md5::new(); |
| hasher.update(&body); |
| builder = builder.header("Content-MD5", BASE64_STANDARD.encode(hasher.finalize())); |
| |
| let response = builder |
| .header(CONTENT_TYPE, "application/xml") |
| .body(body) |
| .with_aws_sigv4(authorizer, Some(digest))? |
| .retryable(&self.config.retry_config) |
| .retry_error_body(true) |
| .send() |
| .await |
| .map_err(|source| Error::DeleteObjectsRequest { |
| source, |
| paths: paths.iter().map(|p| p.to_string()).collect(), |
| })? |
| .into_body() |
| .bytes() |
| .await |
| .map_err(|source| Error::DeleteObjectsResponse { source })?; |
| |
| let response: BatchDeleteResponse = |
| quick_xml::de::from_reader(response.reader()).map_err(|err| { |
| Error::InvalidDeleteObjectsResponse { |
| source: Box::new(err), |
| } |
| })?; |
| |
| // Assume all were ok, then fill in errors. This guarantees output order |
| // matches input order. |
| let mut results: Vec<Result<Path>> = paths.iter().cloned().map(Ok).collect(); |
| for content in response.content.into_iter() { |
| if let DeleteObjectResult::Error(error) = content { |
| let path = |
| Path::parse(&error.key).map_err(|err| Error::InvalidDeleteObjectsResponse { |
| source: Box::new(err), |
| })?; |
| let i = paths.iter().find_position(|&p| p == &path).unwrap().0; |
| results[i] = Err(Error::from(error).into()); |
| } |
| } |
| |
| Ok(results) |
| } |
| |
| /// Make a single-object S3 Delete request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_DeleteObject.html> |
| /// |
| /// Unlike [`bulk_delete_request`](Self::bulk_delete_request), this issues a |
| /// plain `DELETE /key` request, which is part of the core S3 API and is |
| /// supported by every S3-compatible provider. |
| pub(crate) async fn delete_request(&self, path: &Path) -> Result<()> { |
| self.request(Method::DELETE, path).send().await?; |
| Ok(()) |
| } |
| |
| /// Make an S3 Copy request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html> |
| pub(crate) fn copy_request<'a>(&'a self, from: &Path, to: &'a Path) -> Request<'a> { |
| let source = format!("{}/{}", self.config.bucket, encode_path(from)); |
| |
| let mut copy_source_encryption_headers = HeaderMap::new(); |
| if let Some(customer_algorithm) = self |
| .config |
| .encryption_headers |
| .0 |
| .get("x-amz-server-side-encryption-customer-algorithm") |
| { |
| copy_source_encryption_headers.insert( |
| "x-amz-copy-source-server-side-encryption-customer-algorithm", |
| customer_algorithm.clone(), |
| ); |
| } |
| if let Some(customer_key) = self |
| .config |
| .encryption_headers |
| .0 |
| .get("x-amz-server-side-encryption-customer-key") |
| { |
| copy_source_encryption_headers.insert( |
| "x-amz-copy-source-server-side-encryption-customer-key", |
| customer_key.clone(), |
| ); |
| } |
| if let Some(customer_key_md5) = self |
| .config |
| .encryption_headers |
| .0 |
| .get("x-amz-server-side-encryption-customer-key-MD5") |
| { |
| copy_source_encryption_headers.insert( |
| "x-amz-copy-source-server-side-encryption-customer-key-MD5", |
| customer_key_md5.clone(), |
| ); |
| } |
| |
| self.request(Method::PUT, to) |
| .idempotent(true) |
| .retry_error_body(true) |
| .header(©_SOURCE_HEADER, &source) |
| .headers(self.config.encryption_headers.clone().into()) |
| .headers(copy_source_encryption_headers) |
| .with_session_creds(false) |
| } |
| |
| pub(crate) async fn create_multipart( |
| &self, |
| location: &Path, |
| opts: PutMultipartOptions, |
| ) -> Result<MultipartId> { |
| let PutMultipartOptions { |
| tags, |
| attributes, |
| extensions, |
| } = opts; |
| |
| let mut request = self.request(Method::POST, location); |
| if let Some(algorithm) = self.config.checksum { |
| match algorithm { |
| Checksum::SHA256 => { |
| request = request.header(ALGORITHM, "SHA256"); |
| } |
| Checksum::CRC64NVME => { |
| request = request.header(ALGORITHM, "CRC64NVME"); |
| } |
| } |
| } |
| let response = request |
| .query(&[("uploads", "")]) |
| .with_encryption_headers() |
| .with_attributes(attributes) |
| .with_tags(tags) |
| .with_extensions(extensions) |
| .header(CONTENT_LENGTH, "0") |
| .idempotent(true) |
| .send() |
| .await? |
| .into_body() |
| .bytes() |
| .await |
| .map_err(|source| Error::CreateMultipartResponseBody { source })?; |
| |
| let response: InitiateMultipartUploadResult = quick_xml::de::from_reader(response.reader()) |
| .map_err(|source| Error::InvalidMultipartResponse { source })?; |
| |
| Ok(response.upload_id) |
| } |
| |
| pub(crate) async fn put_part( |
| &self, |
| path: &Path, |
| upload_id: &MultipartId, |
| part_idx: usize, |
| data: PutPartPayload<'_>, |
| ) -> Result<PartId> { |
| let is_copy = matches!(data, PutPartPayload::Copy(_)); |
| let part = (part_idx + 1).to_string(); |
| |
| let mut request = self |
| .request(Method::PUT, path) |
| .query(&[("partNumber", &part), ("uploadId", upload_id)]) |
| .idempotent(true); |
| |
| request = match data { |
| PutPartPayload::Part(payload) => request.with_payload(payload)?, |
| PutPartPayload::Copy(path) => request.header( |
| "x-amz-copy-source", |
| &format!("{}/{}", self.config.bucket, encode_path(path)), |
| ), |
| }; |
| |
| if self |
| .config |
| .encryption_headers |
| .0 |
| .contains_key("x-amz-server-side-encryption-customer-algorithm") |
| { |
| // If SSE-C is used, we must include the encryption headers in every upload request. |
| request = request.with_encryption_headers(); |
| } |
| |
| let (parts, body) = request.send().await?.into_parts(); |
| let (e_tag, checksum_sha256, checksum_crc64nvme) = if is_copy { |
| let response = body |
| .bytes() |
| .await |
| .map_err(|source| Error::CreateMultipartResponseBody { source })?; |
| let response: CopyPartResult = quick_xml::de::from_reader(response.reader()) |
| .map_err(|source| Error::InvalidMultipartResponse { source })?; |
| ( |
| response.e_tag, |
| response.checksum_sha256, |
| response.checksum_crc64nvme, |
| ) |
| } else { |
| let e_tag = get_etag(&parts.headers).map_err(|source| Error::Metadata { source })?; |
| let checksum_sha256 = parts |
| .headers |
| .get(SHA256_CHECKSUM) |
| .and_then(|v| v.to_str().ok()) |
| .map(|v| v.to_string()); |
| let checksum_crc64nvme = parts |
| .headers |
| .get(CRC64NVME_CHECKSUM) |
| .and_then(|v| v.to_str().ok()) |
| .map(|v| v.to_string()); |
| (e_tag, checksum_sha256, checksum_crc64nvme) |
| }; |
| |
| let content_id = match self.config.checksum { |
| Some(_) => { |
| let meta = PartMetadata { |
| e_tag, |
| checksum_sha256, |
| checksum_crc64nvme, |
| }; |
| quick_xml::se::to_string(&meta).unwrap() |
| } |
| None => e_tag, |
| }; |
| |
| Ok(PartId { content_id }) |
| } |
| |
| pub(crate) async fn abort_multipart(&self, location: &Path, upload_id: &str) -> Result<()> { |
| self.request(Method::DELETE, location) |
| .query(&[("uploadId", upload_id)]) |
| .with_encryption_headers() |
| .send() |
| .await?; |
| |
| Ok(()) |
| } |
| |
| pub(crate) async fn complete_multipart( |
| &self, |
| location: &Path, |
| upload_id: &str, |
| parts: Vec<PartId>, |
| mode: CompleteMultipartMode, |
| ) -> Result<PutResult> { |
| let parts = if parts.is_empty() { |
| // If no parts were uploaded, upload an empty part |
| // otherwise the completion request will fail |
| let part = self |
| .put_part( |
| location, |
| &upload_id.to_string(), |
| 0, |
| PutPartPayload::default(), |
| ) |
| .await?; |
| vec![part] |
| } else { |
| parts |
| }; |
| let request = CompleteMultipartUpload::from(parts); |
| let body = quick_xml::se::to_string(&request).unwrap(); |
| |
| let credential = self.config.get_session_credential().await?; |
| let authorizer = credential.as_ref().map(|x| x.authorizer()).transpose()?; |
| let url = self.config.path_url(location); |
| |
| let mut builder = self.client.post(url); |
| if let Some(headers) = self.config.client_options.get_default_headers() { |
| builder = builder.headers(headers.clone()); |
| } |
| |
| let request = builder |
| .query(&[("uploadId", upload_id)]) |
| .body(body) |
| .with_aws_sigv4(authorizer, None)?; |
| |
| let request = match mode { |
| CompleteMultipartMode::Overwrite => request, |
| CompleteMultipartMode::Create => request.header("If-None-Match", "*"), |
| }; |
| |
| let response = request |
| .retryable(&self.config.retry_config) |
| .idempotent(true) |
| .retry_error_body(true) |
| .send() |
| .await |
| .map_err(|source| Error::CompleteMultipartRequest { |
| source, |
| path: location.as_ref().to_string(), |
| })?; |
| |
| let (parts, body) = response.into_parts(); |
| |
| let version = get_version(&parts.headers, VERSION_HEADER) |
| .map_err(|source| Error::Metadata { source })?; |
| |
| let data = body |
| .bytes() |
| .await |
| .map_err(|source| Error::CompleteMultipartResponseBody { source })?; |
| |
| let response: CompleteMultipartUploadResult = quick_xml::de::from_reader(data.reader()) |
| .map_err(|source| Error::InvalidMultipartResponse { source })?; |
| |
| Ok(PutResult { |
| e_tag: Some(response.e_tag), |
| version, |
| extensions: parts.extensions, |
| }) |
| } |
| |
| #[cfg(test)] |
| pub(crate) async fn get_object_tagging(&self, path: &Path) -> Result<HttpResponse> { |
| let credential = self.config.get_session_credential().await?; |
| let authorizer = credential.as_ref().map(|x| x.authorizer()).transpose()?; |
| let url = format!("{}?tagging", self.config.path_url(path)); |
| let response = self |
| .client |
| .request(Method::GET, url) |
| .with_aws_sigv4(authorizer, None)? |
| .send_retry(&self.config.retry_config) |
| .await |
| .map_err(|e| e.error(STORE, path.to_string()))?; |
| Ok(response) |
| } |
| } |
| |
| #[async_trait] |
| impl GetClient for S3Client { |
| const STORE: &'static str = STORE; |
| |
| const HEADER_CONFIG: HeaderConfig = HeaderConfig { |
| etag_required: false, |
| last_modified_required: false, |
| version_header: Some(VERSION_HEADER), |
| user_defined_metadata_prefix: Some(USER_DEFINED_METADATA_HEADER_PREFIX), |
| }; |
| |
| fn retry_config(&self) -> &RetryConfig { |
| &self.config.retry_config |
| } |
| |
| /// Make an S3 GET request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html> |
| async fn get_request( |
| &self, |
| ctx: &mut RetryContext, |
| path: &Path, |
| options: GetOptions, |
| ) -> Result<HttpResponse> { |
| let credential = self.config.get_session_credential().await?; |
| let authorizer = credential.as_ref().map(|x| x.authorizer()).transpose()?; |
| let url = self.config.path_url(path); |
| let method = match options.head { |
| true => Method::HEAD, |
| false => Method::GET, |
| }; |
| |
| let mut builder = self.client.request(method, url); |
| if let Some(headers) = self.config.client_options.get_default_headers() { |
| builder = builder.headers(headers.clone()); |
| } |
| if self |
| .config |
| .encryption_headers |
| .0 |
| .contains_key("x-amz-server-side-encryption-customer-algorithm") |
| { |
| builder = builder.headers(self.config.encryption_headers.clone().into()); |
| } |
| |
| if let Some(v) = &options.version { |
| builder = builder.query(&[("versionId", v)]) |
| } |
| |
| let response = builder |
| .with_get_options(options) |
| .with_aws_sigv4(authorizer, None)? |
| .retryable_request() |
| .send(ctx) |
| .await |
| .map_err(|e| e.error(STORE, path.to_string()))?; |
| |
| Ok(response) |
| } |
| } |
| |
| #[async_trait] |
| impl ListClient for Arc<S3Client> { |
| /// Make an S3 List request <https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html> |
| async fn list_request( |
| &self, |
| prefix: Option<&str>, |
| opts: PaginatedListOptions, |
| ) -> Result<PaginatedListResult> { |
| let credential = self.config.get_session_credential().await?; |
| let authorizer = credential.as_ref().map(|x| x.authorizer()).transpose()?; |
| let url = self.config.bucket_endpoint.clone(); |
| |
| let mut query = Vec::with_capacity(4); |
| |
| if let Some(token) = &opts.page_token { |
| query.push(("continuation-token", token.as_ref())) |
| } |
| |
| if let Some(d) = &opts.delimiter { |
| query.push(("delimiter", d.as_ref())) |
| } |
| |
| query.push(("list-type", "2")); |
| |
| if let Some(prefix) = prefix { |
| query.push(("prefix", prefix)) |
| } |
| |
| if let Some(offset) = &opts.offset { |
| query.push(("start-after", offset.as_ref())) |
| } |
| |
| let max_keys_str; |
| if let Some(max_keys) = &opts.max_keys { |
| max_keys_str = max_keys.to_string(); |
| query.push(("max-keys", max_keys_str.as_ref())) |
| } |
| |
| let response = self |
| .client |
| .request(Method::GET, &url) |
| .extensions(opts.extensions) |
| .query(&query) |
| .with_aws_sigv4(authorizer, None)? |
| .send_retry(&self.config.retry_config) |
| .await |
| .map_err(|source| Error::ListRequest { source })?; |
| |
| let (parts, body) = response.into_parts(); |
| |
| let response = body |
| .bytes() |
| .await |
| .map_err(|source| Error::ListResponseBody { source })?; |
| |
| let mut response: ListResponse = quick_xml::de::from_reader(response.reader()) |
| .map_err(|source| Error::InvalidListResponse { source })?; |
| |
| let token = response.next_continuation_token.take(); |
| |
| let mut result: ListResult = response.try_into()?; |
| result.extensions = parts.extensions; |
| |
| Ok(PaginatedListResult { |
| result, |
| page_token: token, |
| }) |
| } |
| } |
| |
| fn encode_path(path: &Path) -> PercentEncode<'_> { |
| utf8_percent_encode(path.as_ref(), &STRICT_PATH_ENCODE_SET) |
| } |
| |
| #[cfg(test)] |
| mod tests { |
| use super::*; |
| use crate::GetOptions; |
| use crate::ObjectStore; |
| use crate::aws::{AmazonS3, AmazonS3Builder}; |
| use crate::client::HttpClient; |
| use crate::client::get::GetClient; |
| use crate::client::mock_server::MockServer; |
| use crate::client::retry::RetryContext; |
| use futures_util::{StreamExt, TryStreamExt}; |
| use http::Response; |
| use http::header::{AUTHORIZATION, CONTENT_LENGTH}; |
| use hyper::Request; |
| use hyper::body::Incoming; |
| |
| #[cfg(feature = "reqwest")] |
| #[tokio::test] |
| async fn test_create_multipart_has_content_length() { |
| let mock = MockServer::new().await; |
| |
| mock.push_fn(|req| { |
| // Verify Content-Length header is present and set to 0 |
| assert_eq!(req.headers().get(CONTENT_LENGTH).unwrap(), "0"); |
| assert!(req.uri().query().unwrap_or("").contains("uploads")); |
| |
| Response::builder() |
| .status(200) |
| .body("<InitiateMultipartUploadResult><UploadId>test-upload-id</UploadId></InitiateMultipartUploadResult>".to_string()) |
| .unwrap() |
| }); |
| |
| let credential = AwsCredential { |
| key_id: "key".to_string(), |
| secret_key: "secret".to_string(), |
| token: None, |
| }; |
| |
| let config = S3Config { |
| bucket_endpoint: mock.url().to_string(), |
| bucket: "test-bucket".to_string(), |
| region: "us-east-1".to_string(), |
| credentials: Arc::new(crate::StaticCredentialProvider::new(credential)), |
| client_options: ClientOptions::new().with_allow_http(true), |
| skip_signature: true, |
| session_provider: None, |
| retry_config: Default::default(), |
| sign_payload: false, |
| disable_tagging: false, |
| disable_bulk_delete: false, |
| checksum: None, |
| copy_if_not_exists: None, |
| conditional_put: Default::default(), |
| encryption_headers: Default::default(), |
| request_payer: false, |
| crypto: None, |
| }; |
| |
| let client = S3Client::new(config, HttpClient::new(reqwest::Client::new())); |
| let result = client |
| .create_multipart(&Path::from("test"), PutMultipartOptions::default()) |
| .await; |
| |
| assert_eq!(result.unwrap(), "test-upload-id"); |
| mock.shutdown().await; |
| } |
| |
| fn assert_default_headers_signed(req: &Request<Incoming>) { |
| assert_eq!(req.headers().get("x-amz-meta-test").unwrap(), "test-value"); |
| assert_eq!(req.headers().get("x-amz-tagging").unwrap(), "key=value"); |
| |
| let auth = req.headers().get(AUTHORIZATION).unwrap().to_str().unwrap(); |
| assert!( |
| auth.contains("x-amz-meta-test"), |
| "x-amz-meta-test not in SignedHeaders: {auth}" |
| ); |
| assert!( |
| auth.contains("x-amz-tagging"), |
| "x-amz-tagging not in SignedHeaders: {auth}" |
| ); |
| } |
| |
| fn default_headers_config(mock: &MockServer) -> S3Config { |
| let credential = AwsCredential { |
| key_id: "AKIAIOSFODNN7EXAMPLE".to_string(), |
| secret_key: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY".to_string(), |
| token: None, |
| }; |
| |
| let mut default_headers = HeaderMap::new(); |
| default_headers.insert("x-amz-meta-test", "test-value".parse().unwrap()); |
| default_headers.insert("x-amz-tagging", "key=value".parse().unwrap()); |
| |
| S3Config { |
| bucket_endpoint: mock.url().to_string(), |
| bucket: "test-bucket".to_string(), |
| region: "us-east-1".to_string(), |
| credentials: Arc::new(crate::StaticCredentialProvider::new(credential)), |
| client_options: ClientOptions::new() |
| .with_allow_http(true) |
| .with_default_headers(default_headers), |
| crypto: None, |
| skip_signature: false, |
| session_provider: None, |
| retry_config: Default::default(), |
| sign_payload: false, |
| disable_tagging: false, |
| disable_bulk_delete: false, |
| checksum: None, |
| copy_if_not_exists: None, |
| conditional_put: Default::default(), |
| encryption_headers: Default::default(), |
| request_payer: false, |
| } |
| } |
| |
| #[cfg(feature = "reqwest")] |
| #[tokio::test] |
| async fn test_default_headers_signed_request() { |
| let mock = MockServer::new().await; |
| mock.push_fn(|req| { |
| assert_default_headers_signed(&req); |
| Response::builder() |
| .status(200) |
| .header("etag", "\"test-etag\"") |
| .body(String::new()) |
| .unwrap() |
| }); |
| |
| let config = default_headers_config(&mock); |
| let client = S3Client::new(config, HttpClient::new(reqwest::Client::new())); |
| let result = client |
| .request(Method::PUT, &Path::from("test")) |
| .with_payload(PutPayload::default()) |
| .unwrap() |
| .do_put() |
| .await; |
| |
| assert!(result.is_ok()); |
| mock.shutdown().await; |
| } |
| |
| #[cfg(feature = "reqwest")] |
| #[tokio::test] |
| async fn test_default_headers_signed_bulk_delete() { |
| let mock = MockServer::new().await; |
| mock.push_fn(|req| { |
| assert_default_headers_signed(&req); |
| Response::builder() |
| .status(200) |
| .body("<DeleteResult><Deleted><Key>test</Key></Deleted></DeleteResult>".to_string()) |
| .unwrap() |
| }); |
| |
| let config = default_headers_config(&mock); |
| let client = S3Client::new(config, HttpClient::new(reqwest::Client::new())); |
| let result = client.bulk_delete_request(vec![Path::from("test")]).await; |
| |
| assert!(result.is_ok()); |
| mock.shutdown().await; |
| } |
| |
| /// `(method, path, query)` captured for assertion outside the mock closure. |
| /// |
| /// `MockServer` swallows panics raised inside its response handler (the |
| /// connection just resets and the S3 retry logic can still surface an `Ok` |
| /// result), so assertions placed inside the closure are silently ignored. |
| /// We capture into shared state and assert in the test body instead. |
| type CapturedRequest = (Method, String, Option<String>); |
| |
| fn capture(captured: &Arc<std::sync::Mutex<Vec<CapturedRequest>>>, req: &Request<Incoming>) { |
| captured.lock().unwrap().push(( |
| req.method().clone(), |
| req.uri().path().to_string(), |
| req.uri().query().map(|s| s.to_string()), |
| )); |
| } |
| |
| /// Build an `AmazonS3` via the public builder so that `bucket_endpoint` |
| /// is computed by the library from the addressing-style option — i.e. |
| /// the option under test actually drives the URL the client emits. |
| fn make_store(mock: &MockServer, virtual_hosted: bool, disable_bulk_delete: bool) -> AmazonS3 { |
| AmazonS3Builder::new() |
| .with_endpoint(mock.url()) |
| .with_bucket_name("test-bucket") |
| .with_region("us-east-1") |
| .with_allow_http(true) |
| .with_skip_signature(true) |
| .with_virtual_hosted_style_request(virtual_hosted) |
| .with_disable_bulk_delete(disable_bulk_delete) |
| .build() |
| .unwrap() |
| } |
| |
| #[tokio::test] |
| async fn test_delete_default() { |
| // Default: path-style + bulk delete enabled. |
| // `delete_stream` must issue a single `POST /{bucket}?delete`. |
| let mock = MockServer::new().await; |
| let captured: Arc<std::sync::Mutex<Vec<CapturedRequest>>> = Default::default(); |
| let c = Arc::clone(&captured); |
| mock.push_fn(move |req| { |
| capture(&c, &req); |
| Response::builder() |
| .status(200) |
| .body("<DeleteResult><Deleted><Key>foo</Key></Deleted></DeleteResult>".to_string()) |
| .unwrap() |
| }); |
| |
| let store = make_store(&mock, false, false); |
| let locations = futures_util::stream::iter(vec![Ok(Path::from("foo"))]).boxed(); |
| let deleted: Vec<_> = store.delete_stream(locations).try_collect().await.unwrap(); |
| assert_eq!(deleted.len(), 1); |
| |
| let captured = captured.lock().unwrap().clone(); |
| assert_eq!(captured.len(), 1, "expected one bulk delete request"); |
| assert_eq!(captured[0].0, Method::POST); |
| assert_eq!(captured[0].1, "/test-bucket"); |
| assert_eq!(captured[0].2.as_deref(), Some("delete")); |
| |
| mock.shutdown().await; |
| } |
| |
| #[tokio::test] |
| async fn test_delete_default_with_disable_bulk() { |
| // Path-style + bulk delete disabled. |
| // `delete_stream` must fan out into `DELETE /{bucket}/{key}` (one per |
| // object, no `?delete` query). |
| let mock = MockServer::new().await; |
| let captured: Arc<std::sync::Mutex<Vec<CapturedRequest>>> = Default::default(); |
| for _ in 0..2 { |
| let c = Arc::clone(&captured); |
| mock.push_fn(move |req| { |
| capture(&c, &req); |
| Response::builder().status(204).body(String::new()).unwrap() |
| }); |
| } |
| |
| let store = make_store(&mock, false, true); |
| let locations = |
| futures_util::stream::iter(vec![Ok(Path::from("foo")), Ok(Path::from("bar"))]).boxed(); |
| let deleted: Vec<_> = store.delete_stream(locations).try_collect().await.unwrap(); |
| assert_eq!(deleted.len(), 2); |
| |
| let mut captured = captured.lock().unwrap().clone(); |
| captured.sort_by(|a, b| a.1.cmp(&b.1)); |
| assert_eq!(captured.len(), 2, "expected one DELETE per object"); |
| assert_eq!( |
| captured[0], |
| (Method::DELETE, "/test-bucket/bar".to_string(), None) |
| ); |
| assert_eq!( |
| captured[1], |
| (Method::DELETE, "/test-bucket/foo".to_string(), None) |
| ); |
| |
| mock.shutdown().await; |
| } |
| |
| #[tokio::test] |
| async fn test_delete_virtual_hosted() { |
| // Virtual-hosted style + bulk delete enabled. |
| // `delete_stream` must issue a single `POST /?delete` (bucket is in |
| // the host, not the path). |
| let mock = MockServer::new().await; |
| let captured: Arc<std::sync::Mutex<Vec<CapturedRequest>>> = Default::default(); |
| let c = Arc::clone(&captured); |
| mock.push_fn(move |req| { |
| capture(&c, &req); |
| Response::builder() |
| .status(200) |
| .body("<DeleteResult><Deleted><Key>foo</Key></Deleted></DeleteResult>".to_string()) |
| .unwrap() |
| }); |
| |
| let store = make_store(&mock, true, false); |
| let locations = futures_util::stream::iter(vec![Ok(Path::from("foo"))]).boxed(); |
| let deleted: Vec<_> = store.delete_stream(locations).try_collect().await.unwrap(); |
| assert_eq!(deleted.len(), 1); |
| |
| let captured = captured.lock().unwrap().clone(); |
| assert_eq!(captured.len(), 1, "expected one bulk delete request"); |
| assert_eq!(captured[0].0, Method::POST); |
| assert_eq!(captured[0].1, "/"); |
| assert_eq!(captured[0].2.as_deref(), Some("delete")); |
| |
| mock.shutdown().await; |
| } |
| |
| #[tokio::test] |
| async fn test_delete_virtual_hosted_with_disable_bulk() { |
| // Virtual-hosted style + bulk delete disabled. |
| // `delete_stream` must fan out into `DELETE /{key}` (no bucket in |
| // path, no `?delete` query). |
| let mock = MockServer::new().await; |
| let captured: Arc<std::sync::Mutex<Vec<CapturedRequest>>> = Default::default(); |
| for _ in 0..2 { |
| let c = Arc::clone(&captured); |
| mock.push_fn(move |req| { |
| capture(&c, &req); |
| Response::builder().status(204).body(String::new()).unwrap() |
| }); |
| } |
| |
| let store = make_store(&mock, true, true); |
| let locations = |
| futures_util::stream::iter(vec![Ok(Path::from("foo")), Ok(Path::from("bar"))]).boxed(); |
| let deleted: Vec<_> = store.delete_stream(locations).try_collect().await.unwrap(); |
| assert_eq!(deleted.len(), 2); |
| |
| let mut captured = captured.lock().unwrap().clone(); |
| captured.sort_by(|a, b| a.1.cmp(&b.1)); |
| assert_eq!(captured.len(), 2, "expected one DELETE per object"); |
| assert_eq!(captured[0], (Method::DELETE, "/bar".to_string(), None)); |
| assert_eq!(captured[1], (Method::DELETE, "/foo".to_string(), None)); |
| |
| mock.shutdown().await; |
| } |
| |
| #[cfg(feature = "reqwest")] |
| #[tokio::test] |
| async fn test_default_headers_signed_get_request() { |
| let mock = MockServer::new().await; |
| mock.push_fn(|req| { |
| assert_default_headers_signed(&req); |
| Response::builder() |
| .status(200) |
| .body("test-body".to_string()) |
| .unwrap() |
| }); |
| |
| let config = default_headers_config(&mock); |
| let client = S3Client::new(config, HttpClient::new(reqwest::Client::new())); |
| let mut ctx = RetryContext::new(&client.config.retry_config); |
| let result = client |
| .get_request(&mut ctx, &Path::from("test"), GetOptions::default()) |
| .await; |
| |
| assert!(result.is_ok()); |
| mock.shutdown().await; |
| } |
| |
| #[cfg(feature = "reqwest")] |
| #[tokio::test] |
| async fn test_default_headers_signed_complete_multipart() { |
| let mock = MockServer::new().await; |
| mock.push_fn(|req| { |
| assert_default_headers_signed(&req); |
| assert!(req.uri().query().unwrap_or("").contains("uploadId")); |
| |
| Response::builder() |
| .status(200) |
| .body("<CompleteMultipartUploadResult><ETag>\"test-etag\"</ETag></CompleteMultipartUploadResult>".to_string()) |
| .unwrap() |
| }); |
| |
| let config = default_headers_config(&mock); |
| let client = S3Client::new(config, HttpClient::new(reqwest::Client::new())); |
| |
| let parts = vec![PartId { |
| content_id: "\"part-etag\"".to_string(), |
| }]; |
| let _ = client |
| .complete_multipart( |
| &Path::from("test"), |
| "test-upload-id", |
| parts, |
| CompleteMultipartMode::Overwrite, |
| ) |
| .await |
| .unwrap(); |
| mock.shutdown().await; |
| } |
| } |
