| 1 |
|
| 2 |
|
| 3 |
|
| 4 |
|
| 5 |
|
| 6 |
|
| 7 |
|
| 8 |
use axum::body::Body; |
| 9 |
use axum::http::{header, Request, StatusCode}; |
| 10 |
use http_body_util::BodyExt; |
| 11 |
use oxy_core::repo::Repo; |
| 12 |
use oxygenhub::config::Config; |
| 13 |
use oxygenhub::repos::RepoName; |
| 14 |
use oxygenhub::{db, Shared}; |
| 15 |
use tower::ServiceExt; |
| 16 |
|
| 17 |
fn hermetic() { |
| 18 |
static ONCE: std::sync::Once = std::sync::Once::new(); |
| 19 |
static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new(); |
| 20 |
ONCE.call_once(|| { |
| 21 |
let d = tempfile::tempdir().unwrap(); |
| 22 |
std::env::set_var("OXY_HOME", d.path()); |
| 23 |
DIR.set(d).ok(); |
| 24 |
}); |
| 25 |
} |
| 26 |
|
| 27 |
struct Hub { |
| 28 |
app: Shared, |
| 29 |
base: String, |
| 30 |
_dir: tempfile::TempDir, |
| 31 |
} |
| 32 |
|
| 33 |
async fn hub() -> Hub { |
| 34 |
hermetic(); |
| 35 |
let dir = tempfile::tempdir().unwrap(); |
| 36 |
let mut config = Config::load(dir.path()).unwrap(); |
| 37 |
|
| 38 |
config.registration.setup_code = "test-setup".into(); |
| 39 |
config.auth.push_tokens = vec!["site".into()]; |
| 40 |
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); |
| 41 |
let port = listener.local_addr().unwrap().port(); |
| 42 |
config.server.base_url = format!("http://127.0.0.1:{port}"); |
| 43 |
let app = oxygenhub::build(config).await.unwrap(); |
| 44 |
let router = oxygenhub::router(app.clone()); |
| 45 |
tokio::spawn(async move { axum::serve(listener, router).await.unwrap() }); |
| 46 |
Hub { app, base: format!("http://127.0.0.1:{port}"), _dir: dir } |
| 47 |
} |
| 48 |
|
| 49 |
async fn send(app: &Shared, method: &str, path: &str, cookie: Option<&str>, form: Option<&str>) -> (StatusCode, String, Option<String>) { |
| 50 |
let mut req = Request::builder().method(method).uri(path); |
| 51 |
if let Some(c) = cookie { |
| 52 |
req = req.header(header::COOKIE, format!("oxyhub={c}")); |
| 53 |
} |
| 54 |
let body = match form { |
| 55 |
Some(f) => { |
| 56 |
req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded"); |
| 57 |
Body::from(f.to_string()) |
| 58 |
} |
| 59 |
None => Body::empty(), |
| 60 |
}; |
| 61 |
let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap(); |
| 62 |
let status = res.status(); |
| 63 |
let cookie = res |
| 64 |
.headers() |
| 65 |
.get_all(header::SET_COOKIE) |
| 66 |
.iter() |
| 67 |
.filter_map(|v| v.to_str().ok()) |
| 68 |
.filter_map(|c| c.split(';').next()) |
| 69 |
.filter_map(|kv| kv.strip_prefix("oxyhub=")) |
| 70 |
.map(String::from) |
| 71 |
.next(); |
| 72 |
let bytes = res.into_body().collect().await.unwrap().to_bytes(); |
| 73 |
(status, String::from_utf8_lossy(&bytes).into_owned(), cookie) |
| 74 |
} |
| 75 |
|
| 76 |
fn csrf_of(page: &str) -> String { |
| 77 |
let i = page.find("name=\"_csrf\" value=\"").expect("csrf field") + "name=\"_csrf\" value=\"".len(); |
| 78 |
page[i..].split('"').next().unwrap().to_string() |
| 79 |
} |
| 80 |
|
| 81 |
fn enc(s: &str) -> String { |
| 82 |
s.bytes() |
| 83 |
.map(|b| if b.is_ascii_alphanumeric() || b"-_.~".contains(&b) { (b as char).to_string() } else { format!("%{b:02X}") }) |
| 84 |
.collect() |
| 85 |
} |
| 86 |
|
| 87 |
async fn andy(app: &Shared) -> String { |
| 88 |
let (st, body, cookie) = send( |
| 89 |
app, |
| 90 |
"POST", |
| 91 |
"/register", |
| 92 |
None, |
| 93 |
Some("username=andy&email=andy%40example.com&password=correct-horse-battery&setup=test-setup&invite="), |
| 94 |
) |
| 95 |
.await; |
| 96 |
assert_eq!(st, StatusCode::SEE_OTHER, "{body}"); |
| 97 |
cookie.expect("session") |
| 98 |
} |
| 99 |
|
| 100 |
async fn new_repo(app: &Shared, me: &str, fields: &str) -> (StatusCode, String) { |
| 101 |
let csrf = csrf_of(&send(app, "GET", "/new", Some(me), None).await.1); |
| 102 |
let (st, body, _) = send(app, "POST", "/new", Some(me), Some(&format!("_csrf={csrf}&owner=andy&{fields}"))).await; |
| 103 |
(st, body) |
| 104 |
} |
| 105 |
|
| 106 |
#[tokio::test(flavor = "multi_thread")] |
| 107 |
async fn a_bucket_the_hub_must_not_use_is_refused_and_nothing_is_made() { |
| 108 |
let h = hub().await; |
| 109 |
let me = andy(&h.app).await; |
| 110 |
for (url, why) in [ |
| 111 |
("s3://b/p?profile=default", "?profile= is not allowed"), |
| 112 |
("s3://b/p?endpoint=http%3A%2F%2F169.254.169.254", "?endpoint= is not allowed"), |
| 113 |
("gs://b/p?credentials=%2Fetc%2Fkey.json", "?credentials= is not allowed"), |
| 114 |
("dir:///var/lib/oxygenhub", "a store is s3://"), |
| 115 |
("az://account", "names the account and the container"), |
| 116 |
] { |
| 117 |
let (st, body) = |
| 118 |
new_repo(&h.app, &me, &format!("name=cloudy&storage=bucket&store_url={}&access_key_id=a&secret_access_key=b", enc(url))).await; |
| 119 |
assert_eq!(st, StatusCode::OK, "{url}"); |
| 120 |
assert!(body.contains(why), "{url}: {body}"); |
| 121 |
} |
| 122 |
let (_, body) = new_repo(&h.app, &me, &format!("name=cloudy&storage=bucket&store_url={}&access_key_id=a", enc("s3://b/p"))).await; |
| 123 |
assert!(body.contains("Credentials: credentials need secret_access_key"), "{body}"); |
| 124 |
let name = RepoName::new("andy", "cloudy").unwrap(); |
| 125 |
assert!(!h.app.repos.exists(&name)); |
| 126 |
assert!(db::repos::get(&h.app.db, "andy", "cloudy").await.unwrap().is_none()); |
| 127 |
|
| 128 |
let (st, body) = new_repo(&h.app, &me, "name=local").await; |
| 129 |
assert_eq!(st, StatusCode::SEE_OTHER, "{body}"); |
| 130 |
let page = send(&h.app, "GET", "/andy/local", Some(&me), None).await.1; |
| 131 |
assert!(!page.contains("class=\"store store-"), "no store icon on a repository on this disk"); |
| 132 |
} |
| 133 |
|
| 134 |
|
| 135 |
fn s3_test() -> Option<(String, String, String)> { |
| 136 |
let url = std::env::var("OXY_TEST_S3").ok().filter(|u| !u.is_empty())?; |
| 137 |
let profile = std::env::var("OXY_TEST_S3_PROFILE").ok().filter(|p| !p.is_empty())?; |
| 138 |
let (base, query) = url.split_once('?').unwrap_or((&url, "")); |
| 139 |
let region: Vec<&str> = query.split('&').filter(|p| p.starts_with("region=")).collect(); |
| 140 |
let home = std::path::PathBuf::from(std::env::var("HOME").ok()?); |
| 141 |
let text = std::fs::read_to_string(home.join(".aws/credentials")).ok()?; |
| 142 |
let (mut section, mut id, mut secret) = (String::new(), None, None); |
| 143 |
for line in text.lines().map(str::trim) { |
| 144 |
if let Some(n) = line.strip_prefix('[').and_then(|l| l.strip_suffix(']')) { |
| 145 |
section = n.to_string(); |
| 146 |
} else if let Some((k, v)) = line.split_once('=') { |
| 147 |
if section == profile { |
| 148 |
match k.trim() { |
| 149 |
"aws_access_key_id" => id = Some(v.trim().to_string()), |
| 150 |
"aws_secret_access_key" => secret = Some(v.trim().to_string()), |
| 151 |
_ => {} |
| 152 |
} |
| 153 |
} |
| 154 |
} |
| 155 |
} |
| 156 |
let base = format!("{base}/hub-{}", std::process::id()); |
| 157 |
let url = if region.is_empty() { base } else { format!("{base}?{}", region.join("&")) }; |
| 158 |
Some((url, id?, secret?)) |
| 159 |
} |
| 160 |
|
| 161 |
#[tokio::test(flavor = "multi_thread")] |
| 162 |
async fn a_repository_in_a_real_bucket_takes_a_push_and_passes_its_check() { |
| 163 |
let Some((url, id, secret)) = s3_test() else { |
| 164 |
eprintln!("skipped: OXY_TEST_S3 and OXY_TEST_S3_PROFILE are not set"); |
| 165 |
return; |
| 166 |
}; |
| 167 |
let h = hub().await; |
| 168 |
let me = andy(&h.app).await; |
| 169 |
let fields = format!( |
| 170 |
"name=cloudy&visibility=private&storage=bucket&store_url={}&access_key_id={}&secret_access_key={}", |
| 171 |
enc(&url), |
| 172 |
enc(&id), |
| 173 |
enc(&secret) |
| 174 |
); |
| 175 |
let (st, body) = new_repo(&h.app, &me, &fields).await; |
| 176 |
let credentials = oxy_strata::Credentials::Aws { access_key_id: id, secret_access_key: secret, session_token: None }; |
| 177 |
let store = oxy_strata::CloudStore::open_with(&url, Some(&credentials)).unwrap(); |
| 178 |
let cleanup = || { |
| 179 |
use oxy_core::strata::Store as _; |
| 180 |
for e in store.list("").unwrap() { |
| 181 |
store.delete(&e.key).unwrap(); |
| 182 |
} |
| 183 |
}; |
| 184 |
if st != StatusCode::SEE_OTHER { |
| 185 |
cleanup(); |
| 186 |
panic!("{body}"); |
| 187 |
} |
| 188 |
let result = std::panic::AssertUnwindSafe(async { |
| 189 |
|
| 190 |
let work = tempfile::tempdir().unwrap(); |
| 191 |
let mut c = Repo::init(work.path()).unwrap(); |
| 192 |
c.config.set("remote.origin.url", &format!("{}/andy/cloudy", h.base)).unwrap(); |
| 193 |
c.config.set("remote.origin.token", "site").unwrap(); |
| 194 |
c.save_config().unwrap(); |
| 195 |
std::fs::write(work.path().join("a.txt"), "kept in a bucket\n").unwrap(); |
| 196 |
let head = c.commit("First", &[]).unwrap(); |
| 197 |
let pushed = tokio::task::spawn_blocking(move || c.push("origin", false).unwrap()).await.unwrap(); |
| 198 |
assert!(pushed.iter().all(|r| r.ok), "{pushed:?}"); |
| 199 |
let hosted = h.app.repos.open(&RepoName::new("andy", "cloudy").unwrap()).unwrap(); |
| 200 |
assert_eq!(hosted.bookmark_heads().unwrap(), vec![head]); |
| 201 |
{ |
| 202 |
use oxy_core::strata::Store as _; |
| 203 |
assert!(store.read_ref("ops/head").unwrap().is_some(), "the operation log is in the bucket"); |
| 204 |
} |
| 205 |
|
| 206 |
let page = send(&h.app, "GET", "/andy/cloudy", Some(&me), None).await.1; |
| 207 |
assert!(page.contains("class=\"store store-s3\"") && page.contains(url.split('?').next().unwrap()), "{page}"); |
| 208 |
let settings = send(&h.app, "GET", "/andy/cloudy/settings/storage", Some(&me), None).await.1; |
| 209 |
assert!(settings.contains("id=\"storage\""), "{settings}"); |
| 210 |
let csrf = csrf_of(&settings); |
| 211 |
let (st, _, _) = |
| 212 |
send(&h.app, "POST", "/andy/cloudy/settings/storage", Some(&me), Some(&format!("_csrf={csrf}&action=check"))).await; |
| 213 |
assert_eq!(st, StatusCode::SEE_OTHER, "back to Settings › Storage"); |
| 214 |
let after = send(&h.app, "GET", "/andy/cloudy/settings/storage", Some(&me), None).await.1; |
| 215 |
assert!(after.contains("8 checks passed"), "{after}"); |
| 216 |
|
| 217 |
let (st, body) = new_repo(&h.app, &me, &fields.replace("name=cloudy", "name=again")).await; |
| 218 |
assert_eq!(st, StatusCode::OK); |
| 219 |
assert!(body.contains("already holds"), "{body}"); |
| 220 |
}); |
| 221 |
let outcome = futures_catch(result).await; |
| 222 |
cleanup(); |
| 223 |
if let Err(p) = outcome { |
| 224 |
std::panic::resume_unwind(p); |
| 225 |
} |
| 226 |
} |
| 227 |
|
| 228 |
async fn futures_catch<F: std::future::Future<Output = ()>>(f: std::panic::AssertUnwindSafe<F>) -> std::thread::Result<()> { |
| 229 |
use futures::FutureExt; |
| 230 |
f.catch_unwind().await |
| 231 |
} |
| 232 |
|
| 233 |
#[tokio::test(flavor = "multi_thread")] |
| 234 |
async fn an_organisation_can_require_its_own_buckets() { |
| 235 |
let h = hub().await; |
| 236 |
let me = andy(&h.app).await; |
| 237 |
let csrf = csrf_of(&send(&h.app, "GET", "/orgs/new", Some(&me), None).await.1); |
| 238 |
let (st, body, _) = send(&h.app, "POST", "/orgs/new", Some(&me), Some(&format!("_csrf={csrf}&name=acme&display_name=Acme"))).await; |
| 239 |
assert_eq!(st, StatusCode::SEE_OTHER, "{body}"); |
| 240 |
let settings = send(&h.app, "GET", "/orgs/acme/settings", Some(&me), None).await.1; |
| 241 |
assert!(settings.contains("id=\"storage\"")); |
| 242 |
let csrf = csrf_of(&settings); |
| 243 |
let set = |allowed: &str| { |
| 244 |
let form = format!("_csrf={csrf}&require_bucket=1&allowed={}", enc(allowed)); |
| 245 |
let app = h.app.clone(); |
| 246 |
let me = me.clone(); |
| 247 |
async move { send(&app, "POST", "/orgs/acme/settings/storage", Some(&me), Some(&form)).await.1 } |
| 248 |
}; |
| 249 |
let page = set("s3://acme-london/repos/?region=eu-west-2").await; |
| 250 |
assert!(page.contains("a prefix names a bucket and a path, with no settings"), "{page}"); |
| 251 |
let page = set("s3://acme-london/repos/\ngs://acme-eu/").await; |
| 252 |
assert!(page.contains("Storage policy saved: every new repository in s3://acme-london/repos/, gs://acme-eu/."), "{page}"); |
| 253 |
|
| 254 |
let new = |fields: String| { |
| 255 |
let app = h.app.clone(); |
| 256 |
let me = me.clone(); |
| 257 |
async move { |
| 258 |
let csrf = csrf_of(&send(&app, "GET", "/new", Some(&me), None).await.1); |
| 259 |
send(&app, "POST", "/new", Some(&me), Some(&format!("_csrf={csrf}&owner=acme&{fields}"))).await.1 |
| 260 |
} |
| 261 |
}; |
| 262 |
let page = new("name=ondisk".into()).await; |
| 263 |
assert!(page.contains("acme keeps every new repository in a bucket"), "{page}"); |
| 264 |
let page = new(format!("name=elsewhere&storage=bucket&store_url={}&access_key_id=a&secret_access_key=b", enc("s3://other/x"))).await; |
| 265 |
assert!(page.contains("acme allows only these stores: s3://acme-london/repos/, gs://acme-eu/"), "{page}"); |
| 266 |
assert!(db::repos::get(&h.app.db, "acme", "ondisk").await.unwrap().is_none()); |
| 267 |
assert!(db::repos::get(&h.app.db, "acme", "elsewhere").await.unwrap().is_none()); |
| 268 |
|
| 269 |
let (st, body) = new_repo(&h.app, &me, "name=mine").await; |
| 270 |
assert_eq!(st, StatusCode::SEE_OTHER, "{body}"); |
| 271 |
} |