feat: keyless GCS backups, and TEST (closes #169) 35fea4237cc9

- **Backups (#169, docs/backups.md):** a user (Settings › Backups) or an organisation's owners copy every repository they own, with its wiki, issues, proposals, releases and settings (never a secret value), hourly or daily, to a directory the hub allows or a Google Cloud Storage bucket with no key at all (the hub signs in as `backup:OWNER` through workload identity federation; TEST checks it). Only what changed travels, runs are kept 7, 30 or 90 days, a failure lands in the owner's inbox, and `oxygenhub admin backup restore` brings repositories back byte for byte.

/// A Cloud Storage bucket, reached keylessly (docs/backups.md §3, §6):

/// the hub mints a short token as `backup:OWNER`, Google's STS exchanges

/// it for an access token through the owner's workload identity

/// provider, and that writes the bucket. No key exists anywhere; the

/// access token lasts an hour and is fetched again before it runs out.

pub struct Gcs {

    bucket: String,

    /// Keys go under this, when set.

    prefix: String,

    storage: String,

    sts: String,

    /// The provider's resource name, `//iam.googleapis.com/projects/…`.

    provider: String,

    subject: String,

    app: Shared,

    client: reqwest::Client,

    rt: tokio::runtime::Handle,

    token: std::sync::Mutex<Option<(String, i64)>>,

}

impl Gcs {

    pub fn new(app: &Shared, s: &Setting) -> Gcs {

        Gcs {

            bucket: s.target.clone(),

            prefix: s.prefix.trim_matches('/').to_string(),

            storage: app.config.backups.storage_url.trim_end_matches('/').to_string(),

            sts: app.config.backups.sts_url.trim_end_matches('/').to_string(),

            provider: s.provider.clone(),

            subject: format!("backup:{}", s.owner),

            app: app.clone(),

            client: reqwest::Client::builder().timeout(std::time::Duration::from_secs(300)).build().unwrap_or_default(),

            rt: tokio::runtime::Handle::current(),

            token: std::sync::Mutex::new(None),

        }

    }

    fn name(&self, key: &str) -> String {

        if self.prefix.is_empty() {

            key.to_string()

        } else {

            format!("{}/{key}", self.prefix)

        }

    }

    /// An access token, exchanged for a fresh one of the hub's when the

    /// last is within five minutes of running out.

    fn access(&self) -> Result<String> {

        let now = db::now_ms();

        if let Some((t, until)) = self.token.lock().unwrap().as_ref() {

            if *until - now > 300_000 {

                return Ok(t.clone());

            }

        }

        let (token, expires_in) = self.rt.block_on(async {

            let issuer = self.app.oidc().await?;

            // Google's default audience for a provider: its name over https.

            let audience = format!("https:{}", self.provider);

            let jwt = issuer.mint_for(self.app.config.base_url(), &self.subject, &audience, db::now_ms(), 600);

            let res = self

                .client

                .post(format!("{}/v1/token", self.sts))

                .json(&serde_json::json!({

                    "grantType": "urn:ietf:params:oauth:grant-type:token-exchange",

                    "audience": self.provider,

                    "scope": "https://www.googleapis.com/auth/devstorage.read_write",

                    "requestedTokenType": "urn:ietf:params:oauth:token-type:access_token",

                    "subjectToken": jwt,

                    "subjectTokenType": "urn:ietf:params:oauth:token-type:jwt",

                }))

                .send()

                .await?;

            let status = res.status();

            let body: serde_json::Value = res.json().await.unwrap_or_default();

            if !status.is_success() {

                bail!(

                    "Google's token exchange refused {}: {} (check the provider's issuer is this hub's address and its attribute condition admits backup:)",

                    self.subject,

                    body["error_description"].as_str().or(body["error"].as_str()).unwrap_or("no reason given")

                );

            }

            let token = body["access_token"].as_str().context("the token exchange gave no access token")?.to_string();

            anyhow::Ok((token, body["expires_in"].as_i64().unwrap_or(3600)))

        })?;

        *self.token.lock().unwrap() = Some((token.clone(), now + expires_in * 1000));

        Ok(token)

    }

    fn object_url(&self, key: &str) -> Result<reqwest::Url> {

        let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;

        u.path_segments_mut().map_err(|_| anyhow::anyhow!("bad storage URL"))?.push(&self.name(key));

        Ok(u)

    }

    fn refused(&self, what: &str, status: reqwest::StatusCode, body: &str) -> anyhow::Error {

        let hint = if status == reqwest::StatusCode::FORBIDDEN || status == reqwest::StatusCode::UNAUTHORIZED {

            format!(" (grant roles/storage.objectAdmin on gs://{} to the principal …/subject/{})", self.bucket, self.subject)

        } else {

            String::new()

        };

        anyhow::anyhow!("GCS {what}: {status}{hint}: {}", body.chars().take(200).collect::<String>())

    }

    /// Every object name under `full` (no delimiter), for removal.

    fn all_under(&self, full: &str) -> Result<Vec<String>> {

        let mut out = Vec::new();

        let mut page: Option<String> = None;

        loop {

            let token = self.access()?;

            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;

            u.query_pairs_mut().append_pair("prefix", full).append_pair("fields", "items(name),nextPageToken");

            if let Some(p) = &page {

                u.query_pairs_mut().append_pair("pageToken", p);

            }

            let body: serde_json::Value = self.rt.block_on(async {

                let res = self.client.get(u).bearer_auth(&token).send().await?;

                let status = res.status();

                let text = res.text().await?;

                if !status.is_success() {

                    return Err(self.refused("list", status, &text));

                }

                anyhow::Ok(serde_json::from_str(&text)?)

            })?;

            out.extend(body["items"].as_array().into_iter().flatten().filter_map(|i| i["name"].as_str().map(String::from)));

            match body["nextPageToken"].as_str() {

                Some(p) => page = Some(p.to_string()),

                None => return Ok(out),

            }

        }

    }

}

impl Destination for Gcs {

    fn exists(&self, key: &str) -> Result<bool> {

        let token = self.access()?;

        let mut u = self.object_url(key)?;

        u.query_pairs_mut().append_pair("fields", "name");

        self.rt.block_on(async {

            let res = self.client.get(u).bearer_auth(&token).send().await?;

            match res.status() {

                s if s.is_success() => Ok(true),

                reqwest::StatusCode::NOT_FOUND => Ok(false),

                s => Err(self.refused("read", s, &res.text().await.unwrap_or_default())),

            }

        })

    }

    fn put(&self, key: &str, bytes: &[u8]) -> Result<()> {

        let token = self.access()?;

        let mut u = reqwest::Url::parse(&format!("{}/upload/storage/v1/b/{}/o", self.storage, self.bucket))?;

        u.query_pairs_mut().append_pair("uploadType", "media").append_pair("name", &self.name(key));

        let body = bytes.to_vec();

        self.rt.block_on(async {

            let res = self.client.post(u).bearer_auth(&token).header("content-type", "application/octet-stream").body(body).send().await?;

            let status = res.status();

            if !status.is_success() {

                return Err(self.refused("write", status, &res.text().await.unwrap_or_default()));

            }

            Ok(())

        })

    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {

        let token = self.access()?;

        let mut u = self.object_url(key)?;

        u.query_pairs_mut().append_pair("alt", "media");

        self.rt.block_on(async {

            let res = self.client.get(u).bearer_auth(&token).send().await?;

            let status = res.status();

            if !status.is_success() {

                return Err(self.refused(&format!("read {key}"), status, &res.text().await.unwrap_or_default()));

            }

            Ok(res.bytes().await?.to_vec())

        })

    }

    fn list(&self, prefix: &str) -> Result<Vec<String>> {

        let full = format!("{}/", self.name(prefix.trim_end_matches('/')));

        let mut out = std::collections::BTreeSet::new();

        let mut page: Option<String> = None;

        loop {

            let token = self.access()?;

            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;

            u.query_pairs_mut()

                .append_pair("prefix", &full)

                .append_pair("delimiter", "/")

                .append_pair("fields", "items(name),prefixes,nextPageToken");

            if let Some(p) = &page {

                u.query_pairs_mut().append_pair("pageToken", p);

            }

            let body: serde_json::Value = self.rt.block_on(async {

                let res = self.client.get(u).bearer_auth(&token).send().await?;

                let status = res.status();

                let text = res.text().await?;

                if !status.is_success() {

                    return Err(self.refused("list", status, &text));

                }

                anyhow::Ok(serde_json::from_str(&text)?)

            })?;

            for n in body["items"].as_array().into_iter().flatten().filter_map(|i| i["name"].as_str()) {

                out.insert(n.trim_start_matches(&full).to_string());

            }

            for p in body["prefixes"].as_array().into_iter().flatten().filter_map(|p| p.as_str()) {

                out.insert(p.trim_start_matches(&full).trim_end_matches('/').to_string());

            }

            match body["nextPageToken"].as_str() {

                Some(p) => page = Some(p.to_string()),

                None => return Ok(out.into_iter().filter(|n| !n.is_empty()).collect()),

            }

        }

    }

    fn remove(&self, key: &str) -> Result<()> {

        let mut names = self.all_under(&format!("{}/", self.name(key)))?;

        names.push(self.name(key));

        for n in names {

            let token = self.access()?;

            let mut u = reqwest::Url::parse(&format!("{}/storage/v1/b/{}/o", self.storage, self.bucket))?;

            u.path_segments_mut().map_err(|_| anyhow::anyhow!("bad storage URL"))?.push(&n);

            self.rt.block_on(async {

                let res = self.client.delete(u).bearer_auth(&token).send().await?;

                match res.status() {

                    s if s.is_success() || s == reqwest::StatusCode::NOT_FOUND => Ok(()),

                    s => Err(self.refused("delete", s, &res.text().await.unwrap_or_default())),

                }

            })?;

        }

        Ok(())

    }

}

/// Check a destination end to end: write a probe, read it back, remove it.

pub fn probe(dest: &dyn Destination) -> Result<()> {

    let marker = format!("probe {}", db::now_ms());

    dest.put(".oxyhub-probe", marker.as_bytes())?;

    if dest.get(".oxyhub-probe")? != marker.as_bytes() {

        bail!("the probe read back differently");

    }

    dest.remove(".oxyhub-probe")

}

        }

        "gcs" => {

            if s.target.is_empty() || s.provider.is_empty() {

                bail!("a GCS destination needs its bucket and its workload identity provider");

            }

            Ok(Box::new(Gcs::new(app, s)))

                    tokio::task::block_in_place(|| dest.put(&key, &serde_json::to_vec(&m)?))?;

            Ok(r) => tokio::task::block_in_place(|| dest.put(&format!("runs/{at}/records/{}.json", row.name), &serde_json::to_vec(&r)?))?,

    // The blocking calls of a destination (GCS drives its client on this

    // runtime) run where blocking is allowed.

    tokio::task::block_in_place(|| -> Result<()> {

        dest.put(&format!("runs/{at}/owner.json"), &serde_json::to_vec(&owner_rows)?)?;

        dest.put(&format!("runs/{at}/DONE"), at.as_bytes())?;

        dest.put("LATEST", at.as_bytes())?;

        prune(dest.as_ref(), s.keep_days)

    })?;

        None => String::from_utf8_lossy(&tokio::task::block_in_place(|| dest.get("LATEST"))?).trim().to_string(),

    for name in tokio::task::block_in_place(|| dest.list(&format!("runs/{at}/records")))? {

        let rec: serde_json::Value =

            serde_json::from_slice(&tokio::task::block_in_place(|| dest.get(&format!("runs/{at}/records/{name}")))?)?;

#[derive(Clone, Debug, Deserialize)]

    /// Google's token exchange and Cloud Storage, for GCS destinations;

    /// set only to point them elsewhere (tests, a private endpoint).

    pub sts_url: String,

    pub storage_url: String,

}

impl Default for Backups {

    fn default() -> Self {

        Backups {

            directories: Vec::new(),

            sts_url: "https://sts.googleapis.com".into(),

            storage_url: "https://storage.googleapis.com".into(),

        }

    }

    /// A signed token for the hub itself acting as `subject` (a backup's

    /// `backup:OWNER`, oxygenhub#169), for `audience`, living `ttl_s`.

    pub fn mint_for(&self, issuer: &str, subject: &str, audience: &str, now_ms: i64, ttl_s: u64) -> String {

        let iat = now_ms / 1000;

        let header = serde_json::json!({ "alg": "RS256", "typ": "JWT", "kid": self.kid });

        let payload = serde_json::json!({

            "iss": issuer,

            "sub": subject,

            "aud": audience,

            "iat": iat,

            "nbf": iat - 30,

            "exp": iat + ttl_s.clamp(60, MAX_TTL_S) as i64,

            "jti": crate::auth::random_hex(16),

        });

        let signing_input = format!("{}.{}", b64url(header.to_string().as_bytes()), b64url(payload.to_string().as_bytes()));

        let sig = self.signing.sign(signing_input.as_bytes());

        format!("{signing_input}.{}", b64url(&sig.to_bytes()))

    }

/// TEST: reach the saved destination and write, read and remove a probe.

async fn test(app: &Shared, shell: Shell, kind: &'static str, owner: &str) -> Result<Page<BackupsPage>, Error> {

    let Some(set) = backups::setting(&app.db, owner).await.map_err(|e| Error::from_db(shell.clone(), e))? else {

        return page(app, shell, kind, owner, None, Some("Save where backups go first.".into())).await;

    };

    let outcome = match backups::destination(app, &set) {

        Ok(dest) => {

            tokio::task::spawn_blocking(move || backups::probe(dest.as_ref())).await.unwrap_or_else(|e| Err(anyhow::anyhow!("{e}")))

        }

        Err(e) => Err(e),

    };

    match outcome {

        Ok(()) => {

            let place = if set.destination == "gcs" { format!("gs://{}", set.target) } else { format!("{}/{owner}", set.target) };

            page(app, shell, kind, owner, Some(format!("TEST passed: wrote, read and removed a probe in {place}.")), None).await

        }

        Err(e) => page(app, shell, kind, owner, None, Some(format!("TEST failed: {e:#}"))).await,

    }

}

}

pub async fn user_test(State(app): State<Shared>, viewer: Viewer, Form(f): Form<Csrf>) -> Result<Page<BackupsPage>, Error> {

    let shell = Shell::new(&app, "/settings/backups", &viewer);

    let me = require(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    test(&app, shell, "user", &me.account.username).await

}

pub async fn org_test(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(name): Path<String>,

    Form(f): Form<Csrf>,

) -> Result<Page<BackupsPage>, Error> {

    let shell = Shell::new(&app, &format!("/orgs/{name}/settings/backups"), &viewer);

    let (_, org) = super::orgs::owned_org(&app, &shell, &viewer, &name).await?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    test(&app, shell, "org", &org.name).await

        .route("/settings/backups/test", post(backups::user_test))

        .route("/orgs/{name}/settings/backups/test", post(backups::org_test))

    hub_with(backups, None).await

}

async fn hub_with(backups: &Path, google: Option<&str>) -> Hub {

    if let Some(g) = google {

        config.backups.sts_url = g.to_string();

        config.backups.storage_url = g.to_string();

    }

}

// ------------------------------------------------------------ a fake Google

/// What the fake has seen and holds.

#[derive(Default)]

struct Google {

    /// Objects by bucket and name.

    objects: std::collections::BTreeMap<(String, String), Vec<u8>>,

    /// The subjects and audiences of the tokens exchanged.

    exchanged: Vec<(String, String)>,

}

type Shared2 = Arc<std::sync::Mutex<Google>>;

fn decode(s: &str) -> String {

    let b = s.as_bytes();

    let mut out = Vec::new();

    let mut i = 0;

    while i < b.len() {

        if b[i] == b'%' && i + 2 < b.len() {

            out.push(u8::from_str_radix(&s[i + 1..i + 3], 16).unwrap());

            i += 3;

        } else {

            out.push(b[i]);

            i += 1;

        }

    }

    String::from_utf8(out).unwrap()

}

fn query(uri: &axum::http::Uri, key: &str) -> Option<String> {

    uri.query()?.split('&').find_map(|kv| kv.split_once('=').filter(|(k, _)| *k == key).map(|(_, v)| decode(v)))

}

/// Google's STS and Cloud Storage, as much as backups use: a token

/// exchange that admits `backup:` subjects for the provider's audience,

/// and buckets that admit only their owner's token (bucket `OWNER-backup`).

async fn fake_google() -> (String, Shared2) {

    use axum::extract::{Request, State};

    use axum::http::StatusCode;

    use axum::response::IntoResponse;

    let state: Shared2 = Default::default();

    async fn handle(State(g): State<Shared2>, req: Request) -> axum::response::Response {

        let (parts, body) = req.into_parts();

        let bytes = http_body_util::BodyExt::collect(body).await.unwrap().to_bytes();

        let path = parts.uri.path().to_string();

        if path == "/v1/token" {

            let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();

            let jwt = v["subjectToken"].as_str().unwrap();

            let payload = jwt.split('.').nth(1).unwrap();

            let claims: serde_json::Value = serde_json::from_slice(&base64url(payload)).unwrap();

            let (sub, aud) = (claims["sub"].as_str().unwrap().to_string(), claims["aud"].as_str().unwrap().to_string());

            let provider = v["audience"].as_str().unwrap();

            g.lock().unwrap().exchanged.push((sub.clone(), aud.clone()));

            if !sub.starts_with("backup:") || aud != format!("https:{provider}") {

                return (

                    StatusCode::BAD_REQUEST,

                    axum::Json(serde_json::json!({"error": "invalid_grant", "error_description": "not admitted"})),

                )

                    .into_response();

            }

            return axum::Json(serde_json::json!({ "access_token": format!("tok-{sub}"), "expires_in": 3600 })).into_response();

        }

        let auth = parts.headers.get("authorization").and_then(|v| v.to_str().ok()).unwrap_or("").to_string();

        let segs: Vec<&str> = path.split('/').collect();

        // /storage/v1/b/{bucket}/o[/{name}]  or  /upload/storage/v1/b/{bucket}/o

        let upload = segs.get(1) == Some(&"upload");

        let at = if upload { 5 } else { 4 };

        let bucket = segs.get(at).unwrap_or(&"").to_string();

        let owner = bucket.trim_end_matches("-backup");

        if auth != format!("Bearer tok-backup:{owner}") {

            return (StatusCode::FORBIDDEN, "caller does not have storage.objects access").into_response();

        }

        let mut g = g.lock().unwrap();

        if upload {

            let name = query(&parts.uri, "name").unwrap();

            g.objects.insert((bucket, name), bytes.to_vec());

            return axum::Json(serde_json::json!({})).into_response();

        }

        let name = segs.get(at + 2).map(|n| decode(n));

        match (parts.method.as_str(), name) {

            ("GET", Some(n)) => match g.objects.get(&(bucket, n.clone())) {

                Some(b) if query(&parts.uri, "alt").as_deref() == Some("media") => b.clone().into_response(),

                Some(_) => axum::Json(serde_json::json!({ "name": n })).into_response(),

                None => StatusCode::NOT_FOUND.into_response(),

            },

            ("DELETE", Some(n)) => {

                g.objects.remove(&(bucket, n));

                StatusCode::NO_CONTENT.into_response()

            }

            ("GET", None) => {

                let prefix = query(&parts.uri, "prefix").unwrap_or_default();

                let delimited = query(&parts.uri, "delimiter").is_some();

                let (mut items, mut prefixes) = (Vec::new(), std::collections::BTreeSet::new());

                for (b, n) in g.objects.keys() {

                    if *b != bucket || !n.starts_with(&prefix) {

                        continue;

                    }

                    let rest = &n[prefix.len()..];

                    match rest.find('/') {

                        Some(i) if delimited => {

                            prefixes.insert(format!("{prefix}{}", &rest[..=i]));

                        }

                        _ => items.push(serde_json::json!({ "name": n })),

                    }

                }

                axum::Json(serde_json::json!({ "items": items, "prefixes": prefixes })).into_response()

            }

            _ => StatusCode::METHOD_NOT_ALLOWED.into_response(),

        }

    }

    let app = axum::Router::new().fallback(handle).with_state(state.clone());

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

    let port = listener.local_addr().unwrap().port();

    tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });

    (format!("http://127.0.0.1:{port}"), state)

}

fn base64url(s: &str) -> Vec<u8> {

    let table = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";

    let mut bits = 0u32;

    let mut n = 0;

    let mut out = Vec::new();

    for c in s.bytes() {

        let v = table.iter().position(|t| *t == c).unwrap() as u32;

        bits = (bits << 6) | v;

        n += 6;

        if n >= 8 {

            n -= 8;

            out.push((bits >> n) as u8);

            bits &= (1 << n) - 1;

        }

    }

    out

}

const PROVIDER: &str = "//iam.googleapis.com/projects/1/locations/global/workloadIdentityPools/oxygenhub/providers/oxygenhub-io";

fn gcs(owner: &str, bucket: &str) -> Setting {

    let mut s = Setting::off(owner, "user");

    s.every = "hourly".into();

    s.destination = "gcs".into();

    s.target = bucket.into();

    s.provider = PROVIDER.into();

    s

}

#[tokio::test(flavor = "multi_thread")]

async fn a_gcs_backup_needs_no_key_and_restores_the_same_repository() {

    let (google, seen) = fake_google().await;

    let store = tempfile::tempdir().unwrap();

    let h = hub_with(store.path(), Some(&google)).await;

    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", false).await.unwrap();

    db::accounts::create_account(&h.app.db, "bob", "[email protected]", "x", "salt", false).await.unwrap();

    create(&h.app, "andy", "demo", hub_api::Visibility::Public).await;

    let work = tempfile::tempdir().unwrap();

    push(&format!("{}/andy/demo", h.base), work.path(), &[("README.md", "hello\n")]);

    backups::save(&h.app.db, &gcs("andy", "andy-backup")).await.unwrap();

    let first = backups::run(&h.app, "andy", "now").await.unwrap();

    assert_eq!((first.status.as_str(), first.repos), ("ok", 1), "{}", first.detail);

    {

        let g = seen.lock().unwrap();

        assert!(g.objects.keys().any(|(b, n)| b == "andy-backup" && n == &format!("runs/{}/DONE", first.at)));

        assert!(g.objects.keys().all(|(b, _)| b == "andy-backup"));

        let (sub, aud) = g.exchanged.first().unwrap();

        assert_eq!(

            (sub.as_str(), aud.as_str()),

            ("backup:andy", format!("https:{PROVIDER}").as_str()),

            "the hub signs in as the owner, for the provider"

        );

    }

    tokio::time::sleep(std::time::Duration::from_millis(1100)).await;

    let second = backups::run(&h.app, "andy", "now").await.unwrap();

    assert_eq!((second.status.as_str(), second.sent), ("ok", 0), "{}", second.detail);

    // The probe writes, reads and removes; the bucket is as it was.

    let before = seen.lock().unwrap().objects.len();

    let dest = backups::destination(&h.app, &gcs("andy", "andy-backup")).unwrap();

    tokio::task::spawn_blocking(move || backups::probe(dest.as_ref())).await.unwrap().unwrap();

    assert_eq!(seen.lock().unwrap().objects.len(), before);

    // Restored from the bucket into a fresh hub: the same repository.

    let empty = tempfile::tempdir().unwrap();

    let fresh = hub_with(empty.path(), Some(&google)).await;

    let dest: Arc<dyn Destination> = backups::destination(&fresh.app, &gcs("andy", "andy-backup")).unwrap().into();

    assert_eq!(backups::restore_into_hub(&fresh.app, dest, None).await.unwrap(), vec!["andy/demo (repos)"]);

    let name = RepoName::parse("andy/demo").unwrap();

    assert_eq!(files(&fresh.app.repos.dir(&name)), files(&h.app.repos.dir(&name)), "byte for byte");

    // Bob cannot write Andy's bucket: Google refuses his token, and the

    // run says what to grant.

    backups::save(&h.app.db, &gcs("bob", "andy-backup")).await.unwrap();

    let refused = backups::run(&h.app, "bob", "now").await.unwrap();

    assert_eq!(refused.status, "failed");

    assert!(refused.detail.contains("roles/storage.objectAdmin") && refused.detail.contains("backup:bob"), "{}", refused.detail);

Status: built 2026-10-13/14 (oxygenhub#169): user and organisation

settings, the hourly and daily schedule with "back up now" and TEST,

directory destinations and keyless GCS destinations, every repository's

and wiki's store byte for byte, the records as JSON, retention, failure

notices, and restore of the stores and repository rows

(`oxygenhub admin backup run|restore`). Next: restoring issues and

proposals from the records, Admin › Backups, per-repository settings.

  §13.5). Built: the token lives 10 minutes, its audience is the

  provider's name over https (Google's default), the exchange asks for

  `devstorage.read_write`, and the access token is exchanged again five

  minutes before it runs out. A refusal says what to grant. For a backup it mints a short-lived token with

**Done for doomsdayone-oxy on 2026-10-13:** the pool `oxygenhub`, the

provider `oxygenhub-io` and the buckets `andymccall-oxygen-backup` and

`doomsdayone-oxygen-backup`, each granted to its own `backup:` subject.

### 13.5.1 Backups (oxygenhub#169, built 2026-10-13/14)

Each user and organisation backs up its repositories hourly to a place it controls: a GCS bucket reached keylessly (the hub mints a short-lived token with `sub = backup:OWNER`, which GCP's workload identity federation exchanges for an access token, so no key exists), or a directory the site allows. A run sends only the content-addressed objects the destination lacks, the refs at that moment and the hub's records (issues, proposals, releases, wiki, settings; never secret values), and is whole once its `DONE` marker is written; `oxygenhub admin restore` brings repositories back from any whole run. Built: Settings › Backups and an organisation's Settings › Backups (how often: off, hourly or daily; a directory from `hub.toml` `[backups] directories` or a GCS bucket with its workload identity provider; a prefix; runs kept 7, 30 or 90 days; "back up now"; the last 50 runs with their outcome), migration 0060 (`backup_settings`, `backup_runs`), the hub's scheduler (one run at a time; a run left running by a restart is marked failed), each repository's and wiki's store copied file by file under the repository's read lock for the final pass, the records as JSON (every table keyed by the repository, and three levels of rows pointing at them; tables and columns holding secrets, tokens, sessions or passwords never), pruning of old runs and unreferenced blobs, an inbox notice to the owner (an organisation's owners) once per spell of failures, and `oxygenhub admin backup run OWNER` and `oxygenhub admin backup restore --from DIR [--run AT]`, which restores the stores and the repositories' rows. A GCS destination is keyless: the hub mints a 10-minute token (`Issuer::mint_for`, `sub = backup:OWNER`, `aud` the provider's name over https), Google's STS exchanges it (`[backups] sts_url`) for an access token scoped `devstorage.read_write`, used with the Cloud Storage JSON API (`[backups] storage_url`) and exchanged again before it expires; TEST writes, reads and removes a probe. The design, the layout at a destination, setup and restore are in `docs/backups.md`.

    <form class="inline more" method="post" action="{{ base }}/test"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button type="submit" class="small"{% if setting.updated_ms == 0 %} disabled title="Save where backups go first"{% endif %} title="Write, read and remove a probe at the destination">TEST</button></form>

    <form class="inline" method="post" action="{{ base }}/run"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button type="submit" class="small primary-btn"{% if setting.updated_ms == 0 %} disabled title="Save where backups go first"{% endif %}>Back up now</button></form>