rustlavel_storage/
s3.rs

1//! S3-compatible object storage.
2//!
3//! Written against the S3 REST API directly, so it works with AWS S3,
4//! Cloudflare R2, MinIO, Backblaze B2 and anything else that speaks it —
5//! only the endpoint changes.
6
7use crate::sigv4::{self, Signing};
8use std::time::Duration;
9use crate::{Entry, Storage, Visibility, content_type, normalize};
10use rustlavel_client::Client;
11use rustlavel_core::{Config, Error, Result};
12
13#[derive(Debug, Clone)]
14pub struct S3Config {
15    pub bucket: String,
16    pub region: String,
17    pub access_key: String,
18    pub secret_key: String,
19    /// For a non-AWS service: `https://<account>.r2.cloudflarestorage.com`.
20    pub endpoint: Option<String>,
21    /// MinIO and most self-hosted services need the bucket in the path rather
22    /// than in the hostname.
23    pub path_style: bool,
24    /// Where public objects are reached, when that differs from the API.
25    pub public_url: Option<String>,
26}
27
28impl Default for S3Config {
29    fn default() -> Self {
30        S3Config {
31            bucket: String::new(),
32            region: "us-east-1".into(),
33            access_key: String::new(),
34            secret_key: String::new(),
35            endpoint: None,
36            path_style: false,
37            public_url: None,
38        }
39    }
40}
41
42impl S3Config {
43    pub fn from_config(config: &Config) -> Result<S3Config> {
44        let settings = S3Config {
45            bucket: config.string("storage.bucket", ""),
46            region: config.string("storage.region", "us-east-1"),
47            access_key: config.string("storage.access_key", ""),
48            secret_key: config.string("storage.secret_key", ""),
49            endpoint: non_empty(config.string("storage.endpoint", "")),
50            path_style: config.bool("storage.path_style", false),
51            public_url: non_empty(config.string("storage.public_url", "")),
52        };
53
54        if settings.bucket.is_empty() {
55            return Err(Error::msg(
56                "the s3 storage driver needs `storage.bucket`. Set it in config/storage.json or \
57                 point STORAGE_BUCKET at it in .env."
58                    .to_string(),
59            ));
60        }
61        Ok(settings)
62    }
63
64    /// The host requests go to.
65    fn host(&self) -> String {
66        match &self.endpoint {
67            Some(endpoint) => endpoint
68                .trim_start_matches("https://")
69                .trim_start_matches("http://")
70                .trim_end_matches('/')
71                .to_string(),
72            None if self.path_style => format!("s3.{}.amazonaws.com", self.region),
73            None => format!("{}.s3.{}.amazonaws.com", self.bucket, self.region),
74        }
75    }
76
77    fn scheme(&self) -> &'static str {
78        match &self.endpoint {
79            Some(endpoint) if endpoint.starts_with("http://") => "http",
80            _ => "https",
81        }
82    }
83
84    /// The request path, which includes the bucket in path style.
85    fn path_for(&self, key: &str) -> String {
86        if self.path_style || self.endpoint.is_some() {
87            format!("/{}/{key}", self.bucket)
88        } else {
89            format!("/{key}")
90        }
91    }
92}
93
94fn non_empty(value: String) -> Option<String> {
95    (!value.is_empty()).then_some(value)
96}
97
98pub struct S3Storage {
99    config: S3Config,
100    client: Client,
101}
102
103impl S3Storage {
104    pub fn new(config: S3Config) -> Self {
105        S3Storage { config, client: Client::new().retries(2) }
106    }
107
108    /// Use a prepared client — how a test drives this against a fake.
109    pub fn with_client(config: S3Config, client: Client) -> Self {
110        S3Storage { config, client }
111    }
112
113    pub fn config(&self) -> &S3Config {
114        &self.config
115    }
116
117    /// Build a signed request for one object operation.
118    fn signed(
119        &self,
120        method: &str,
121        key: &str,
122        query: &str,
123        body: &[u8],
124        extra: &[(String, String)],
125    ) -> (String, Vec<(String, String)>) {
126        let host = self.config.host();
127        let path = self.config.path_for(key);
128        let timestamp = timestamp();
129        let payload_hash = sigv4::sha256_hex(body);
130
131        let mut headers = vec![
132            ("host".to_string(), host.clone()),
133            ("x-amz-content-sha256".to_string(), payload_hash.clone()),
134            ("x-amz-date".to_string(), timestamp.clone()),
135        ];
136        headers.extend_from_slice(extra);
137
138        let signing = Signing {
139            access_key: &self.config.access_key,
140            secret_key: &self.config.secret_key,
141            region: &self.config.region,
142            service: "s3",
143            timestamp: &timestamp,
144        };
145
146        let authorization =
147            sigv4::authorization(&signing, method, &path, query, &headers, &payload_hash);
148        headers.push(("authorization".to_string(), authorization));
149
150        let url = if query.is_empty() {
151            format!("{}://{host}{}", self.config.scheme(), sigv4::encode_path(&path))
152        } else {
153            format!("{}://{host}{}?{query}", self.config.scheme(), sigv4::encode_path(&path))
154        };
155
156        (url, headers)
157    }
158}
159
160/// A multipart upload in progress: the id the bucket assigned, and the key.
161///
162/// Hand the browser presigned URLs for each part, collect the `ETag` header
163/// each PUT returns, and finish with [`S3Storage::complete_multipart`]. The
164/// server never holds a byte of the file.
165#[derive(Debug, Clone, PartialEq, Eq)]
166pub struct Multipart {
167    pub key: String,
168    pub upload_id: String,
169}
170
171/// One uploaded part, as the bucket reported it.
172#[derive(Debug, Clone, PartialEq, Eq)]
173pub struct Part {
174    /// 1-based, and the order the parts are joined in.
175    pub number: u32,
176    /// The `ETag` response header from the part's PUT, quotes included or not.
177    pub etag: String,
178}
179
180/// The smallest part S3 accepts for every part but the last.
181pub const MIN_PART_BYTES: u64 = 5 * 1024 * 1024;
182/// The largest number of parts one upload may have.
183pub const MAX_PARTS: u32 = 10_000;
184/// The longest a presigned URL may be valid for; AWS refuses more.
185pub const MAX_PRESIGN: Duration = Duration::from_secs(7 * 24 * 60 * 60);
186
187impl S3Storage {
188    fn signing<'a>(&'a self, timestamp: &'a str) -> Signing<'a> {
189        Signing {
190            access_key: &self.config.access_key,
191            secret_key: &self.config.secret_key,
192            region: &self.config.region,
193            service: "s3",
194            timestamp,
195        }
196    }
197
198    /// A URL that lets whoever holds it perform `method` on `key` until
199    /// `expires` runs out, with no credential of their own.
200    fn presigned(&self, method: &str, key: &str, expires: Duration, extra: &[(&str, &str)]) -> Result<String> {
201        if expires > MAX_PRESIGN {
202            return Err(Error::msg(format!(
203                "a presigned URL may be valid for at most seven days; {} seconds was asked for",
204                expires.as_secs()
205            )));
206        }
207        let key = normalize(key)?;
208        let host = self.config.host();
209        let path = self.config.path_for(&key);
210        let timestamp = timestamp();
211        let signing = self.signing(&timestamp);
212
213        let query = sigv4::presigned_query(&signing, method, &host, &path, expires.as_secs(), extra);
214        Ok(format!("{}://{host}{}?{query}", self.config.scheme(), sigv4::encode_path(&path)))
215    }
216
217    /// A URL a browser can `PUT` a file to directly.
218    ///
219    /// **This is how a large upload should reach the bucket.** A video that
220    /// passes through the application costs the server the whole file in
221    /// memory and the transfer twice. The server issues this URL, the browser
222    /// uploads, and the server is told the key afterwards — it never holds a
223    /// byte. For anything over a few hundred megabytes, use
224    /// [`S3Storage::create_multipart`] instead, so a dropped connection costs
225    /// one part rather than the whole file.
226    ///
227    /// The browser may send any headers it likes; only the host is signed.
228    pub fn presigned_put(&self, key: &str, expires: Duration) -> Result<String> {
229        self.presigned("PUT", key, expires, &[])
230    }
231
232    /// A URL that reads a private object for a while, for handing a visitor a
233    /// download without making the object public.
234    pub fn presigned_get(&self, key: &str, expires: Duration) -> Result<String> {
235        self.presigned("GET", key, expires, &[])
236    }
237
238    /// Begin a multipart upload. Returns the id every part must carry.
239    pub async fn create_multipart(&self, key: &str) -> Result<Multipart> {
240        let key = normalize(key)?;
241        let extra = vec![("content-type".to_string(), content_type(&key).to_string())];
242        let (url, headers) = self.signed("POST", &key, "uploads=", b"", &extra);
243
244        let mut request = self.client.post(url);
245        for (name, value) in headers {
246            request = request.header(&name, value);
247        }
248        let body = request.send().await?.error_for_status()?.text();
249
250        let upload_id = between(&body, "<UploadId>", "</UploadId>")
251            .map(unescape_xml)
252            .ok_or_else(|| Error::msg("the bucket began a multipart upload and sent no UploadId"))?;
253        Ok(Multipart { key, upload_id })
254    }
255
256    /// A URL the browser `PUT`s one part to. Parts are numbered from 1, every
257    /// part but the last must be at least [`MIN_PART_BYTES`], and there may be
258    /// at most [`MAX_PARTS`]. The `ETag` header of the response is what
259    /// [`S3Storage::complete_multipart`] needs back.
260    pub fn presigned_part(&self, upload: &Multipart, number: u32, expires: Duration) -> Result<String> {
261        if number == 0 || number > MAX_PARTS {
262            return Err(Error::msg(format!(
263                "part numbers run from 1 to {MAX_PARTS}; {number} is not one"
264            )));
265        }
266        let number = number.to_string();
267        self.presigned(
268            "PUT",
269            &upload.key,
270            expires,
271            &[("partNumber", &number), ("uploadId", &upload.upload_id)],
272        )
273    }
274
275    /// Join the parts into the object. `parts` in any order; they are joined
276    /// by number.
277    pub async fn complete_multipart(&self, upload: &Multipart, mut parts: Vec<Part>) -> Result<()> {
278        if parts.is_empty() {
279            return Err(Error::msg("a multipart upload cannot be completed with no parts"));
280        }
281        parts.sort_by_key(|part| part.number);
282
283        let mut body = String::from("<CompleteMultipartUpload>");
284        for part in &parts {
285            // Quoted, whichever way the caller passed it on: the bucket sends
286            // `"abc"` and some browsers strip the quotes before handing it over.
287            let etag = part.etag.trim_matches('"');
288            body.push_str(&format!(
289                "<Part><PartNumber>{}</PartNumber><ETag>\"{etag}\"</ETag></Part>",
290                part.number
291            ));
292        }
293        body.push_str("</CompleteMultipartUpload>");
294
295        let query = format!("uploadId={}", sigv4::encode_segment(&upload.upload_id));
296        let extra = vec![("content-type".to_string(), "application/xml".to_string())];
297        let (url, headers) = self.signed("POST", &upload.key, &query, body.as_bytes(), &extra);
298
299        let mut request = self.client.post(url).body(body.into_bytes());
300        for (name, value) in headers {
301            request = request.header(&name, value);
302        }
303        let response = request.send().await?.error_for_status()?;
304
305        // S3 answers 200 and *then* may put an error in the body, because it
306        // starts the response before it has finished joining the parts. A
307        // `<Error>` in a 200 is a failure.
308        let text = response.text();
309        if text.contains("<Error>") {
310            let code = between(&text, "<Code>", "</Code>").unwrap_or("unknown");
311            return Err(Error::msg(format!("completing the multipart upload failed: {code}")));
312        }
313        Ok(())
314    }
315
316    /// Give up on a multipart upload, freeing the parts already stored.
317    ///
318    /// Call it when the browser goes away. Parts of an abandoned upload are
319    /// billed until this is called or a lifecycle rule removes them.
320    pub async fn abort_multipart(&self, upload: &Multipart) -> Result<()> {
321        let query = format!("uploadId={}", sigv4::encode_segment(&upload.upload_id));
322        let (url, headers) = self.signed("DELETE", &upload.key, &query, b"", &[]);
323
324        let mut request = self.client.delete(url);
325        for (name, value) in headers {
326            request = request.header(&name, value);
327        }
328        request.send().await?.error_for_status()?;
329        Ok(())
330    }
331
332    /// Tell the bucket to delete objects under `prefix` after `days`, and to
333    /// clean up multipart uploads nobody finished after the same time.
334    ///
335    /// **This replaces the bucket's whole lifecycle configuration.** A bucket
336    /// has one, and S3 has no way to add a rule without sending the rest. If
337    /// somebody has configured rules in the console, this overwrites them —
338    /// which is why the prefix is required rather than defaulting to
339    /// everything.
340    pub async fn expire_after(&self, prefix: &str, days: u32) -> Result<()> {
341        if prefix.trim().is_empty() {
342            return Err(Error::msg(
343                "an expiry rule needs a prefix: one for the whole bucket deletes everything in it",
344            ));
345        }
346        if days == 0 {
347            return Err(Error::msg("an expiry of zero days would delete objects as they arrive"));
348        }
349        // Normalised for safety and then given its trailing slash back.
350        // `normalize` is for keys and trims the slash; for a prefix the slash
351        // is the difference between `videos/raw/` and every key that merely
352        // starts with `videos/raw` — `videos/rawfootage/…` included, which a
353        // rule would then delete.
354        let trailing = prefix.trim_end().ends_with('/');
355        let mut prefix = normalize(prefix)?;
356        if trailing {
357            prefix.push('/');
358        }
359
360        let body = format!(
361            "<LifecycleConfiguration><Rule><ID>rustlavel-expire-{}</ID><Filter><Prefix>{}</Prefix></Filter>\
362             <Status>Enabled</Status><Expiration><Days>{days}</Days></Expiration>\
363             <AbortIncompleteMultipartUpload><DaysAfterInitiation>{days}</DaysAfterInitiation>\
364             </AbortIncompleteMultipartUpload></Rule></LifecycleConfiguration>",
365            prefix.trim_end_matches('/').replace('/', "-"),
366            escape_xml(&prefix),
367        );
368
369        // A bucket-level request: the key is empty and the query names the
370        // sub-resource. Path style puts the bucket in the path; virtual-host
371        // style has it in the host and the path is just `/`.
372        let extra = vec![
373            ("content-type".to_string(), "application/xml".to_string()),
374            ("content-md5".to_string(), md5_base64(body.as_bytes())),
375        ];
376        let (url, headers) = self.signed("PUT", "", "lifecycle=", body.as_bytes(), &extra);
377
378        let mut request = self.client.put(url).body(body.into_bytes());
379        for (name, value) in headers {
380            request = request.header(&name, value);
381        }
382        request.send().await?.error_for_status()?;
383        Ok(())
384    }
385}
386
387impl Storage for S3Storage {
388    async fn put(&self, path: &str, contents: Vec<u8>) -> Result<()> {
389        self.put_with(path, contents, Visibility::Private).await
390    }
391
392    async fn put_with(&self, path: &str, contents: Vec<u8>, visibility: Visibility) -> Result<()> {
393        let key = normalize(path)?;
394        let mut extra = vec![("content-type".to_string(), content_type(&key).to_string())];
395        if visibility == Visibility::Public {
396            extra.push(("x-amz-acl".to_string(), "public-read".to_string()));
397        }
398
399        let (url, headers) = self.signed("PUT", &key, "", &contents, &extra);
400
401        let mut request = self.client.put(url).body(contents);
402        for (name, value) in headers {
403            request = request.header(&name, value);
404        }
405
406        request.send().await?.error_for_status()?;
407        Ok(())
408    }
409
410    async fn get(&self, path: &str) -> Result<Vec<u8>> {
411        let key = normalize(path)?;
412        let (url, headers) = self.signed("GET", &key, "", b"", &[]);
413
414        let mut request = self.client.get(url);
415        for (name, value) in headers {
416            request = request.header(&name, value);
417        }
418
419        let response = request.send().await?;
420        if response.status.code() == 404 {
421            return Err(Error::msg(format!("`{path}` is not in bucket {}", self.config.bucket)));
422        }
423        Ok(response.error_for_status()?.body)
424    }
425
426    async fn exists(&self, path: &str) -> Result<bool> {
427        let key = normalize(path)?;
428        let (url, headers) = self.signed("HEAD", &key, "", b"", &[]);
429
430        let mut request = self.client.request(rustlavel_http::Method::Head, url);
431        for (name, value) in headers {
432            request = request.header(&name, value);
433        }
434
435        Ok(request.send().await?.is_success())
436    }
437
438    async fn delete(&self, path: &str) -> Result<()> {
439        let key = normalize(path)?;
440        let (url, headers) = self.signed("DELETE", &key, "", b"", &[]);
441
442        let mut request = self.client.delete(url);
443        for (name, value) in headers {
444            request = request.header(&name, value);
445        }
446
447        let response = request.send().await?;
448        // S3 answers 204 for a delete, and does not mind if it was already gone.
449        if response.status.code() == 404 || response.is_success() {
450            return Ok(());
451        }
452        response.error_for_status()?;
453        Ok(())
454    }
455
456    async fn size(&self, path: &str) -> Result<u64> {
457        let key = normalize(path)?;
458        let (url, headers) = self.signed("HEAD", &key, "", b"", &[]);
459
460        let mut request = self.client.request(rustlavel_http::Method::Head, url);
461        for (name, value) in headers {
462            request = request.header(&name, value);
463        }
464
465        let response = request.send().await?;
466        if !response.is_success() {
467            return Err(Error::msg(format!("`{path}` is not in bucket {}", self.config.bucket)));
468        }
469        Ok(response.headers.content_length().unwrap_or(0) as u64)
470    }
471
472    async fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
473        // ListObjectsV2 addresses the bucket, not an object.
474        let query = format!(
475            "list-type=2&prefix={}",
476            sigv4::encode_segment(prefix.trim_matches('/'))
477        );
478        let (url, headers) = self.signed("GET", "", &query, b"", &[]);
479
480        let mut request = self.client.get(url);
481        for (name, value) in headers {
482            request = request.header(&name, value);
483        }
484
485        let response = request.send().await?.error_for_status()?;
486        Ok(parse_listing(&response.text()))
487    }
488
489    fn url(&self, path: &str) -> Option<String> {
490        let key = normalize(path).ok()?;
491
492        match &self.config.public_url {
493            Some(base) => Some(format!("{}/{key}", base.trim_end_matches('/'))),
494            None => Some(format!(
495                "{}://{}{}",
496                self.config.scheme(),
497                self.config.host(),
498                sigv4::encode_path(&self.config.path_for(&key))
499            )),
500        }
501    }
502}
503
504/// Pull keys and sizes out of a ListObjectsV2 response.
505///
506/// S3 answers in XML and this is the only XML the framework meets, so it reads
507/// the two tags it needs rather than carrying a parser for the rest.
508fn parse_listing(xml: &str) -> Vec<Entry> {
509    let mut entries = Vec::new();
510
511    for chunk in xml.split("<Contents>").skip(1) {
512        let Some(key) = between(chunk, "<Key>", "</Key>") else { continue };
513        let size = between(chunk, "<Size>", "</Size>")
514            .and_then(|s| s.trim().parse().ok())
515            .unwrap_or(0);
516
517        entries.push(Entry { path: unescape_xml(key), size, is_directory: false });
518    }
519
520    for chunk in xml.split("<CommonPrefixes>").skip(1) {
521        if let Some(prefix) = between(chunk, "<Prefix>", "</Prefix>") {
522            entries.push(Entry {
523                path: unescape_xml(prefix.trim_end_matches('/')),
524                size: 0,
525                is_directory: true,
526            });
527        }
528    }
529
530    entries.sort_by(|a, b| a.path.cmp(&b.path));
531    entries
532}
533
534fn between<'a>(text: &'a str, open: &str, close: &str) -> Option<&'a str> {
535    let start = text.find(open)? + open.len();
536    let end = text[start..].find(close)? + start;
537    Some(&text[start..end])
538}
539
540fn escape_xml(text: &str) -> String {
541    text.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;")
542}
543
544/// S3 requires `Content-MD5` on lifecycle configuration, and on nothing else
545/// this crate sends. MD5 is not used for anything a person's security depends
546/// on here: it is the checksum the API demands, and the `md-5` crate is one of
547/// the cryptography exceptions the project already carries.
548fn md5_base64(bytes: &[u8]) -> String {
549    use md5::{Digest, Md5};
550    rustlavel_core::base64::encode(&Md5::digest(bytes))
551}
552
553fn unescape_xml(text: &str) -> String {
554    text.replace("&amp;", "&")
555        .replace("&lt;", "<")
556        .replace("&gt;", ">")
557        .replace("&quot;", "\"")
558        .replace("&apos;", "'")
559}
560
561/// `YYYYMMDDTHHMMSSZ` in UTC, the only format SigV4 accepts.
562fn timestamp() -> String {
563    let now = std::time::SystemTime::now()
564        .duration_since(std::time::UNIX_EPOCH)
565        .map(|d| d.as_secs() as i64)
566        .unwrap_or(0);
567
568    let (year, month, day) = civil_from_days(now.div_euclid(86_400));
569    let seconds = now.rem_euclid(86_400);
570
571    format!(
572        "{year:04}{month:02}{day:02}T{:02}{:02}{:02}Z",
573        seconds / 3600,
574        (seconds % 3600) / 60,
575        seconds % 60
576    )
577}
578
579fn civil_from_days(days: i64) -> (i64, u32, u32) {
580    let z = days + 719_468;
581    let era = z.div_euclid(146_097);
582    let day_of_era = z.rem_euclid(146_097);
583    let year_of_era =
584        (day_of_era - day_of_era / 1460 + day_of_era / 36_524 - day_of_era / 146_096) / 365;
585    let year = year_of_era + era * 400;
586    let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100);
587    let mp = (5 * day_of_year + 2) / 153;
588    let day = (day_of_year - (153 * mp + 2) / 5 + 1) as u32;
589    let month = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
590    (year + i64::from(month <= 2), month, day)
591}
592
593#[cfg(test)]
594mod tests {
595    use super::*;
596    use rustlavel_client::{Fake, FakeResponse};
597
598    fn config() -> S3Config {
599        S3Config {
600            bucket: "uploads".into(),
601            region: "eu-west-1".into(),
602            access_key: "AKIAEXAMPLE".into(),
603            secret_key: "secret".into(),
604            ..S3Config::default()
605        }
606    }
607
608    #[test]
609    fn addresses_aws_in_virtual_host_style_by_default() {
610        let storage = S3Storage::new(config());
611
612        assert_eq!(storage.config.host(), "uploads.s3.eu-west-1.amazonaws.com");
613        assert_eq!(storage.config.path_for("a/b.png"), "/a/b.png");
614        assert_eq!(
615            storage.url("a/b.png").as_deref(),
616            Some("https://uploads.s3.eu-west-1.amazonaws.com/a/b.png")
617        );
618    }
619
620    #[test]
621    fn a_custom_endpoint_switches_to_path_style() {
622        let storage = S3Storage::new(S3Config {
623            endpoint: Some("http://127.0.0.1:9000".into()),
624            ..config()
625        });
626
627        assert_eq!(storage.config.host(), "127.0.0.1:9000");
628        assert_eq!(storage.config.path_for("a.png"), "/uploads/a.png");
629        assert_eq!(storage.url("a.png").as_deref(), Some("http://127.0.0.1:9000/uploads/a.png"));
630    }
631
632    #[test]
633    fn a_public_url_overrides_the_api_host() {
634        let storage = S3Storage::new(S3Config {
635            public_url: Some("https://cdn.example.com".into()),
636            ..config()
637        });
638
639        assert_eq!(storage.url("a/b.png").as_deref(), Some("https://cdn.example.com/a/b.png"));
640    }
641
642    /// The whole point of a presigned URL: the credential is not in it, the
643    /// signature is, and it expires. A browser holding this can upload and
644    /// nothing else.
645    #[test]
646    fn a_presigned_put_carries_a_signature_and_no_secret() {
647        let storage = S3Storage::new(config());
648        let url = storage.presigned_put("videos/raw/abc.mp4", Duration::from_secs(900)).unwrap();
649
650        assert!(url.starts_with("https://uploads.s3.eu-west-1.amazonaws.com/videos/raw/abc.mp4?"), "{url}");
651        assert!(url.contains("X-Amz-Algorithm=AWS4-HMAC-SHA256"), "{url}");
652        assert!(url.contains("X-Amz-Credential=AKIAEXAMPLE%2F"), "{url}");
653        assert!(url.contains("X-Amz-Expires=900"), "{url}");
654        assert!(url.contains("X-Amz-SignedHeaders=host"), "{url}");
655        assert!(url.contains("&X-Amz-Signature="), "{url}");
656        assert!(!url.contains(&config().secret_key), "the secret is in the URL");
657        // The signature is the last thing, 64 hex characters.
658        let signature = url.rsplit("X-Amz-Signature=").next().unwrap();
659        assert_eq!(signature.len(), 64, "{signature}");
660        assert!(signature.bytes().all(|b| b.is_ascii_hexdigit()));
661    }
662
663    /// AWS refuses anything past seven days, and a URL that fails on the
664    /// bucket's side fails for the browser holding it, not for the developer
665    /// who minted it.
666    #[test]
667    fn a_presigned_url_cannot_outlive_what_the_bucket_allows() {
668        let storage = S3Storage::new(config());
669        let error = storage
670            .presigned_put("x", MAX_PRESIGN + Duration::from_secs(1))
671            .unwrap_err()
672            .to_string();
673        assert!(error.contains("seven days"), "{error}");
674        assert!(storage.presigned_put("x", MAX_PRESIGN).is_ok());
675    }
676
677    /// A path that escapes the root is refused before it is signed. Signing
678    /// `../` would hand a browser a valid URL to a key outside the prefix.
679    #[test]
680    fn a_presigned_url_refuses_an_escaping_path() {
681        let storage = S3Storage::new(config());
682        assert!(storage.presigned_put("../other-bucket-key", Duration::from_secs(60)).is_err());
683    }
684
685    #[tokio::test]
686    async fn a_multipart_upload_is_begun_and_completed_in_the_shape_s3_expects() {
687        let client = Client::new().faking(
688            Fake::new()
689                .on("uploads=", FakeResponse::text(
690                    "<InitiateMultipartUploadResult><Bucket>uploads</Bucket><Key>videos/big.mp4</Key>\
691                     <UploadId>2~abc.DEF</UploadId></InitiateMultipartUploadResult>",
692                ))
693                .fallback(FakeResponse::text("<CompleteMultipartUploadResult/>")),
694        );
695        let storage = S3Storage::with_client(config(), client);
696
697        let upload = storage.create_multipart("videos/big.mp4").await.unwrap();
698        assert_eq!(upload.upload_id, "2~abc.DEF");
699        assert_eq!(upload.key, "videos/big.mp4");
700
701        // Each part is a presigned PUT naming the part and the upload.
702        let url = storage.presigned_part(&upload, 3, Duration::from_secs(600)).unwrap();
703        assert!(url.contains("partNumber=3"), "{url}");
704        assert!(url.contains("uploadId=2~abc.DEF"), "{url}");
705        assert!(url.contains("X-Amz-Signature="), "{url}");
706
707        // Parts given out of order, ETags quoted and unquoted: S3 wants them by
708        // number and quoted, and the caller should not have to know either.
709        storage
710            .complete_multipart(&upload, vec![
711                Part { number: 2, etag: "\"e2\"".into() },
712                Part { number: 1, etag: "e1".into() },
713            ])
714            .await
715            .unwrap();
716
717        let fake = storage.client.fake().unwrap();
718        let complete = fake.recorded().into_iter().last().unwrap();
719        assert!(complete.url.contains("uploadId=2~abc.DEF"), "{}", complete.url);
720        let body = complete.body_text();
721        let one = body.find("<PartNumber>1</PartNumber>").expect("part 1");
722        let two = body.find("<PartNumber>2</PartNumber>").expect("part 2");
723        assert!(one < two, "parts were not joined by number: {body}");
724        assert!(body.contains("<ETag>\"e1\"</ETag>"), "an unquoted ETag was sent bare: {body}");
725        assert!(body.contains("<ETag>\"e2\"</ETag>"), "a quoted ETag was double-quoted: {body}");
726    }
727
728    /// S3 answers 200 and then puts the failure in the body, because it starts
729    /// the response before it has finished joining the parts.
730    #[tokio::test]
731    async fn a_failure_hidden_in_a_200_is_still_a_failure() {
732        let client = Client::new().faking(Fake::new().fallback(FakeResponse::text(
733            "<Error><Code>InvalidPart</Code><Message>One or more parts could not be found</Message></Error>",
734        )));
735        let storage = S3Storage::with_client(config(), client);
736        let upload = Multipart { key: "k".into(), upload_id: "u".into() };
737
738        let error = storage
739            .complete_multipart(&upload, vec![Part { number: 1, etag: "e".into() }])
740            .await
741            .unwrap_err()
742            .to_string();
743        assert!(error.contains("InvalidPart"), "{error}");
744    }
745
746    #[test]
747    fn a_part_number_outside_what_s3_accepts_is_refused_before_signing() {
748        let storage = S3Storage::new(config());
749        let upload = Multipart { key: "k".into(), upload_id: "u".into() };
750        assert!(storage.presigned_part(&upload, 0, Duration::from_secs(60)).is_err());
751        assert!(storage.presigned_part(&upload, MAX_PARTS + 1, Duration::from_secs(60)).is_err());
752        assert!(storage.presigned_part(&upload, MAX_PARTS, Duration::from_secs(60)).is_ok());
753    }
754
755    /// A lifecycle rule replaces the bucket's whole configuration, so one with
756    /// no prefix would delete everything in the bucket on a timer.
757    #[tokio::test]
758    async fn an_expiry_rule_needs_a_prefix_and_carries_the_checksum_s3_demands() {
759        let client = Client::new().faking(Fake::new().fallback(FakeResponse::text("")));
760        let storage = S3Storage::with_client(config(), client);
761
762        assert!(storage.expire_after("", 30).await.is_err(), "an empty prefix was accepted");
763        assert!(storage.expire_after("   ", 30).await.is_err());
764        assert!(storage.expire_after("tmp/", 0).await.is_err(), "zero days was accepted");
765
766        storage.expire_after("videos/raw/", 60).await.unwrap();
767
768        let fake = storage.client.fake().unwrap();
769        let sent = fake.recorded().into_iter().last().unwrap();
770        assert!(sent.url.ends_with("?lifecycle="), "{}", sent.url);
771        assert!(sent.headers.contains("content-md5"), "S3 refuses a lifecycle PUT without it");
772        let body = sent.body_text();
773        assert!(body.contains("<Prefix>videos/raw/</Prefix>"), "{body}");
774        assert!(body.contains("<Days>60</Days>"), "{body}");
775        assert!(body.contains("<DaysAfterInitiation>60</DaysAfterInitiation>"), "abandoned uploads are not cleaned up: {body}");
776    }
777
778    #[tokio::test]
779    async fn a_put_is_signed_and_typed() {
780        let client = Client::new().faking(Fake::new().fallback(FakeResponse::text("")));
781        let storage = S3Storage::with_client(config(), client);
782
783        storage.put("photos/holiday.png", b"bytes".to_vec()).await.unwrap();
784
785        let fake = storage.client.fake().unwrap();
786        let sent = &fake.recorded()[0];
787
788        assert_eq!(sent.url, "https://uploads.s3.eu-west-1.amazonaws.com/photos/holiday.png");
789        assert_eq!(sent.headers.get("content-type"), Some("image/png"));
790        assert!(sent.headers.get("authorization").unwrap().starts_with("AWS4-HMAC-SHA256 Credential=AKIAEXAMPLE/"));
791        assert!(sent.headers.contains("x-amz-content-sha256"));
792        assert_eq!(sent.headers.get("x-amz-acl"), None, "private by default");
793    }
794
795    #[tokio::test]
796    async fn a_public_put_asks_for_a_public_acl() {
797        let client = Client::new().faking(Fake::new().fallback(FakeResponse::text("")));
798        let storage = S3Storage::with_client(config(), client);
799
800        storage.put_with("a.txt", b"x".to_vec(), Visibility::Public).await.unwrap();
801
802        assert_eq!(
803            storage.client.fake().unwrap().recorded()[0].headers.get("x-amz-acl"),
804            Some("public-read")
805        );
806    }
807
808    #[tokio::test]
809    async fn a_missing_object_names_the_bucket() {
810        let client =
811            Client::new().faking(Fake::new().fallback(FakeResponse::text("no").status(404)));
812        let storage = S3Storage::with_client(config(), client);
813
814        let error = storage.get("gone.txt").await.unwrap_err().to_string();
815
816        assert!(error.contains("gone.txt"));
817        assert!(error.contains("uploads"));
818    }
819
820    #[tokio::test]
821    async fn deleting_something_absent_is_not_an_error() {
822        let client =
823            Client::new().faking(Fake::new().fallback(FakeResponse::text("no").status(404)));
824        let storage = S3Storage::with_client(config(), client);
825
826        assert!(storage.delete("gone.txt").await.is_ok());
827    }
828
829    #[tokio::test]
830    async fn a_listing_is_parsed_out_of_the_xml() {
831        let xml = r#"<?xml version="1.0"?>
832<ListBucketResult>
833  <Contents><Key>photos/a.png</Key><Size>1024</Size></Contents>
834  <Contents><Key>photos/b &amp; c.png</Key><Size>2048</Size></Contents>
835  <CommonPrefixes><Prefix>photos/thumbs/</Prefix></CommonPrefixes>
836</ListBucketResult>"#;
837
838        let client = Client::new().faking(Fake::new().fallback(FakeResponse::text(xml)));
839        let storage = S3Storage::with_client(config(), client);
840
841        let entries = storage.list("photos").await.unwrap();
842
843        assert_eq!(entries.len(), 3);
844        assert_eq!(entries[0].path, "photos/a.png");
845        assert_eq!(entries[0].size, 1024);
846        assert_eq!(entries[1].path, "photos/b & c.png");
847        assert!(entries[2].is_directory);
848    }
849
850    #[tokio::test]
851    async fn the_secret_key_never_reaches_a_request_or_an_error() {
852        let client =
853            Client::new().faking(Fake::new().fallback(FakeResponse::text("denied").status(403)));
854        let storage = S3Storage::with_client(config(), client);
855
856        let error = storage.get("a.txt").await.unwrap_err().to_string();
857        assert!(!error.contains("secret"));
858
859        let sent = &storage.client.fake().unwrap().recorded()[0];
860        for (_, value) in sent.headers.iter() {
861            assert!(!value.contains("secret"), "the secret key leaked into {value}");
862        }
863    }
864
865    #[test]
866    fn a_bucketless_configuration_says_what_to_set() {
867        let config = rustlavel_core::Config::new();
868        config.set("storage.driver", "s3");
869
870        let error = S3Config::from_config(&config).unwrap_err().to_string();
871        assert!(error.contains("storage.bucket"));
872    }
873
874    #[test]
875    fn timestamps_are_in_the_format_sigv4_requires() {
876        let stamp = timestamp();
877
878        assert_eq!(stamp.len(), 16);
879        assert!(stamp.ends_with('Z'));
880        assert_eq!(&stamp[8..9], "T");
881    }
882}
rustlavel_storage/s3.rs

rustlavel_storage/
s3.rs
