//! Repositories kept in their owner's bucket (hub spec §4, oxygenhub#96, //! oxygen#68). The repository's objects and operation log live in an S3, //! Google Cloud Storage or Azure bucket the owner chose; the hub keeps a //! bare repository directory with a cache in front (`storage.url` in its //! config) and the credentials, sealed under `keys/secrets.key`. Stores //! are opened by URL, and the credentials are handed to them through //! oxy-strata's lookup, so nothing from the hub host's own environment or //! cloud profiles is ever used for a repository. use crate::db::{now_ms, Pool}; use crate::Shared; use oxy_strata::Credentials; use std::collections::HashMap; use std::sync::RwLock; #[derive(Clone, Debug, sqlx::FromRow)] pub struct Storage { pub repo_id: i64, pub url: String, /// Sealed; `App.secret_key` opens it. pub credentials: Vec, pub created_ms: i64, pub checked_ms: Option, pub check_result: String, } /// What the stores opened by URL get: each repository's credentials by /// its store URL, and the URL by repository id for pages. #[derive(Default)] struct Known { credentials: HashMap, /// By the repository's directory, unique even between hubs in one process. urls: HashMap, } static KNOWN: RwLock> = RwLock::new(None); fn lookup(url: &str) -> Option { KNOWN.read().ok()?.as_ref()?.credentials.get(url).cloned() } fn remember(dir: std::path::PathBuf, url: &str, credentials: Credentials) { let mut known = KNOWN.write().unwrap_or_else(|p| p.into_inner()); let known = known.get_or_insert_with(Known::default); known.credentials.insert(url.to_string(), credentials); known.urls.insert(dir, url.to_string()); } fn forget(dir: &std::path::Path) { let mut known = KNOWN.write().unwrap_or_else(|p| p.into_inner()); if let Some(k) = known.as_mut() { if let Some(url) = k.urls.remove(dir) { k.credentials.remove(&url); } } } /// Where a repository's data lives, when it is not on this hub's disk. /// What the repository header shows of a store (#123): which provider, /// and, for the repository's admins, where in it. #[derive(Clone, Debug, PartialEq)] pub struct StoreBadge { /// `gcs`, `s3` or `azure`: picks the icon. pub kind: &'static str, pub provider: &'static str, /// The bucket, or for Azure the storage account and container. pub place: String, pub prefix: String, pub region: Option, /// The URL as stored, query and all. pub url: String, } impl StoreBadge { /// The parts of a store URL; `None` for a scheme the hub does not /// keep repositories on. pub fn of(url: &str) -> Option { let (scheme, rest) = url.split_once("://")?; let (path, query) = rest.split_once('?').unwrap_or((rest, "")); let region = query.split('&').find_map(|kv| kv.strip_prefix("region=")).filter(|r| !r.is_empty()).map(str::to_string); let mut parts = path.trim_matches('/').splitn(if scheme == "az" { 3 } else { 2 }, '/'); let (kind, provider, place) = match scheme { "gs" => ("gcs", "Google Cloud Storage", parts.next()?.to_string()), "s3" => ("s3", "Amazon S3", parts.next()?.to_string()), "az" => ("azure", "Azure Blob Storage", format!("{} / {}", parts.next()?, parts.next()?)), _ => return None, }; Some(StoreBadge { kind, provider, place, prefix: parts.next().unwrap_or("").to_string(), region, url: url.to_string() }) } } pub fn location(app: &Shared, owner: &str, name: &str) -> Option { let dir = app.repos.dir(&crate::repos::RepoName::new(owner, name)?); KNOWN.read().ok()?.as_ref()?.urls.get(&dir).cloned() } /// The directory of the repository with row `repo_id`. async fn dir_of(app: &Shared, repo_id: i64) -> Option { let row = crate::db::repos::get_by_id(&app.db, repo_id).await.ok()??; Some(app.repos.dir(&crate::repos::RepoName::new(&row.owner, &row.name)?)) } /// Hand the cloud stores to oxy-core, and every repository's credentials /// to them. At start. pub async fn load(app: &Shared) -> sqlx::Result<()> { oxy_strata::register(); oxy_strata::set_credentials_lookup(lookup); for s in all(&app.db).await? { match app.secret_key.open(&s.credentials).map(|j| Credentials::from_json(&j)) { Some(Ok(c)) => match dir_of(app, s.repo_id).await { Some(dir) => remember(dir, &s.url, c), None => tracing::warn!("storage: repository {} has a store but no row", s.repo_id), }, _ => tracing::warn!("storage: the credentials of repository {} do not open under keys/secrets.key", s.repo_id), } } Ok(()) } pub async fn all(pool: &Pool) -> sqlx::Result> { sqlx::query_as("SELECT * FROM repo_storage").fetch_all(pool).await } pub async fn of(pool: &Pool, repo_id: i64) -> sqlx::Result> { sqlx::query_as("SELECT * FROM repo_storage WHERE repo_id = $1").bind(repo_id).fetch_optional(pool).await } /// Check a store URL a user gave: a bucket on S3, Google Cloud Storage or /// Azure, with no setting but `region`. A profile, a key file or an /// endpoint would make the hub use its own host's credentials or reach /// addresses on its network, and `dir://` its own disk. pub fn check_url(url: &str) -> Result { let scheme = url.trim().split_once("://").map(|(s, _)| s).unwrap_or(""); if !matches!(scheme, "s3" | "gs" | "az") { return Err("a store is s3://BUCKET/PREFIX, gs://BUCKET/PREFIX or az://ACCOUNT/CONTAINER/PREFIX".into()); } let parsed = oxy_strata::StoreUrl::parse(url.trim()).map_err(|e| e.to_string())?; if parsed.scheme == "az" && parsed.path.is_empty() { return Err("an Azure store names the account and the container: az://ACCOUNT/CONTAINER/PREFIX".into()); } if let Some(k) = parsed.params.keys().find(|k| k.as_str() != "region") { return Err(format!("?{k}= is not allowed here: the only setting a hub accepts is ?region=")); } Ok(parsed) } /// The credentials a form gave for a store of `scheme`. pub fn credentials_from( scheme: &str, access_key_id: &str, secret_access_key: &str, service_account_key: &str, tenant_id: &str, client_id: &str, client_secret: &str, ) -> Result { let json = match scheme { "s3" => serde_json::json!({ "kind": "aws", "access_key_id": access_key_id, "secret_access_key": secret_access_key }), "gs" => serde_json::json!({ "kind": "gcp", "service_account_key": service_account_key }), "az" => serde_json::json!({ "kind": "azure", "tenant_id": tenant_id, "client_id": client_id, "client_secret": client_secret }), _ => return Err("unknown kind of store".into()), }; Credentials::from_json(&json.to_string()).map_err(|e| e.to_string()) } /// Run the conformance check against a store under a probe prefix of its /// own, which it leaves empty: what passed, or what failed. With /// `must_be_empty` the store must hold nothing yet (a new repository's). pub async fn probe(url: &str, credentials: &Credentials, must_be_empty: bool) -> Result { let (url, credentials) = (url.to_string(), credentials.clone()); tokio::task::spawn_blocking(move || { use oxy_core::strata::Store as _; let store = oxy_strata::CloudStore::open_with(&url, Some(&credentials)).map_err(|e| e.to_string())?; if must_be_empty { let found = store.list("").map_err(|e| e.to_string())?; if !found.is_empty() { return Err(format!( "{} already holds {} object(s): a new repository needs an empty bucket or prefix", store.describe(), found.len() )); } } let mut nonce = [0u8; 6]; rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce); let prefix = format!("oxy-probe-{}", hex::encode(nonce)); let passed = oxy_core::strata::conformance(&store, &prefix).map_err(|e| { for e in store.list(&prefix).unwrap_or_default() { let _ = store.delete(&e.key); } e.to_string() })?; Ok(format!("{} checks passed", passed.len())) }) .await .map_err(|e| e.to_string())? } /// Create repository `name` (row `repo_id`, already in the database) in /// the store at `url`: checked first, then the credentials sealed and /// kept, then the bare repository made over the store. pub async fn create(app: &Shared, name: &crate::repos::RepoName, repo_id: i64, url: &str, credentials: Credentials) -> Result<(), String> { let result = probe(url, &credentials, true).await?; let sealed = app.secret_key.seal(&credentials.to_json()); let dir = app.repos.dir(name); remember(dir.clone(), url, credentials); let made = { let (app, name, url) = (app.clone(), name.clone(), url.to_string()); tokio::task::spawn_blocking(move || app.repos.create_on(&name, &url).map(drop).map_err(|e| e.to_string())) .await .map_err(|e| e.to_string()) .and_then(|r| r) }; if let Err(e) = made { forget(&dir); return Err(format!("the repository could not be made in {url}: {e}")); } let now = now_ms(); sqlx::query( "INSERT INTO repo_storage (repo_id, url, credentials, created_ms, checked_ms, check_result) VALUES ($1, $2, $3, $4, $5, $6)", ) .bind(repo_id) .bind(url) .bind(sealed) .bind(now) .bind(now) .bind(&result) .execute(&app.db) .await .map_err(|e| e.to_string())?; Ok(()) } /// Check a repository's store again and record the result. pub async fn check(app: &Shared, s: &Storage) -> Result { let json = app.secret_key.open(&s.credentials).ok_or("the credentials do not open under this hub's key")?; let credentials = Credentials::from_json(&json).map_err(|e| e.to_string())?; let result = probe(&s.url, &credentials, false).await; let text = match &result { Ok(r) => r.clone(), Err(e) => format!("failed: {e}"), }; sqlx::query("UPDATE repo_storage SET checked_ms = $1, check_result = $2 WHERE repo_id = $3") .bind(now_ms()) .bind(&text) .bind(s.repo_id) .execute(&app.db) .await .map_err(|e| e.to_string())?; result } /// Replace a repository's credentials (a rotated key), once they pass the /// check against its store. pub async fn set_credentials(app: &Shared, s: &Storage, credentials: Credentials) -> Result { let result = probe(&s.url, &credentials, false).await?; sqlx::query("UPDATE repo_storage SET credentials = $1, checked_ms = $2, check_result = $3 WHERE repo_id = $4") .bind(app.secret_key.seal(&credentials.to_json())) .bind(now_ms()) .bind(&result) .bind(s.repo_id) .execute(&app.db) .await .map_err(|e| e.to_string())?; if let Some(dir) = dir_of(app, s.repo_id).await { remember(dir, &s.url, credentials); } Ok(result) } /// A repository is gone (its store row with it): forget its credentials. pub fn removed(app: &Shared, name: &crate::repos::RepoName) { forget(&app.repos.dir(name)); } /// A repository moved to another name (a transfer): its store goes with it. pub fn renamed(app: &Shared, from: &crate::repos::RepoName, to: &crate::repos::RepoName) { let mut known = KNOWN.write().unwrap_or_else(|p| p.into_inner()); if let Some(k) = known.as_mut() { if let Some(url) = k.urls.remove(&app.repos.dir(from)) { k.urls.insert(app.repos.dir(to), url); } } } /// An organisation's storage policy: every new repository in a bucket, /// and only in stores under these URL prefixes when there are any. The /// buckets an organisation lists are its own, so where they are (the /// region, the provider, the country) is where its data is. #[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct Policy { pub require_bucket: bool, pub allowed: Vec, } impl Policy { /// Whether a new repository may be kept at `store` (`None`: this /// hub's disk); otherwise what to tell its creator. pub fn permits(&self, org: &str, store: Option<&str>) -> Result<(), String> { match store { None if self.require_bucket => Err(format!( "{org} keeps every new repository in a bucket: choose “my own bucket” under Storage{}", if self.allowed.is_empty() { String::new() } else { format!(" ({})", self.allowed.join(", ")) } )), Some(url) if !self.allowed.is_empty() && !self.allowed.iter().any(|p| url.trim().starts_with(p.as_str())) => { Err(format!("{org} allows only these stores: {}", self.allowed.join(", "))) } _ => Ok(()), } } } pub async fn policy(pool: &Pool, org_id: i64) -> sqlx::Result { let row: Option<(bool, String)> = sqlx::query_as("SELECT require_bucket, allowed FROM org_storage_policy WHERE org_id = $1") .bind(org_id) .fetch_optional(pool) .await?; Ok(row .map(|(r, a)| Policy { require_bucket: r, allowed: a.lines().map(str::trim).filter(|l| !l.is_empty()).map(String::from).collect() }) .unwrap_or_default()) } /// Set a policy from what a form gave: the allowed prefixes one per line /// (or comma-separated), each a store URL of a bucket. pub async fn set_policy(pool: &Pool, org_id: i64, require_bucket: bool, allowed: &str) -> Result { let mut prefixes = Vec::new(); for p in allowed.split([',', '\n']).map(str::trim).filter(|p| !p.is_empty()) { if p.contains('?') { return Err(format!("{p}: a prefix names a bucket and a path, with no settings")); } check_url(p).map_err(|e| format!("{p}: {e}"))?; prefixes.push(p.to_string()); } sqlx::query( "INSERT INTO org_storage_policy (org_id, require_bucket, allowed) VALUES ($1, $2, $3) \ ON CONFLICT (org_id) DO UPDATE SET require_bucket = excluded.require_bucket, allowed = excluded.allowed", ) .bind(org_id) .bind(require_bucket) .bind(prefixes.join("\n")) .execute(pool) .await .map_err(|e| e.to_string())?; Ok(Policy { require_bucket, allowed: prefixes }) } #[cfg(test)] mod tests { use super::*; #[test] fn store_urls_a_hub_accepts() { assert!(check_url("s3://bucket/repos/x?region=eu-west-2").is_ok()); assert!(check_url("gs://bucket/x").is_ok()); assert!(check_url("az://account/container/x").is_ok()); for (bad, why) in [ ("s3://b/x?profile=prod", "?profile="), ("s3://b/x?region=a&endpoint=http://10.0.0.1", "?endpoint="), ("gs://b/x?credentials=/etc/key.json", "?credentials="), ("dir:///srv", "a store is"), ("file:///srv", "a store is"), ("az://account", "the container"), ] { let e = check_url(bad).unwrap_err(); assert!(e.contains(why), "{bad}: {e}"); } } #[test] fn a_policy_says_what_it_allows() { let open = Policy::default(); assert!(open.permits("acme", None).is_ok()); let strict = Policy { require_bucket: true, allowed: vec!["s3://acme-london/".into(), "gs://acme-eu/".into()] }; let e = strict.permits("acme", None).unwrap_err(); assert!(e.contains("acme keeps every new repository in a bucket") && e.contains("s3://acme-london/"), "{e}"); assert!(strict.permits("acme", Some("s3://acme-london/repos/x?region=eu-west-2")).is_ok()); let e = strict.permits("acme", Some("s3://acme-virginia/x")).unwrap_err(); assert!(e.contains("allows only these stores"), "{e}"); } } #[cfg(test)] mod badge_tests { use super::StoreBadge; #[test] fn a_store_url_names_its_provider_place_prefix_and_region() { let g = StoreBadge::of("gs://doomsdayone-oxy-repos/repos/tetris").unwrap(); assert_eq!( (g.kind, g.provider, g.place.as_str(), g.prefix.as_str(), g.region), ("gcs", "Google Cloud Storage", "doomsdayone-oxy-repos", "repos/tetris", None) ); let s = StoreBadge::of("s3://my-bucket/repos/name?region=eu-west-2").unwrap(); assert_eq!( (s.kind, s.place.as_str(), s.prefix.as_str(), s.region.as_deref()), ("s3", "my-bucket", "repos/name", Some("eu-west-2")) ); let a = StoreBadge::of("az://account/container/repos/x").unwrap(); assert_eq!((a.kind, a.place.as_str(), a.prefix.as_str()), ("azure", "account / container", "repos/x")); assert_eq!(StoreBadge::of("s3://bucket").unwrap().prefix, ""); assert!(StoreBadge::of("dir:///tmp/x").is_none()); assert!(StoreBadge::of("az://only-account").is_none()); } }