OxygenFlow claims list the state's pinned Oxygen subrepos that the pusher can read, and the job token may read those repositories too fd3db644df22

-- Repositories a job's token may also read: those its state pins as

-- subrepos and its pusher can read (OxygenFlow: subrepo inputs).

ALTER TABLE flow_jobs ADD COLUMN also_read TEXT NOT NULL DEFAULT '';

    /// Comma-separated ids of other repositories the job token may read.

    pub also_read: String,

/// Let a job's token also read these repositories (its pinned subrepos).

pub async fn set_also_read(pool: &Pool, job_id: i64, repo_ids: &[i64]) -> sqlx::Result<()> {

    let list: Vec<String> = repo_ids.iter().map(|i| i.to_string()).collect();

    sqlx::query("UPDATE flow_jobs SET also_read = ? WHERE id = ?").bind(list.join(",")).bind(job_id).execute(pool).await?;

    Ok(())

}

impl Job {

    pub fn also_read_ids(&self) -> Vec<i64> {

        self.also_read.split(',').filter_map(|x| x.trim().parse().ok()).collect()

    }

}

        subrepos: vec![],

    let mut claim = claim;

    let (inputs, ids) = subrepo_inputs(&app, &row, &run).await;

    if !ids.is_empty() {

        fdb::set_also_read(&app.db, job.id, &ids).await?;

    }

    claim.subrepos = inputs;

}

/// The Oxygen subrepos the run's state pins that live on this hub and that

/// its pusher may read: what the job's workspace needs beside its own tree.

/// A pin the hub cannot serve is left out; the job then fails on the missing

/// files with an ordinary error.

async fn subrepo_inputs(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run) -> (Vec<flow_core::api::SubrepoInput>, Vec<i64>) {

    let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) else { return (vec![], vec![]) };

    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();

    let pins = app

        .repos

        .read(&name, move |repo, cache| {

            let loaded = cache.load(repo, &heads)?;

            let mut out = Vec::new();

            for (path, f) in &loaded.snapshot.files {

                if f.kind != oxy_core::objects::NodeKind::Subrepo {

                    continue;

                }

                if let oxy_core::state::FileContent::Opaque { blobs } = &f.content {

                    if let Some(b) = blobs.first() {

                        if let Ok(oxy_core::subrepo::Pin::Oxy { repo: id, heads, .. }) = repo

                            .read_blob(b)

                            .map_err(crate::repos::RepoError::Oxy)

                            .and_then(|bytes| oxy_core::subrepo::Pin::decode(&bytes).map_err(|e| crate::repos::RepoError::Oxy(e.into())))

                        {

                            out.push((path.clone(), id, heads));

                        }

                    }

                }

            }

            Ok(out)

        })

        .await

        .unwrap_or_default();

    if pins.is_empty() {

        return (vec![], vec![]);

    }

    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();

    let mut inputs = Vec::new();

    let mut ids = Vec::new();

    let rows = db::repos::list(&app.db).await.unwrap_or_default();

    for (path, id, heads) in pins {

        for other in &rows {

            let Some(other_name) = crate::repos::RepoName::new(&other.owner, &other.name) else { continue };

            let matches = app.repos.read(&other_name, move |r, _| Ok(r.id().ok() == Some(id))).await.unwrap_or(false);

            if !matches {

                continue;

            }

            let role = crate::auth::roles::role_for(&app.db, pusher.as_ref(), other).await.ok().flatten();

            if crate::auth::roles::allows(role, crate::auth::Access::Read) {

                inputs.push(flow_core::api::SubrepoInput {

                    path: path.clone(),

                    repo: format!("{}/{}", other.owner, other.name),

                    clone_url: format!("{}/{}/{}", app.config.base_url(), other.owner, other.name),

                    heads: heads.iter().map(|h| h.to_hex()).collect(),

                });

                ids.push(other.id);

            }

            break;

        }

    }

    (inputs, ids)

    /// A running OxygenFlow job: reads its own repository and the subrepos

    /// its state pins, nothing else (hub spec §13.2).

        also: Vec<i64>,

            return Ok(Caller::Job { repo_id: run.repo_id, also: job.also_read_ids() });

        if let Caller::Job { repo_id, also } = &caller {

            // Only reads: of the job's own repository, and of the subrepos its

            // state pins that its pusher could read when the job was claimed.

                oxy_server::Access::Read if *repo_id == row.id || also.contains(&row.id) => Ok(()),

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

async fn claims_carry_pinned_subrepos_and_the_job_token_reads_only_those() {

    let h = hub().await;

    for n in ["lib", "vault", "other", "app"] {

        create(&h.app, "andy", n).await;

    }

    let push = |name: &str, file: &str| {

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

        let c = client(&format!("{}/andy/{name}", h.base), work.path());

        write(work.path(), file, "x\n");

        c.snapshot("t").unwrap();

        c.commit("first", &[]).unwrap();

        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

        let heads = c.bookmark_heads().unwrap();

        (work, heads)

    };

    let (_l, lib_heads) = push("lib", "lib.rs");

    let (_v, _) = push("vault", "secret.rs");

    let (_o, _) = push("other", "o.rs");

    // The app pins lib and vault as subrepos and pushes as the site token.

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

    let app = client(&format!("{}/andy/app", h.base), work.path());

    app.submodule_add(&format!("{}/andy/lib", h.base), "deps/lib", Some("main")).unwrap();

    app.submodule_add(&format!("{}/andy/vault", h.base), "deps/vault", Some("main")).unwrap();

    // Vault goes private after it was pinned: this pusher cannot read it.

    sqlx::query("UPDATE repositories SET visibility = 'private' WHERE name = 'vault'").execute(&h.app.db).await.unwrap();

    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));

    write(work.path(), "Makefile", "all:\n");

    app.snapshot("t").unwrap();

    app.commit("app with deps", &[]).unwrap();

    assert!(app.push("origin", false).unwrap().iter().all(|r| r.ok));

    assert!(!runs_of(&h.app, "andy", "app").await.is_empty());

    let job = claim(&h.app).await.unwrap();

    let subs = job["subrepos"].as_array().cloned().unwrap_or_default();

    assert_eq!(subs.len(), 1, "vault is private and the pusher has no account: {subs:?}");

    assert_eq!(subs[0]["path"], "deps/lib");

    assert_eq!(subs[0]["repo"], "andy/lib");

    assert_eq!(subs[0]["heads"], serde_json::json!(lib_heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>()));

    assert!(subs[0]["clone_url"].as_str().unwrap().ends_with("/andy/lib"));

    // The job token reads its repository and the pinned lib, nothing else.

    let token = job["token"].as_str().unwrap().to_string();

    let read = |name: &str| {

        let opts = RemoteOptions { token: Some(token.clone()), ..Default::default() };

        open_remote(&format!("{}/andy/{name}", h.base), &opts).and_then(|r| r.refs()).is_ok()

    };

    assert!(read("app") && read("lib"));

    assert!(!read("vault"), "not pinned for this pusher");

    assert!(!read("other"), "public, but not the job's to read");

}