OXYGENHUB
# 4bae285c6ed8 oxygenhub / crates / hub / src / storage.rs
10.2 KiB · Rust · last changed A repository can live in its owner's bucket 1 month ago source blame history raw
1
//! Repositories kept in their owner's bucket (hub spec §4, oxygenhub#96,
2
//! oxygen#68). The repository's objects and operation log live in an S3,
3
//! Google Cloud Storage or Azure bucket the owner chose; the hub keeps a
4
//! bare repository directory with a cache in front (`storage.url` in its
5
//! config) and the credentials, sealed under `keys/secrets.key`. Stores
6
//! are opened by URL, and the credentials are handed to them through
7
//! oxy-strata's lookup, so nothing from the hub host's own environment or
8
//! cloud profiles is ever used for a repository.
9
10
use crate::db::{now_ms, Pool};
11
use crate::Shared;
12
use oxy_strata::Credentials;
13
use std::collections::HashMap;
14
use std::sync::RwLock;
15
16
#[derive(Clone, Debug, sqlx::FromRow)]
17
pub struct Storage {
18
    pub repo_id: i64,
19
    pub url: String,
20
    /// Sealed; `App.secret_key` opens it.
21
    pub credentials: Vec<u8>,
22
    pub created_ms: i64,
23
    pub checked_ms: Option<i64>,
24
    pub check_result: String,
25
}
26
27
/// What the stores opened by URL get: each repository's credentials by
28
/// its store URL, and the URL by repository id for pages.
29
#[derive(Default)]
30
struct Known {
31
    credentials: HashMap<String, Credentials>,
32
    /// By the repository's directory, unique even between hubs in one process.
33
    urls: HashMap<std::path::PathBuf, String>,
34
}
35
36
static KNOWN: RwLock<Option<Known>> = RwLock::new(None);
37
38
fn lookup(url: &str) -> Option<Credentials> {
39
    KNOWN.read().ok()?.as_ref()?.credentials.get(url).cloned()
40
}
41
42
fn remember(dir: std::path::PathBuf, url: &str, credentials: Credentials) {
43
    let mut known = KNOWN.write().unwrap_or_else(|p| p.into_inner());
44
    let known = known.get_or_insert_with(Known::default);
45
    known.credentials.insert(url.to_string(), credentials);
46
    known.urls.insert(dir, url.to_string());
47
}
48
49
fn forget(dir: &std::path::Path) {
50
    let mut known = KNOWN.write().unwrap_or_else(|p| p.into_inner());
51
    if let Some(k) = known.as_mut() {
52
        if let Some(url) = k.urls.remove(dir) {
53
            k.credentials.remove(&url);
54
        }
55
    }
56
}
57
58
/// Where a repository's data lives, when it is not on this hub's disk.
59
pub fn location(app: &Shared, owner: &str, name: &str) -> Option<String> {
60
    let dir = app.repos.dir(&crate::repos::RepoName::new(owner, name)?);
61
    KNOWN.read().ok()?.as_ref()?.urls.get(&dir).cloned()
62
}
63
64
/// The directory of the repository with row `repo_id`.
65
async fn dir_of(app: &Shared, repo_id: i64) -> Option<std::path::PathBuf> {
66
    let row = crate::db::repos::get_by_id(&app.db, repo_id).await.ok()??;
67
    Some(app.repos.dir(&crate::repos::RepoName::new(&row.owner, &row.name)?))
68
}
69
70
/// Hand the cloud stores to oxy-core, and every repository's credentials
71
/// to them. At start.
72
pub async fn load(app: &Shared) -> sqlx::Result<()> {
73
    oxy_strata::register();
74
    oxy_strata::set_credentials_lookup(lookup);
75
    for s in all(&app.db).await? {
76
        match app.secret_key.open(&s.credentials).map(|j| Credentials::from_json(&j)) {
77
            Some(Ok(c)) => match dir_of(app, s.repo_id).await {
78
                Some(dir) => remember(dir, &s.url, c),
79
                None => tracing::warn!("storage: repository {} has a store but no row", s.repo_id),
80
            },
81
            _ => tracing::warn!("storage: the credentials of repository {} do not open under keys/secrets.key", s.repo_id),
82
        }
83
    }
84
    Ok(())
85
}
86
87
pub async fn all(pool: &Pool) -> sqlx::Result<Vec<Storage>> {
88
    sqlx::query_as("SELECT * FROM repo_storage").fetch_all(pool).await
89
}
90
91
pub async fn of(pool: &Pool, repo_id: i64) -> sqlx::Result<Option<Storage>> {
92
    sqlx::query_as("SELECT * FROM repo_storage WHERE repo_id = ?").bind(repo_id).fetch_optional(pool).await
93
}
94
95
/// Check a store URL a user gave: a bucket on S3, Google Cloud Storage or
96
/// Azure, with no setting but `region`. A profile, a key file or an
97
/// endpoint would make the hub use its own host's credentials or reach
98
/// addresses on its network, and `dir://` its own disk.
99
pub fn check_url(url: &str) -> Result<oxy_strata::StoreUrl, String> {
100
    let scheme = url.trim().split_once("://").map(|(s, _)| s).unwrap_or("");
101
    if !matches!(scheme, "s3" | "gs" | "az") {
102
        return Err("a store is s3://BUCKET/PREFIX, gs://BUCKET/PREFIX or az://ACCOUNT/CONTAINER/PREFIX".into());
103
    }
104
    let parsed = oxy_strata::StoreUrl::parse(url.trim()).map_err(|e| e.to_string())?;
105
    if parsed.scheme == "az" && parsed.path.is_empty() {
106
        return Err("an Azure store names the account and the container: az://ACCOUNT/CONTAINER/PREFIX".into());
107
    }
108
    if let Some(k) = parsed.params.keys().find(|k| k.as_str() != "region") {
109
        return Err(format!("?{k}= is not allowed here: the only setting a hub accepts is ?region="));
110
    }
111
    Ok(parsed)
112
}
113
114
/// The credentials a form gave for a store of `scheme`.
115
pub fn credentials_from(
116
    scheme: &str,
117
    access_key_id: &str,
118
    secret_access_key: &str,
119
    service_account_key: &str,
120
    tenant_id: &str,
121
    client_id: &str,
122
    client_secret: &str,
123
) -> Result<Credentials, String> {
124
    let json = match scheme {
125
        "s3" => serde_json::json!({ "kind": "aws", "access_key_id": access_key_id, "secret_access_key": secret_access_key }),
126
        "gs" => serde_json::json!({ "kind": "gcp", "service_account_key": service_account_key }),
127
        "az" => serde_json::json!({ "kind": "azure", "tenant_id": tenant_id, "client_id": client_id, "client_secret": client_secret }),
128
        _ => return Err("unknown kind of store".into()),
129
    };
130
    Credentials::from_json(&json.to_string()).map_err(|e| e.to_string())
131
}
132
133
/// Run the conformance check against a store under a probe prefix of its
134
/// own, which it leaves empty: what passed, or what failed. With
135
/// `must_be_empty` the store must hold nothing yet (a new repository's).
136
pub async fn probe(url: &str, credentials: &Credentials, must_be_empty: bool) -> Result<String, String> {
137
    let (url, credentials) = (url.to_string(), credentials.clone());
138
    tokio::task::spawn_blocking(move || {
139
        use oxy_core::strata::Store as _;
140
        let store = oxy_strata::CloudStore::open_with(&url, Some(&credentials)).map_err(|e| e.to_string())?;
141
        if must_be_empty {
142
            let found = store.list("").map_err(|e| e.to_string())?;
143
            if !found.is_empty() {
144
                return Err(format!(
145
                    "{} already holds {} object(s): a new repository needs an empty bucket or prefix",
146
                    store.describe(),
147
                    found.len()
148
                ));
149
            }
150
        }
151
        let mut nonce = [0u8; 6];
152
        rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce);
153
        let prefix = format!("oxy-probe-{}", hex::encode(nonce));
154
        let passed = oxy_core::strata::conformance(&store, &prefix).map_err(|e| {
155
            for e in store.list(&prefix).unwrap_or_default() {
156
                let _ = store.delete(&e.key);
157
            }
158
            e.to_string()
159
        })?;
160
        Ok(format!("{} checks passed", passed.len()))
161
    })
162
    .await
163
    .map_err(|e| e.to_string())?
164
}
165
166
/// Create repository `name` (row `repo_id`, already in the database) in
167
/// the store at `url`: checked first, then the credentials sealed and
168
/// kept, then the bare repository made over the store.
169
pub async fn create(app: &Shared, name: &crate::repos::RepoName, repo_id: i64, url: &str, credentials: Credentials) -> Result<(), String> {
170
    let result = probe(url, &credentials, true).await?;
171
    let sealed = app.secret_key.seal(&credentials.to_json());
172
    let dir = app.repos.dir(name);
173
    remember(dir.clone(), url, credentials);
174
    let made = {
175
        let (app, name, url) = (app.clone(), name.clone(), url.to_string());
176
        tokio::task::spawn_blocking(move || app.repos.create_on(&name, &url).map(drop).map_err(|e| e.to_string()))
177
            .await
178
            .map_err(|e| e.to_string())
179
            .and_then(|r| r)
180
    };
181
    if let Err(e) = made {
182
        forget(&dir);
183
        return Err(format!("the repository could not be made in {url}: {e}"));
184
    }
185
    let now = now_ms();
186
    sqlx::query("INSERT INTO repo_storage (repo_id, url, credentials, created_ms, checked_ms, check_result) VALUES (?, ?, ?, ?, ?, ?)")
187
        .bind(repo_id)
188
        .bind(url)
189
        .bind(sealed)
190
        .bind(now)
191
        .bind(now)
192
        .bind(&result)
193
        .execute(&app.db)
194
        .await
195
        .map_err(|e| e.to_string())?;
196
    Ok(())
197
}
198
199
/// Check a repository's store again and record the result.
200
pub async fn check(app: &Shared, s: &Storage) -> Result<String, String> {
201
    let json = app.secret_key.open(&s.credentials).ok_or("the credentials do not open under this hub's key")?;
202
    let credentials = Credentials::from_json(&json).map_err(|e| e.to_string())?;
203
    let result = probe(&s.url, &credentials, false).await;
204
    let text = match &result {
205
        Ok(r) => r.clone(),
206
        Err(e) => format!("failed: {e}"),
207
    };
208
    sqlx::query("UPDATE repo_storage SET checked_ms = ?, check_result = ? WHERE repo_id = ?")
209
        .bind(now_ms())
210
        .bind(&text)
211
        .bind(s.repo_id)
212
        .execute(&app.db)
213
        .await
214
        .map_err(|e| e.to_string())?;
215
    result
216
}
217
218
/// Replace a repository's credentials (a rotated key), once they pass the
219
/// check against its store.
220
pub async fn set_credentials(app: &Shared, s: &Storage, credentials: Credentials) -> Result<String, String> {
221
    let result = probe(&s.url, &credentials, false).await?;
222
    sqlx::query("UPDATE repo_storage SET credentials = ?, checked_ms = ?, check_result = ? WHERE repo_id = ?")
223
        .bind(app.secret_key.seal(&credentials.to_json()))
224
        .bind(now_ms())
225
        .bind(&result)
226
        .bind(s.repo_id)
227
        .execute(&app.db)
228
        .await
229
        .map_err(|e| e.to_string())?;
230
    if let Some(dir) = dir_of(app, s.repo_id).await {
231
        remember(dir, &s.url, credentials);
232
    }
233
    Ok(result)
234
}
235
236
/// A repository is gone (its store row with it): forget its credentials.
237
pub fn removed(app: &Shared, name: &crate::repos::RepoName) {
238
    forget(&app.repos.dir(name));
239
}
240
241
/// A repository moved to another name (a transfer): its store goes with it.
242
pub fn renamed(app: &Shared, from: &crate::repos::RepoName, to: &crate::repos::RepoName) {
243
    let mut known = KNOWN.write().unwrap_or_else(|p| p.into_inner());
244
    if let Some(k) = known.as_mut() {
245
        if let Some(url) = k.urls.remove(&app.repos.dir(from)) {
246
            k.urls.insert(app.repos.dir(to), url);
247
        }
248
    }
249
}