#![allow(dead_code)]
use anyhow::{anyhow, bail, Context, Result};
use rusty_s3::actions::ListObjectsV2;
use rusty_s3::{Bucket, Credentials, S3Action, UrlStyle};
use std::fs::File;
use std::io::Read;
use std::path::Path;
use std::time::Duration;
const OP_TTL: Duration = Duration::from_secs(60);
pub struct Store {
bucket: Bucket,
creds: Credentials,
}
impl Store {
pub fn new(bucket: &str, region: &str, endpoint: Option<&str>) -> Result<Self> {
let (endpoint, style) = match endpoint {
Some(e) => (e.to_string(), UrlStyle::Path), None => (
format!("https://s3.{region}.amazonaws.com"),
UrlStyle::VirtualHost,
),
};
let bucket = Bucket::new(
endpoint.parse().context("parsing S3 endpoint URL")?,
style,
bucket.to_string(),
region.to_string(),
)
.map_err(|e| anyhow!("bucket config: {e}"))?;
let secrets = crate::secrets::Secrets::load()?;
let creds = Credentials::new(secrets.access_key_id, secrets.secret_access_key);
Ok(Self { bucket, creds })
}
pub fn presign_get(&self, key: &str, ttl: Duration) -> String {
sign_get(&self.bucket, &self.creds, key, ttl)
}
pub fn put_object(&self, key: &str, body: &[u8]) -> Result<()> {
let url = self.bucket.put_object(Some(&self.creds), key).sign(OP_TTL);
let resp = ureq::put(url.as_str())
.send_bytes(body)
.map_err(|e| anyhow!("PutObject {key} failed: {}", s3_err(e)))?;
if resp.status() >= 300 {
bail!("PutObject {key}: HTTP {}", resp.status());
}
Ok(())
}
pub fn delete_object(&self, key: &str) -> Result<()> {
let url = self
.bucket
.delete_object(Some(&self.creds), key)
.sign(OP_TTL);
let resp = ureq::delete(url.as_str())
.call()
.map_err(|e| anyhow!("DeleteObject {key} failed: {}", s3_err(e)))?;
if resp.status() >= 300 {
bail!("DeleteObject {key}: HTTP {}", resp.status());
}
Ok(())
}
pub fn list(&self, prefix: &str) -> Result<Vec<String>> {
let mut keys = Vec::new();
let mut continuation: Option<String> = None;
loop {
let mut action = self.bucket.list_objects_v2(Some(&self.creds));
action.with_prefix(prefix);
if let Some(token) = continuation.clone() {
action.with_continuation_token(token);
}
let url = action.sign(OP_TTL);
let resp = ureq::get(url.as_str())
.call()
.map_err(|e| anyhow!("ListObjectsV2 failed: {}", s3_err(e)))?;
if resp.status() >= 300 {
bail!("ListObjectsV2: HTTP {}", resp.status());
}
let text = resp.into_string()?;
let parsed = ListObjectsV2::parse_response(&text)
.map_err(|e| anyhow!("parsing ListObjectsV2 response: {e}"))?;
keys.extend(parsed.contents.into_iter().map(|o| o.key));
match parsed.next_continuation_token {
Some(token) if !token.is_empty() => continuation = Some(token),
_ => break,
}
}
Ok(keys)
}
pub fn put_file(
&self,
key: &str,
path: &Path,
progress: &dyn crate::progress::Progress,
) -> Result<u64> {
let file = File::open(path).with_context(|| format!("opening {}", path.display()))?;
let total = file.metadata()?.len();
let url = self.bucket.put_object(Some(&self.creds), key).sign(OP_TTL);
let reader = ProgressReader {
inner: file,
uploaded: 0,
total,
progress,
};
let resp = ureq::put(url.as_str())
.set("Content-Length", &total.to_string())
.send(reader)
.map_err(|e| anyhow!("PutObject {key} failed: {}", s3_err(e)))?;
if resp.status() >= 300 {
bail!("PutObject {key}: HTTP {}", resp.status());
}
Ok(total)
}
}
struct ProgressReader<'a, R> {
inner: R,
uploaded: u64,
total: u64,
progress: &'a dyn crate::progress::Progress,
}
impl<R: Read> Read for ProgressReader<'_, R> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let n = self.inner.read(buf)?;
self.uploaded += n as u64;
self.progress.bytes(self.uploaded, self.total);
Ok(n)
}
}
fn sign_get(bucket: &Bucket, creds: &Credentials, key: &str, ttl: Duration) -> String {
bucket.get_object(Some(creds), key).sign(ttl).to_string()
}
fn s3_err(e: ureq::Error) -> String {
match e {
ureq::Error::Status(code, _) => format!("HTTP {code}"),
ureq::Error::Transport(t) => t.kind().to_string(),
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn presign_get_signs_a_url_with_key_and_expiry() {
let bucket = Bucket::new(
"https://s3.us-east-1.amazonaws.com".parse().unwrap(),
UrlStyle::VirtualHost,
"dove-shares-example".to_string(),
"us-east-1".to_string(),
)
.unwrap();
let creds = Credentials::new("AKIAEXAMPLE", "secretexample");
let url = sign_get(
&bucket,
&creds,
"abc123/report.pdf",
Duration::from_secs(3600),
);
assert!(url.contains("report.pdf"), "key not in URL: {url}");
assert!(
url.contains("X-Amz-Expires=3600"),
"expiry not in URL: {url}"
);
assert!(url.contains("X-Amz-Signature="), "not signed: {url}");
}
#[test]
fn progress_reader_reports_cumulative_bytes() {
use std::cell::Cell;
struct Recorder {
last: Cell<u64>,
}
impl crate::progress::Progress for Recorder {
fn step(&self, _: &str) {}
fn done(&self, _: &str) {}
fn field(&self, _: &str, _: &str) {}
fn bytes(&self, uploaded: u64, _total: u64) {
self.last.set(uploaded);
}
}
let recorder = Recorder { last: Cell::new(0) };
let mut r = ProgressReader {
inner: std::io::Cursor::new(vec![0u8; 100]),
uploaded: 0,
total: 100,
progress: &recorder,
};
let mut buf = [0u8; 30];
let mut total = 0;
loop {
let n = r.read(&mut buf).unwrap();
if n == 0 {
break;
}
total += n;
}
drop(r); assert_eq!(total, 100);
assert_eq!(recorder.last.get(), 100);
}
#[test]
fn s3_err_does_not_leak_signed_params() {
let err = ureq::get("http://127.0.0.1:1/x?X-Amz-Signature=leak")
.call()
.unwrap_err();
let msg = s3_err(err);
assert!(!msg.contains("X-Amz"), "leaked signed params: {msg}");
}
}