s3/request/
tokio_backend.rs

1extern crate base64;
2extern crate md5;
3
4use bytes::Bytes;
5use futures_util::TryStreamExt;
6use maybe_async::maybe_async;
7use std::collections::HashMap;
8use std::str::FromStr as _;
9use time::OffsetDateTime;
10
11use super::request_trait::{Request, ResponseData, ResponseDataStream};
12use crate::bucket::Bucket;
13use crate::command::Command;
14use crate::command::HttpMethod;
15use crate::error::S3Error;
16use crate::utils::now_utc;
17
18use tokio_stream::StreamExt;
19
20#[derive(Clone, Debug, Default)]
21pub(crate) struct ClientOptions {
22    pub proxy: Option<reqwest::Proxy>,
23    #[cfg(any(
24        feature = "tokio-native-tls",
25        feature = "tokio-rustls-tls",
26        feature = "tokio-rustls-tls-ring"
27    ))]
28    pub accept_invalid_certs: bool,
29    #[cfg(any(
30        feature = "tokio-native-tls",
31        feature = "tokio-rustls-tls",
32        feature = "tokio-rustls-tls-ring"
33    ))]
34    pub accept_invalid_hostnames: bool,
35}
36
37#[cfg(feature = "with-tokio")]
38pub(crate) fn client(options: &ClientOptions) -> Result<reqwest::Client, S3Error> {
39    // Request timeouts are applied to individual requests so changing a
40    // Bucket's timeout does not require rebuilding this client (or lose its
41    // proxy/TLS configuration).
42    let client = reqwest::Client::builder();
43
44    let client = if let Some(ref proxy) = options.proxy {
45        client.proxy(proxy.clone())
46    } else {
47        client
48    };
49
50    cfg_if::cfg_if! {
51        if #[cfg(any(
52            feature = "tokio-native-tls",
53            feature = "tokio-rustls-tls",
54            feature = "tokio-rustls-tls-ring"
55        ))] {
56            let client = client.danger_accept_invalid_certs(options.accept_invalid_certs);
57        }
58    }
59
60    cfg_if::cfg_if! {
61        if #[cfg(any(
62            feature = "tokio-native-tls",
63            feature = "tokio-rustls-tls",
64            feature = "tokio-rustls-tls-ring"
65        ))] {
66            let client = client.danger_accept_invalid_hostnames(options.accept_invalid_hostnames);
67        }
68    }
69
70    Ok(client.build()?)
71}
72// Temporary structure for making a request
73pub struct ReqwestRequest<'a> {
74    pub bucket: &'a Bucket,
75    pub path: &'a str,
76    pub command: Command<'a>,
77    pub datetime: OffsetDateTime,
78    pub sync: bool,
79}
80
81async fn request_response_attempt(
82    request: &ReqwestRequest<'_>,
83) -> Result<reqwest::Response, crate::request::RequestAttemptError> {
84    let headers = request
85        .headers()
86        .await?
87        .iter()
88        .map(|(k, v)| {
89            (
90                reqwest::header::HeaderName::from_str(k.as_str()),
91                reqwest::header::HeaderValue::from_str(v.to_str().unwrap_or_default()),
92            )
93        })
94        .filter(|(k, v)| k.is_ok() && v.is_ok())
95        .map(|(k, v)| (k.unwrap(), v.unwrap()))
96        .collect();
97
98    let client = request.bucket.http_client();
99    let method = match request.command.http_verb() {
100        HttpMethod::Delete => reqwest::Method::DELETE,
101        HttpMethod::Get => reqwest::Method::GET,
102        HttpMethod::Post => reqwest::Method::POST,
103        HttpMethod::Put => reqwest::Method::PUT,
104        HttpMethod::Head => reqwest::Method::HEAD,
105    };
106
107    let request_builder = client
108        .request(method, request.url()?.as_str())
109        .headers(headers)
110        .body(request.request_body()?);
111    let request_builder = if let Some(timeout) = request.bucket.request_timeout {
112        request_builder.timeout(timeout)
113    } else {
114        request_builder
115    };
116    let response = client.execute(request_builder.build()?).await?;
117
118    if cfg!(feature = "fail-on-err") && !response.status().is_success() {
119        let status = response.status().as_u16();
120        let text = response.text().await.map_err(|error| {
121            crate::request::RequestAttemptError::do_not_retry(S3Error::Reqwest(error))
122        })?;
123        return Err(S3Error::HttpFailWithBody(status, text).into());
124    }
125
126    Ok(response)
127}
128
129#[maybe_async]
130impl<'a> Request for ReqwestRequest<'a> {
131    type Response = reqwest::Response;
132    type HeaderMap = reqwest::header::HeaderMap;
133
134    async fn response(&self) -> Result<Self::Response, S3Error> {
135        request_response_attempt(self)
136            .await
137            .map_err(|error| error.error)
138    }
139
140    async fn response_status(&self) -> Result<u16, S3Error> {
141        crate::request::retry_request(&self.command, || async {
142            let headers = self
143                .headers()
144                .await?
145                .iter()
146                .map(|(k, v)| {
147                    (
148                        reqwest::header::HeaderName::from_str(k.as_str()),
149                        reqwest::header::HeaderValue::from_str(v.to_str().unwrap_or_default()),
150                    )
151                })
152                .filter(|(k, v)| k.is_ok() && v.is_ok())
153                .map(|(k, v)| (k.unwrap(), v.unwrap()))
154                .collect();
155
156            let client = self.bucket.http_client();
157
158            let method = match self.command.http_verb() {
159                HttpMethod::Delete => reqwest::Method::DELETE,
160                HttpMethod::Get => reqwest::Method::GET,
161                HttpMethod::Post => reqwest::Method::POST,
162                HttpMethod::Put => reqwest::Method::PUT,
163                HttpMethod::Head => reqwest::Method::HEAD,
164            };
165
166            let request = client
167                .request(method, self.url()?.as_str())
168                .headers(headers)
169                .body(self.request_body()?);
170
171            let request = if let Some(timeout) = self.bucket.request_timeout {
172                request.timeout(timeout)
173            } else {
174                request
175            };
176
177            let request = request.build()?;
178            let response = client.execute(request).await.map_err(S3Error::from)?;
179            let status = response.status().as_u16();
180
181            if status == 404 {
182                return Ok(status);
183            }
184
185            if cfg!(feature = "fail-on-err") && !response.status().is_success() {
186                let text = response.text().await.map_err(|error| {
187                    crate::request::RequestAttemptError::do_not_retry(S3Error::Reqwest(error))
188                })?;
189                return Err(S3Error::HttpFailWithBody(status, text).into());
190            }
191
192            Ok(status)
193        })
194        .await
195    }
196
197    async fn response_data(&self, etag: bool) -> Result<ResponseData, S3Error> {
198        let response =
199            crate::request::retry_request(&self.command, || request_response_attempt(self)).await?;
200        let status_code = response.status().as_u16();
201        let mut headers = response.headers().clone();
202        let response_headers = headers
203            .clone()
204            .iter()
205            .map(|(k, v)| {
206                (
207                    k.to_string(),
208                    v.to_str()
209                        .unwrap_or("could-not-decode-header-value")
210                        .to_string(),
211                )
212            })
213            .collect::<HashMap<String, String>>();
214        // When etag=true, we extract the ETag header and return it as the body.
215        // This is used for PUT operations (regular puts, multipart chunks) where:
216        // 1. S3 returns an empty or non-useful response body
217        // 2. The ETag header contains the essential information we need
218        // 3. The calling code expects to get the ETag via response_data.as_str()
219        //
220        // Note: This approach means we discard any actual response body when etag=true,
221        // but for the operations that use this (PUTs), the body is typically empty
222        // or contains redundant information already available in headers.
223        //
224        // TODO: Refactor this to properly return the response body and access ETag
225        // from headers instead of replacing the body. This would be a breaking change.
226        let body_vec = if etag {
227            if let Some(etag) = headers.remove("ETag") {
228                Bytes::from(etag.to_str()?.to_string())
229            } else {
230                Bytes::from("")
231            }
232        } else {
233            response.bytes().await?
234        };
235        Ok(ResponseData::new(body_vec, status_code, response_headers))
236    }
237
238    async fn response_data_to_writer<T: tokio::io::AsyncWrite + Send + Unpin + ?Sized>(
239        &self,
240        writer: &mut T,
241    ) -> Result<u16, S3Error> {
242        use tokio::io::AsyncWriteExt;
243        let response =
244            crate::request::retry_request(&self.command, || request_response_attempt(self)).await?;
245
246        let status_code = response.status();
247        let mut stream = response.bytes_stream();
248
249        while let Some(item) = stream.next().await {
250            writer.write_all(&item?).await?;
251        }
252
253        Ok(status_code.as_u16())
254    }
255
256    async fn response_data_to_stream(&self) -> Result<ResponseDataStream, S3Error> {
257        let response =
258            crate::request::retry_request(&self.command, || request_response_attempt(self)).await?;
259        let status_code = response.status();
260        let stream = response.bytes_stream().map_err(S3Error::Reqwest);
261
262        Ok(ResponseDataStream {
263            bytes: Box::pin(stream),
264            status_code: status_code.as_u16(),
265        })
266    }
267
268    async fn response_header(&self) -> Result<(Self::HeaderMap, u16), S3Error> {
269        let response =
270            crate::request::retry_request(&self.command, || request_response_attempt(self)).await?;
271        let status_code = response.status().as_u16();
272        let headers = response.headers().clone();
273        Ok((headers, status_code))
274    }
275
276    fn datetime(&self) -> OffsetDateTime {
277        self.datetime
278    }
279
280    fn bucket(&self) -> Bucket {
281        self.bucket.clone()
282    }
283
284    fn command(&self) -> Command<'_> {
285        self.command.clone()
286    }
287
288    fn path(&self) -> String {
289        self.path.to_string()
290    }
291}
292
293impl<'a> ReqwestRequest<'a> {
294    pub async fn new(
295        bucket: &'a Bucket,
296        path: &'a str,
297        command: Command<'a>,
298    ) -> Result<ReqwestRequest<'a>, S3Error> {
299        bucket.credentials_refresh().await?;
300        Ok(Self {
301            bucket,
302            path,
303            command,
304            datetime: now_utc(),
305            sync: false,
306        })
307    }
308}
309
310#[cfg(test)]
311mod tests {
312    use crate::bucket::Bucket;
313    use crate::command::Command;
314    use crate::request::Request;
315    use crate::request::tokio_backend::ReqwestRequest;
316    use awscreds::Credentials;
317    use http::header::{HOST, RANGE};
318    use std::io::{Read, Write};
319    use std::net::TcpListener;
320    use std::sync::mpsc;
321    use std::thread;
322    use std::time::Duration;
323    use tokio_stream::StreamExt;
324
325    // Fake keys - otherwise using Credentials::default will use actual user
326    // credentials if they exist.
327    fn fake_credentials() -> Credentials {
328        let access_key = "AKIAIOSFODNN7EXAMPLE";
329        let secert_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY";
330        Credentials::new(Some(access_key), Some(secert_key), None, None, None).unwrap()
331    }
332
333    fn timeout_test_bucket(endpoint: String, timeout: Option<Duration>) -> Bucket {
334        let mut bucket = *Bucket::new(
335            "timeout-test",
336            crate::region::Region::Custom {
337                region: "us-east-1".to_owned(),
338                endpoint,
339            },
340            fake_credentials(),
341        )
342        .unwrap()
343        .with_path_style();
344        bucket.set_request_timeout(timeout);
345        bucket
346    }
347
348    fn delayed_http_response(
349        response: &'static [u8],
350        delay: Duration,
351    ) -> (String, mpsc::Sender<()>, thread::JoinHandle<()>) {
352        mock_http_response(response, None, delay)
353    }
354
355    fn stalled_http_body(
356        initial_response: &'static [u8],
357        trailing_body: &'static [u8],
358        delay: Duration,
359    ) -> (String, mpsc::Sender<()>, thread::JoinHandle<()>) {
360        mock_http_response(initial_response, Some(trailing_body), delay)
361    }
362
363    fn mock_http_response(
364        initial_response: &'static [u8],
365        trailing_body: Option<&'static [u8]>,
366        delay: Duration,
367    ) -> (String, mpsc::Sender<()>, thread::JoinHandle<()>) {
368        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
369        listener.set_nonblocking(true).unwrap();
370        let endpoint = format!("http://{}", listener.local_addr().unwrap());
371        let (stop_tx, stop_rx) = mpsc::channel();
372        let server = thread::spawn(move || {
373            let accept_deadline = std::time::Instant::now() + Duration::from_secs(8);
374            let (mut stream, _) = loop {
375                if !matches!(stop_rx.try_recv(), Err(mpsc::TryRecvError::Empty)) {
376                    return;
377                }
378                match listener.accept() {
379                    Ok(connection) => break connection,
380                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
381                        if std::time::Instant::now() >= accept_deadline {
382                            return;
383                        }
384                        thread::sleep(Duration::from_millis(5));
385                    }
386                    Err(_) => return,
387                }
388            };
389            stream.set_nonblocking(false).unwrap();
390            stream
391                .set_read_timeout(Some(Duration::from_secs(2)))
392                .unwrap();
393            stream
394                .set_write_timeout(Some(Duration::from_secs(2)))
395                .unwrap();
396            let mut request = Vec::new();
397            let mut byte = [0u8; 1];
398            while !request.ends_with(b"\r\n\r\n") {
399                if stream.read(&mut byte).unwrap_or(0) == 0 {
400                    return;
401                }
402                request.push(byte[0]);
403                assert!(
404                    request.len() < 16 * 1024,
405                    "request header exceeded fixture bound"
406                );
407            }
408            if trailing_body.is_none()
409                && !matches!(
410                    stop_rx.recv_timeout(delay),
411                    Err(mpsc::RecvTimeoutError::Timeout)
412                )
413            {
414                return;
415            }
416            if stream.write_all(initial_response).is_err() {
417                return;
418            }
419            if let Some(trailing_body) = trailing_body {
420                if !matches!(
421                    stop_rx.recv_timeout(delay),
422                    Err(mpsc::RecvTimeoutError::Timeout)
423                ) {
424                    return;
425                }
426                let _ = stream.write_all(trailing_body);
427            }
428            let _ = stream.flush();
429        });
430        (endpoint, stop_tx, server)
431    }
432
433    fn timeout_then_success_status_server() -> (String, mpsc::Sender<()>, thread::JoinHandle<usize>)
434    {
435        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
436        listener.set_nonblocking(true).unwrap();
437        let endpoint = format!("http://{}", listener.local_addr().unwrap());
438        let (stop_tx, stop_rx) = mpsc::channel();
439        let server = thread::spawn(move || {
440            let mut attempts = 0;
441            let deadline = std::time::Instant::now() + Duration::from_secs(8);
442            let Some(mut first) = accept_mock_request(&listener, &stop_rx, deadline) else {
443                return attempts;
444            };
445            attempts += 1;
446            if !read_mock_request(&mut first) {
447                return attempts;
448            }
449            let response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n";
450            let first_response_deadline = std::time::Instant::now() + Duration::from_secs(3);
451            loop {
452                if !matches!(stop_rx.try_recv(), Err(mpsc::TryRecvError::Empty)) {
453                    return attempts;
454                }
455                match listener.accept() {
456                    Ok((mut second, _)) => {
457                        let _ = second.set_nonblocking(false);
458                        let _ = second.set_read_timeout(Some(Duration::from_secs(2)));
459                        let _ = second.set_write_timeout(Some(Duration::from_secs(2)));
460                        attempts += 1;
461                        if read_mock_request(&mut second) {
462                            let _ = second.write_all(response);
463                            let _ = second.flush();
464                        }
465                        return attempts;
466                    }
467                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
468                        if std::time::Instant::now() >= first_response_deadline {
469                            let _ = first.write_all(response);
470                            let _ = first.flush();
471                            return attempts;
472                        }
473                        thread::sleep(Duration::from_millis(5));
474                    }
475                    Err(_) => return attempts,
476                }
477            }
478        });
479        (endpoint, stop_tx, server)
480    }
481
482    fn accept_mock_request(
483        listener: &TcpListener,
484        stop: &mpsc::Receiver<()>,
485        deadline: std::time::Instant,
486    ) -> Option<std::net::TcpStream> {
487        loop {
488            if !matches!(stop.try_recv(), Err(mpsc::TryRecvError::Empty))
489                || std::time::Instant::now() >= deadline
490            {
491                return None;
492            }
493            match listener.accept() {
494                Ok((stream, _)) => {
495                    stream.set_nonblocking(false).ok()?;
496                    let _ = stream.set_read_timeout(Some(Duration::from_secs(2)));
497                    let _ = stream.set_write_timeout(Some(Duration::from_secs(2)));
498                    return Some(stream);
499                }
500                Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
501                    thread::sleep(Duration::from_millis(5));
502                }
503                Err(_) => return None,
504            }
505        }
506    }
507
508    fn read_mock_request(stream: &mut std::net::TcpStream) -> bool {
509        let mut request = Vec::new();
510        let mut byte = [0u8; 1];
511        while !request.ends_with(b"\r\n\r\n") {
512            match stream.read(&mut byte) {
513                Ok(0) | Err(_) => return false,
514                Ok(_) => request.push(byte[0]),
515            }
516            if request.len() >= 16 * 1024 {
517                return false;
518            }
519        }
520        true
521    }
522
523    #[tokio::test]
524    async fn request_timeout_covers_delayed_response_headers() {
525        let (endpoint, stop, server) = delayed_http_response(
526            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
527            Duration::from_millis(1500),
528        );
529        let bucket = timeout_test_bucket(endpoint, Some(Duration::from_millis(500)));
530        let request = ReqwestRequest::new(&bucket, "/object", Command::GetObject)
531            .await
532            .unwrap();
533
534        let result = tokio::time::timeout(Duration::from_secs(5), request.response()).await;
535        let _ = stop.send(());
536        server.join().unwrap();
537        let result = result.expect("request should finish within the fixture bound");
538        assert!(
539            matches!(&result, Err(crate::error::S3Error::Reqwest(error)) if error.is_timeout()),
540            "delayed headers should exceed the request timeout"
541        );
542    }
543
544    #[tokio::test]
545    async fn request_timeout_covers_body_stream_and_writer_consumption() {
546        let (endpoint, stop, server) = stalled_http_body(
547            b"HTTP/1.1 200 OK\r\nContent-Length: 6\r\nConnection: close\r\n\r\nabc",
548            b"def",
549            Duration::from_millis(1500),
550        );
551        let bucket = timeout_test_bucket(endpoint, Some(Duration::from_millis(500)));
552        let mut body_stream = bucket.get_object_stream("/object").await.unwrap();
553        let mut prefix = Vec::new();
554        while prefix.len() < 3 {
555            let chunk = tokio::time::timeout(Duration::from_secs(1), body_stream.bytes.next())
556                .await
557                .expect("first body bytes should arrive promptly")
558                .expect("response body should contain its initial bytes")
559                .unwrap();
560            prefix.extend_from_slice(&chunk);
561        }
562        assert_eq!(&prefix[..3], b"abc");
563        let next = tokio::time::timeout(Duration::from_secs(2), body_stream.bytes.next())
564            .await
565            .expect("body timeout should be bounded");
566        assert!(matches!(
567            next.expect("body should report its timeout"),
568            Err(crate::error::S3Error::Reqwest(ref error)) if error.is_timeout()
569        ));
570        let _ = stop.send(());
571        server.join().unwrap();
572
573        let (endpoint, stop, server) = stalled_http_body(
574            b"HTTP/1.1 200 OK\r\nContent-Length: 6\r\nConnection: close\r\n\r\nabc",
575            b"def",
576            Duration::from_millis(1500),
577        );
578        let bucket = timeout_test_bucket(endpoint, Some(Duration::from_millis(500)));
579        let mut writer = Vec::new();
580        let result = tokio::time::timeout(
581            Duration::from_secs(2),
582            bucket.get_object_to_writer("/object", &mut writer),
583        )
584        .await;
585        let _ = stop.send(());
586        server.join().unwrap();
587        let result = result.expect("writer timeout should be bounded");
588        assert!(matches!(
589            result,
590            Err(crate::error::S3Error::Reqwest(ref error)) if error.is_timeout()
591        ));
592        assert_eq!(writer, b"abc", "the complete prefix should be written once");
593    }
594
595    #[tokio::test]
596    async fn request_timeout_can_be_removed_after_bucket_creation() {
597        let (endpoint, stop, server) = delayed_http_response(
598            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
599            Duration::from_millis(1500),
600        );
601        let mut bucket = *timeout_test_bucket(endpoint, None)
602            .with_request_timeout(Duration::from_millis(500))
603            .unwrap();
604        bucket.set_request_timeout(None);
605        let request = ReqwestRequest::new(&bucket, "/object", Command::GetObject)
606            .await
607            .unwrap();
608
609        let response = tokio::time::timeout(Duration::from_secs(5), request.response()).await;
610        let _ = stop.send(());
611        server.join().unwrap();
612        let response = response
613            .expect("request should finish within the fixture bound")
614            .expect("None should remove the prior per-request timeout");
615        assert_eq!(response.status(), reqwest::StatusCode::OK);
616    }
617
618    #[tokio::test]
619    async fn response_status_applies_timeout_and_retries_the_request() {
620        let (endpoint, stop, server) = timeout_then_success_status_server();
621        let bucket = timeout_test_bucket(endpoint, Some(Duration::from_millis(500)));
622        let request = ReqwestRequest::new(&bucket, "/object", Command::GetObject)
623            .await
624            .unwrap();
625
626        let status = tokio::time::timeout(Duration::from_secs(5), request.response_status()).await;
627        let _ = stop.send(());
628        let attempts = server.join().unwrap();
629        let status = status
630            .expect("status retry should remain bounded")
631            .expect("the second response should succeed");
632        assert_eq!(status, 200);
633        assert_eq!(attempts, 2);
634    }
635
636    #[tokio::test]
637    async fn url_uses_https_by_default() {
638        let region = "custom-region".parse().unwrap();
639        let bucket = Bucket::new("my-first-bucket", region, fake_credentials()).unwrap();
640        let path = "/my-first/path";
641        let request = ReqwestRequest::new(&bucket, path, Command::GetObject)
642            .await
643            .unwrap();
644
645        assert_eq!(request.url().unwrap().scheme(), "https");
646
647        let headers = request.headers().await.unwrap();
648        let host = headers.get(HOST).unwrap();
649
650        assert_eq!(*host, "my-first-bucket.custom-region".to_string());
651    }
652
653    #[tokio::test]
654    async fn url_uses_https_by_default_path_style() {
655        let region = "custom-region".parse().unwrap();
656        let bucket = Bucket::new("my-first-bucket", region, fake_credentials())
657            .unwrap()
658            .with_path_style();
659        let path = "/my-first/path";
660        let request = ReqwestRequest::new(&bucket, path, Command::GetObject)
661            .await
662            .unwrap();
663
664        assert_eq!(request.url().unwrap().scheme(), "https");
665
666        let headers = request.headers().await.unwrap();
667        let host = headers.get(HOST).unwrap();
668
669        assert_eq!(*host, "custom-region".to_string());
670    }
671
672    #[tokio::test]
673    async fn url_uses_scheme_from_custom_region_if_defined() {
674        let region = "http://custom-region".parse().unwrap();
675        let bucket = Bucket::new("my-second-bucket", region, fake_credentials()).unwrap();
676        let path = "/my-second/path";
677        let request = ReqwestRequest::new(&bucket, path, Command::GetObject)
678            .await
679            .unwrap();
680
681        assert_eq!(request.url().unwrap().scheme(), "http");
682
683        let headers = request.headers().await.unwrap();
684        let host = headers.get(HOST).unwrap();
685        assert_eq!(*host, "my-second-bucket.custom-region".to_string());
686    }
687
688    #[tokio::test]
689    async fn url_uses_scheme_from_custom_region_if_defined_with_path_style() {
690        let region = "http://custom-region".parse().unwrap();
691        let bucket = Bucket::new("my-second-bucket", region, fake_credentials())
692            .unwrap()
693            .with_path_style();
694        let path = "/my-second/path";
695        let request = ReqwestRequest::new(&bucket, path, Command::GetObject)
696            .await
697            .unwrap();
698
699        assert_eq!(request.url().unwrap().scheme(), "http");
700
701        let headers = request.headers().await.unwrap();
702        let host = headers.get(HOST).unwrap();
703        assert_eq!(*host, "custom-region".to_string());
704    }
705
706    #[tokio::test]
707    async fn test_get_object_range_header() {
708        let region = "http://custom-region".parse().unwrap();
709        let bucket = Bucket::new("my-second-bucket", region, fake_credentials())
710            .unwrap()
711            .with_path_style();
712        let path = "/my-second/path";
713
714        let request = ReqwestRequest::new(
715            &bucket,
716            path,
717            Command::GetObjectRange {
718                start: 0,
719                end: None,
720            },
721        )
722        .await
723        .unwrap();
724        let headers = request.headers().await.unwrap();
725        let range = headers.get(RANGE).unwrap();
726        assert_eq!(range, "bytes=0-");
727        assert!(!headers.contains_key("Content-Length"));
728        assert!(!headers.contains_key("Content-Type"));
729        assert_eq!(headers.get("Accept").unwrap(), "application/octet-stream");
730        let authorization = headers.get("Authorization").unwrap().to_str().unwrap();
731        let signed_headers = authorization
732            .split("SignedHeaders=")
733            .nth(1)
734            .and_then(|value| value.split(',').next())
735            .unwrap();
736        assert!(signed_headers.contains("range"));
737        assert!(!signed_headers.contains("content-length"));
738        assert!(!signed_headers.contains("content-type"));
739
740        let request = ReqwestRequest::new(
741            &bucket,
742            path,
743            Command::GetObjectRange {
744                start: 0,
745                end: Some(1),
746            },
747        )
748        .await
749        .unwrap();
750        let headers = request.headers().await.unwrap();
751        let range = headers.get(RANGE).unwrap();
752        assert_eq!(range, "bytes=0-1");
753        assert!(!headers.contains_key("Content-Length"));
754        assert!(!headers.contains_key("Content-Type"));
755        let authorization = headers.get("Authorization").unwrap().to_str().unwrap();
756        let signed_headers = authorization
757            .split("SignedHeaders=")
758            .nth(1)
759            .and_then(|value| value.split(',').next())
760            .unwrap();
761        assert!(signed_headers.contains("range"));
762        assert!(!signed_headers.contains("content-length"));
763        assert!(!signed_headers.contains("content-type"));
764
765        let get = ReqwestRequest::new(&bucket, path, Command::GetObject)
766            .await
767            .unwrap();
768        let headers = get.headers().await.unwrap();
769        assert!(!headers.contains_key("Content-Length"));
770        assert!(!headers.contains_key("Content-Type"));
771
772        let put = ReqwestRequest::new(
773            &bucket,
774            path,
775            Command::PutObject {
776                content: b"abc",
777                content_type: "application/test",
778                custom_headers: None,
779                multipart: None,
780            },
781        )
782        .await
783        .unwrap();
784        let headers = put.headers().await.unwrap();
785        assert_eq!(headers.get("Content-Length").unwrap(), "3");
786        assert_eq!(headers.get("Content-Type").unwrap(), "application/test");
787        let authorization = headers.get("Authorization").unwrap().to_str().unwrap();
788        let signed_headers = authorization
789            .split("SignedHeaders=")
790            .nth(1)
791            .and_then(|value| value.split(',').next())
792            .unwrap();
793        assert!(signed_headers.contains("content-length"));
794        assert!(signed_headers.contains("content-type"));
795    }
796}
s3/request/tokio_backend.rs

s3/request/
tokio_backend.rs
