rws_storage/storage/
s3.rs

1use super::aws_credentials::CredentialsProvider;
2use super::aws_sigv4;
3use super::{Storage, StorageError};
4use crate::http_client::{Client, Response};
5use std::env;
6use std::time::{SystemTime, UNIX_EPOCH};
7
8/// Connection details for an S3-compatible bucket.
9#[derive(Clone, Debug)]
10pub struct S3Config {
11    pub bucket: String,
12    pub region: String,
13    /// Empty string means "no static key — auto-detect workload identity
14    /// (EKS IRSA, ECS task role, EC2 IMDSv2) instead". See
15    /// [`S3Storage::new`].
16    pub access_key: String,
17    /// Empty string means "no static key — auto-detect workload identity".
18    /// See [`S3Storage::new`].
19    pub secret_key: String,
20    /// Scheme + host, e.g. `https://s3.us-east-1.amazonaws.com`. Point this
21    /// at a custom host to use Cloudflare R2, MinIO, or any other
22    /// S3-compatible provider.
23    pub endpoint: String,
24}
25
26impl S3Config {
27    /// Read configuration from environment variables.
28    ///
29    /// | Variable | Default |
30    /// |---|---|
31    /// | `RWS_S3_BUCKET` | **(required)** |
32    /// | `RWS_S3_REGION` | `us-east-1` |
33    /// | `RWS_S3_ACCESS_KEY` | optional — falls back to workload identity (IRSA/ECS/IMDS) when unset |
34    /// | `RWS_S3_SECRET_KEY` | optional — falls back to workload identity (IRSA/ECS/IMDS) when unset |
35    /// | `RWS_S3_ENDPOINT` | `https://s3.{region}.amazonaws.com` (`http://` works too) |
36    ///
37    /// [`S3Storage::from_env`] also reads `RWS_S3_CA_FILE`.
38    pub fn from_env() -> Result<Self, StorageError> {
39        let bucket = env::var("RWS_S3_BUCKET")
40            .map_err(|_| StorageError::new("RWS_S3_BUCKET environment variable is required"))?;
41        let region = env::var("RWS_S3_REGION").unwrap_or_else(|_| "us-east-1".to_string());
42        let access_key = env::var("RWS_S3_ACCESS_KEY").unwrap_or_default();
43        let secret_key = env::var("RWS_S3_SECRET_KEY").unwrap_or_default();
44        let endpoint = env::var("RWS_S3_ENDPOINT")
45            .unwrap_or_else(|_| format!("https://s3.{region}.amazonaws.com"));
46        Ok(S3Config { bucket, region, access_key, secret_key, endpoint })
47    }
48}
49
50/// S3-compatible object storage (AWS S3, Cloudflare R2, MinIO, ...).
51///
52/// Signs every request with AWS Signature Version 4 using the outbound HTTP
53/// client (`crate::http_client::Client`) — no AWS SDK dependency. Uses
54/// path-style addressing (`{endpoint}/{bucket}/{key}`), which every
55/// S3-compatible provider supports; virtual-hosted-style (`{bucket}.{host}`)
56/// is not used since custom endpoints for R2/MinIO don't reliably support it.
57///
58/// Credentials come from `config.access_key`/`config.secret_key` when both
59/// are non-empty; otherwise they're auto-detected from EKS IRSA, ECS task
60/// role, or EC2 IMDSv2 (in that order), cached, and refreshed shortly before
61/// expiry. See `crate::storage::aws_credentials`.
62pub struct S3Storage {
63    config: S3Config,
64    client: Client,
65    credentials: CredentialsProvider,
66}
67
68impl S3Storage {
69    pub fn new(config: S3Config) -> Self {
70        let access_key = non_empty(&config.access_key);
71        let secret_key = non_empty(&config.secret_key);
72        let credentials = CredentialsProvider::detect(&config.region, access_key, secret_key);
73        S3Storage { config, client: Client::new(), credentials }
74    }
75
76    /// Trust the CA certificates in the PEM file at `path` for an
77    /// `https://` endpoint signed by a private CA, instead of the public
78    /// roots.
79    pub fn with_ca_file(mut self, path: &str) -> Result<Self, StorageError> {
80        self.client = Client::new().ca_file(path).map_err(|e| StorageError::new(e.to_string()))?;
81        Ok(self)
82    }
83
84    /// `S3Storage::new(S3Config::from_env()?)`, with
85    /// [`with_ca_file`](Self::with_ca_file) when `RWS_S3_CA_FILE` is set.
86    pub fn from_env() -> Result<Self, StorageError> {
87        let storage = S3Storage::new(S3Config::from_env()?);
88        match env::var("RWS_S3_CA_FILE").ok().filter(|v| !v.trim().is_empty()) {
89            Some(path) => storage.with_ca_file(path.trim()),
90            None => Ok(storage),
91        }
92    }
93
94    /// The `Host` the request goes out with — the endpoint's host, with its
95    /// port unless it's the scheme's default, as `http_client::Client`
96    /// writes it (since 18.21.0) — so the SigV4 signature covers exactly
97    /// that value. Before, the port was left out, and every request to an
98    /// endpoint on another port (MinIO's `:9000`) failed the signature check.
99    fn host(&self) -> String {
100        let endpoint = &self.config.endpoint;
101        let (default_port, rest) = match (endpoint.strip_prefix("https://"), endpoint.strip_prefix("http://")) {
102            (Some(rest), _) => (":443", rest),
103            (None, Some(rest)) => (":80", rest),
104            (None, None) => (":443", endpoint.as_str()),
105        };
106        let authority = rest.split('/').next().unwrap_or(rest);
107        authority.strip_suffix(default_port).unwrap_or(authority).to_string()
108    }
109
110    /// Percent-encoded `/{bucket}/{key}` — used both as the request path and
111    /// as the canonical URI signed by SigV4, so the two always match.
112    fn canonical_path(&self, key: &str) -> String {
113        aws_sigv4::uri_encode_path(&format!("/{}/{}", self.config.bucket, key))
114    }
115
116    fn request_url(&self, key: &str) -> String {
117        format!("{}{}", self.config.endpoint.trim_end_matches('/'), self.canonical_path(key))
118    }
119
120    fn signed_headers(&self, method: &str, key: &str, payload: &[u8]) -> Result<Vec<(String, String)>, StorageError> {
121        let creds = self.credentials.get()?;
122        let epoch_secs = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
123        Ok(aws_sigv4::sign(
124            method,
125            &self.host(),
126            &self.canonical_path(key),
127            payload,
128            &self.config.region,
129            &creds.access_key,
130            &creds.secret_key,
131            creds.session_token.as_deref(),
132            epoch_secs,
133            "s3",
134        ))
135    }
136
137    fn error_from_response(action: &str, key: &str, resp: &Response) -> StorageError {
138        StorageError::new(format!(
139            "S3 {action} '{key}' failed: HTTP {} {}",
140            resp.status(),
141            resp.text().unwrap_or_default()
142        ))
143    }
144}
145
146impl Storage for S3Storage {
147    fn put(&self, key: &str, data: &[u8], content_type: &str) -> Result<String, StorageError> {
148        let url = self.request_url(key);
149        let mut builder = self.client.put(&url).header("Content-Type", content_type);
150        for (name, value) in self.signed_headers("PUT", key, data)? {
151            // The `host` entry documents what was signed; the actual `Host:`
152            // header is already sent by `Client`, derived from `url`.
153            if name.eq_ignore_ascii_case("host") {
154                continue;
155            }
156            builder = builder.header(&name, &value);
157        }
158        let resp = builder
159            .body(data.to_vec())
160            .send()
161            .map_err(|e| StorageError::new(format!("S3 PUT '{key}' failed: {e}")))?;
162        if !resp.is_success() {
163            return Err(Self::error_from_response("PUT", key, &resp));
164        }
165        Ok(key.to_string())
166    }
167
168    fn get(&self, key: &str) -> Result<Vec<u8>, StorageError> {
169        let url = self.request_url(key);
170        let mut builder = self.client.get(&url);
171        for (name, value) in self.signed_headers("GET", key, b"")? {
172            if name.eq_ignore_ascii_case("host") {
173                continue;
174            }
175            builder = builder.header(&name, &value);
176        }
177        let resp = builder.send().map_err(|e| StorageError::new(format!("S3 GET '{key}' failed: {e}")))?;
178        if !resp.is_success() {
179            return Err(Self::error_from_response("GET", key, &resp));
180        }
181        Ok(resp.bytes().to_vec())
182    }
183
184    fn delete(&self, key: &str) -> Result<(), StorageError> {
185        let url = self.request_url(key);
186        let mut builder = self.client.delete(&url);
187        for (name, value) in self.signed_headers("DELETE", key, b"")? {
188            if name.eq_ignore_ascii_case("host") {
189                continue;
190            }
191            builder = builder.header(&name, &value);
192        }
193        let resp = builder.send().map_err(|e| StorageError::new(format!("S3 DELETE '{key}' failed: {e}")))?;
194        // S3 returns 204 whether or not the key existed — matches
195        // `Storage::delete`'s no-op-on-missing contract.
196        if !resp.is_success() {
197            return Err(Self::error_from_response("DELETE", key, &resp));
198        }
199        Ok(())
200    }
201
202    fn url(&self, key: &str) -> String {
203        self.request_url(key)
204    }
205
206    /// S3 multipart upload: parts of [`S3_PART_SIZE`] read and sent one at
207    /// a time, so memory stays at one part however large the object. An
208    /// object that fits in one part is a plain `PUT`. Any failure aborts the
209    /// upload, so S3 keeps no orphaned parts.
210    fn put_stream(&self, key: &str, reader: &mut dyn std::io::Read, content_type: &str) -> Result<String, StorageError> {
211        let mut part = read_part(reader, key)?;
212        if part.len() < S3_PART_SIZE {
213            return self.put(key, &part, content_type);
214        }
215        let upload_id = self.create_multipart_upload(key, content_type)?;
216        let mut etags: Vec<String> = Vec::new();
217        let uploaded = (|| -> Result<(), StorageError> {
218            loop {
219                let number = (etags.len() + 1).to_string();
220                etags.push(self.upload_part(key, &upload_id, &number, &part)?);
221                part = read_part(reader, key)?;
222                if part.is_empty() {
223                    return Ok(());
224                }
225            }
226        })()
227        .and_then(|_| self.complete_multipart_upload(key, &upload_id, &etags));
228        if let Err(e) = uploaded {
229            let _ = self.abort_multipart_upload(key, &upload_id);
230            return Err(e);
231        }
232        Ok(key.to_string())
233    }
234}
235
236/// Size of each part of an S3 multipart upload (S3's minimum is 5 MiB,
237/// except for the last part). 10 000 parts — S3's maximum — make objects of
238/// up to ~78 GiB.
239pub const S3_PART_SIZE: usize = 8 * 1024 * 1024;
240
241/// Up to [`S3_PART_SIZE`] bytes from `reader` (fewer only at the end).
242fn read_part(reader: &mut dyn std::io::Read, key: &str) -> Result<Vec<u8>, StorageError> {
243    let mut part = vec![0u8; S3_PART_SIZE];
244    let mut filled = 0;
245    while filled < part.len() {
246        match reader.read(&mut part[filled..]) {
247            Ok(0) => break,
248            Ok(n) => filled += n,
249            Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {}
250            Err(e) => return Err(StorageError::new(format!("failed to read the data for '{key}': {e}"))),
251        }
252    }
253    part.truncate(filled);
254    Ok(part)
255}
256
257impl S3Storage {
258    /// Send a request to `/{bucket}/{key}?{params}`, signed with the query.
259    fn send_with_query(&self, method: &str, key: &str, params: &[(&str, &str)], body: Vec<u8>, content_type: Option<&str>) -> Result<Response, StorageError> {
260        let query = aws_sigv4::canonical_query(params);
261        let url = format!("{}?{}", self.request_url(key), query);
262        let creds = self.credentials.get()?;
263        let epoch_secs = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
264        let headers = aws_sigv4::sign_with_query(
265            method, &self.host(), &self.canonical_path(key), params, &body, &self.config.region,
266            &creds.access_key, &creds.secret_key, creds.session_token.as_deref(), epoch_secs, "s3",
267        );
268        let mut builder = self.client.request(method, &url);
269        if let Some(content_type) = content_type {
270            builder = builder.header("Content-Type", content_type);
271        }
272        for (name, value) in headers {
273            if !name.eq_ignore_ascii_case("host") {
274                builder = builder.header(&name, &value);
275            }
276        }
277        builder.body(body).send().map_err(|e| StorageError::new(format!("S3 {method} '{key}' failed: {e}")))
278    }
279
280    fn create_multipart_upload(&self, key: &str, content_type: &str) -> Result<String, StorageError> {
281        let resp = self.send_with_query("POST", key, &[("uploads", "")], Vec::new(), Some(content_type))?;
282        if !resp.is_success() {
283            return Err(Self::error_from_response("CreateMultipartUpload", key, &resp));
284        }
285        xml_element(&resp.text().unwrap_or_default(), "UploadId")
286            .ok_or_else(|| StorageError::new(format!("S3 CreateMultipartUpload '{key}': no UploadId in the response")))
287    }
288
289    fn upload_part(&self, key: &str, upload_id: &str, number: &str, data: &[u8]) -> Result<String, StorageError> {
290        let resp = self.send_with_query("PUT", key, &[("partNumber", number), ("uploadId", upload_id)], data.to_vec(), None)?;
291        if !resp.is_success() {
292            return Err(Self::error_from_response("UploadPart", key, &resp));
293        }
294        resp.header("etag").map(str::to_string)
295            .ok_or_else(|| StorageError::new(format!("S3 UploadPart '{key}' part {number}: no ETag in the response")))
296    }
297
298    fn complete_multipart_upload(&self, key: &str, upload_id: &str, etags: &[String]) -> Result<(), StorageError> {
299        let mut body = String::from("<CompleteMultipartUpload>");
300        for (i, etag) in etags.iter().enumerate() {
301            body.push_str(&format!("<Part><PartNumber>{}</PartNumber><ETag>{}</ETag></Part>", i + 1, xml_escape(etag)));
302        }
303        body.push_str("</CompleteMultipartUpload>");
304        let resp = self.send_with_query("POST", key, &[("uploadId", upload_id)], body.into_bytes(), Some("application/xml"))?;
305        // S3 can answer 200 and still report an error in the body.
306        let text = resp.text().unwrap_or_default();
307        if !resp.is_success() || text.contains("<Error>") {
308            return Err(StorageError::new(format!("S3 CompleteMultipartUpload '{key}' failed: HTTP {} {text}", resp.status())));
309        }
310        Ok(())
311    }
312
313    fn abort_multipart_upload(&self, key: &str, upload_id: &str) -> Result<(), StorageError> {
314        let resp = self.send_with_query("DELETE", key, &[("uploadId", upload_id)], Vec::new(), None)?;
315        if !resp.is_success() {
316            return Err(Self::error_from_response("AbortMultipartUpload", key, &resp));
317        }
318        Ok(())
319    }
320}
321
322/// The text of the first `<name>…</name>` in `xml`.
323fn xml_element(xml: &str, name: &str) -> Option<String> {
324    let open = format!("<{name}>");
325    let start = xml.find(&open)? + open.len();
326    let end = xml[start..].find(&format!("</{name}>"))?;
327    Some(xml[start..start + end].to_string())
328}
329
330fn xml_escape(text: &str) -> String {
331    text.replace('&', "&amp;").replace('<', "&lt;").replace('>', "&gt;").replace('"', "&quot;")
332}
333
334fn non_empty(s: &str) -> Option<String> {
335    if s.is_empty() { None } else { Some(s.to_string()) }
336}
337
338#[cfg(test)]
339mod tests;
rws_storage/storage/s3.rs

rws_storage/storage/
s3.rs
