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 crate::{Entry, Storage, Visibility, content_type, normalize};
9use rustlavel_client::Client;
10use rustlavel_core::{Config, Error, Result};
11
12#[derive(Debug, Clone)]
13pub struct S3Config {
14    pub bucket: String,
15    pub region: String,
16    pub access_key: String,
17    pub secret_key: String,
18    /// For a non-AWS service: `https://<account>.r2.cloudflarestorage.com`.
19    pub endpoint: Option<String>,
20    /// MinIO and most self-hosted services need the bucket in the path rather
21    /// than in the hostname.
22    pub path_style: bool,
23    /// Where public objects are reached, when that differs from the API.
24    pub public_url: Option<String>,
25}
26
27impl Default for S3Config {
28    fn default() -> Self {
29        S3Config {
30            bucket: String::new(),
31            region: "us-east-1".into(),
32            access_key: String::new(),
33            secret_key: String::new(),
34            endpoint: None,
35            path_style: false,
36            public_url: None,
37        }
38    }
39}
40
41impl S3Config {
42    pub fn from_config(config: &Config) -> Result<S3Config> {
43        let settings = S3Config {
44            bucket: config.string("storage.bucket", ""),
45            region: config.string("storage.region", "us-east-1"),
46            access_key: config.string("storage.access_key", ""),
47            secret_key: config.string("storage.secret_key", ""),
48            endpoint: non_empty(config.string("storage.endpoint", "")),
49            path_style: config.bool("storage.path_style", false),
50            public_url: non_empty(config.string("storage.public_url", "")),
51        };
52
53        if settings.bucket.is_empty() {
54            return Err(Error::msg(
55                "the s3 storage driver needs `storage.bucket`. Set it in config/storage.json or \
56                 point STORAGE_BUCKET at it in .env."
57                    .to_string(),
58            ));
59        }
60        Ok(settings)
61    }
62
63    /// The host requests go to.
64    fn host(&self) -> String {
65        match &self.endpoint {
66            Some(endpoint) => endpoint
67                .trim_start_matches("https://")
68                .trim_start_matches("http://")
69                .trim_end_matches('/')
70                .to_string(),
71            None if self.path_style => format!("s3.{}.amazonaws.com", self.region),
72            None => format!("{}.s3.{}.amazonaws.com", self.bucket, self.region),
73        }
74    }
75
76    fn scheme(&self) -> &'static str {
77        match &self.endpoint {
78            Some(endpoint) if endpoint.starts_with("http://") => "http",
79            _ => "https",
80        }
81    }
82
83    /// The request path, which includes the bucket in path style.
84    fn path_for(&self, key: &str) -> String {
85        if self.path_style || self.endpoint.is_some() {
86            format!("/{}/{key}", self.bucket)
87        } else {
88            format!("/{key}")
89        }
90    }
91}
92
93fn non_empty(value: String) -> Option<String> {
94    (!value.is_empty()).then_some(value)
95}
96
97pub struct S3Storage {
98    config: S3Config,
99    client: Client,
100}
101
102impl S3Storage {
103    pub fn new(config: S3Config) -> Self {
104        S3Storage { config, client: Client::new().retries(2) }
105    }
106
107    /// Use a prepared client — how a test drives this against a fake.
108    pub fn with_client(config: S3Config, client: Client) -> Self {
109        S3Storage { config, client }
110    }
111
112    pub fn config(&self) -> &S3Config {
113        &self.config
114    }
115
116    /// Build a signed request for one object operation.
117    fn signed(
118        &self,
119        method: &str,
120        key: &str,
121        query: &str,
122        body: &[u8],
123        extra: &[(String, String)],
124    ) -> (String, Vec<(String, String)>) {
125        let host = self.config.host();
126        let path = self.config.path_for(key);
127        let timestamp = timestamp();
128        let payload_hash = sigv4::sha256_hex(body);
129
130        let mut headers = vec![
131            ("host".to_string(), host.clone()),
132            ("x-amz-content-sha256".to_string(), payload_hash.clone()),
133            ("x-amz-date".to_string(), timestamp.clone()),
134        ];
135        headers.extend_from_slice(extra);
136
137        let signing = Signing {
138            access_key: &self.config.access_key,
139            secret_key: &self.config.secret_key,
140            region: &self.config.region,
141            service: "s3",
142            timestamp: &timestamp,
143        };
144
145        let authorization =
146            sigv4::authorization(&signing, method, &path, query, &headers, &payload_hash);
147        headers.push(("authorization".to_string(), authorization));
148
149        let url = if query.is_empty() {
150            format!("{}://{host}{}", self.config.scheme(), sigv4::encode_path(&path))
151        } else {
152            format!("{}://{host}{}?{query}", self.config.scheme(), sigv4::encode_path(&path))
153        };
154
155        (url, headers)
156    }
157}
158
159impl Storage for S3Storage {
160    async fn put(&self, path: &str, contents: Vec<u8>) -> Result<()> {
161        self.put_with(path, contents, Visibility::Private).await
162    }
163
164    async fn put_with(&self, path: &str, contents: Vec<u8>, visibility: Visibility) -> Result<()> {
165        let key = normalize(path)?;
166        let mut extra = vec![("content-type".to_string(), content_type(&key).to_string())];
167        if visibility == Visibility::Public {
168            extra.push(("x-amz-acl".to_string(), "public-read".to_string()));
169        }
170
171        let (url, headers) = self.signed("PUT", &key, "", &contents, &extra);
172
173        let mut request = self.client.put(url).body(contents);
174        for (name, value) in headers {
175            request = request.header(&name, value);
176        }
177
178        request.send().await?.error_for_status()?;
179        Ok(())
180    }
181
182    async fn get(&self, path: &str) -> Result<Vec<u8>> {
183        let key = normalize(path)?;
184        let (url, headers) = self.signed("GET", &key, "", b"", &[]);
185
186        let mut request = self.client.get(url);
187        for (name, value) in headers {
188            request = request.header(&name, value);
189        }
190
191        let response = request.send().await?;
192        if response.status.code() == 404 {
193            return Err(Error::msg(format!("`{path}` is not in bucket {}", self.config.bucket)));
194        }
195        Ok(response.error_for_status()?.body)
196    }
197
198    async fn exists(&self, path: &str) -> Result<bool> {
199        let key = normalize(path)?;
200        let (url, headers) = self.signed("HEAD", &key, "", b"", &[]);
201
202        let mut request = self.client.request(rustlavel_http::Method::Head, url);
203        for (name, value) in headers {
204            request = request.header(&name, value);
205        }
206
207        Ok(request.send().await?.is_success())
208    }
209
210    async fn delete(&self, path: &str) -> Result<()> {
211        let key = normalize(path)?;
212        let (url, headers) = self.signed("DELETE", &key, "", b"", &[]);
213
214        let mut request = self.client.delete(url);
215        for (name, value) in headers {
216            request = request.header(&name, value);
217        }
218
219        let response = request.send().await?;
220        // S3 answers 204 for a delete, and does not mind if it was already gone.
221        if response.status.code() == 404 || response.is_success() {
222            return Ok(());
223        }
224        response.error_for_status()?;
225        Ok(())
226    }
227
228    async fn size(&self, path: &str) -> Result<u64> {
229        let key = normalize(path)?;
230        let (url, headers) = self.signed("HEAD", &key, "", b"", &[]);
231
232        let mut request = self.client.request(rustlavel_http::Method::Head, url);
233        for (name, value) in headers {
234            request = request.header(&name, value);
235        }
236
237        let response = request.send().await?;
238        if !response.is_success() {
239            return Err(Error::msg(format!("`{path}` is not in bucket {}", self.config.bucket)));
240        }
241        Ok(response.headers.content_length().unwrap_or(0) as u64)
242    }
243
244    async fn list(&self, prefix: &str) -> Result<Vec<Entry>> {
245        // ListObjectsV2 addresses the bucket, not an object.
246        let query = format!(
247            "list-type=2&prefix={}",
248            sigv4::encode_segment(prefix.trim_matches('/'))
249        );
250        let (url, headers) = self.signed("GET", "", &query, b"", &[]);
251
252        let mut request = self.client.get(url);
253        for (name, value) in headers {
254            request = request.header(&name, value);
255        }
256
257        let response = request.send().await?.error_for_status()?;
258        Ok(parse_listing(&response.text()))
259    }
260
261    fn url(&self, path: &str) -> Option<String> {
262        let key = normalize(path).ok()?;
263
264        match &self.config.public_url {
265            Some(base) => Some(format!("{}/{key}", base.trim_end_matches('/'))),
266            None => Some(format!(
267                "{}://{}{}",
268                self.config.scheme(),
269                self.config.host(),
270                sigv4::encode_path(&self.config.path_for(&key))
271            )),
272        }
273    }
274}
275
276/// Pull keys and sizes out of a ListObjectsV2 response.
277///
278/// S3 answers in XML and this is the only XML the framework meets, so it reads
279/// the two tags it needs rather than carrying a parser for the rest.
280fn parse_listing(xml: &str) -> Vec<Entry> {
281    let mut entries = Vec::new();
282
283    for chunk in xml.split("<Contents>").skip(1) {
284        let Some(key) = between(chunk, "<Key>", "</Key>") else { continue };
285        let size = between(chunk, "<Size>", "</Size>")
286            .and_then(|s| s.trim().parse().ok())
287            .unwrap_or(0);
288
289        entries.push(Entry { path: unescape_xml(key), size, is_directory: false });
290    }
291
292    for chunk in xml.split("<CommonPrefixes>").skip(1) {
293        if let Some(prefix) = between(chunk, "<Prefix>", "</Prefix>") {
294            entries.push(Entry {
295                path: unescape_xml(prefix.trim_end_matches('/')),
296                size: 0,
297                is_directory: true,
298            });
299        }
300    }
301
302    entries.sort_by(|a, b| a.path.cmp(&b.path));
303    entries
304}
305
306fn between<'a>(text: &'a str, open: &str, close: &str) -> Option<&'a str> {
307    let start = text.find(open)? + open.len();
308    let end = text[start..].find(close)? + start;
309    Some(&text[start..end])
310}
311
312fn unescape_xml(text: &str) -> String {
313    text.replace("&amp;", "&")
314        .replace("&lt;", "<")
315        .replace("&gt;", ">")
316        .replace("&quot;", "\"")
317        .replace("&apos;", "'")
318}
319
320/// `YYYYMMDDTHHMMSSZ` in UTC, the only format SigV4 accepts.
321fn timestamp() -> String {
322    let now = std::time::SystemTime::now()
323        .duration_since(std::time::UNIX_EPOCH)
324        .map(|d| d.as_secs() as i64)
325        .unwrap_or(0);
326
327    let (year, month, day) = civil_from_days(now.div_euclid(86_400));
328    let seconds = now.rem_euclid(86_400);
329
330    format!(
331        "{year:04}{month:02}{day:02}T{:02}{:02}{:02}Z",
332        seconds / 3600,
333        (seconds % 3600) / 60,
334        seconds % 60
335    )
336}
337
338fn civil_from_days(days: i64) -> (i64, u32, u32) {
339    let z = days + 719_468;
340    let era = z.div_euclid(146_097);
341    let day_of_era = z.rem_euclid(146_097);
342    let year_of_era =
343        (day_of_era - day_of_era / 1460 + day_of_era / 36_524 - day_of_era / 146_096) / 365;
344    let year = year_of_era + era * 400;
345    let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100);
346    let mp = (5 * day_of_year + 2) / 153;
347    let day = (day_of_year - (153 * mp + 2) / 5 + 1) as u32;
348    let month = if mp < 10 { mp + 3 } else { mp - 9 } as u32;
349    (year + i64::from(month <= 2), month, day)
350}
351
352#[cfg(test)]
353mod tests {
354    use super::*;
355    use rustlavel_client::{Fake, FakeResponse};
356
357    fn config() -> S3Config {
358        S3Config {
359            bucket: "uploads".into(),
360            region: "eu-west-1".into(),
361            access_key: "AKIAEXAMPLE".into(),
362            secret_key: "secret".into(),
363            ..S3Config::default()
364        }
365    }
366
367    #[test]
368    fn addresses_aws_in_virtual_host_style_by_default() {
369        let storage = S3Storage::new(config());
370
371        assert_eq!(storage.config.host(), "uploads.s3.eu-west-1.amazonaws.com");
372        assert_eq!(storage.config.path_for("a/b.png"), "/a/b.png");
373        assert_eq!(
374            storage.url("a/b.png").as_deref(),
375            Some("https://uploads.s3.eu-west-1.amazonaws.com/a/b.png")
376        );
377    }
378
379    #[test]
380    fn a_custom_endpoint_switches_to_path_style() {
381        let storage = S3Storage::new(S3Config {
382            endpoint: Some("http://127.0.0.1:9000".into()),
383            ..config()
384        });
385
386        assert_eq!(storage.config.host(), "127.0.0.1:9000");
387        assert_eq!(storage.config.path_for("a.png"), "/uploads/a.png");
388        assert_eq!(storage.url("a.png").as_deref(), Some("http://127.0.0.1:9000/uploads/a.png"));
389    }
390
391    #[test]
392    fn a_public_url_overrides_the_api_host() {
393        let storage = S3Storage::new(S3Config {
394            public_url: Some("https://cdn.example.com".into()),
395            ..config()
396        });
397
398        assert_eq!(storage.url("a/b.png").as_deref(), Some("https://cdn.example.com/a/b.png"));
399    }
400
401    #[tokio::test]
402    async fn a_put_is_signed_and_typed() {
403        let client = Client::new().faking(Fake::new().fallback(FakeResponse::text("")));
404        let storage = S3Storage::with_client(config(), client);
405
406        storage.put("photos/holiday.png", b"bytes".to_vec()).await.unwrap();
407
408        let fake = storage.client.fake().unwrap();
409        let sent = &fake.recorded()[0];
410
411        assert_eq!(sent.url, "https://uploads.s3.eu-west-1.amazonaws.com/photos/holiday.png");
412        assert_eq!(sent.headers.get("content-type"), Some("image/png"));
413        assert!(sent.headers.get("authorization").unwrap().starts_with("AWS4-HMAC-SHA256 Credential=AKIAEXAMPLE/"));
414        assert!(sent.headers.contains("x-amz-content-sha256"));
415        assert_eq!(sent.headers.get("x-amz-acl"), None, "private by default");
416    }
417
418    #[tokio::test]
419    async fn a_public_put_asks_for_a_public_acl() {
420        let client = Client::new().faking(Fake::new().fallback(FakeResponse::text("")));
421        let storage = S3Storage::with_client(config(), client);
422
423        storage.put_with("a.txt", b"x".to_vec(), Visibility::Public).await.unwrap();
424
425        assert_eq!(
426            storage.client.fake().unwrap().recorded()[0].headers.get("x-amz-acl"),
427            Some("public-read")
428        );
429    }
430
431    #[tokio::test]
432    async fn a_missing_object_names_the_bucket() {
433        let client =
434            Client::new().faking(Fake::new().fallback(FakeResponse::text("no").status(404)));
435        let storage = S3Storage::with_client(config(), client);
436
437        let error = storage.get("gone.txt").await.unwrap_err().to_string();
438
439        assert!(error.contains("gone.txt"));
440        assert!(error.contains("uploads"));
441    }
442
443    #[tokio::test]
444    async fn deleting_something_absent_is_not_an_error() {
445        let client =
446            Client::new().faking(Fake::new().fallback(FakeResponse::text("no").status(404)));
447        let storage = S3Storage::with_client(config(), client);
448
449        assert!(storage.delete("gone.txt").await.is_ok());
450    }
451
452    #[tokio::test]
453    async fn a_listing_is_parsed_out_of_the_xml() {
454        let xml = r#"<?xml version="1.0"?>
455<ListBucketResult>
456  <Contents><Key>photos/a.png</Key><Size>1024</Size></Contents>
457  <Contents><Key>photos/b &amp; c.png</Key><Size>2048</Size></Contents>
458  <CommonPrefixes><Prefix>photos/thumbs/</Prefix></CommonPrefixes>
459</ListBucketResult>"#;
460
461        let client = Client::new().faking(Fake::new().fallback(FakeResponse::text(xml)));
462        let storage = S3Storage::with_client(config(), client);
463
464        let entries = storage.list("photos").await.unwrap();
465
466        assert_eq!(entries.len(), 3);
467        assert_eq!(entries[0].path, "photos/a.png");
468        assert_eq!(entries[0].size, 1024);
469        assert_eq!(entries[1].path, "photos/b & c.png");
470        assert!(entries[2].is_directory);
471    }
472
473    #[tokio::test]
474    async fn the_secret_key_never_reaches_a_request_or_an_error() {
475        let client =
476            Client::new().faking(Fake::new().fallback(FakeResponse::text("denied").status(403)));
477        let storage = S3Storage::with_client(config(), client);
478
479        let error = storage.get("a.txt").await.unwrap_err().to_string();
480        assert!(!error.contains("secret"));
481
482        let sent = &storage.client.fake().unwrap().recorded()[0];
483        for (_, value) in sent.headers.iter() {
484            assert!(!value.contains("secret"), "the secret key leaked into {value}");
485        }
486    }
487
488    #[test]
489    fn a_bucketless_configuration_says_what_to_set() {
490        let config = rustlavel_core::Config::new();
491        config.set("storage.driver", "s3");
492
493        let error = S3Config::from_config(&config).unwrap_err().to_string();
494        assert!(error.contains("storage.bucket"));
495    }
496
497    #[test]
498    fn timestamps_are_in_the_format_sigv4_requires() {
499        let stamp = timestamp();
500
501        assert_eq!(stamp.len(), 16);
502        assert!(stamp.ends_with('Z'));
503        assert_eq!(&stamp[8..9], "T");
504    }
505}
rustlavel_storage/s3.rs

rustlavel_storage/
s3.rs
