Another repository's pass is taken only on its signed receipt, re-verified from that repository's store and its key not revoked; only passes that ran there (closes #296's C2, refs #296) 06b756358f7e

- **Another repository's pass is taken only on its signed receipt (#296).** A pass from upstream or from any other repository on the hub now stands only if its runner's signed receipt was kept (the hub checks it against the runner's bound key when the job finishes) and, read again from that repository's store, still verifies and its key is not revoked; otherwise the job runs here. Only a pass that ran in the repository offering it is taken, so the receipt is where it is checked.

    passed_in(pool, repo_id, state, name, spec, sandbox, false).await

}

/// As [`passed_on_state`]; with `ran_here`, only a job that ran in that

/// repository, not one it reused: its signed receipt is in that

/// repository's store, where another repository's reuse checks it (#296).

async fn passed_in(

    pool: &Pool,

    repo_id: i64,

    state: &str,

    name: &str,

    spec: &str,

    sandbox: &str,

    ran_here: bool,

) -> sqlx::Result<Option<Reused>> {

           AND (NOT $6 OR j.error NOT LIKE 'reused from %') \

    .bind(ran_here)

        if let Some(mut r) = passed_in(pool, up.id, state, name, spec, sandbox, true).await? {

           AND o.id <> $5 AND (o.visibility = 'public' OR o.owner = $6) AND j.error NOT LIKE 'reused from %' \

                    // Another repository's pass is taken only on its signed

                    // receipt, checked again now: otherwise it runs here.

                    if let Some((from, attest)) = j.reused.as_ref().and_then(|r| Some((r.repo.clone()?, r.receipt_attest.clone()))) {

                        if let Err(why) = receipt_holds(app, &from, &attest).await {

                            tracing::info!("flow: not reusing {from}'s pass of {} on {state}: {why}", j.name);

                            j.reused = None;

                        }

                    }

    }

    Ok(())

}

/// Whether another repository's pass may stand here (#296): its runner's

/// signed receipt was kept (the hub checked it against the runner's bound

/// key when the job finished), and read again from that repository's store

/// it still verifies and its key is not revoked.

pub async fn receipt_holds(app: &Shared, from: &str, attest: &str) -> Result<(), String> {

    if attest.is_empty() {

        return Err("no signed receipt was kept".into());

    }

    let repo = crate::repos::RepoName::parse(from).ok_or("not a repository name")?;

    let at = oxy_core::hash::ObjectRef::from_hex(attest).ok_or("not a receipt's hash")?;

    let info = app.repos.read(&repo, move |r, _| Ok(r.receipt(&at)?)).await.map_err(|e| format!("the receipt: {e}"))?;

    if let Some(problem) = info.problem {

        return Err(format!("its receipt does not verify: {problem}"));

    }

    if info.revoked {

        return Err("its runner's key is revoked".into());

    // Upstream's jobs pass on a runner, each with its signed receipt, which

    // the hub checks against the runner's key and keeps (#296's C2).

    {

        use oxy_core::hash::{hash_object, ChangeHash, ObjectType};

        use oxy_core::identity::SecretKey;

        use oxy_core::objects::StateObject;

        use oxy_core::receipt::SignedReceipt;

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

        let state = hash_object(ObjectType::State, &StateObject::new(heads).encode());

        let repo_id = h.app.repos.open(&RepoName::parse("andy/game").unwrap()).unwrap().id().unwrap();

        let sk = SecretKey::generate();

        let runner = sk.identity("test-1 (OxygenFlow runner)", "", 1);

        let run_url = format!("{}/andy/game/flow/{}", h.base, first.number);

        while let Some(j) = claim(&h.app).await {

            let r = receipt(j["job"]["name"].as_str().unwrap(), j["pipeline_hash"].as_str().unwrap(), 0);

            let parsed: flow_core::Receipt = serde_json::from_value(r.clone()).unwrap();

            let signed = hex::encode(SignedReceipt::sign(parsed.to_oxy(), repo_id, state, &run_url, &runner, &sk).encode());

            let body = serde_json::json!({ "exit_code": 0, "duration_ms": 5, "receipt": r, "signed_receipt": signed });

            assert_eq!(finish(&h.app, j["job_id"].as_i64().unwrap(), body).await, StatusCode::NO_CONTENT);

        }

    }

    let up_jobs = db::flow::jobs(&h.app.db, first.id).await.unwrap();

    assert!(up_jobs.iter().all(|j| j.status == "passed" && j.receipt_attest.len() == 64), "{up_jobs:?}");

        let from = up_jobs.iter().find(|u| u.name == j.name).unwrap();

        assert_eq!(j.receipt_attest, from.receipt_attest, "upstream's signed receipt stands for it");

    // A pass whose receipt does not hold (here, one the store never kept)

    // is not taken: that job runs in the fork; the others are reused.

    let lint = up_jobs.iter().find(|j| j.name == "lint").unwrap();

    sqlx::query("UPDATE flow_jobs SET receipt_attest = $1 WHERE id = $2")

        .bind("ab".repeat(32))

        .bind(lint.id)

        .execute(&h.app.db)

        .await

        .unwrap();

    create(&h.app, "bob", "game2").await;

    let fork2 = db::repos::get(&h.app.db, "bob", "game2").await.unwrap().unwrap();

    db::repos::set_forked_from(&h.app.db, fork2.id, up.id).await.unwrap();

    sqlx::query("UPDATE repositories SET flow = 'on' WHERE id = $1").bind(fork2.id).execute(&h.app.db).await.unwrap();

    c.config.set("remote.origin.url", &format!("{}/bob/game2", h.base)).unwrap();

    c.save_config().unwrap();

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

    let jobs2 = db::flow::jobs(&h.app.db, runs_of(&h.app, "bob", "game2").await.remove(0).id).await.unwrap();

    let status = |name: &str| jobs2.iter().find(|j| j.name == name).map(|j| (j.status.clone(), j.reused_repo().map(String::from))).unwrap();

    assert_eq!(status("lint"), ("queued".to_string(), None), "{jobs2:?}");

    assert_eq!(status("build"), ("passed".to_string(), Some("andy/game".to_string())), "{jobs2:?}");

    sqlx::query("UPDATE flow_jobs SET receipt_attest = $1 WHERE id = $2")

        .bind(&lint.receipt_attest)

        .bind(lint.id)

        .execute(&h.app.db)

        .await

        .unwrap();