storage.rs 52.5%

156 of 297 lines run, 33 of 117 functions · run #500 on main · run missed

//! 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<u8>,
    pub created_ms: i64,
    pub checked_ms: Option<i64>,
    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<String, Credentials>,
    /// By the repository's directory, unique even between hubs in one process.
    urls: HashMap<std::path::PathBuf, String>,
}

static KNOWN: RwLock<Option<Known>> = RwLock::new(None);

fn lookup(url: &str) -> Option<Credentials> {
    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<String>,
    /// 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<StoreBadge> {
        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<String> {
    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<std::path::PathBuf> {
    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<Vec<Storage>> {
    sqlx::query_as("SELECT * FROM repo_storage").fetch_all(pool).await
}

pub async fn of(pool: &Pool, repo_id: i64) -> sqlx::Result<Option<Storage>> {
    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<oxy_strata::StoreUrl, String> {
    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<Credentials, String> {
    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<String, String> {
    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<String, String> {
    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<String, String> {
    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<String>,
}

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<Policy> {
    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<Policy, String> {
    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());
    }
}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

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	0×	fn lookup(url: &str) -> Option<Credentials> {
39	0×	KNOWN.read().ok()?.as_ref()?.credentials.get(url).cloned()
40	0×	}
41
42	0×	fn remember(dir: std::path::PathBuf, url: &str, credentials: Credentials) {
43	0×	let mut known = KNOWN.write().unwrap_or_else(\|p\| p.into_inner());
44	0×	let known = known.get_or_insert_with(Known::default);
45	0×	known.credentials.insert(url.to_string(), credentials);
46	0×	known.urls.insert(dir, url.to_string());
47	0×	}
48
49	1×	fn forget(dir: &std::path::Path) {
50	1×	let mut known = KNOWN.write().unwrap_or_else(\|p\| p.into_inner());
51	1×	if let Some(k) = known.as_mut() {
52	0×	if let Some(url) = k.urls.remove(dir) {
53	0×	k.credentials.remove(&url);
54	0×	}
55	1×	}
56	1×	}
57
58		/// Where a repository's data lives, when it is not on this hub's disk.
59		/// What the repository header shows of a store (#123): which provider,
60		/// and, for the repository's admins, where in it.
61		#[derive(Clone, Debug, PartialEq)]
62		pub struct StoreBadge {
63		/// `gcs`, `s3` or `azure`: picks the icon.
64		pub kind: &'static str,
65		pub provider: &'static str,
66		/// The bucket, or for Azure the storage account and container.
67		pub place: String,
68		pub prefix: String,
69		pub region: Option<String>,
70		/// The URL as stored, query and all.
71		pub url: String,
72		}
73
74		impl StoreBadge {
75		/// The parts of a store URL; `None` for a scheme the hub does not
76		/// keep repositories on.
77	6×	pub fn of(url: &str) -> Option<StoreBadge> {
78	6×	let (scheme, rest) = url.split_once("://")?;
79	6×	let (path, query) = rest.split_once('?').unwrap_or((rest, ""));
80	6×	let region = query.split('&').find_map(\|kv\| kv.strip_prefix("region=")).filter(\|r\| !r.is_empty()).map(str::to_string);
81	6×	let mut parts = path.trim_matches('/').splitn(if scheme == "az" { 3 } else { 2 }, '/');
82	6×	let (kind, provider, place) = match scheme {
83	6×	"gs" => ("gcs", "Google Cloud Storage", parts.next()?.to_string()),
84	5×	"s3" => ("s3", "Amazon S3", parts.next()?.to_string()),
85	3×	"az" => ("azure", "Azure Blob Storage", format!("{} / {}", parts.next()?, parts.next()?)),
86	1×	_ => return None,
87		};
88	4×	Some(StoreBadge { kind, provider, place, prefix: parts.next().unwrap_or("").to_string(), region, url: url.to_string() })
89	6×	}
90		}
91
92	524×	pub fn location(app: &Shared, owner: &str, name: &str) -> Option<String> {
93	524×	let dir = app.repos.dir(&crate::repos::RepoName::new(owner, name)?);
94	524×	KNOWN.read().ok()?.as_ref()?.urls.get(&dir).cloned()
95	524×	}
96
97		/// The directory of the repository with row `repo_id`.
98	0×	async fn dir_of(app: &Shared, repo_id: i64) -> Option<std::path::PathBuf> {
99	0×	let row = crate::db::repos::get_by_id(&app.db, repo_id).await.ok()??;
100	0×	Some(app.repos.dir(&crate::repos::RepoName::new(&row.owner, &row.name)?))
101	0×	}
102
103		/// Hand the cloud stores to oxy-core, and every repository's credentials
104		/// to them. At start.
105	194×	pub async fn load(app: &Shared) -> sqlx::Result<()> {
106	194×	oxy_strata::register();
107	194×	oxy_strata::set_credentials_lookup(lookup);
108	194×	for s in all(&app.db).await? {
109	0×	match app.secret_key.open(&s.credentials).map(\|j\| Credentials::from_json(&j)) {
110	0×	Some(Ok(c)) => match dir_of(app, s.repo_id).await {
111	0×	Some(dir) => remember(dir, &s.url, c),
112	0×	None => tracing::warn!("storage: repository {} has a store but no row", s.repo_id),
113		},
114	0×	_ => tracing::warn!("storage: the credentials of repository {} do not open under keys/secrets.key", s.repo_id),
115		}
116		}
117	194×	Ok(())
118	194×	}
119
120	194×	pub async fn all(pool: &Pool) -> sqlx::Result<Vec<Storage>> {
121	194×	sqlx::query_as("SELECT * FROM repo_storage").fetch_all(pool).await
122	194×	}
123
124	141×	pub async fn of(pool: &Pool, repo_id: i64) -> sqlx::Result<Option<Storage>> {
125	141×	sqlx::query_as("SELECT * FROM repo_storage WHERE repo_id = $1").bind(repo_id).fetch_optional(pool).await
126	141×	}
127
128		/// Check a store URL a user gave: a bucket on S3, Google Cloud Storage or
129		/// Azure, with no setting but `region`. A profile, a key file or an
130		/// endpoint would make the hub use its own host's credentials or reach
131		/// addresses on its network, and `dir://` its own disk.
132	17×	pub fn check_url(url: &str) -> Result<oxy_strata::StoreUrl, String> {
133	17×	let scheme = url.trim().split_once("://").map(\|(s, _)\| s).unwrap_or("");
134	17×	if !matches!(scheme, "s3" \| "gs" \| "az") {
135	3×	return Err("a store is s3://BUCKET/PREFIX, gs://BUCKET/PREFIX or az://ACCOUNT/CONTAINER/PREFIX".into());
136	14×	}
137	14×	let parsed = oxy_strata::StoreUrl::parse(url.trim()).map_err(\|e\| e.to_string())?;
138	14×	if parsed.scheme == "az" && parsed.path.is_empty() {
139	2×	return Err("an Azure store names the account and the container: az://ACCOUNT/CONTAINER/PREFIX".into());
140	12×	}
141	12×	if let Some(k) = parsed.params.keys().find(\|k\| k.as_str() != "region") {
142	6×	return Err(format!("?{k}= is not allowed here: the only setting a hub accepts is ?region="));
143	6×	}
144	6×	Ok(parsed)
145	17×	}
146
147		/// The credentials a form gave for a store of `scheme`.
148	1×	pub fn credentials_from(
149	1×	scheme: &str,
150	1×	access_key_id: &str,
151	1×	secret_access_key: &str,
152	1×	service_account_key: &str,
153	1×	tenant_id: &str,
154	1×	client_id: &str,
155	1×	client_secret: &str,
156	1×	) -> Result<Credentials, String> {
157	1×	let json = match scheme {
158	1×	"s3" => serde_json::json!({ "kind": "aws", "access_key_id": access_key_id, "secret_access_key": secret_access_key }),
159	0×	"gs" => serde_json::json!({ "kind": "gcp", "service_account_key": service_account_key }),
160	0×	"az" => serde_json::json!({ "kind": "azure", "tenant_id": tenant_id, "client_id": client_id, "client_secret": client_secret }),
161	0×	_ => return Err("unknown kind of store".into()),
162		};
163	1×	Credentials::from_json(&json.to_string()).map_err(\|e\| e.to_string())
164	1×	}
165
166		/// Run the conformance check against a store under a probe prefix of its
167		/// own, which it leaves empty: what passed, or what failed. With
168		/// `must_be_empty` the store must hold nothing yet (a new repository's).
169	0×	pub async fn probe(url: &str, credentials: &Credentials, must_be_empty: bool) -> Result<String, String> {
170	0×	let (url, credentials) = (url.to_string(), credentials.clone());
171	0×	tokio::task::spawn_blocking(move \|\| {
172		use oxy_core::strata::Store as _;
173	0×	let store = oxy_strata::CloudStore::open_with(&url, Some(&credentials)).map_err(\|e\| e.to_string())?;
174	0×	if must_be_empty {
175	0×	let found = store.list("").map_err(\|e\| e.to_string())?;
176	0×	if !found.is_empty() {
177	0×	return Err(format!(
178	0×	"{} already holds {} object(s): a new repository needs an empty bucket or prefix",
179	0×	store.describe(),
180	0×	found.len()
181	0×	));
182	0×	}
183	0×	}
184	0×	let mut nonce = [0u8; 6];
185	0×	rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce);
186	0×	let prefix = format!("oxy-probe-{}", hex::encode(nonce));
187	0×	let passed = oxy_core::strata::conformance(&store, &prefix).map_err(\|e\| {
188	0×	for e in store.list(&prefix).unwrap_or_default() {
189	0×	let _ = store.delete(&e.key);
190	0×	}
191	0×	e.to_string()
192	0×	})?;
193	0×	Ok(format!("{} checks passed", passed.len()))
194	0×	})
195	0×	.await
196	0×	.map_err(\|e\| e.to_string())?
197	0×	}
198
199		/// Create repository `name` (row `repo_id`, already in the database) in
200		/// the store at `url`: checked first, then the credentials sealed and
201		/// kept, then the bare repository made over the store.
202	0×	pub async fn create(app: &Shared, name: &crate::repos::RepoName, repo_id: i64, url: &str, credentials: Credentials) -> Result<(), String> {
203	0×	let result = probe(url, &credentials, true).await?;
204	0×	let sealed = app.secret_key.seal(&credentials.to_json());
205	0×	let dir = app.repos.dir(name);
206	0×	remember(dir.clone(), url, credentials);
207	0×	let made = {
208	0×	let (app, name, url) = (app.clone(), name.clone(), url.to_string());
209	0×	tokio::task::spawn_blocking(move \|\| app.repos.create_on(&name, &url).map(drop).map_err(\|e\| e.to_string()))
210	0×	.await
211	0×	.map_err(\|e\| e.to_string())
212	0×	.and_then(\|r\| r)
213		};
214	0×	if let Err(e) = made {
215	0×	forget(&dir);
216	0×	return Err(format!("the repository could not be made in {url}: {e}"));
217	0×	}
218	0×	let now = now_ms();
219	0×	sqlx::query(
220	0×	"INSERT INTO repo_storage (repo_id, url, credentials, created_ms, checked_ms, check_result) VALUES ($1, $2, $3, $4, $5, $6)",
221	0×	)
222	0×	.bind(repo_id)
223	0×	.bind(url)
224	0×	.bind(sealed)
225	0×	.bind(now)
226	0×	.bind(now)
227	0×	.bind(&result)
228	0×	.execute(&app.db)
229	0×	.await
230	0×	.map_err(\|e\| e.to_string())?;
231	0×	Ok(())
232	0×	}
233
234		/// Check a repository's store again and record the result.
235	0×	pub async fn check(app: &Shared, s: &Storage) -> Result<String, String> {
236	0×	let json = app.secret_key.open(&s.credentials).ok_or("the credentials do not open under this hub's key")?;
237	0×	let credentials = Credentials::from_json(&json).map_err(\|e\| e.to_string())?;
238	0×	let result = probe(&s.url, &credentials, false).await;
239	0×	let text = match &result {
240	0×	Ok(r) => r.clone(),
241	0×	Err(e) => format!("failed: {e}"),
242		};
243	0×	sqlx::query("UPDATE repo_storage SET checked_ms = $1, check_result = $2 WHERE repo_id = $3")
244	0×	.bind(now_ms())
245	0×	.bind(&text)
246	0×	.bind(s.repo_id)
247	0×	.execute(&app.db)
248	0×	.await
249	0×	.map_err(\|e\| e.to_string())?;
250	0×	result
251	0×	}
252
253		/// Replace a repository's credentials (a rotated key), once they pass the
254		/// check against its store.
255	0×	pub async fn set_credentials(app: &Shared, s: &Storage, credentials: Credentials) -> Result<String, String> {
256	0×	let result = probe(&s.url, &credentials, false).await?;
257	0×	sqlx::query("UPDATE repo_storage SET credentials = $1, checked_ms = $2, check_result = $3 WHERE repo_id = $4")
258	0×	.bind(app.secret_key.seal(&credentials.to_json()))
259	0×	.bind(now_ms())
260	0×	.bind(&result)
261	0×	.bind(s.repo_id)
262	0×	.execute(&app.db)
263	0×	.await
264	0×	.map_err(\|e\| e.to_string())?;
265	0×	if let Some(dir) = dir_of(app, s.repo_id).await {
266	0×	remember(dir, &s.url, credentials);
267	0×	}
268	0×	Ok(result)
269	0×	}
270
271		/// A repository is gone (its store row with it): forget its credentials.
272	1×	pub fn removed(app: &Shared, name: &crate::repos::RepoName) {
273	1×	forget(&app.repos.dir(name));
274	1×	}
275
276		/// A repository moved to another name (a transfer): its store goes with it.
277	1×	pub fn renamed(app: &Shared, from: &crate::repos::RepoName, to: &crate::repos::RepoName) {
278	1×	let mut known = KNOWN.write().unwrap_or_else(\|p\| p.into_inner());
279	1×	if let Some(k) = known.as_mut() {
280	0×	if let Some(url) = k.urls.remove(&app.repos.dir(from)) {
281	0×	k.urls.insert(app.repos.dir(to), url);
282	0×	}
283	1×	}
284	1×	}
285
286		/// An organisation's storage policy: every new repository in a bucket,
287		/// and only in stores under these URL prefixes when there are any. The
288		/// buckets an organisation lists are its own, so where they are (the
289		/// region, the provider, the country) is where its data is.
290		#[derive(Clone, Debug, Default, PartialEq, Eq)]
291		pub struct Policy {
292		pub require_bucket: bool,
293		pub allowed: Vec<String>,
294		}
295
296		impl Policy {
297		/// Whether a new repository may be kept at `store` (`None`: this
298		/// hub's disk); otherwise what to tell its creator.
299	9×	pub fn permits(&self, org: &str, store: Option<&str>) -> Result<(), String> {
300	6×	match store {
301	2×	None if self.require_bucket => Err(format!(
302		"{org} keeps every new repository in a bucket: choose “my own bucket” under Storage{}",
303	2×	if self.allowed.is_empty() { String::new() } else { format!(" ({})", self.allowed.join(", ")) }
304		)),
305	5×	Some(url) if !self.allowed.is_empty() && !self.allowed.iter().any(\|p\| url.trim().starts_with(p.as_str())) => {
306	2×	Err(format!("{org} allows only these stores: {}", self.allowed.join(", ")))
307		}
308	5×	_ => Ok(()),
309		}
310	9×	}
311		}
312
313	36×	pub async fn policy(pool: &Pool, org_id: i64) -> sqlx::Result<Policy> {
314	36×	let row: Option<(bool, String)> = sqlx::query_as("SELECT require_bucket, allowed FROM org_storage_policy WHERE org_id = $1")
315	36×	.bind(org_id)
316	36×	.fetch_optional(pool)
317	36×	.await?;
318	36×	Ok(row
319	36×	.map(\|(r, a)\| Policy { require_bucket: r, allowed: a.lines().map(str::trim).filter(\|l\| !l.is_empty()).map(String::from).collect() })
320	36×	.unwrap_or_default())
321	36×	}
322
323		/// Set a policy from what a form gave: the allowed prefixes one per line
324		/// (or comma-separated), each a store URL of a bucket.
325	2×	pub async fn set_policy(pool: &Pool, org_id: i64, require_bucket: bool, allowed: &str) -> Result<Policy, String> {
326	2×	let mut prefixes = Vec::new();
327	3×	for p in allowed.split([',', '\n']).map(str::trim).filter(\|p\| !p.is_empty()) {
328	3×	if p.contains('?') {
329	1×	return Err(format!("{p}: a prefix names a bucket and a path, with no settings"));
330	2×	}
331	2×	check_url(p).map_err(\|e\| format!("{p}: {e}"))?;
332	2×	prefixes.push(p.to_string());
333		}
334	1×	sqlx::query(
335	1×	"INSERT INTO org_storage_policy (org_id, require_bucket, allowed) VALUES ($1, $2, $3) \
336	1×	ON CONFLICT (org_id) DO UPDATE SET require_bucket = excluded.require_bucket, allowed = excluded.allowed",
337	1×	)
338	1×	.bind(org_id)
339	1×	.bind(require_bucket)
340	1×	.bind(prefixes.join("\n"))
341	1×	.execute(pool)
342	1×	.await
343	1×	.map_err(\|e\| e.to_string())?;
344	1×	Ok(Policy { require_bucket, allowed: prefixes })
345	2×	}
346
347		#[cfg(test)]
348		mod tests {
349		use super::*;
350
351		#[test]
352	1×	fn store_urls_a_hub_accepts() {
353	1×	assert!(check_url("s3://bucket/repos/x?region=eu-west-2").is_ok());
354	1×	assert!(check_url("gs://bucket/x").is_ok());
355	1×	assert!(check_url("az://account/container/x").is_ok());
356	6×	for (bad, why) in [
357	1×	("s3://b/x?profile=prod", "?profile="),
358	1×	("s3://b/x?region=a&endpoint=http://10.0.0.1", "?endpoint="),
359	1×	("gs://b/x?credentials=/etc/key.json", "?credentials="),
360	1×	("dir:///srv", "a store is"),
361	1×	("file:///srv", "a store is"),
362	1×	("az://account", "the container"),
363	1×	] {
364	6×	let e = check_url(bad).unwrap_err();
365	6×	assert!(e.contains(why), "{bad}: {e}");
366		}
367	1×	}
368
369		#[test]
370	1×	fn a_policy_says_what_it_allows() {
371	1×	let open = Policy::default();
372	1×	assert!(open.permits("acme", None).is_ok());
373	1×	let strict = Policy { require_bucket: true, allowed: vec!["s3://acme-london/".into(), "gs://acme-eu/".into()] };
374	1×	let e = strict.permits("acme", None).unwrap_err();
375	1×	assert!(e.contains("acme keeps every new repository in a bucket") && e.contains("s3://acme-london/"), "{e}");
376	1×	assert!(strict.permits("acme", Some("s3://acme-london/repos/x?region=eu-west-2")).is_ok());
377	1×	let e = strict.permits("acme", Some("s3://acme-virginia/x")).unwrap_err();
378	1×	assert!(e.contains("allows only these stores"), "{e}");
379	1×	}
380		}
381
382		#[cfg(test)]
383		mod badge_tests {
384		use super::StoreBadge;
385
386		#[test]
387	1×	fn a_store_url_names_its_provider_place_prefix_and_region() {
388	1×	let g = StoreBadge::of("gs://doomsdayone-oxy-repos/repos/tetris").unwrap();
389	1×	assert_eq!(
390	1×	(g.kind, g.provider, g.place.as_str(), g.prefix.as_str(), g.region),
391		("gcs", "Google Cloud Storage", "doomsdayone-oxy-repos", "repos/tetris", None)
392		);
393	1×	let s = StoreBadge::of("s3://my-bucket/repos/name?region=eu-west-2").unwrap();
394	1×	assert_eq!(
395	1×	(s.kind, s.place.as_str(), s.prefix.as_str(), s.region.as_deref()),
396		("s3", "my-bucket", "repos/name", Some("eu-west-2"))
397		);
398	1×	let a = StoreBadge::of("az://account/container/repos/x").unwrap();
399	1×	assert_eq!((a.kind, a.place.as_str(), a.prefix.as_str()), ("azure", "account / container", "repos/x"));
400	1×	assert_eq!(StoreBadge::of("s3://bucket").unwrap().prefix, "");
401	1×	assert!(StoreBadge::of("dir:///tmp/x").is_none());
402	1×	assert!(StoreBadge::of("az://only-account").is_none());
403	1×	}
404		}

Coverage / crates/hub/src/storage.rs 52.5%

Keyboard