s3/request/
request_trait.rs

1use base64::engine::general_purpose;
2use base64::Engine;
3use hmac::Mac;
4use http_body_util::BodyStream;
5use hyper::body::Incoming;
6use std::collections::HashMap;
7use time::format_description::well_known::Rfc2822;
8use time::OffsetDateTime;
9use url::Url;
10
11use crate::bucket::Bucket;
12use crate::command::Command;
13use crate::error::S3Error;
14use crate::signing;
15use crate::LONG_DATETIME;
16use bytes::Bytes;
17use http::header::{
18    HeaderName, ACCEPT, AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE, DATE, HOST, RANGE,
19};
20use http::HeaderMap;
21use std::fmt::Write as _;
22
23#[derive(Debug)]
24
25pub struct ResponseData {
26    bytes: Bytes,
27    status_code: u16,
28    headers: HashMap<String, String>,
29}
30
31impl From<ResponseData> for Vec<u8> {
32    fn from(data: ResponseData) -> Vec<u8> {
33        data.to_vec()
34    }
35}
36
37impl ResponseData {
38    pub fn new(bytes: Bytes, status_code: u16, headers: HashMap<String, String>) -> ResponseData {
39        ResponseData {
40            bytes,
41            status_code,
42            headers,
43        }
44    }
45
46    pub fn as_slice(&self) -> &[u8] {
47        &self.bytes
48    }
49
50    pub fn to_vec(self) -> Vec<u8> {
51        self.bytes.to_vec()
52    }
53
54    pub fn bytes(&self) -> &Bytes {
55        &self.bytes
56    }
57
58    pub fn status_code(&self) -> u16 {
59        self.status_code
60    }
61
62    pub fn as_str(&self) -> Result<&str, std::str::Utf8Error> {
63        std::str::from_utf8(self.as_slice())
64    }
65
66    pub fn to_string(&self) -> Result<String, std::str::Utf8Error> {
67        std::str::from_utf8(self.as_slice()).map(|s| s.to_string())
68    }
69
70    pub fn headers(&self) -> HashMap<String, String> {
71        self.headers.clone()
72    }
73}
74
75use std::fmt;
76
77impl fmt::Display for ResponseData {
78    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
79        write!(
80            f,
81            "Status code: {}\n Data: {}",
82            self.status_code(),
83            self.to_string()
84                .unwrap_or_else(|_| "Data could not be cast to UTF string".to_string())
85        )
86    }
87}
88
89pub type DataStream = BodyStream<Incoming>;
90
91pub struct ResponseDataStream {
92    pub body_stream: DataStream,
93    pub status_code: u16,
94}
95
96impl ResponseDataStream {
97    pub fn body_stream(&mut self) -> &mut DataStream {
98        &mut self.body_stream
99    }
100}
101
102pub trait Request: Send + Sync {
103    type Response;
104    type HeaderMap;
105
106    fn response(&self)
107        -> impl std::future::Future<Output = Result<Self::Response, S3Error>> + Send;
108    fn response_data(
109        &self,
110        etag: bool,
111    ) -> impl std::future::Future<Output = Result<ResponseData, S3Error>> + Send;
112    fn response_data_to_writer<T: tokio::io::AsyncWrite + Send + Unpin>(
113        &self,
114        writer: &mut T,
115    ) -> impl std::future::Future<Output = Result<u16, S3Error>> + Send;
116    fn response_data_to_stream(
117        &self,
118    ) -> impl std::future::Future<Output = Result<ResponseDataStream, S3Error>> + Send;
119    fn response_header(
120        &self,
121    ) -> impl std::future::Future<Output = Result<(Self::HeaderMap, u16), S3Error>> + Send;
122    fn datetime(&self) -> OffsetDateTime;
123    fn bucket(&self) -> Bucket;
124    fn command(&self) -> Command;
125    fn path(&self) -> String;
126
127    fn signing_key(&self) -> Result<Vec<u8>, S3Error> {
128        signing::signing_key(
129            &self.datetime(),
130            &self
131                .bucket()
132                .secret_key()?
133                .expect("Secret key must be provided to sign headers, found None"),
134            &self.bucket().region(),
135            "s3",
136        )
137    }
138
139    fn request_body(&self) -> Vec<u8> {
140        match self.command() {
141            Command::PutObject { content, .. } => Vec::from(content),
142            Command::PutObjectTagging { tags } => Vec::from(tags),
143            Command::UploadPart { content, .. } => Vec::from(content),
144            Command::CompleteMultipartUpload { data, .. } => data.to_string().as_bytes().to_vec(),
145            Command::CreateBucket { config } => config
146                .location_constraint_payload()
147                .map(Vec::from)
148                .unwrap_or_default(),
149            _ => vec![],
150        }
151    }
152
153    fn long_date(&self) -> Result<String, S3Error> {
154        Ok(self.datetime().format(LONG_DATETIME)?)
155    }
156
157    fn string_to_sign(&self, request: &str) -> Result<String, S3Error> {
158        match self.command() {
159            Command::PresignPost { post_policy, .. } => Ok(post_policy),
160            _ => Ok(signing::string_to_sign(
161                &self.datetime(),
162                &self.bucket().region(),
163                request,
164            )?),
165        }
166    }
167
168    fn host_header(&self) -> String {
169        self.bucket().host()
170    }
171
172    fn presigned(&self) -> Result<String, S3Error> {
173        let (expiry, custom_headers, custom_queries) = match self.command() {
174            Command::PresignGet {
175                expiry_secs,
176                custom_queries,
177            } => (expiry_secs, None, custom_queries),
178            Command::PresignPut {
179                expiry_secs,
180                custom_headers,
181            } => (expiry_secs, custom_headers, None),
182            Command::PresignDelete { expiry_secs } => (expiry_secs, None, None),
183            _ => unreachable!(),
184        };
185
186        Ok(format!(
187            "{}&X-Amz-Signature={}",
188            self.presigned_url_no_sig(expiry, custom_headers.as_ref(), custom_queries.as_ref())?,
189            self.presigned_authorization(custom_headers.as_ref())?
190        ))
191    }
192
193    fn presigned_authorization(
194        &self,
195        custom_headers: Option<&HeaderMap>,
196    ) -> Result<String, S3Error> {
197        let mut headers = HeaderMap::new();
198        let host_header = self.host_header();
199        headers.insert(HOST, host_header.parse()?);
200        if let Some(custom_headers) = custom_headers {
201            for (k, v) in custom_headers.iter() {
202                headers.insert(k.clone(), v.clone());
203            }
204        }
205        let canonical_request = self.presigned_canonical_request(&headers)?;
206        let string_to_sign = self.string_to_sign(&canonical_request)?;
207        let mut hmac = signing::HmacSha256::new_from_slice(&self.signing_key()?)?;
208        hmac.update(string_to_sign.as_bytes());
209        let signature = hex::encode(hmac.finalize().into_bytes());
210        // let signed_header = signing::signed_header_string(&headers);
211        Ok(signature)
212    }
213
214    fn presigned_canonical_request(&self, headers: &HeaderMap) -> Result<String, S3Error> {
215        let (expiry, custom_headers, custom_queries) = match self.command() {
216            Command::PresignGet {
217                expiry_secs,
218                custom_queries,
219            } => (expiry_secs, None, custom_queries),
220            Command::PresignPut {
221                expiry_secs,
222                custom_headers,
223            } => (expiry_secs, custom_headers, None),
224            Command::PresignDelete { expiry_secs } => (expiry_secs, None, None),
225            _ => unreachable!(),
226        };
227
228        signing::canonical_request(
229            &self.command().http_verb().to_string(),
230            &self.presigned_url_no_sig(expiry, custom_headers.as_ref(), custom_queries.as_ref())?,
231            headers,
232            "UNSIGNED-PAYLOAD",
233        )
234    }
235
236    fn presigned_url_no_sig(
237        &self,
238        expiry: u32,
239        custom_headers: Option<&HeaderMap>,
240        custom_queries: Option<&HashMap<String, String>>,
241    ) -> Result<Url, S3Error> {
242        let bucket = self.bucket();
243        let token = if let Some(security_token) = bucket.security_token()? {
244            Some(security_token)
245        } else {
246            bucket.session_token()?
247        };
248        let url = Url::parse(&format!(
249            "{}{}{}",
250            self.url()?,
251            &signing::authorization_query_params_no_sig(
252                &self.bucket().access_key()?.unwrap_or_default(),
253                &self.datetime(),
254                &self.bucket().region(),
255                expiry,
256                custom_headers,
257                token.as_ref()
258            )?,
259            &signing::flatten_queries(custom_queries)?,
260        ))?;
261
262        Ok(url)
263    }
264
265    fn url(&self) -> Result<Url, S3Error> {
266        let mut url_str = self.bucket().url();
267
268        if let Command::ListBuckets { .. } = self.command() {
269            return Ok(Url::parse(&url_str)?);
270        }
271
272        if let Command::CreateBucket { .. } = self.command() {
273            return Ok(Url::parse(&url_str)?);
274        }
275
276        let path = if self.path().starts_with('/') {
277            self.path()[1..].to_string()
278        } else {
279            self.path()[..].to_string()
280        };
281
282        url_str.push('/');
283        url_str.push_str(&signing::uri_encode(&path, false));
284
285        // Append to url_path
286        #[allow(clippy::collapsible_match)]
287        match self.command() {
288            Command::InitiateMultipartUpload { .. } | Command::ListMultipartUploads { .. } => {
289                url_str.push_str("?uploads")
290            }
291            Command::AbortMultipartUpload { upload_id } => {
292                write!(url_str, "?uploadId={}", upload_id).expect("Could not write to url_str");
293            }
294            Command::CompleteMultipartUpload { upload_id, .. } => {
295                write!(url_str, "?uploadId={}", upload_id).expect("Could not write to url_str");
296            }
297            Command::GetObjectTorrent => url_str.push_str("?torrent"),
298            Command::PutObject { multipart, .. } => {
299                if let Some(multipart) = multipart {
300                    url_str.push_str(&multipart.query_string())
301                }
302            }
303            _ => {}
304        }
305
306        let mut url = Url::parse(&url_str)?;
307
308        for (key, value) in &self.bucket().extra_query {
309            url.query_pairs_mut().append_pair(key, value);
310        }
311
312        if let Command::ListObjectsV2 {
313            prefix,
314            delimiter,
315            continuation_token,
316            start_after,
317            max_keys,
318        } = self.command().clone()
319        {
320            let mut query_pairs = url.query_pairs_mut();
321            delimiter.map(|d| query_pairs.append_pair("delimiter", &d));
322
323            query_pairs.append_pair("prefix", &prefix);
324            query_pairs.append_pair("list-type", "2");
325            if let Some(token) = continuation_token {
326                query_pairs.append_pair("continuation-token", &token);
327            }
328            if let Some(start_after) = start_after {
329                query_pairs.append_pair("start-after", &start_after);
330            }
331            if let Some(max_keys) = max_keys {
332                query_pairs.append_pair("max-keys", &max_keys.to_string());
333            }
334        }
335
336        if let Command::ListObjects {
337            prefix,
338            delimiter,
339            marker,
340            max_keys,
341        } = self.command().clone()
342        {
343            let mut query_pairs = url.query_pairs_mut();
344            delimiter.map(|d| query_pairs.append_pair("delimiter", &d));
345
346            query_pairs.append_pair("prefix", &prefix);
347            if let Some(marker) = marker {
348                query_pairs.append_pair("marker", &marker);
349            }
350            if let Some(max_keys) = max_keys {
351                query_pairs.append_pair("max-keys", &max_keys.to_string());
352            }
353        }
354
355        match self.command() {
356            Command::ListMultipartUploads {
357                prefix,
358                delimiter,
359                key_marker,
360                max_uploads,
361            } => {
362                let mut query_pairs = url.query_pairs_mut();
363                delimiter.map(|d| query_pairs.append_pair("delimiter", d));
364                if let Some(prefix) = prefix {
365                    query_pairs.append_pair("prefix", prefix);
366                }
367                if let Some(key_marker) = key_marker {
368                    query_pairs.append_pair("key-marker", &key_marker);
369                }
370                if let Some(max_uploads) = max_uploads {
371                    query_pairs.append_pair("max-uploads", max_uploads.to_string().as_str());
372                }
373            }
374            Command::PutObjectTagging { .. }
375            | Command::GetObjectTagging
376            | Command::DeleteObjectTagging => {
377                url.query_pairs_mut().append_pair("tagging", "");
378            }
379            _ => {}
380        }
381
382        Ok(url)
383    }
384
385    fn canonical_request(&self, headers: &HeaderMap) -> Result<String, S3Error> {
386        signing::canonical_request(
387            &self.command().http_verb().to_string(),
388            &self.url()?,
389            headers,
390            &self.command().sha256(),
391        )
392    }
393
394    fn authorization(&self, headers: &HeaderMap) -> Result<String, S3Error> {
395        let canonical_request = self.canonical_request(headers)?;
396        let string_to_sign = self.string_to_sign(&canonical_request)?;
397        let mut hmac = signing::HmacSha256::new_from_slice(&self.signing_key()?)?;
398        hmac.update(string_to_sign.as_bytes());
399        let signature = hex::encode(hmac.finalize().into_bytes());
400        let signed_header = signing::signed_header_string(headers);
401        signing::authorization_header(
402            &self.bucket().access_key()?.expect("No access_key provided"),
403            &self.datetime(),
404            &self.bucket().region(),
405            &signed_header,
406            &signature,
407        )
408    }
409
410    fn headers(&self) -> Result<HeaderMap, S3Error> {
411        // Generate this once, but it's used in more than one place.
412        let sha256 = self.command().sha256();
413
414        // Start with extra_headers, that way our headers replace anything with
415        // the same name.
416
417        let mut headers = HeaderMap::new();
418
419        for (k, v) in self.bucket().extra_headers.iter() {
420            headers.insert(k.clone(), v.clone());
421        }
422
423        let host_header = self.host_header();
424
425        headers.insert(HOST, host_header.parse()?);
426
427        match self.command() {
428            Command::CopyObject { from } => {
429                headers.insert(HeaderName::from_static("x-amz-copy-source"), from.parse()?);
430            }
431            Command::ListObjects { .. } => {}
432            Command::ListObjectsV2 { .. } => {}
433            Command::GetObject => {}
434            Command::GetObjectTagging => {}
435            Command::GetBucketLocation => {}
436            _ => {
437                headers.insert(
438                    CONTENT_LENGTH,
439                    self.command().content_length().to_string().parse()?,
440                );
441                headers.insert(CONTENT_TYPE, self.command().content_type().parse()?);
442            }
443        }
444        headers.insert(
445            HeaderName::from_static("x-amz-content-sha256"),
446            sha256.parse()?,
447        );
448        headers.insert(
449            HeaderName::from_static("x-amz-date"),
450            self.long_date()?.parse()?,
451        );
452
453        if let Some(session_token) = self.bucket().session_token()? {
454            headers.insert(
455                HeaderName::from_static("x-amz-security-token"),
456                session_token.parse()?,
457            );
458        } else if let Some(security_token) = self.bucket().security_token()? {
459            headers.insert(
460                HeaderName::from_static("x-amz-security-token"),
461                security_token.parse()?,
462            );
463        }
464
465        if let Command::PutObjectTagging { tags } = self.command() {
466            let digest = md5::compute(tags);
467            let hash = general_purpose::STANDARD.encode(digest.as_ref());
468            headers.insert(HeaderName::from_static("content-md5"), hash.parse()?);
469        } else if let Command::PutObject { content, .. } = self.command() {
470            let digest = md5::compute(content);
471            let hash = general_purpose::STANDARD.encode(digest.as_ref());
472            headers.insert(HeaderName::from_static("content-md5"), hash.parse()?);
473        } else if let Command::UploadPart { content, .. } = self.command() {
474            let digest = md5::compute(content);
475            let hash = general_purpose::STANDARD.encode(digest.as_ref());
476            headers.insert(HeaderName::from_static("content-md5"), hash.parse()?);
477        } else if let Command::GetObject {} = self.command() {
478            headers.insert(ACCEPT, "application/octet-stream".to_string().parse()?);
479        // headers.insert(header::ACCEPT_CHARSET, HeaderValue::from_str("UTF-8")?);
480        } else if let Command::GetObjectRange { start, end } = self.command() {
481            headers.insert(ACCEPT, "application/octet-stream".to_string().parse()?);
482
483            let mut range = format!("bytes={}-", start);
484
485            if let Some(end) = end {
486                range.push_str(&end.to_string());
487            }
488
489            headers.insert(RANGE, range.parse()?);
490        } else if let Command::CreateBucket { ref config } = self.command() {
491            config.add_headers(&mut headers)?;
492        }
493
494        // This must be last, as it signs the other headers, omitted if no secret key is provided
495        if self.bucket().secret_key()?.is_some() {
496            let authorization = self.authorization(&headers)?;
497            headers.insert(AUTHORIZATION, authorization.parse()?);
498        }
499
500        // The format of RFC2822 is somewhat malleable, so including it in
501        // signed headers can cause signature mismatches. We do include the
502        // X-Amz-Date header, so requests are still properly limited to a date
503        // range and can't be used again e.g. reply attacks. Adding this header
504        // after the generation of the Authorization header leaves it out of
505        // the signed headers.
506        headers.insert(DATE, self.datetime().format(&Rfc2822)?.parse()?);
507
508        Ok(headers)
509    }
510}
s3/request/request_trait.rs

s3/request/
request_trait.rs
