OXYGENHUB
# f4e138301198 oxygenhub / crates / hub / tests / storage.rs
12.5 KiB · Rust · last changed Repository settings: one page per setting under a grouped sidebar (#278) 16 days ago source blame history raw
1
//! Repositories kept in their owner's bucket (oxygenhub#96): the New
2
//! repository form refuses what the hub must not accept and creates
3
//! nothing then; with a real S3 test bucket configured (`OXY_TEST_S3` and
4
//! `OXY_TEST_S3_PROFILE`, keys read from that profile and handed over), a
5
//! repository is made in the bucket, takes a push, shows where it lives,
6
//! passes Settings → Storage → Check, and the test removes what it wrote.
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
    // The first account's setup code (oxygenhub#245).
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
    // The hub's own disk stays the default.
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
/// The S3 test bucket's URL with only its region kept, and the profile's keys.
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
        // A push from a client lands in the bucket.
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
        // The pages say where it lives; Settings → Storage checks the bucket.
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
        // A second repository cannot start in the same place.
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
    // A user's own repositories are not the organisation's business.
269
    let (st, body) = new_repo(&h.app, &me, "name=mine").await;
270
    assert_eq!(st, StatusCode::SEE_OTHER, "{body}");
271
}