use std::net::Ipv4Addr;
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use tokio::net::TcpListener;
pub const DEFAULT_BUCKET: &str = "dev";
const DEV_ACCESS_KEY: &str = "dev";
const DEV_SECRET_KEY: &str = "dev";
#[derive(Debug, Clone)]
pub struct DevS3 {
pub endpoint: String,
pub bucket: String,
pub root: PathBuf,
}
impl DevS3 {
pub async fn start(root: impl Into<PathBuf>, bucket: &str) -> Result<DevS3> {
let root = root.into();
let bucket_dir = root.join(bucket);
tokio::fs::create_dir_all(&bucket_dir)
.await
.with_context(|| format!("creating dev S3 bucket dir {}", bucket_dir.display()))?;
let service = build_service(&root)?;
let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0))
.await
.context("binding dev S3 listener")?;
let addr = listener.local_addr().context("dev S3 local_addr")?;
let endpoint = format!("http://{addr}");
tokio::spawn(serve_loop(listener, service));
Ok(DevS3 {
endpoint,
bucket: bucket.to_string(),
root,
})
}
pub fn env_vars(&self) -> Vec<(String, String)> {
vec![
("R2_ENDPOINT".to_string(), self.endpoint.clone()),
("R2_BUCKET".to_string(), self.bucket.clone()),
("R2_ACCESS_KEY_ID".to_string(), DEV_ACCESS_KEY.to_string()),
("R2_SECRET_ACCESS_KEY".to_string(), DEV_SECRET_KEY.to_string()),
]
}
}
struct AllowAllAccess;
#[async_trait::async_trait]
impl s3s::access::S3Access for AllowAllAccess {
async fn check(&self, _cx: &mut s3s::access::S3AccessContext<'_>) -> s3s::S3Result<()> {
Ok(())
}
}
fn build_service(root: &Path) -> Result<s3s::service::S3Service> {
use s3s::auth::SimpleAuth;
use s3s::service::S3ServiceBuilder;
let fs = s3s_fs::FileSystem::new(root)
.map_err(|e| anyhow::anyhow!("opening s3s-fs at {}: {e:?}", root.display()))?;
let mut builder = S3ServiceBuilder::new(fs);
builder.set_auth(SimpleAuth::from_single(DEV_ACCESS_KEY, DEV_SECRET_KEY));
builder.set_access(AllowAllAccess);
Ok(builder.build())
}
async fn serve_loop(listener: TcpListener, service: s3s::service::S3Service) {
use hyper_util::rt::{TokioExecutor, TokioIo};
use hyper_util::server::conn::auto::Builder as ConnBuilder;
let http = ConnBuilder::new(TokioExecutor::new());
loop {
let socket = match listener.accept().await {
Ok((socket, _)) => socket,
Err(e) => {
tracing::warn!(error = %e, "dev S3: accept failed");
continue;
}
};
let conn = http
.serve_connection(TokioIo::new(socket), service.clone())
.into_owned();
tokio::spawn(async move {
if let Err(e) = conn.await {
tracing::debug!(error = %e, "dev S3: connection ended");
}
});
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn start_creates_bucket_dir() {
let tmp = tempfile::tempdir().unwrap();
let s3 = DevS3::start(tmp.path().join("s3"), DEFAULT_BUCKET)
.await
.unwrap();
assert!(s3.root.join(DEFAULT_BUCKET).is_dir());
assert!(s3.endpoint.starts_with("http://127.0.0.1:"));
assert_eq!(s3.bucket, DEFAULT_BUCKET);
}
#[tokio::test]
async fn put_then_get_round_trips() {
let tmp = tempfile::tempdir().unwrap();
let s3 = DevS3::start(tmp.path().join("s3"), DEFAULT_BUCKET)
.await
.unwrap();
let url = format!("{}/{}/hello.txt", s3.endpoint, s3.bucket);
let client = reqwest::Client::new();
let put = client
.put(&url)
.body("WORLD")
.send()
.await
.expect("PUT request");
assert!(
put.status().is_success(),
"PUT status: {} body: {:?}",
put.status(),
put.text().await
);
let get = client.get(&url).send().await.expect("GET request");
assert!(get.status().is_success(), "GET status: {}", get.status());
assert_eq!(get.text().await.unwrap(), "WORLD");
assert!(s3.root.join(DEFAULT_BUCKET).join("hello.txt").is_file());
}
}