sofie_storage/
s3.rs

1mod signature;
2
3use std::sync::Arc;
4
5use bytes::{Bytes, BytesMut};
6use sofie_client::{Outgoing, Response, Vendor};
7use sofie_core::{Error, Result, Timestamp};
8
9use crate::{DownloadLink, Pending, Reader, StorageKey, Store, Writer};
10use signature::{Canonical, Credentials, query_string, sha256_hex, uri_encode};
11
12/// S3 takes parts of 5 MiB or more but the last; one part is held in memory at a time.
13const PART_SIZE: usize = 8 * 1024 * 1024;
14/// A first part grows from here instead of from empty by doubling.
15const FIRST_BUFFER: usize = 256 * 1024;
16const READ_CHUNK: u64 = 8 * 1024 * 1024;
17/// Enough for a type sniff, so a download opened and dropped does not hold a whole chunk.
18const FIRST_READ_CHUNK: u64 = 256 * 1024;
19const EMPTY_BODY_HASH: &str = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855";
20
21/// `endpoint` defaults to AWS's for the region; another S3-compatible service names its own,
22/// usually with `path_style`. On AWS the credentials need `s3:ListBucket` besides the object
23/// permissions: without it a missing file answers 403, read as a failure rather than as absent.
24pub struct S3Settings {
25    pub endpoint: Option<String>,
26    pub region: String,
27    pub bucket: String,
28    pub access_key_id: String,
29    pub secret_access_key: String,
30    pub path_style: bool,
31}
32
33impl S3Settings {
34    /// What the adapter's `Outgoing` is built with; `private` allows a service inside the network.
35    pub fn vendor(&self, private: bool) -> Result<Vendor> {
36        Ok(Vendor {
37            base_url: Some(self.base_url()?),
38            private: Some(private),
39            max_response_bytes: Some(READ_CHUNK),
40            ..Vendor::default()
41        })
42    }
43
44    fn base_url(&self) -> Result<String> {
45        let place = self.place()?;
46        Ok(format!("{}://{}", place.scheme, place.host))
47    }
48
49    fn place(&self) -> Result<Place> {
50        let is_bucket = (3..=63).contains(&self.bucket.len())
51            && self.bucket.bytes().all(|byte| {
52                byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'.' | b'-')
53            });
54        if !is_bucket {
55            return Err(Error::internal_message(format!(
56                "the bucket `{}` is not a bucket name: 3 to 63 lowercase letters, digits, `.` and `-`",
57                self.bucket
58            )));
59        }
60        let endpoint = self
61            .endpoint
62            .clone()
63            .unwrap_or_else(|| format!("https://s3.{}.amazonaws.com", self.region));
64        let (scheme, host) = split_endpoint(&endpoint)?;
65        Ok(if self.path_style {
66            Place {
67                scheme,
68                host,
69                prefix: format!("/{}", self.bucket),
70            }
71        } else {
72            Place {
73                scheme,
74                host: format!("{}.{host}", self.bucket),
75                prefix: String::new(),
76            }
77        })
78    }
79}
80
81struct Place {
82    scheme: &'static str,
83    /// As sofie-client sends it in `Host`: lowercase, the port only when not the scheme's own.
84    host: String,
85    prefix: String,
86}
87
88fn split_endpoint(endpoint: &str) -> Result<(&'static str, String)> {
89    let refused = |reason: &str| {
90        Error::internal_message(format!("the S3 endpoint `{endpoint}` is refused: {reason}"))
91    };
92    let (scheme, rest) = endpoint
93        .split_once("://")
94        .ok_or_else(|| refused("it is not an absolute URL"))?;
95    let (scheme, default_port) = match scheme {
96        "https" => ("https", ":443"),
97        "http" => ("http", ":80"),
98        _ => return Err(refused("its scheme is neither https nor http")),
99    };
100    let authority = rest.strip_suffix('/').unwrap_or(rest).to_ascii_lowercase();
101    if authority.is_empty() || authority.contains(['/', '@', '?', '#']) {
102        return Err(refused("it must be a scheme and a host, without a path"));
103    }
104    let host = authority
105        .strip_suffix(default_port)
106        .map_or_else(|| authority.clone(), str::to_owned);
107    Ok((scheme, host))
108}
109
110#[derive(Clone)]
111pub struct S3 {
112    shared: Arc<Shared>,
113}
114
115struct Shared {
116    http: Outgoing,
117    credentials: Credentials,
118    place: Place,
119}
120
121impl S3 {
122    /// `http` must be built with `settings.vendor(..)`.
123    pub fn new(http: Outgoing, settings: S3Settings) -> Result<Self> {
124        let place = settings.place()?;
125        Ok(Self {
126            shared: Arc::new(Shared {
127                http,
128                credentials: Credentials {
129                    access_key_id: settings.access_key_id,
130                    secret_access_key: settings.secret_access_key,
131                    region: settings.region,
132                },
133                place,
134            }),
135        })
136    }
137}
138
139#[derive(Clone, Copy)]
140enum Verb {
141    Get,
142    Put,
143    Post,
144    Delete,
145}
146
147impl Verb {
148    fn as_str(self) -> &'static str {
149        match self {
150            Self::Get => "GET",
151            Self::Put => "PUT",
152            Self::Post => "POST",
153            Self::Delete => "DELETE",
154        }
155    }
156}
157
158struct Call<'a> {
159    verb: Verb,
160    key: &'a StorageKey,
161    query: Vec<(String, String)>,
162    range: Option<String>,
163    if_match: Option<String>,
164    body: Bytes,
165}
166
167impl<'a> Call<'a> {
168    fn new(verb: Verb, key: &'a StorageKey) -> Self {
169        Self {
170            verb,
171            key,
172            query: Vec::new(),
173            range: None,
174            if_match: None,
175            body: Bytes::new(),
176        }
177    }
178
179    fn query(mut self, name: &str, value: &str) -> Self {
180        self.query.push((name.to_owned(), value.to_owned()));
181        self
182    }
183
184    fn body(self, body: Bytes) -> Self {
185        Self { body, ..self }
186    }
187}
188
189impl Shared {
190    fn path(&self, key: &StorageKey) -> String {
191        uri_encode(&format!("{}/{key}", self.place.prefix), true)
192    }
193
194    async fn send(&self, call: Call<'_>) -> Result<Response> {
195        let path = self.path(call.key);
196        let payload_hash = payload_hash(&call.body).await?;
197        let now = Timestamp::now();
198        let mut headers = vec![
199            ("host".to_owned(), self.place.host.clone()),
200            ("x-amz-content-sha256".to_owned(), payload_hash.clone()),
201            ("x-amz-date".to_owned(), signature::amz_date(now)),
202        ];
203        if let Some(range) = &call.range {
204            headers.push(("range".to_owned(), range.clone()));
205        }
206        if let Some(etag) = &call.if_match {
207            headers.push(("if-match".to_owned(), etag.clone()));
208        }
209        let query = query_string(&call.query);
210        let authorization = self.credentials.authorization(
211            Canonical {
212                method: call.verb.as_str(),
213                path: &path,
214                query: call.query,
215                headers: headers.clone(),
216                payload_hash: &payload_hash,
217            },
218            now,
219        );
220        let target = if query.is_empty() {
221            path
222        } else {
223            format!("{path}?{query}")
224        };
225        let mut request = match call.verb {
226            Verb::Get => self.http.get(&target),
227            Verb::Put => self.http.put(&target),
228            Verb::Post => self.http.post(&target),
229            Verb::Delete => self.http.delete(&target),
230        };
231        for (name, value) in headers.iter().skip(1) {
232            request = request.header(name, value);
233        }
234        request = request.header("authorization", &authorization);
235        if !call.body.is_empty() {
236            request = request.body("application/octet-stream", call.body);
237        }
238        request.send().await
239    }
240
241    async fn expect(&self, call: Call<'_>, operation: &str) -> Result<Response> {
242        let response = self.send(call).await?;
243        if (200..300).contains(&response.status()) && !is_error_document(response.body()) {
244            return Ok(response);
245        }
246        Err(refused(&response, operation))
247    }
248}
249
250async fn payload_hash(body: &Bytes) -> Result<String> {
251    if body.is_empty() {
252        return Ok(EMPTY_BODY_HASH.to_owned());
253    }
254    // An 8 MiB part hashes for milliseconds, which would stall the async worker.
255    let body = body.clone();
256    tokio::task::spawn_blocking(move || sha256_hex(&body))
257        .await
258        .map_err(|error| Error::internal("hashing an S3 request body", error))
259}
260
261/// S3 may answer 200 to completing an upload and still have failed it, after whitespace it sends
262/// to keep the connection open; the root element decides, not a substring.
263fn is_error_document(body: &[u8]) -> bool {
264    let mut body = body.trim_ascii_start();
265    if body.starts_with(b"<?xml") {
266        let Some(end) = body.windows(2).position(|pair| pair == b"?>") else {
267            return false;
268        };
269        body = body[end + 2..].trim_ascii_start();
270    }
271    body.strip_prefix(b"<Error")
272        .is_some_and(|rest| matches!(rest.first(), Some(b'>' | b' ' | b'\t' | b'\r' | b'\n')))
273}
274
275/// S3's error code names the problem; its message and resource may carry more than a log should.
276fn refused(response: &Response, operation: &str) -> Error {
277    let code = element(response.body(), "Code").unwrap_or_default();
278    Error::internal_message(format!(
279        "S3 refused to {operation}: {} {code}",
280        response.status()
281    ))
282}
283
284/// The element's text with XML's five entities read; an S3-compatible server may escape an id.
285fn element(body: &[u8], name: &str) -> Option<String> {
286    let text = std::str::from_utf8(body).ok()?;
287    let start = text.find(&format!("<{name}>"))? + name.len() + 2;
288    let length = text[start..].find(&format!("</{name}>"))?;
289    Some(unescaped(&text[start..start + length]))
290}
291
292fn unescaped(text: &str) -> String {
293    let mut out = String::with_capacity(text.len());
294    let mut rest = text;
295    while let Some(start) = rest.find('&') {
296        out.push_str(&rest[..start]);
297        let after = &rest[start..];
298        let entity = [
299            ("&amp;", '&'),
300            ("&lt;", '<'),
301            ("&gt;", '>'),
302            ("&quot;", '"'),
303            ("&apos;", '\''),
304        ]
305        .into_iter()
306        .find(|(name, _)| after.starts_with(name));
307        let (character, length) =
308            entity.map_or(('&', 1), |(name, character)| (character, name.len()));
309        out.push(character);
310        rest = &after[length..];
311    }
312    out.push_str(rest);
313    out
314}
315
316impl Store for S3 {
317    fn writer(&self, key: &StorageKey) -> Pending<'_, Box<dyn Writer>> {
318        let writer = S3Writer {
319            shared: Arc::clone(&self.shared),
320            key: key.clone(),
321            buffer: BytesMut::with_capacity(FIRST_BUFFER),
322            upload: None,
323            settled: false,
324            broken: false,
325        };
326        Box::pin(std::future::ready(Ok(Box::new(writer) as Box<dyn Writer>)))
327    }
328
329    fn reader(&self, key: &StorageKey) -> Pending<'_, Option<Box<dyn Reader>>> {
330        let shared = Arc::clone(&self.shared);
331        let key = key.clone();
332        Box::pin(self.shared.http.operation("get", async move {
333            let mut reader = S3Reader {
334                shared,
335                key,
336                next: 0,
337                total: 0,
338                etag: None,
339                first: None,
340            };
341            Ok(reader
342                .open()
343                .await?
344                .then(|| Box::new(reader) as Box<dyn Reader>))
345        }))
346    }
347
348    fn delete(&self, key: &StorageKey) -> Pending<'_, ()> {
349        let key = key.clone();
350        Box::pin(self.shared.http.operation("delete", async move {
351            self.shared
352                .expect(Call::new(Verb::Delete, &key), "delete a file")
353                .await
354                .map(drop)
355        }))
356    }
357
358    fn exists(&self, key: &StorageKey) -> Pending<'_, bool> {
359        let key = key.clone();
360        Box::pin(self.shared.http.operation("exists", async move {
361            let mut call = Call::new(Verb::Get, &key);
362            call.range = Some("bytes=0-0".to_owned());
363            let response = self.shared.send(call).await?;
364            match response.status() {
365                200 | 206 | 416 => Ok(true),
366                404 => Ok(false),
367                _ => Err(refused(&response, "look for a file")),
368            }
369        }))
370    }
371
372    fn download_url(&self, link: &DownloadLink<'_>) -> Result<String> {
373        let seconds = link.valid_for.as_secs();
374        let (content_type, disposition) = link.content_type.served();
375        let place = &self.shared.place;
376        let path = self.shared.path(link.key);
377        let query = self.shared.credentials.presigned_query(
378            "GET",
379            &path,
380            &place.host,
381            vec![
382                ("response-content-type".to_owned(), content_type.to_owned()),
383                (
384                    "response-content-disposition".to_owned(),
385                    crate::content_disposition(disposition, link.filename),
386                ),
387            ],
388            link.now,
389            seconds,
390        );
391        Ok(format!("{}://{}{path}?{query}", place.scheme, place.host))
392    }
393}
394
395struct Upload {
396    id: String,
397    etags: Vec<String>,
398}
399
400struct S3Writer {
401    shared: Arc<Shared>,
402    key: StorageKey,
403    buffer: BytesMut,
404    upload: Option<Upload>,
405    settled: bool,
406    broken: bool,
407}
408
409impl S3Writer {
410    async fn start_upload(&self) -> Result<Upload> {
411        let call = Call::new(Verb::Post, &self.key).query("uploads", "");
412        let response = self.shared.expect(call, "start an upload").await?;
413        let id = element(response.body(), "UploadId")
414            .ok_or_else(|| Error::internal_message("S3 started an upload without an id"))?;
415        Ok(Upload {
416            id,
417            etags: Vec::new(),
418        })
419    }
420
421    async fn send_part(&mut self, part: Bytes) -> Result<()> {
422        let upload = match self.upload.take() {
423            Some(upload) => upload,
424            None => self.start_upload().await?,
425        };
426        let upload = self.upload.insert(upload);
427        let number = (upload.etags.len() + 1).to_string();
428        let call = Call::new(Verb::Put, &self.key)
429            .query("partNumber", &number)
430            .query("uploadId", &upload.id)
431            .body(part);
432        let response = self.shared.expect(call, "store a part").await?;
433        let etag = response
434            .header("etag")
435            .filter(|etag| !etag.contains(['<', '&']))
436            .ok_or_else(|| Error::internal_message("S3 stored a part without an ETag"))?;
437        upload.etags.push(etag.to_owned());
438        Ok(())
439    }
440
441    async fn complete(&mut self) -> Result<()> {
442        let rest = self.buffer.split().freeze();
443        let unsent = |upload: &Upload| !rest.is_empty() || upload.etags.is_empty();
444        if self.upload.as_ref().is_some_and(unsent) {
445            self.send_part(rest.clone()).await?;
446        }
447        let Some(upload) = &self.upload else {
448            let call = Call::new(Verb::Put, &self.key).body(rest);
449            return self.shared.expect(call, "store a file").await.map(drop);
450        };
451        let mut parts = String::from("<CompleteMultipartUpload>");
452        for (index, etag) in upload.etags.iter().enumerate() {
453            parts.push_str("<Part><PartNumber>");
454            parts.push_str(&(index + 1).to_string());
455            parts.push_str("</PartNumber><ETag>");
456            parts.push_str(etag);
457            parts.push_str("</ETag></Part>");
458        }
459        parts.push_str("</CompleteMultipartUpload>");
460        let call = Call::new(Verb::Post, &self.key)
461            .query("uploadId", &upload.id)
462            .body(Bytes::from(parts));
463        self.shared
464            .expect(call, "complete an upload")
465            .await
466            .map(drop)
467    }
468}
469
470impl Writer for S3Writer {
471    fn write(&mut self, chunk: Bytes) -> Pending<'_, ()> {
472        self.buffer.extend_from_slice(&chunk);
473        Box::pin(async move {
474            if self.broken {
475                return Err(broken());
476            }
477            while self.buffer.len() >= PART_SIZE {
478                let part = self.buffer.split_to(PART_SIZE).freeze();
479                if let Err(error) = self.send_part(part).await {
480                    self.broken = true;
481                    return Err(error);
482                }
483            }
484            Ok(())
485        })
486    }
487
488    fn finish(mut self: Box<Self>) -> Pending<'static, ()> {
489        let http = self.shared.http.clone();
490        Box::pin(async move {
491            if self.broken {
492                return Err(broken());
493            }
494            let completing = async move {
495                self.complete().await?;
496                self.settled = true;
497                Ok(())
498            };
499            http.operation("put", Box::pin(completing)).await
500        })
501    }
502
503    fn abort(mut self: Box<Self>) -> Pending<'static, ()> {
504        self.settled = true;
505        let upload = self.upload.take();
506        Box::pin(abort(Arc::clone(&self.shared), self.key.clone(), upload))
507    }
508}
509
510fn broken() -> Error {
511    Error::internal_message("an S3 upload lost a part to a failed write, so it cannot be finished")
512}
513
514async fn abort(shared: Arc<Shared>, key: StorageKey, upload: Option<Upload>) -> Result<()> {
515    let Some(upload) = upload else {
516        return Ok(());
517    };
518    let call = Call::new(Verb::Delete, &key).query("uploadId", &upload.id);
519    shared.expect(call, "abort an upload").await.map(drop)
520}
521
522/// A writer dropped half way leaves no parts for S3 to keep and bill.
523impl Drop for S3Writer {
524    fn drop(&mut self) {
525        let Some(upload) = self.upload.take().filter(|_| !self.settled) else {
526            return;
527        };
528        let Ok(runtime) = tokio::runtime::Handle::try_current() else {
529            return;
530        };
531        let aborting = abort(Arc::clone(&self.shared), self.key.clone(), Some(upload));
532        runtime.spawn(async move {
533            if let Err(error) = aborting.await {
534                tracing::warn!(error.kind = "storage", error = %error, "aborting an S3 upload failed");
535            }
536        });
537    }
538}
539
540struct S3Reader {
541    shared: Arc<Shared>,
542    key: StorageKey,
543    next: u64,
544    total: u64,
545    /// Every later range must match it, so an overwrite meanwhile is an error, not a mix.
546    etag: Option<String>,
547    first: Option<Bytes>,
548}
549
550impl S3Reader {
551    /// `false` when nothing is stored under the key.
552    async fn open(&mut self) -> Result<bool> {
553        let response = self.shared.send(self.range_call()).await?;
554        match response.status() {
555            404 => return Ok(false),
556            416 => return Ok(true),
557            200 => self.total = u64::try_from(response.body().len()).unwrap_or(u64::MAX),
558            206 => self.total = self.checked_range(&response)?,
559            _ => return Err(refused(&response, "read a file")),
560        }
561        self.etag = response.header("etag").map(str::to_owned);
562        self.next = u64::try_from(response.body().len()).unwrap_or(u64::MAX);
563        self.first = Some(response.into_body());
564        Ok(true)
565    }
566
567    fn range_call(&self) -> Call<'_> {
568        let length = if self.next == 0 {
569            FIRST_READ_CHUNK
570        } else {
571            READ_CHUNK
572        };
573        let mut call = Call::new(Verb::Get, &self.key);
574        call.range = Some(format!("bytes={}-{}", self.next, self.next + length - 1));
575        call.if_match.clone_from(&self.etag);
576        call
577    }
578
579    fn checked_range(&self, response: &Response) -> Result<u64> {
580        let refused = |reason: &str| {
581            Error::internal_message(format!(
582                "S3 answered the range from byte {} {reason}",
583                self.next
584            ))
585        };
586        if response.status() != 206 {
587            return Err(refused(&format!("with {}", response.status())));
588        }
589        let (start, total) = response
590            .header("content-range")
591            .and_then(|range| range.strip_prefix("bytes "))
592            .and_then(|range| range.split_once('-'))
593            .and_then(|(start, rest)| Some((start, rest.rsplit_once('/')?.1)))
594            .and_then(|(start, total)| Some((start.parse::<u64>().ok()?, total.parse().ok()?)))
595            .ok_or_else(|| refused("without a readable Content-Range"))?;
596        if start != self.next {
597            return Err(refused(&format!("with one from byte {start}")));
598        }
599        Ok(total)
600    }
601}
602
603impl Reader for S3Reader {
604    fn size(&self) -> Option<u64> {
605        Some(self.total)
606    }
607
608    fn chunk(&mut self) -> Pending<'_, Option<Bytes>> {
609        Box::pin(async move {
610            if let Some(first) = self.first.take() {
611                return Ok(Some(first).filter(|first| !first.is_empty()));
612            }
613            if self.next >= self.total {
614                return Ok(None);
615            }
616            let response = self.shared.expect(self.range_call(), "read a file").await?;
617            self.checked_range(&response)?;
618            if response.body().is_empty() {
619                return Err(Error::internal_message("S3 answered a range with no bytes"));
620            }
621            self.next += u64::try_from(response.body().len()).unwrap_or(u64::MAX);
622            Ok(Some(response.into_body()))
623        })
624    }
625}
626
627#[cfg(test)]
628mod tests;
sofie_storage/s3.rs

sofie_storage/
s3.rs
