s3/
client.rs

1use std::{
2    borrow::Cow,
3    collections::BTreeMap,
4    fmt,
5    pin::Pin,
6    time::{SystemTime, UNIX_EPOCH},
7};
8
9use bytes::Bytes;
10use crypto::{Hasher, hmac::Hmac, sha2::Sha256};
11use futures_util::{Stream, StreamExt};
12use url::Url;
13
14pub(crate) const EMPTY_PAYLOAD_SHA256: &str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
15
16#[derive(Debug, Clone)]
17pub struct StaticCredentials<'a> {
18    pub access_key_id: &'a str,
19    pub secret_access_key: &'a str,
20    pub session_token: &'a str,
21}
22
23#[derive(Debug, Clone)]
24pub struct ClientConfig<'a> {
25    pub endpoint: &'a str,
26    pub credentials: StaticCredentials<'a>,
27    pub region: &'a str,
28    pub virtual_hosted: bool,
29}
30
31#[derive(Debug, Clone)]
32pub(crate) struct OwnedCredentials {
33    pub(crate) access_key_id: String,
34    pub(crate) secret_access_key: String,
35    pub(crate) session_token: Option<String>,
36}
37
38#[derive(Debug)]
39pub struct Client<H: HttpClient> {
40    pub(crate) endpoint: Url,
41    pub(crate) region: String,
42    pub(crate) credentials: OwnedCredentials,
43    pub(crate) virtual_hosted: bool,
44    pub(crate) http: H,
45}
46
47#[derive(Debug)]
48pub enum Error {
49    InvalidConfig(&'static str),
50    Http(Box<dyn core::error::Error + Send + Sync>),
51    Time(std::time::SystemTimeError),
52    Xml(quick_xml::DeError),
53    Api { status: u16, body: String },
54}
55
56impl fmt::Display for Error {
57    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
58        match self {
59            Error::InvalidConfig(msg) => write!(f, "invalid config: {msg}"),
60            Error::Http(err) => write!(f, "http error: {err}"),
61            Error::Time(err) => write!(f, "time error: {err}"),
62            Error::Xml(err) => write!(f, "xml error: {err}"),
63            Error::Api {
64                status,
65                body,
66            } => write!(f, "s3 api error (status {status}): {body}"),
67        }
68    }
69}
70
71impl core::error::Error for Error {
72    fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
73        match self {
74            Error::Http(err) => Some(&**err),
75            _ => None,
76        }
77    }
78}
79
80impl From<std::time::SystemTimeError> for Error {
81    fn from(value: std::time::SystemTimeError) -> Self {
82        Self::Time(value)
83    }
84}
85
86impl From<quick_xml::DeError> for Error {
87    fn from(value: quick_xml::DeError) -> Self {
88        Self::Xml(value)
89    }
90}
91
92pub type HttpError = Box<dyn core::error::Error + Send + Sync>;
93
94/// A pinned, boxed async stream of byte chunks.
95pub type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, HttpError>> + Send>>;
96
97#[derive(Debug, Clone, Copy)]
98pub enum HttpMethod {
99    Get,
100    Put,
101    Post,
102    Delete,
103    Head,
104}
105
106impl HttpMethod {
107    pub(crate) fn as_str(self) -> &'static str {
108        match self {
109            HttpMethod::Get => "GET",
110            HttpMethod::Put => "PUT",
111            HttpMethod::Post => "POST",
112            HttpMethod::Delete => "DELETE",
113            HttpMethod::Head => "HEAD",
114        }
115    }
116}
117
118#[derive(Debug, Clone)]
119pub struct HttpRequest {
120    pub method: HttpMethod,
121    pub url: String,
122    pub headers: Vec<(String, String)>,
123    pub body: Vec<u8>,
124}
125
126pub struct HttpResponseData {
127    pub status_code: u16,
128    pub headers: Vec<(String, String)>,
129    pub body: ByteStream,
130}
131
132impl HttpResponseData {
133    pub fn header(&self, name: &str) -> Option<&str> {
134        self.headers
135            .iter()
136            .find(|(k, _)| k.eq_ignore_ascii_case(name))
137            .map(|(_, v)| v.as_str())
138    }
139}
140
141pub trait HttpClient: Send + Sync {
142    fn send(
143        &self,
144        request: HttpRequest,
145    ) -> impl std::future::Future<Output = Result<HttpResponseData, HttpError>> + Send;
146}
147
148#[cfg(feature = "reqwest")]
149#[derive(Debug, Clone)]
150pub struct ReqwestHttpClient {
151    inner: reqwest::Client,
152}
153
154#[cfg(feature = "reqwest")]
155impl Default for ReqwestHttpClient {
156    fn default() -> Self {
157        Self::new()
158    }
159}
160
161#[cfg(feature = "reqwest")]
162impl ReqwestHttpClient {
163    /// Builds an HTTP client backed by `reqwest`.
164    ///
165    /// Installs the stdx rustls `CryptoProvider` as the process default (if
166    /// none is installed yet) before creating the underlying `reqwest` client,
167    /// because `reqwest` is compiled with the `rustls-no-provider` feature and
168    /// would otherwise panic for lack of a cryptography provider.
169    pub fn new() -> Self {
170        crypto_rustls::install_default_rustls_provider();
171        Self {
172            inner: reqwest::Client::new(),
173        }
174    }
175}
176
177#[cfg(feature = "reqwest")]
178impl HttpClient for ReqwestHttpClient {
179    async fn send(&self, request: HttpRequest) -> Result<HttpResponseData, HttpError> {
180        let method = reqwest::Method::from_bytes(request.method.as_str().as_bytes())?;
181        let mut req = self.inner.request(method, &request.url);
182        for (name, value) in request.headers {
183            req = req.header(&name, &value);
184        }
185        if !request.body.is_empty() {
186            req = req.body(request.body);
187        }
188        let response = req.send().await?;
189        let status_code = response.status().as_u16();
190        let headers = response
191            .headers()
192            .iter()
193            .filter_map(|(name, value)| value.to_str().ok().map(|v| (name.as_str().to_string(), v.to_string())))
194            .collect();
195        let body: ByteStream = Box::pin(
196            response
197                .bytes_stream()
198                .map(|r| r.map_err(|e| -> HttpError { Box::new(e) })),
199        );
200        Ok(HttpResponseData {
201            status_code,
202            headers,
203            body,
204        })
205    }
206}
207
208impl<H: HttpClient> Client<H> {
209    pub fn with_http_client(config: &ClientConfig<'_>, http: H) -> Result<Self, Error> {
210        if config.endpoint.trim().is_empty() {
211            return Err(Error::InvalidConfig("endpoint must not be empty"));
212        }
213        if config.region.trim().is_empty() {
214            return Err(Error::InvalidConfig("region must not be empty"));
215        }
216        if config.credentials.access_key_id.trim().is_empty() {
217            return Err(Error::InvalidConfig("access key id must not be empty"));
218        }
219        if config.credentials.secret_access_key.trim().is_empty() {
220            return Err(Error::InvalidConfig("secret access key must not be empty"));
221        }
222
223        let endpoint = Url::parse(config.endpoint).map_err(|_| Error::InvalidConfig("invalid endpoint URL"))?;
224
225        Ok(Self {
226            endpoint,
227            region: config.region.to_string(),
228            credentials: OwnedCredentials {
229                access_key_id: config.credentials.access_key_id.to_string(),
230                secret_access_key: config.credentials.secret_access_key.to_string(),
231                session_token: if config.credentials.session_token.is_empty() {
232                    None
233                } else {
234                    Some(config.credentials.session_token.to_string())
235                },
236            },
237            virtual_hosted: config.virtual_hosted,
238            http,
239        })
240    }
241
242    pub(crate) async fn execute(
243        &self,
244        method: HttpMethod,
245        canonical_uri: &str,
246        canonical_query: &str,
247        body: &[u8],
248        bucket: &str,
249    ) -> Result<HttpResponseData, Error> {
250        self.execute_with_headers(method, canonical_uri, canonical_query, body, &[], bucket)
251            .await
252    }
253
254    pub(crate) async fn execute_with_headers(
255        &self,
256        method: HttpMethod,
257        canonical_uri: &str,
258        canonical_query: &str,
259        body: &[u8],
260        extra_headers: &[(String, String)],
261        bucket: &str,
262    ) -> Result<HttpResponseData, Error> {
263        fn canonical_header_value(value: &str) -> String {
264            let trimmed = value.trim();
265            let mut out = String::with_capacity(trimmed.len());
266            let mut in_space = false;
267            for ch in trimmed.chars() {
268                if ch.is_whitespace() {
269                    if !in_space {
270                        out.push(' ');
271                        in_space = true;
272                    }
273                } else {
274                    out.push(ch);
275                    in_space = false;
276                }
277            }
278            out
279        }
280
281        let (date, amz_datetime) = amz_datetime(SystemTime::now())?;
282        let credential_scope = format!("{}/{}/s3/aws4_request", date, self.region);
283
284        let base_host = endpoint_host(&self.endpoint);
285        let host = if self.virtual_hosted && !bucket.is_empty() {
286            Cow::Owned(format!("{bucket}.{base_host}"))
287        } else {
288            Cow::Borrowed(&base_host)
289        };
290        let payload_hash = if body.is_empty() {
291            EMPTY_PAYLOAD_SHA256.to_string()
292        } else {
293            hex::encode(sha256(body))
294        };
295
296        let mut headers = vec![
297            ("host".to_string(), host.to_string()),
298            ("x-amz-date".to_string(), amz_datetime.clone()),
299            ("x-amz-content-sha256".to_string(), payload_hash.clone()),
300        ];
301        headers.extend(extra_headers.iter().cloned());
302        if let Some(token) = self.credentials.session_token.as_deref() {
303            headers.push(("x-amz-security-token".to_string(), token.to_string()));
304        }
305
306        let mut canonical_headers = headers
307            .iter()
308            .map(|(name, value)| format!("{}:{}\n", name.to_ascii_lowercase(), canonical_header_value(value)))
309            .collect::<Vec<_>>();
310        let mut signed_headers = headers
311            .iter()
312            .map(|(name, _)| name.to_ascii_lowercase())
313            .collect::<Vec<_>>();
314
315        canonical_headers.sort();
316        signed_headers.sort();
317
318        let canonical_headers_joined = canonical_headers.concat();
319        let signed_headers_joined = signed_headers.join(";");
320
321        let canonical_request = format!(
322            "{}\n{}\n{}\n{}\n{}\n{}",
323            method.as_str(),
324            canonical_uri,
325            canonical_query,
326            canonical_headers_joined,
327            signed_headers_joined,
328            payload_hash
329        );
330
331        let string_to_sign = format!(
332            "AWS4-HMAC-SHA256\n{}\n{}\n{}",
333            amz_datetime,
334            credential_scope,
335            hex::encode(sha256(canonical_request.as_bytes()))
336        );
337
338        let signature = hex::encode(sign_v4(
339            &self.credentials.secret_access_key,
340            &date,
341            &self.region,
342            &string_to_sign,
343        ));
344
345        let authorization = format!(
346            "AWS4-HMAC-SHA256 Credential={}/{}, SignedHeaders={}, Signature={}",
347            self.credentials.access_key_id, credential_scope, signed_headers_joined, signature
348        );
349
350        let url = if self.virtual_hosted && !bucket.is_empty() {
351            let scheme = self.endpoint.scheme();
352            let host_str = endpoint_host(&self.endpoint);
353            let path = canonical_uri;
354            if canonical_query.is_empty() {
355                format!("{scheme}://{bucket}.{host_str}{path}")
356            } else {
357                format!("{scheme}://{bucket}.{host_str}{path}?{canonical_query}")
358            }
359        } else if canonical_query.is_empty() {
360            format!("{}{}", self.endpoint.as_str().trim_end_matches('/'), canonical_uri)
361        } else {
362            format!(
363                "{}{}?{}",
364                self.endpoint.as_str().trim_end_matches('/'),
365                canonical_uri,
366                canonical_query
367            )
368        };
369
370        headers.push(("authorization".to_string(), authorization));
371
372        let request = HttpRequest {
373            method,
374            url,
375            headers,
376            body: body.to_vec(),
377        };
378        let response = self.http.send(request).await.map_err(Error::Http)?;
379
380        if (200..300).contains(&response.status_code) {
381            return Ok(response);
382        }
383
384        let status = response.status_code;
385        let body_bytes = collect_body(response.body).await.unwrap_or_default();
386        let body = String::from_utf8_lossy(&body_bytes).into_owned();
387        Err(Error::Api {
388            status,
389            body,
390        })
391    }
392}
393
394#[cfg(feature = "reqwest")]
395impl Client<ReqwestHttpClient> {
396    pub fn new(config: &ClientConfig<'_>) -> Result<Self, Error> {
397        Self::with_http_client(config, ReqwestHttpClient::new())
398    }
399}
400
401pub(crate) fn consume_empty(_response: HttpResponseData) -> Result<(), Error> {
402    Ok(())
403}
404
405pub(crate) fn canonical_bucket_uri(bucket: &str) -> String {
406    canonical_uri(&format!("/{bucket}"))
407}
408
409pub(crate) fn canonical_object_uri(bucket: &str, key: &str) -> String {
410    canonical_uri(&format!("/{bucket}/{key}"))
411}
412
413pub(crate) fn canonical_uri(path: &str) -> String {
414    path.split('/').map(percent_encode).collect::<Vec<_>>().join("/")
415}
416
417pub(crate) fn canonical_query_string(params: &BTreeMap<String, String>) -> String {
418    params
419        .iter()
420        .map(|(k, v)| format!("{}={}", percent_encode(k), percent_encode(v)))
421        .collect::<Vec<_>>()
422        .join("&")
423}
424
425fn endpoint_host(url: &Url) -> String {
426    match (url.host_str(), url.port()) {
427        (Some(host), Some(port)) => format!("{host}:{port}"),
428        (Some(host), None) => host.to_string(),
429        _ => String::new(),
430    }
431}
432
433pub(crate) fn percent_encode(input: &str) -> String {
434    let mut out = String::with_capacity(input.len());
435    for &b in input.as_bytes() {
436        if b.is_ascii_uppercase()
437            || b.is_ascii_lowercase()
438            || b.is_ascii_digit()
439            || matches!(b, b'-' | b'_' | b'.' | b'~')
440        {
441            out.push(b as char);
442        } else {
443            out.push('%');
444            out.push(hex_nibble_upper((b >> 4) & 0x0f));
445            out.push(hex_nibble_upper(b & 0x0f));
446        }
447    }
448    out
449}
450
451fn hex_nibble_upper(value: u8) -> char {
452    match value {
453        0..=9 => (b'0' + value) as char,
454        10..=15 => (b'A' + (value - 10)) as char,
455        _ => unreachable!(),
456    }
457}
458
459fn sign_v4(secret_access_key: &str, date: &str, region: &str, string_to_sign: &str) -> [u8; 32] {
460    let k_date = hmac_sha256(format!("AWS4{secret_access_key}").as_bytes(), date.as_bytes());
461    let k_region = hmac_sha256(&k_date, region.as_bytes());
462    let k_service = hmac_sha256(&k_region, b"s3");
463    let k_signing = hmac_sha256(&k_service, b"aws4_request");
464    hmac_sha256(&k_signing, string_to_sign.as_bytes())
465}
466
467pub(crate) fn hmac_sha256(key: &[u8], data: &[u8]) -> [u8; Sha256::OUTPUT_SIZE] {
468    let mut mac = Hmac::<Sha256>::new(key);
469    mac.update(data);
470    *mac.finalize().as_ref().as_array().unwrap()
471}
472
473pub(crate) fn sha256(data: &[u8]) -> [u8; Sha256::OUTPUT_SIZE] {
474    let mut hasher = Sha256::new();
475    hasher.update(data);
476    hasher.sum().as_ref().try_into().unwrap()
477}
478
479fn amz_datetime(now: SystemTime) -> Result<(String, String), Error> {
480    let elapsed = now.duration_since(UNIX_EPOCH)?;
481    let total_seconds = elapsed.as_secs() as i64;
482
483    let days = total_seconds.div_euclid(86_400);
484    let seconds_of_day = total_seconds.rem_euclid(86_400);
485
486    let (year, month, day) = civil_from_days(days);
487    let hour = seconds_of_day / 3_600;
488    let minute = (seconds_of_day % 3_600) / 60;
489    let second = seconds_of_day % 60;
490
491    let date = format!("{year:04}{month:02}{day:02}");
492    let datetime = format!("{date}T{hour:02}{minute:02}{second:02}Z");
493
494    Ok((date, datetime))
495}
496
497fn civil_from_days(days_since_unix_epoch: i64) -> (i32, i64, i64) {
498    let z = days_since_unix_epoch + 719_468;
499    let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
500    let doe = z - era * 146_097;
501    let yoe = (doe - doe / 1_460 + doe / 36_524 - doe / 146_096) / 365;
502    let y = yoe + era * 400;
503    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
504    let mp = (5 * doy + 2) / 153;
505    let day = doy - (153 * mp + 2) / 5 + 1;
506    let month = mp + if mp < 10 { 3 } else { -9 };
507    let year = y + if month <= 2 { 1 } else { 0 };
508
509    (year as i32, month, day)
510}
511
512pub(crate) async fn collect_body(mut stream: ByteStream) -> Result<Vec<u8>, Error> {
513    let mut out = Vec::new();
514    while let Some(chunk) = stream.next().await {
515        let bytes = chunk.map_err(Error::Http)?;
516        out.extend_from_slice(&bytes);
517    }
518    Ok(out)
519}
520
521pub(crate) fn bytes_to_string(bytes: Vec<u8>) -> Result<String, Error> {
522    String::from_utf8(bytes).map_err(|e| Error::Http(Box::new(e)))
523}
524
525pub(crate) fn header_to_string(response: &HttpResponseData, name: &str) -> Option<String> {
526    response.header(name).map(ToString::to_string)
527}
528
529pub(crate) fn header_to_u64(response: &HttpResponseData, name: &str) -> Option<u64> {
530    response.header(name).and_then(|s| s.parse::<u64>().ok())
531}
532
533pub(crate) fn xml_escape(s: &str) -> String {
534    s.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;")
535}
536
537#[cfg(test)]
538mod tests {
539    use super::*;
540
541    #[derive(Default)]
542    struct NoopHttpClient;
543
544    impl HttpClient for NoopHttpClient {
545        async fn send(&self, _request: HttpRequest) -> Result<HttpResponseData, HttpError> {
546            Ok(HttpResponseData {
547                status_code: 200,
548                headers: Vec::new(),
549                body: Box::pin(futures_util::stream::empty()),
550            })
551        }
552    }
553
554    #[test]
555    fn sha256_known_vectors() {
556        assert_eq!(
557            hex::encode(sha256(b"")),
558            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
559        );
560        assert_eq!(
561            hex::encode(sha256(b"abc")),
562            "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
563        );
564    }
565
566    #[test]
567    fn hmac_sha256_known_vector() {
568        let key = [0x0b; 20];
569        let sig = hmac_sha256(&key, b"Hi There");
570        assert_eq!(
571            hex::encode(sig),
572            "b0344c61d8db38535ca8afceaf0bf12b881dc200c9833da726e9376c2e32cff7"
573        );
574    }
575
576    #[test]
577    fn percent_encoding_works() {
578        assert_eq!(percent_encode("abcXYZ-_.~"), "abcXYZ-_.~");
579        assert_eq!(percent_encode("hello world/é"), "hello%20world%2F%C3%A9");
580    }
581
582    #[test]
583    fn canonical_uri_preserves_slashes() {
584        assert_eq!(
585            canonical_object_uri("my-bucket", "folder/my file.txt"),
586            "/my-bucket/folder/my%20file.txt"
587        );
588    }
589
590    #[test]
591    fn amz_datetime_format_works() {
592        let ts = UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_000);
593        let (date, datetime) = amz_datetime(ts).unwrap();
594        assert_eq!(date.len(), 8);
595        assert_eq!(datetime.len(), 16);
596        assert!(datetime.ends_with('Z'));
597    }
598
599    #[test]
600    fn client_config_validation_works() {
601        let cfg = ClientConfig {
602            endpoint: "",
603            credentials: StaticCredentials {
604                access_key_id: "a",
605                secret_access_key: "b",
606                session_token: "",
607            },
608            region: "auto",
609            virtual_hosted: false,
610        };
611        assert!(matches!(
612            Client::with_http_client(&cfg, NoopHttpClient),
613            Err(Error::InvalidConfig(_))
614        ));
615    }
616
617    #[test]
618    fn sign_v4_known_output_length() {
619        let sig = sign_v4(
620            "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
621            "20130524",
622            "auto",
623            "AWS4-HMAC-SHA256\n20130524T000000Z\n20130524/auto/s3/aws4_request\nabc",
624        );
625        assert_eq!(sig.len(), 32);
626    }
627
628    #[test]
629    fn empty_payload_sha256_constant_is_correct() {
630        assert_eq!(EMPTY_PAYLOAD_SHA256, hex::encode(sha256(b"")));
631    }
632}
s3/client.rs

s3/
client.rs
