s3/request/
request_trait.rs

1use base64::Engine;
2use base64::engine::general_purpose;
3use hmac::Mac;
4use quick_xml::se::to_string;
5use std::collections::HashMap;
6#[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
7use std::pin::Pin;
8use time::OffsetDateTime;
9use time::format_description::well_known::Rfc2822;
10use url::Url;
11
12use crate::LONG_DATETIME;
13use crate::bucket::Bucket;
14use crate::command::Command;
15use crate::error::S3Error;
16use crate::signing;
17use bytes::Bytes;
18use http::HeaderMap;
19use http::header::{
20    ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE, DATE, HOST, HeaderName, RANGE,
21};
22use std::fmt::Write as _;
23
24#[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
25use futures_util::Stream;
26#[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
27use futures_util::stream::{self, StreamExt};
28
29#[derive(Debug)]
30
31pub struct ResponseData {
32    bytes: Bytes,
33    status_code: u16,
34    headers: HashMap<String, String>,
35}
36
37#[cfg(all(
38    test,
39    any(feature = "sync", feature = "with-tokio", feature = "with-async-std")
40))]
41mod generated_bodyless_header_tests {
42    use super::Request;
43    use crate::bucket::Bucket;
44    use crate::command::Command;
45    use crate::error::S3Error;
46    use crate::region::Region;
47    use awscreds::Credentials;
48    use http::header::{AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE};
49
50    #[cfg(feature = "sync")]
51    type BackendRequest<'a> = crate::request::blocking::AttoRequest<'a>;
52    #[cfg(all(not(feature = "sync"), feature = "with-tokio"))]
53    type BackendRequest<'a> = crate::request::tokio_backend::ReqwestRequest<'a>;
54    #[cfg(all(
55        not(feature = "sync"),
56        not(feature = "with-tokio"),
57        feature = "with-async-std"
58    ))]
59    type BackendRequest<'a> = crate::request::async_std_backend::SurfRequest<'a>;
60
61    #[maybe_async::maybe_async]
62    async fn make_request<'a>(
63        bucket: &'a Bucket,
64        command: Command<'a>,
65    ) -> Result<BackendRequest<'a>, S3Error> {
66        BackendRequest::new(bucket, "key", command).await
67    }
68
69    #[maybe_async::test(
70        feature = "sync",
71        async(all(not(feature = "sync"), feature = "with-tokio"), tokio::test),
72        async(
73            all(not(feature = "sync"), feature = "with-async-std"),
74            async_std::test
75        )
76    )]
77    async fn custom_endpoint_path_stays_in_url_but_not_host_authority() {
78        use http::header::{AUTHORIZATION, HOST};
79        use url::Url;
80
81        let credentials =
82            Credentials::new(Some("test-key"), Some("test-secret"), None, None, None).unwrap();
83        let bucket = Bucket::new(
84            "bucket",
85            Region::Custom {
86                region: "us-east-1".to_owned(),
87                endpoint: "https://project.example:8443/storage/v1/s3".to_owned(),
88            },
89            credentials,
90        )
91        .unwrap()
92        .with_path_style();
93
94        let request = make_request(&bucket, Command::GetObject).await.unwrap();
95        let request_url = request.url().unwrap();
96        assert_eq!(
97            request_url.as_str(),
98            "https://project.example:8443/storage/v1/s3/bucket/key"
99        );
100
101        let headers = request.headers().await.unwrap();
102        assert_eq!(headers[HOST], "project.example:8443");
103        let mut signing_headers = headers.clone();
104        signing_headers.remove(AUTHORIZATION);
105        signing_headers.remove(http::header::DATE);
106        let canonical_request = request.canonical_request(&signing_headers).unwrap();
107        let canonical_lines = canonical_request.lines().collect::<Vec<_>>();
108        assert_eq!(canonical_lines[0], "GET");
109        assert_eq!(canonical_lines[1], "/storage/v1/s3/bucket/key");
110        assert!(canonical_request.contains("\nhost:project.example:8443\n"));
111        assert!(!canonical_request.contains("\nhost:project.example:8443/storage"));
112        let signed_headers = headers[AUTHORIZATION]
113            .to_str()
114            .unwrap()
115            .split("SignedHeaders=")
116            .nth(1)
117            .unwrap()
118            .split(',')
119            .next()
120            .unwrap();
121        assert!(signed_headers.split(';').any(|header| header == "host"));
122
123        let presigned = bucket.presign_get("/key", 3600, None).await.unwrap();
124        let presigned_url = Url::parse(&presigned).unwrap();
125        assert_eq!(presigned_url.host_str(), Some("project.example"));
126        assert_eq!(presigned_url.port(), Some(8443));
127        assert_eq!(presigned_url.path(), "/storage/v1/s3/bucket/key");
128        assert!(
129            presigned_url
130                .query_pairs()
131                .any(|(key, _)| key == "X-Amz-Signature")
132        );
133    }
134
135    #[maybe_async::test(
136        feature = "sync",
137        async(all(not(feature = "sync"), feature = "with-tokio"), tokio::test),
138        async(
139            all(not(feature = "sync"), feature = "with-async-std"),
140            async_std::test
141        )
142    )]
143    async fn delete_object_does_not_sign_generated_body_headers() {
144        let credentials =
145            Credentials::new(Some("test-key"), Some("test-secret"), None, None, None).unwrap();
146        let mut bucket = Bucket::new(
147            "bucket",
148            "us-east-1".parse::<Region>().unwrap(),
149            credentials,
150        )
151        .unwrap();
152        bucket.add_header("x-test-caller-extra", "kept");
153        let request = make_request(&bucket, Command::DeleteObject).await.unwrap();
154        let headers = request.headers().await.unwrap();
155
156        assert!(!headers.contains_key(CONTENT_LENGTH));
157        assert!(!headers.contains_key(CONTENT_TYPE));
158        assert_eq!(headers["x-test-caller-extra"], "kept");
159        let authorization = headers[AUTHORIZATION].to_str().unwrap();
160        assert!(!authorization.contains("content-length"));
161        assert!(!authorization.contains("content-type"));
162        assert!(authorization.contains("x-test-caller-extra"));
163
164        for command in [Command::GetObject, Command::HeadObject] {
165            let request = make_request(&bucket, command).await.unwrap();
166            let headers = request.headers().await.unwrap();
167            assert!(!headers.contains_key(CONTENT_LENGTH));
168            assert!(!headers.contains_key(CONTENT_TYPE));
169        }
170
171        bucket.add_header("content-length", "0");
172        bucket.add_header("content-type", "application/caller");
173        let request = make_request(&bucket, Command::DeleteObject).await.unwrap();
174        let headers = request.headers().await.unwrap();
175        assert_eq!(headers[CONTENT_LENGTH], "0");
176        assert_eq!(headers[CONTENT_TYPE], "application/caller");
177    }
178
179    #[maybe_async::test(
180        feature = "sync",
181        async(all(not(feature = "sync"), feature = "with-tokio"), tokio::test),
182        async(
183            all(not(feature = "sync"), feature = "with-async-std"),
184            async_std::test
185        )
186    )]
187    async fn generated_body_headers_remain_for_put_and_xml_post() {
188        let credentials =
189            Credentials::new(Some("test-key"), Some("test-secret"), None, None, None).unwrap();
190        let bucket = Bucket::new(
191            "bucket",
192            "us-east-1".parse::<Region>().unwrap(),
193            credentials,
194        )
195        .unwrap();
196
197        let empty_put = Command::PutObject {
198            content: b"",
199            content_type: "application/test",
200            custom_headers: None,
201            multipart: None,
202        };
203        let request = make_request(&bucket, empty_put).await.unwrap();
204        let headers = request.headers().await.unwrap();
205        assert_eq!(headers[CONTENT_LENGTH], "0");
206        assert_eq!(headers[CONTENT_TYPE], "application/test");
207
208        let put = Command::PutObject {
209            content: b"abc",
210            content_type: "application/test",
211            custom_headers: None,
212            multipart: None,
213        };
214        let request = make_request(&bucket, put).await.unwrap();
215        let headers = request.headers().await.unwrap();
216        assert_eq!(headers[CONTENT_LENGTH], "3");
217        assert_eq!(headers[CONTENT_TYPE], "application/test");
218        assert!(
219            headers[AUTHORIZATION]
220                .to_str()
221                .unwrap()
222                .contains("content-length")
223        );
224        assert!(
225            headers[AUTHORIZATION]
226                .to_str()
227                .unwrap()
228                .contains("content-type")
229        );
230
231        let request = make_request(&bucket, Command::CopyObject { from: "/source" })
232            .await
233            .unwrap();
234        let headers = request.headers().await.unwrap();
235        assert!(!headers.contains_key(CONTENT_LENGTH));
236        assert!(!headers.contains_key(CONTENT_TYPE));
237        assert_eq!(headers["x-amz-copy-source"], "/source");
238
239        let post = Command::CompleteMultipartUpload {
240            upload_id: "upload-id",
241            data: crate::serde_types::CompleteMultipartUploadData {
242                parts: vec![crate::serde_types::Part {
243                    part_number: 1,
244                    etag: "etag".into(),
245                }],
246            },
247        };
248        let expected_length = post.content_length().unwrap();
249        let request = make_request(&bucket, post).await.unwrap();
250        let headers = request.headers().await.unwrap();
251        assert_eq!(headers[CONTENT_LENGTH], expected_length.to_string());
252        assert_eq!(headers[CONTENT_TYPE], "application/xml");
253        assert!(
254            headers[AUTHORIZATION]
255                .to_str()
256                .unwrap()
257                .contains("content-length")
258        );
259        assert!(
260            headers[AUTHORIZATION]
261                .to_str()
262                .unwrap()
263                .contains("content-type")
264        );
265    }
266}
267
268#[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
269pub type DataStream = Pin<Box<dyn Stream<Item = StreamItem> + Send>>;
270#[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
271pub type StreamItem = Result<Bytes, S3Error>;
272
273#[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
274pub struct ResponseDataStream {
275    pub bytes: DataStream,
276    pub status_code: u16,
277}
278
279#[cfg(any(feature = "with-tokio", feature = "with-async-std"))]
280impl ResponseDataStream {
281    pub fn bytes(&mut self) -> &mut DataStream {
282        &mut self.bytes
283    }
284}
285
286impl From<ResponseData> for Vec<u8> {
287    fn from(data: ResponseData) -> Vec<u8> {
288        data.to_vec()
289    }
290}
291
292impl ResponseData {
293    pub fn new(bytes: Bytes, status_code: u16, headers: HashMap<String, String>) -> ResponseData {
294        ResponseData {
295            bytes,
296            status_code,
297            headers,
298        }
299    }
300
301    pub fn as_slice(&self) -> &[u8] {
302        &self.bytes
303    }
304
305    pub fn to_vec(self) -> Vec<u8> {
306        self.bytes.to_vec()
307    }
308
309    pub fn bytes(&self) -> &Bytes {
310        &self.bytes
311    }
312
313    pub fn bytes_mut(&mut self) -> &mut Bytes {
314        &mut self.bytes
315    }
316
317    pub fn into_bytes(self) -> Bytes {
318        self.bytes
319    }
320
321    pub fn status_code(&self) -> u16 {
322        self.status_code
323    }
324
325    pub fn as_str(&self) -> Result<&str, std::str::Utf8Error> {
326        std::str::from_utf8(self.as_slice())
327    }
328
329    pub fn to_string(&self) -> Result<String, std::str::Utf8Error> {
330        std::str::from_utf8(self.as_slice()).map(|s| s.to_string())
331    }
332
333    pub fn headers(&self) -> HashMap<String, String> {
334        self.headers.clone()
335    }
336}
337
338use std::fmt;
339
340impl fmt::Display for ResponseData {
341    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
342        write!(
343            f,
344            "Status code: {}\n Data: {}",
345            self.status_code(),
346            self.to_string()
347                .unwrap_or_else(|_| "Data could not be cast to UTF string".to_string())
348        )
349    }
350}
351
352#[cfg(feature = "with-tokio")]
353impl tokio::io::AsyncRead for ResponseDataStream {
354    fn poll_read(
355        mut self: Pin<&mut Self>,
356        cx: &mut std::task::Context<'_>,
357        buf: &mut tokio::io::ReadBuf<'_>,
358    ) -> std::task::Poll<std::io::Result<()>> {
359        if buf.remaining() == 0 {
360            return std::task::Poll::Ready(Ok(()));
361        }
362
363        loop {
364            match Stream::poll_next(self.bytes.as_mut(), cx) {
365                std::task::Poll::Ready(Some(Ok(chunk))) => {
366                    if chunk.is_empty() {
367                        continue;
368                    }
369
370                    let amt = std::cmp::min(chunk.len(), buf.remaining());
371                    buf.put_slice(&chunk[..amt]);
372
373                    if amt < chunk.len() {
374                        let remainder = chunk.slice(amt..);
375                        let previous_stream =
376                            std::mem::replace(&mut self.bytes, Box::pin(stream::empty()));
377                        self.bytes = Box::pin(
378                            stream::once(async move { Ok(remainder) }).chain(previous_stream),
379                        );
380                    }
381
382                    return std::task::Poll::Ready(Ok(()));
383                }
384                std::task::Poll::Ready(Some(Err(error))) => {
385                    return std::task::Poll::Ready(Err(std::io::Error::other(error)));
386                }
387                std::task::Poll::Ready(None) => {
388                    return std::task::Poll::Ready(Ok(()));
389                }
390                std::task::Poll::Pending => return std::task::Poll::Pending,
391            }
392        }
393    }
394}
395
396#[cfg(feature = "with-async-std")]
397impl async_std::io::Read for ResponseDataStream {
398    fn poll_read(
399        mut self: Pin<&mut Self>,
400        cx: &mut std::task::Context<'_>,
401        buf: &mut [u8],
402    ) -> std::task::Poll<std::io::Result<usize>> {
403        if buf.is_empty() {
404            return std::task::Poll::Ready(Ok(0));
405        }
406
407        loop {
408            match Stream::poll_next(self.bytes.as_mut(), cx) {
409                std::task::Poll::Ready(Some(Ok(chunk))) => {
410                    if chunk.is_empty() {
411                        continue;
412                    }
413
414                    let amt = std::cmp::min(chunk.len(), buf.len());
415                    buf[..amt].copy_from_slice(&chunk[..amt]);
416
417                    if amt < chunk.len() {
418                        let remainder = chunk.slice(amt..);
419                        let previous_stream =
420                            std::mem::replace(&mut self.bytes, Box::pin(stream::empty()));
421                        self.bytes = Box::pin(
422                            stream::once(async move { Ok(remainder) }).chain(previous_stream),
423                        );
424                    }
425
426                    return std::task::Poll::Ready(Ok(amt));
427                }
428                std::task::Poll::Ready(Some(Err(error))) => {
429                    return std::task::Poll::Ready(Err(std::io::Error::other(error)));
430                }
431                std::task::Poll::Ready(None) => {
432                    return std::task::Poll::Ready(Ok(0));
433                }
434                std::task::Poll::Pending => return std::task::Poll::Pending,
435            }
436        }
437    }
438}
439
440/// The authority of an endpoint host, dropping any path that follows it.
441///
442/// A custom endpoint may carry a path, as Supabase Storage's
443/// `https://<project>.supabase.co/storage/v1/s3` does, and `Region::host`
444/// keeps it so that `Bucket::url` can place the bucket and key underneath.
445/// RFC 9110 §7.2 defines `Host` as `uri-host [ ":" port ]`, so that path
446/// cannot travel in the header.
447fn authority_of(host: &str) -> &str {
448    match host.find('/') {
449        Some(position) => &host[..position],
450        None => host,
451    }
452}
453
454#[maybe_async::maybe_async]
455pub trait Request {
456    type Response;
457    type HeaderMap;
458
459    async fn response(&self) -> Result<Self::Response, S3Error>;
460    async fn response_data(&self, etag: bool) -> Result<ResponseData, S3Error>;
461    #[cfg(feature = "with-tokio")]
462    async fn response_data_to_writer<T: tokio::io::AsyncWrite + Send + Unpin + ?Sized>(
463        &self,
464        writer: &mut T,
465    ) -> Result<u16, S3Error>;
466    #[cfg(feature = "with-async-std")]
467    async fn response_data_to_writer<T: async_std::io::Write + Send + Unpin + ?Sized>(
468        &self,
469        writer: &mut T,
470    ) -> Result<u16, S3Error>;
471    #[cfg(feature = "sync")]
472    fn response_data_to_writer<T: std::io::Write + Send + ?Sized>(
473        &self,
474        writer: &mut T,
475    ) -> Result<u16, S3Error>;
476    #[cfg(any(feature = "with-async-std", feature = "with-tokio"))]
477    async fn response_data_to_stream(&self) -> Result<ResponseDataStream, S3Error>;
478    async fn response_header(&self) -> Result<(Self::HeaderMap, u16), S3Error>;
479    async fn response_status(&self) -> Result<u16, S3Error> {
480        let (_, status_code) = self.response_header().await?;
481        Ok(status_code)
482    }
483    fn datetime(&self) -> OffsetDateTime;
484    fn bucket(&self) -> Bucket;
485    fn command(&self) -> Command<'_>;
486    fn path(&self) -> String;
487
488    async fn signing_key(&self) -> Result<Vec<u8>, S3Error> {
489        signing::signing_key(
490            &self.datetime(),
491            &self
492                .bucket()
493                .secret_key()
494                .await?
495                .expect("Secret key must be provided to sign headers, found None"),
496            &self.bucket().region(),
497            "s3",
498        )
499    }
500
501    fn request_body(&self) -> Result<Vec<u8>, S3Error> {
502        let result = if let Command::PutObject { content, .. } = self.command() {
503            Vec::from(content)
504        } else if let Command::PutObjectTagging { tags } = self.command() {
505            Vec::from(tags)
506        } else if let Command::UploadPart { content, .. } = self.command() {
507            Vec::from(content)
508        } else if let Command::CompleteMultipartUpload { data, .. } = &self.command() {
509            let body = data.to_string();
510            body.as_bytes().to_vec()
511        } else if let Command::CreateBucket { config } = &self.command() {
512            if let Some(payload) = config.location_constraint_payload() {
513                Vec::from(payload)
514            } else {
515                Vec::new()
516            }
517        } else if let Command::PutBucketLifecycle { configuration, .. } = &self.command() {
518            quick_xml::se::to_string(configuration)?.as_bytes().to_vec()
519        } else if let Command::PutBucketCors { configuration, .. } = &self.command() {
520            let cors = configuration.to_string();
521            cors.as_bytes().to_vec()
522        } else if let Command::PutBucketPolicy { policy } = &self.command() {
523            policy.as_bytes().to_vec()
524        } else if let Command::DeleteObjects { data } = &self.command() {
525            data.to_string().as_bytes().to_vec()
526        } else {
527            Vec::new()
528        };
529        Ok(result)
530    }
531
532    fn long_date(&self) -> Result<String, S3Error> {
533        Ok(self.datetime().format(LONG_DATETIME)?)
534    }
535
536    fn string_to_sign(&self, request: &str) -> Result<String, S3Error> {
537        signing::string_to_sign(&self.datetime(), &self.bucket().region(), request)
538    }
539
540    fn host_header(&self) -> String {
541        authority_of(&self.bucket().host()).to_string()
542    }
543
544    #[maybe_async::async_impl]
545    async fn presigned(&self) -> Result<String, S3Error> {
546        let (expiry, custom_headers, custom_queries) = match self.command() {
547            Command::PresignGet {
548                expiry_secs,
549                custom_queries,
550            } => (expiry_secs, None, custom_queries),
551            Command::PresignPut {
552                expiry_secs,
553                custom_headers,
554                custom_queries,
555            } => (expiry_secs, custom_headers, custom_queries),
556            Command::PresignDelete { expiry_secs } => (expiry_secs, None, None),
557            _ => unreachable!(),
558        };
559
560        let url = self
561            .presigned_url_no_sig(expiry, custom_headers.as_ref(), custom_queries.as_ref())
562            .await?;
563
564        // Build the URL string preserving the original host (including standard ports)
565        // The Url type drops standard ports when converting to string, but we need them
566        // for signature validation
567        let url_str = if let awsregion::Region::Custom { ref endpoint, .. } = self.bucket().region()
568        {
569            // Check if we need to preserve a standard port
570            if (endpoint.contains(":80") && url.scheme() == "http" && url.port().is_none())
571                || (endpoint.contains(":443") && url.scheme() == "https" && url.port().is_none())
572            {
573                // Rebuild the URL with the original host from the endpoint
574                let host = self.bucket().host();
575                format!(
576                    "{}://{}{}{}",
577                    url.scheme(),
578                    host,
579                    url.path(),
580                    url.query().map(|q| format!("?{}", q)).unwrap_or_default()
581                )
582            } else {
583                url.to_string()
584            }
585        } else {
586            url.to_string()
587        };
588
589        Ok(format!(
590            "{}&X-Amz-Signature={}",
591            url_str,
592            self.presigned_authorization(custom_headers.as_ref())
593                .await?
594        ))
595    }
596
597    #[maybe_async::sync_impl]
598    async fn presigned(&self) -> Result<String, S3Error> {
599        let (expiry, custom_headers, custom_queries) = match self.command() {
600            Command::PresignGet {
601                expiry_secs,
602                custom_queries,
603            } => (expiry_secs, None, custom_queries),
604            Command::PresignPut {
605                expiry_secs,
606                custom_headers,
607                ..
608            } => (expiry_secs, custom_headers, None),
609            Command::PresignDelete { expiry_secs } => (expiry_secs, None, None),
610            _ => unreachable!(),
611        };
612
613        let url =
614            self.presigned_url_no_sig(expiry, custom_headers.as_ref(), custom_queries.as_ref())?;
615
616        // Build the URL string preserving the original host (including standard ports)
617        // The Url type drops standard ports when converting to string, but we need them
618        // for signature validation
619        let url_str = if let awsregion::Region::Custom { ref endpoint, .. } = self.bucket().region()
620        {
621            // Check if we need to preserve a standard port
622            if (endpoint.contains(":80") && url.scheme() == "http" && url.port().is_none())
623                || (endpoint.contains(":443") && url.scheme() == "https" && url.port().is_none())
624            {
625                // Rebuild the URL with the original host from the endpoint
626                let host = self.bucket().host();
627                format!(
628                    "{}://{}{}{}",
629                    url.scheme(),
630                    host,
631                    url.path(),
632                    url.query().map(|q| format!("?{}", q)).unwrap_or_default()
633                )
634            } else {
635                url.to_string()
636            }
637        } else {
638            url.to_string()
639        };
640
641        Ok(format!(
642            "{}&X-Amz-Signature={}",
643            url_str,
644            self.presigned_authorization(custom_headers.as_ref())?
645        ))
646    }
647
648    async fn presigned_authorization(
649        &self,
650        custom_headers: Option<&HeaderMap>,
651    ) -> Result<String, S3Error> {
652        let mut headers = HeaderMap::new();
653        let host_header = self.host_header();
654        headers.insert(HOST, host_header.parse()?);
655        if let Some(custom_headers) = custom_headers {
656            for (k, v) in custom_headers.iter() {
657                headers.insert(k.clone(), v.clone());
658            }
659        }
660        let canonical_request = self.presigned_canonical_request(&headers).await?;
661        let string_to_sign = self.string_to_sign(&canonical_request)?;
662        let mut hmac = signing::HmacSha256::new_from_slice(&self.signing_key().await?)?;
663        hmac.update(string_to_sign.as_bytes());
664        let signature = hex::encode(hmac.finalize().into_bytes());
665        // let signed_header = signing::signed_header_string(&headers);
666        Ok(signature)
667    }
668
669    async fn presigned_canonical_request(&self, headers: &HeaderMap) -> Result<String, S3Error> {
670        let (expiry, custom_headers, custom_queries) = match self.command() {
671            Command::PresignGet {
672                expiry_secs,
673                custom_queries,
674            } => (expiry_secs, None, custom_queries),
675            Command::PresignPut {
676                expiry_secs,
677                custom_headers,
678                custom_queries,
679            } => (expiry_secs, custom_headers, custom_queries),
680            Command::PresignDelete { expiry_secs } => (expiry_secs, None, None),
681            _ => unreachable!(),
682        };
683
684        signing::canonical_request(
685            &self.command().http_verb().to_string(),
686            &self
687                .presigned_url_no_sig(expiry, custom_headers.as_ref(), custom_queries.as_ref())
688                .await?,
689            headers,
690            "UNSIGNED-PAYLOAD",
691        )
692    }
693
694    #[maybe_async::async_impl]
695    async fn presigned_url_no_sig(
696        &self,
697        expiry: u32,
698        custom_headers: Option<&HeaderMap>,
699        custom_queries: Option<&HashMap<String, String>>,
700    ) -> Result<Url, S3Error> {
701        let bucket = self.bucket();
702        let token = if let Some(security_token) = bucket.security_token().await? {
703            Some(security_token)
704        } else {
705            bucket.session_token().await?
706        };
707        let url = Url::parse(&format!(
708            "{}{}{}",
709            self.url()?,
710            signing::authorization_query_params_no_sig(
711                &self.bucket().access_key().await?.unwrap_or_default(),
712                &self.datetime(),
713                &self.bucket().region(),
714                expiry,
715                custom_headers,
716                token.as_ref()
717            )?,
718            signing::flatten_queries(custom_queries)?,
719        ))?;
720
721        Ok(url)
722    }
723
724    #[maybe_async::sync_impl]
725    fn presigned_url_no_sig(
726        &self,
727        expiry: u32,
728        custom_headers: Option<&HeaderMap>,
729        custom_queries: Option<&HashMap<String, String>>,
730    ) -> Result<Url, S3Error> {
731        let bucket = self.bucket();
732        let token = if let Some(security_token) = bucket.security_token()? {
733            Some(security_token)
734        } else {
735            bucket.session_token()?
736        };
737        let url = Url::parse(&format!(
738            "{}{}{}",
739            self.url()?,
740            signing::authorization_query_params_no_sig(
741                &self.bucket().access_key()?.unwrap_or_default(),
742                &self.datetime(),
743                &self.bucket().region(),
744                expiry,
745                custom_headers,
746                token.as_ref()
747            )?,
748            signing::flatten_queries(custom_queries)?,
749        ))?;
750
751        Ok(url)
752    }
753
754    fn url(&self) -> Result<Url, S3Error> {
755        let mut url_str = self.bucket().url();
756
757        if let Command::ListBuckets { .. } = self.command() {
758            return Ok(Url::parse(&url_str)?);
759        }
760
761        if let Command::CreateBucket { .. } = self.command() {
762            return Ok(Url::parse(&url_str)?);
763        }
764
765        let path = if self.path().starts_with('/') {
766            self.path()[1..].to_string()
767        } else {
768            self.path()[..].to_string()
769        };
770
771        url_str.push('/');
772        url_str.push_str(&signing::uri_encode(&path, false));
773
774        // Append to url_path
775        #[allow(clippy::collapsible_match)]
776        match self.command() {
777            Command::InitiateMultipartUpload { .. } | Command::ListMultipartUploads { .. } => {
778                url_str.push_str("?uploads")
779            }
780            Command::AbortMultipartUpload { upload_id } => {
781                write!(url_str, "?uploadId={}", upload_id).expect("Could not write to url_str");
782            }
783            Command::CompleteMultipartUpload { upload_id, .. } => {
784                write!(url_str, "?uploadId={}", upload_id).expect("Could not write to url_str");
785            }
786            Command::GetObjectTorrent => url_str.push_str("?torrent"),
787            Command::PutObject { multipart, .. } => {
788                if let Some(multipart) = multipart {
789                    url_str.push_str(&multipart.query_string())
790                }
791            }
792            Command::GetBucketLifecycle
793            | Command::PutBucketLifecycle { .. }
794            | Command::DeleteBucketLifecycle => {
795                url_str.push_str("?lifecycle");
796            }
797            Command::GetBucketCors { .. }
798            | Command::PutBucketCors { .. }
799            | Command::DeleteBucketCors { .. } => {
800                url_str.push_str("?cors");
801            }
802            Command::GetBucketPolicy
803            | Command::PutBucketPolicy { .. }
804            | Command::DeleteBucketPolicy => {
805                url_str.push_str("?policy");
806            }
807            Command::GetObjectAttributes { version_id, .. } => {
808                if let Some(version_id) = version_id {
809                    url_str.push_str(&format!("?attributes&versionId={}", version_id));
810                } else {
811                    url_str.push_str("?attributes&versionId=null");
812                }
813            }
814            Command::HeadObject => {}
815            Command::DeleteObject => {}
816            Command::DeleteObjectTagging => {}
817            Command::GetObject => {}
818            Command::GetObjectRange { .. } => {}
819            Command::GetObjectTagging => {}
820            Command::ListObjects { .. } => {}
821            Command::ListObjectsV2 { .. } => {}
822            Command::GetBucketLocation => {}
823            Command::PresignGet { .. } => {}
824            Command::PresignPut { .. } => {}
825            Command::PresignDelete { .. } => {}
826            Command::DeleteBucket => {}
827            Command::ListBuckets => {}
828            Command::CopyObject { .. } => {}
829            Command::PutObjectTagging { .. } => {}
830            Command::UploadPart { .. } => {}
831            Command::CreateBucket { .. } => {}
832            Command::DeleteObjects { .. } => {
833                url_str.push_str("?delete");
834            }
835        }
836
837        let mut url = Url::parse(&url_str)?;
838
839        for (key, value) in &self.bucket().extra_query {
840            url.query_pairs_mut().append_pair(key, value);
841        }
842
843        if let Command::ListObjectsV2 {
844            prefix,
845            delimiter,
846            continuation_token,
847            start_after,
848            max_keys,
849        } = self.command().clone()
850        {
851            let mut query_pairs = url.query_pairs_mut();
852            delimiter.map(|d| query_pairs.append_pair("delimiter", &d));
853
854            query_pairs.append_pair("prefix", &prefix);
855            query_pairs.append_pair("list-type", "2");
856            if let Some(token) = continuation_token {
857                query_pairs.append_pair("continuation-token", &token);
858            }
859            if let Some(start_after) = start_after {
860                query_pairs.append_pair("start-after", &start_after);
861            }
862            if let Some(max_keys) = max_keys {
863                query_pairs.append_pair("max-keys", &max_keys.to_string());
864            }
865        }
866
867        if let Command::ListObjects {
868            prefix,
869            delimiter,
870            marker,
871            max_keys,
872        } = self.command().clone()
873        {
874            let mut query_pairs = url.query_pairs_mut();
875            delimiter.map(|d| query_pairs.append_pair("delimiter", &d));
876
877            query_pairs.append_pair("prefix", &prefix);
878            if let Some(marker) = marker {
879                query_pairs.append_pair("marker", &marker);
880            }
881            if let Some(max_keys) = max_keys {
882                query_pairs.append_pair("max-keys", &max_keys.to_string());
883            }
884        }
885
886        match self.command() {
887            Command::ListMultipartUploads {
888                prefix,
889                delimiter,
890                key_marker,
891                upload_id_marker,
892                max_uploads,
893            } => {
894                let mut query_pairs = url.query_pairs_mut();
895                delimiter.map(|d| query_pairs.append_pair("delimiter", d));
896                if let Some(prefix) = prefix {
897                    query_pairs.append_pair("prefix", prefix);
898                }
899                if let Some(key_marker) = key_marker {
900                    query_pairs.append_pair("key-marker", &key_marker);
901                }
902                if let Some(upload_id_marker) = upload_id_marker {
903                    query_pairs.append_pair("upload-id-marker", &upload_id_marker);
904                }
905                if let Some(max_uploads) = max_uploads {
906                    query_pairs.append_pair("max-uploads", max_uploads.to_string().as_str());
907                }
908            }
909            Command::PutObjectTagging { .. }
910            | Command::GetObjectTagging
911            | Command::DeleteObjectTagging => {
912                url.query_pairs_mut().append_pair("tagging", "");
913            }
914            _ => {}
915        }
916
917        Ok(url)
918    }
919
920    fn canonical_request(&self, headers: &HeaderMap) -> Result<String, S3Error> {
921        signing::canonical_request(
922            &self.command().http_verb().to_string(),
923            &self.url()?,
924            headers,
925            &self.command().sha256()?,
926        )
927    }
928
929    #[maybe_async::maybe_async]
930    async fn authorization(&self, headers: &HeaderMap) -> Result<String, S3Error> {
931        let canonical_request = self.canonical_request(headers)?;
932        let string_to_sign = self.string_to_sign(&canonical_request)?;
933        let mut hmac = signing::HmacSha256::new_from_slice(&self.signing_key().await?)?;
934        hmac.update(string_to_sign.as_bytes());
935        let signature = hex::encode(hmac.finalize().into_bytes());
936        let signed_header = signing::signed_header_string(headers);
937        signing::authorization_header(
938            &self
939                .bucket()
940                .access_key()
941                .await?
942                .expect("No access_key provided"),
943            &self.datetime(),
944            &self.bucket().region(),
945            &signed_header,
946            &signature,
947        )
948    }
949
950    #[maybe_async::maybe_async]
951    async fn headers(&self) -> Result<HeaderMap, S3Error> {
952        // Generate this once, but it's used in more than one place.
953        let sha256 = self.command().sha256()?;
954
955        // Start with extra_headers, that way our headers replace anything with
956        // the same name.
957
958        let mut headers = HeaderMap::new();
959
960        for (k, v) in self.bucket().extra_headers.iter() {
961            if k.as_str().starts_with("x-amz-meta-") {
962                // metadata is invalid on any multipart command other than initiate
963                match self.command() {
964                    Command::UploadPart { .. }
965                    | Command::AbortMultipartUpload { .. }
966                    | Command::CompleteMultipartUpload { .. }
967                    | Command::PutObject {
968                        multipart: Some(_), ..
969                    } => continue,
970                    _ => (),
971                }
972            }
973            headers.insert(k.clone(), v.clone());
974        }
975
976        // Append custom headers for PUT request if any
977        if let Command::PutObject { custom_headers, .. } = self.command()
978            && let Some(custom_headers) = custom_headers
979        {
980            for (k, v) in custom_headers.iter() {
981                headers.insert(k.clone(), v.clone());
982            }
983        }
984
985        let host_header = self.host_header();
986
987        headers.insert(HOST, host_header.parse()?);
988
989        if let Command::CopyObject { from } = self.command() {
990            headers.insert(HeaderName::from_static("x-amz-copy-source"), from.parse()?);
991        } else if !matches!(
992            self.command().http_verb(),
993            crate::command::HttpMethod::Get
994                | crate::command::HttpMethod::Head
995                | crate::command::HttpMethod::Delete
996        ) {
997            headers.insert(
998                CONTENT_LENGTH,
999                self.command().content_length()?.to_string().parse()?,
1000            );
1001            headers.insert(CONTENT_TYPE, self.command().content_type().parse()?);
1002        }
1003        headers.insert(
1004            HeaderName::from_static("x-amz-content-sha256"),
1005            sha256.parse()?,
1006        );
1007        headers.insert(
1008            HeaderName::from_static("x-amz-date"),
1009            self.long_date()?.parse()?,
1010        );
1011
1012        if let Some(session_token) = self.bucket().session_token().await? {
1013            headers.insert(
1014                HeaderName::from_static("x-amz-security-token"),
1015                session_token.parse()?,
1016            );
1017        } else if let Some(security_token) = self.bucket().security_token().await? {
1018            headers.insert(
1019                HeaderName::from_static("x-amz-security-token"),
1020                security_token.parse()?,
1021            );
1022        }
1023
1024        if let Command::PutObjectTagging { tags } = self.command() {
1025            let digest = md5::compute(tags);
1026            let hash = general_purpose::STANDARD.encode(digest.as_ref());
1027            headers.insert(HeaderName::from_static("content-md5"), hash.parse()?);
1028        } else if let Command::PutObject { content, .. } = self.command() {
1029            let digest = md5::compute(content);
1030            let hash = general_purpose::STANDARD.encode(digest.as_ref());
1031            headers.insert(HeaderName::from_static("content-md5"), hash.parse()?);
1032        } else if let Command::UploadPart { content, .. } = self.command() {
1033            let digest = md5::compute(content);
1034            let hash = general_purpose::STANDARD.encode(digest.as_ref());
1035            headers.insert(HeaderName::from_static("content-md5"), hash.parse()?);
1036        } else if let Command::GetObject {} = self.command() {
1037            headers.insert(ACCEPT, "application/octet-stream".to_string().parse()?);
1038        // headers.insert(header::ACCEPT_CHARSET, HeaderValue::from_str("UTF-8")?);
1039        } else if let Command::GetObjectRange { start, end } = self.command() {
1040            headers.insert(ACCEPT, "application/octet-stream".to_string().parse()?);
1041
1042            let mut range = format!("bytes={}-", start);
1043
1044            if let Some(end) = end {
1045                range.push_str(&end.to_string());
1046            }
1047
1048            headers.insert(RANGE, range.parse()?);
1049        } else if let Command::CreateBucket { ref config } = self.command() {
1050            config.add_headers(&mut headers)?;
1051        } else if let Command::PutBucketLifecycle { ref configuration } = self.command() {
1052            let digest = md5::compute(to_string(configuration)?.as_bytes());
1053            let hash = general_purpose::STANDARD.encode(digest.as_ref());
1054            headers.insert(HeaderName::from_static("content-md5"), hash.parse()?);
1055        } else if let Command::PutBucketCors {
1056            expected_bucket_owner,
1057            configuration,
1058            ..
1059        } = self.command()
1060        {
1061            let digest = md5::compute(configuration.to_string().as_bytes());
1062            let hash = general_purpose::STANDARD.encode(digest.as_ref());
1063            headers.insert(HeaderName::from_static("content-md5"), hash.parse()?);
1064
1065            headers.insert(
1066                HeaderName::from_static("x-amz-expected-bucket-owner"),
1067                expected_bucket_owner.parse()?,
1068            );
1069        } else if let Command::PutBucketPolicy { policy } = self.command() {
1070            let digest = md5::compute(policy.as_bytes());
1071            let hash = general_purpose::STANDARD.encode(digest.as_ref());
1072            headers.insert(HeaderName::from_static("content-md5"), hash.parse()?);
1073        } else if let Command::GetBucketCors {
1074            expected_bucket_owner,
1075        } = self.command()
1076        {
1077            headers.insert(
1078                HeaderName::from_static("x-amz-expected-bucket-owner"),
1079                expected_bucket_owner.parse()?,
1080            );
1081        } else if let Command::DeleteBucketCors {
1082            expected_bucket_owner,
1083        } = self.command()
1084        {
1085            headers.insert(
1086                HeaderName::from_static("x-amz-expected-bucket-owner"),
1087                expected_bucket_owner.parse()?,
1088            );
1089        } else if let Command::GetObjectAttributes {
1090            expected_bucket_owner,
1091            ..
1092        } = self.command()
1093        {
1094            headers.insert(
1095                HeaderName::from_static("x-amz-expected-bucket-owner"),
1096                expected_bucket_owner.parse()?,
1097            );
1098            headers.insert(
1099                HeaderName::from_static("x-amz-object-attributes"),
1100                "ETag".parse()?,
1101            );
1102        } else if let Command::DeleteObjects { ref data } = self.command() {
1103            let body = data.to_string();
1104            let digest = md5::compute(body.as_bytes());
1105            let hash = general_purpose::STANDARD.encode(digest.as_ref());
1106            headers.insert(HeaderName::from_static("content-md5"), hash.parse()?);
1107        }
1108
1109        // This must be last, as it signs the other headers, omitted if no secret key is provided
1110        if self.bucket().secret_key().await?.is_some() {
1111            let authorization = self.authorization(&headers).await?;
1112            headers.insert(AUTHORIZATION, authorization.parse()?);
1113        }
1114
1115        // The format of RFC2822 is somewhat malleable, so including it in
1116        // signed headers can cause signature mismatches. We do include the
1117        // X-Amz-Date header, so requests are still properly limited to a date
1118        // range and can't be used again e.g. reply attacks. Adding this header
1119        // after the generation of the Authorization header leaves it out of
1120        // the signed headers.
1121        headers.insert(DATE, self.datetime().format(&Rfc2822)?.parse()?);
1122
1123        Ok(headers)
1124    }
1125}
1126
1127#[cfg(all(test, feature = "with-tokio"))]
1128mod tests {
1129    use super::*;
1130    use bytes::Bytes;
1131    use futures_util::stream;
1132    use tokio::io::AsyncReadExt;
1133
1134    #[test]
1135    fn test_authority_of_keeps_a_plain_host() {
1136        assert_eq!(
1137            authority_of("s3.eu-central-1.amazonaws.com"),
1138            "s3.eu-central-1.amazonaws.com"
1139        );
1140        assert_eq!(authority_of("127.0.0.1:9000"), "127.0.0.1:9000");
1141    }
1142
1143    #[test]
1144    fn test_authority_of_drops_an_endpoint_path() {
1145        assert_eq!(
1146            authority_of("project.supabase.co/storage/v1/s3"),
1147            "project.supabase.co"
1148        );
1149        assert_eq!(
1150            authority_of("127.0.0.1:54321/storage/v1/s3"),
1151            "127.0.0.1:54321"
1152        );
1153    }
1154
1155    #[tokio::test]
1156    async fn test_async_read_implementation() {
1157        // Create a mock stream with test data
1158        let chunks = vec![
1159            Ok(Bytes::from("Hello, ")),
1160            Ok(Bytes::from("World!")),
1161            Ok(Bytes::from(" This is a test.")),
1162        ];
1163
1164        let stream = stream::iter(chunks);
1165        let data_stream: DataStream = Box::pin(stream);
1166
1167        let mut response_stream = ResponseDataStream {
1168            bytes: data_stream,
1169            status_code: 200,
1170        };
1171
1172        // Read all data using AsyncRead
1173        let mut buffer = Vec::new();
1174        response_stream.read_to_end(&mut buffer).await.unwrap();
1175
1176        assert_eq!(buffer, b"Hello, World! This is a test.");
1177    }
1178
1179    #[tokio::test]
1180    async fn test_async_read_with_small_buffer_preserves_large_chunk() {
1181        let expected = b"This is a much longer string that won't fit in a small buffer";
1182        let chunks = vec![Ok(Bytes::copy_from_slice(expected))];
1183
1184        let stream = stream::iter(chunks);
1185        let data_stream: DataStream = Box::pin(stream);
1186
1187        let mut response_stream = ResponseDataStream {
1188            bytes: data_stream,
1189            status_code: 200,
1190        };
1191
1192        let mut output = Vec::new();
1193        let mut buffer = [0u8; 10];
1194
1195        loop {
1196            let n = response_stream.read(&mut buffer).await.unwrap();
1197            if n == 0 {
1198                break;
1199            }
1200            output.extend_from_slice(&buffer[..n]);
1201        }
1202
1203        assert_eq!(output, expected);
1204    }
1205
1206    #[tokio::test]
1207    async fn test_async_read_with_error() {
1208        use crate::error::S3Error;
1209
1210        // Create a stream that returns an error
1211        let chunks: Vec<Result<Bytes, S3Error>> = vec![
1212            Ok(Bytes::from("Some data")),
1213            Err(S3Error::Io(std::io::Error::new(
1214                std::io::ErrorKind::Other,
1215                "Test error",
1216            ))),
1217        ];
1218
1219        let stream = stream::iter(chunks);
1220        let data_stream: DataStream = Box::pin(stream);
1221
1222        let mut response_stream = ResponseDataStream {
1223            bytes: data_stream,
1224            status_code: 200,
1225        };
1226
1227        // First read should succeed
1228        let mut buffer = [0u8; 20];
1229        let n = response_stream.read(&mut buffer).await.unwrap();
1230        assert_eq!(n, 9);
1231        assert_eq!(&buffer[..n], b"Some data");
1232
1233        // Second read should fail with an error
1234        let result = response_stream.read(&mut buffer).await;
1235        assert!(result.is_err());
1236    }
1237
1238    #[tokio::test]
1239    async fn test_async_read_copy() {
1240        // Test using tokio::io::copy which is a common use case
1241        let chunks = vec![
1242            Ok(Bytes::from("First chunk\n")),
1243            Ok(Bytes::from("Second chunk\n")),
1244            Ok(Bytes::from("Third chunk\n")),
1245        ];
1246
1247        let stream = stream::iter(chunks);
1248        let data_stream: DataStream = Box::pin(stream);
1249
1250        let mut response_stream = ResponseDataStream {
1251            bytes: data_stream,
1252            status_code: 200,
1253        };
1254
1255        let mut output = Vec::new();
1256        tokio::io::copy(&mut response_stream, &mut output)
1257            .await
1258            .unwrap();
1259
1260        assert_eq!(output, b"First chunk\nSecond chunk\nThird chunk\n");
1261    }
1262}
1263
1264#[cfg(all(test, feature = "with-async-std"))]
1265mod async_std_tests {
1266    use super::*;
1267    use async_std::io::ReadExt;
1268    use bytes::Bytes;
1269    use futures_util::stream;
1270
1271    #[async_std::test]
1272    async fn test_async_read_implementation() {
1273        // Create a mock stream with test data
1274        let chunks = vec![
1275            Ok(Bytes::from("Hello, ")),
1276            Ok(Bytes::from("World!")),
1277            Ok(Bytes::from(" This is a test.")),
1278        ];
1279
1280        let stream = stream::iter(chunks);
1281        let data_stream: DataStream = Box::pin(stream);
1282
1283        let mut response_stream = ResponseDataStream {
1284            bytes: data_stream,
1285            status_code: 200,
1286        };
1287
1288        // Read all data using AsyncRead
1289        let mut buffer = Vec::new();
1290        response_stream.read_to_end(&mut buffer).await.unwrap();
1291
1292        assert_eq!(buffer, b"Hello, World! This is a test.");
1293    }
1294
1295    #[async_std::test]
1296    async fn test_async_read_with_small_buffer_preserves_large_chunk() {
1297        let expected = b"This is a much longer string that won't fit in a small buffer";
1298        let chunks = vec![Ok(Bytes::copy_from_slice(expected))];
1299
1300        let stream = stream::iter(chunks);
1301        let data_stream: DataStream = Box::pin(stream);
1302
1303        let mut response_stream = ResponseDataStream {
1304            bytes: data_stream,
1305            status_code: 200,
1306        };
1307
1308        let mut output = Vec::new();
1309        let mut buffer = [0u8; 10];
1310
1311        loop {
1312            let n = response_stream.read(&mut buffer).await.unwrap();
1313            if n == 0 {
1314                break;
1315            }
1316            output.extend_from_slice(&buffer[..n]);
1317        }
1318
1319        assert_eq!(output, expected);
1320    }
1321
1322    #[async_std::test]
1323    async fn test_async_read_with_error() {
1324        use crate::error::S3Error;
1325
1326        // Create a stream that returns an error
1327        let chunks: Vec<Result<Bytes, S3Error>> = vec![
1328            Ok(Bytes::from("Some data")),
1329            Err(S3Error::Io(std::io::Error::new(
1330                std::io::ErrorKind::Other,
1331                "Test error",
1332            ))),
1333        ];
1334
1335        let stream = stream::iter(chunks);
1336        let data_stream: DataStream = Box::pin(stream);
1337
1338        let mut response_stream = ResponseDataStream {
1339            bytes: data_stream,
1340            status_code: 200,
1341        };
1342
1343        // First read should succeed
1344        let mut buffer = [0u8; 20];
1345        let n = response_stream.read(&mut buffer).await.unwrap();
1346        assert_eq!(n, 9);
1347        assert_eq!(&buffer[..n], b"Some data");
1348
1349        // Second read should fail with an error
1350        let result = response_stream.read(&mut buffer).await;
1351        assert!(result.is_err());
1352    }
1353
1354    #[async_std::test]
1355    async fn test_async_read_copy() {
1356        // Test using async_std::io::copy which is a common use case
1357        let chunks = vec![
1358            Ok(Bytes::from("First chunk\n")),
1359            Ok(Bytes::from("Second chunk\n")),
1360            Ok(Bytes::from("Third chunk\n")),
1361        ];
1362
1363        let stream = stream::iter(chunks);
1364        let data_stream: DataStream = Box::pin(stream);
1365
1366        let mut response_stream = ResponseDataStream {
1367            bytes: data_stream,
1368            status_code: 200,
1369        };
1370
1371        let mut output = Vec::new();
1372        async_std::io::copy(&mut response_stream, &mut output)
1373            .await
1374            .unwrap();
1375
1376        assert_eq!(output, b"First chunk\nSecond chunk\nThird chunk\n");
1377    }
1378}
s3/request/request_trait.rs

s3/request/
request_trait.rs
