Verify and keep signed Flow receipts cc19697f8871

-- Signed receipts (Oxygen spec §14.6): the receipt attestation a job's

-- receipt is kept in the repository as, the runner that signed it, and why a

-- signed receipt was not kept.

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

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

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

-- The key each runner token signs with, learned from its first signed

-- receipt; later receipts under that token must carry the same key.

CREATE TABLE flow_runner_keys (

    token_hash TEXT PRIMARY KEY,

    key TEXT NOT NULL,

    name TEXT NOT NULL,

    first_ms INTEGER NOT NULL

);

    /// The receipt attestation kept in the repository (Oxygen spec §14.6).

    pub receipt_attest: String,

    /// Who signed it: the runner's identity name.

    pub receipt_signer: String,

    /// Why a signed receipt was not kept.

    pub receipt_note: String,

}

/// The key runner token `token_hash` signs with: the one already bound, or

/// `key` (named `name`) when this is the first.

pub async fn bind_runner_key(pool: &Pool, token_hash: &str, key: &str, name: &str) -> sqlx::Result<String> {

    sqlx::query("INSERT OR IGNORE INTO flow_runner_keys (token_hash, key, name, first_ms) VALUES (?, ?, ?, ?)")

        .bind(token_hash)

        .bind(key)

        .bind(name)

        .bind(crate::db::now_ms())

        .execute(pool)

        .await?;

    sqlx::query_scalar("SELECT key FROM flow_runner_keys WHERE token_hash = ?").bind(token_hash).fetch_one(pool).await

}

/// Record what became of job `job_id`'s signed receipt.

pub async fn set_receipt(pool: &Pool, job_id: i64, attest: &str, signer: &str, note: &str) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_jobs SET receipt_attest = ?, receipt_signer = ?, receipt_note = ? WHERE id = ?")

        .bind(attest)

        .bind(signer)

        .bind(note)

        .bind(job_id)

        .execute(pool)

        .await?;

    Ok(())

/// The runner's token, checked, as the BLAKE3 hex that names it.

fn runner_auth(app: &Shared, headers: &HeaderMap) -> Result<String, ApiError> {

        Ok(blake3::hash(t.as_bytes()).to_hex().to_string())

    let token = runner_auth(&app, &headers)?;

            if let (Some(r), Some(signed)) = (&f.receipt, &f.signed_receipt) {

                let (attest, signer, note) = match keep_receipt(&app, &token, &row, &run, r, signed).await {

                    Ok((attest, signer)) => (attest, signer, String::new()),

                    Err(e) => {

                        tracing::warn!("flow: job {} receipt not kept: {e}", job.id);

                        (String::new(), String::new(), e)

                    }

                };

                fdb::set_receipt(&app.db, job.id, &attest, &signer, &note).await?;

            }

}

/// Check a runner's signed receipt for a job of `run` and keep it in the

/// repository (Oxygen spec §14.6): the attestation's hex and the runner's

/// name, or why it was not kept. It must be the job's receipt, for the

/// run's state and repository, signed by the key the runner's token first

/// signed with.

pub async fn keep_receipt(

    app: &Shared,

    token_hash: &str,

    row: &db::repos::RepoRow,

    run: &fdb::Run,

    receipt: &flow_core::Receipt,

    signed_hex: &str,

) -> Result<(String, String), String> {

    use oxy_core::receipt::SignedReceipt;

    let bytes = hex::decode(signed_hex).map_err(|_| "the signed receipt is not hex".to_string())?;

    let sr = SignedReceipt::decode(&bytes).map_err(|e| format!("not a signed receipt: {e}"))?;

    sr.verify().map_err(|e| format!("it does not verify: {e}"))?;

    if sr.receipt.hash().0 != receipt.hash() {

        return Err("it signs another receipt than the job's".into());

    }

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

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

    if sr.state() != state {

        return Err("it is for another state than the run's".into());

    }

    let key = sr.runner.hash().to_hex();

    let bound = fdb::bind_runner_key(&app.db, token_hash, &key, &sr.runner.name).await.map_err(|e| e.to_string())?;

    if bound != key {

        return Err(format!("signed by {}, not the key this runner signs with ({})", &key[..12], &bound[..12.min(bound.len())]));

    }

    let name = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;

    let signer = sr.runner.name.clone();

    let attest = app.repos.write(&name, move |repo, _| Ok(repo.add_receipt(&sr)?)).await.map_err(|e| e.to_string())?;

    Ok((attest.to_hex(), signer))

/// A receipt kept in a repository (Oxygen spec §14.6), as pages show it.

#[derive(Clone, Debug)]

pub struct ReceiptRow {

    pub attest: String,

    pub job: String,

    pub exit_code: u32,

    pub duration: String,

    pub runner: String,

    pub run: String,

    /// Why it does not verify, or that its runner's key was revoked.

    pub problem: String,

    pub made_ms: i64,

}

impl ReceiptRow {

    pub fn passed(&self) -> bool {

        self.exit_code == 0

    }

}

/// The receipts kept for the state of `heads`.

pub async fn receipts_of(app: &Shared, name: &crate::repos::RepoName, heads: Vec<oxy_core::hash::ChangeHash>) -> Vec<ReceiptRow> {

    let base = app.config.base_url();

    let found = app.repos.read(name, move |repo, _| Ok(repo.receipts(&heads)?)).await.unwrap_or_default();

    found

        .into_iter()

        .map(|i| {

            let r = &i.signed.receipt;

            let run = i.signed.run().to_string();

            ReceiptRow {

                attest: i.attest.to_hex(),

                job: r.job_name.clone(),

                exit_code: r.exit_code,

                duration: crate::flow::human_ms(r.duration_ms as i64),

                runner: i.signed.runner.name.clone(),

                // Only links to this hub are followed from the page.

                run: if run.starts_with(base) { run } else { String::new() },

                problem: match (i.problem, i.revoked) {

                    (Some(p), _) => p,

                    (None, true) => "the runner's key was revoked".into(),

                    (None, false) => String::new(),

                },

                made_ms: i.signed.attest.hlc.wall_ms as i64,

            }

        })

        .collect()

}

/// `GET /{owner}/{name}/receipts/{attest}`: a signed receipt, for

/// `oxy receipt verify` or `oxy receipt import`.

pub async fn receipt_file(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name, attest)): Path<(String, String, String)>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/receipts/{attest}"), &viewer);

    let (repo, _row, _role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let Some(r) = oxy_core::hash::ObjectRef::from_hex(attest.trim_end_matches(".receipt")) else {

        return Err(Error::not_found(shell));

    };

    let (job, bytes) = app

        .repos

        .read(&repo, move |repo, _| {

            let i = repo.receipt(&r)?;

            Ok((i.signed.receipt.job_name.clone(), i.signed.encode()))

        })

        .await

        .map_err(|_| Error::not_found(shell.clone()))?;

    let file: String = job.chars().filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')).collect();

    Ok((

        [

            (axum::http::header::CONTENT_TYPE, "application/oxy-receipt".to_string()),

            (axum::http::header::CONTENT_DISPOSITION, format!("attachment; filename=\"{file}-{}.receipt\"", r.short(12))),

        ],

        bytes,

    )

        .into_response())

}

    /// Receipts runners signed for the state whose only head is this change.

    receipts: Vec<super::flow::ReceiptRow>,

    let receipts = match oxy_core::hash::ChangeHash::from_hex(&c.view.hash) {

        Some(h) => super::flow::receipts_of(&app, &repo, vec![h]).await,

        None => vec![],

    };

    Ok(Page(ChangePage { shell, repo: RepoShell::new(&app, &row, "changes").with_role(role), c, files, body_html, ignore_ws, receipts }))

        .route("/{owner}/{name}/receipts/{attest}", get(flow::receipt_file))

async fn get_bytes(app: &Shared, path: &str) -> (StatusCode, Vec<u8>) {

    let res = oxygenhub::router(app.clone()).oneshot(Request::get(path).body(Body::empty()).unwrap()).await.unwrap();

    let status = res.status();

    (status, res.into_body().collect().await.unwrap().to_bytes().to_vec())

}

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

async fn signed_receipts_are_checked_kept_and_shown() {

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

    use oxy_core::identity::SecretKey;

    use oxy_core::objects::StateObject;

    use oxy_core::receipt::SignedReceipt;

    let h = hub().await;

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

    let url = format!("{}/andy/demo", h.base);

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

    let c = client(&url, work.path());

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

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

    c.commit("Add a pipeline", &[]).unwrap();

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

    let run = runs_of(&h.app, "andy", "demo").await.remove(0);

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

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

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

    let sk = SecretKey::generate();

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

    let sign = |r: &serde_json::Value, state: ObjectRef, sk: &SecretKey, id: &oxy_core::identity::Identity| {

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

        let run_url = format!("{url}/flow/{}", run.number);

        hex::encode(SignedReceipt::sign(r.to_oxy(), repo_id, state, &run_url, id, sk).encode())

    };

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

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

    let ph = a["pipeline_hash"].as_str().unwrap().to_string();

    let (a_id, b_id) = (a["job_id"].as_i64().unwrap(), b["job_id"].as_i64().unwrap());

    // The first signed receipt binds the runner token to its key, and is kept.

    let rb = receipt("build", &ph, 0);

    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": rb, "signed_receipt": sign(&rb, state, &sk, &runner) });

    assert_eq!(finish(&h.app, a_id, body).await, StatusCode::NO_CONTENT);

    // Another key under the same token: the result counts, the receipt is not kept.

    let other = SecretKey::generate();

    let rl = receipt("lint", &ph, 0);

    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 300, "receipt": rl,

        "signed_receipt": sign(&rl, state, &other, &other.identity("mallory", "", 1)) });

    assert_eq!(finish(&h.app, b_id, body).await, StatusCode::NO_CONTENT);

    // The right key, but for another state.

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

    let rt = receipt("test", &ph, 0);

    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 500, "receipt": rt,

        "signed_receipt": sign(&rt, ObjectRef([9; 32]), &sk, &runner) });

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

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

    let job = |n: &str| jobs.iter().find(|j| j.name == n).unwrap().clone();

    assert_eq!(job("build").status, "passed");

    assert_eq!((job("build").receipt_attest.len(), job("build").receipt_signer.as_str()), (64, "test-1 (OxygenFlow runner)"));

    assert_eq!(job("lint").status, "passed", "the job's result stands");

    assert!(job("lint").receipt_attest.is_empty());

    assert!(job("lint").receipt_note.contains("not the key this runner signs with"), "{}", job("lint").receipt_note);

    assert!(job("test").receipt_note.contains("another state"), "{}", job("test").receipt_note);

    // The repository keeps it, and anyone who can read the repository can

    // download it and check it on their own.

    let kept = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap().receipts(&heads).unwrap();

    assert_eq!(kept.len(), 1);

    let attest = job("build").receipt_attest;

    let (st, bytes) = get_bytes(&h.app, &format!("/andy/demo/receipts/{attest}")).await;

    assert_eq!(st, StatusCode::OK);

    let sr = SignedReceipt::decode(&bytes).unwrap();

    sr.verify().unwrap();

    assert_eq!((sr.state(), sr.receipt.job_name.as_str()), (state, "build"));

    assert_eq!(get_bytes(&h.app, &format!("/andy/demo/receipts/{}", "0".repeat(64))).await.0, StatusCode::NOT_FOUND);

    // Pages: the run names the signer or why a receipt was not kept; the

    // change whose state it is lists it.

    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;

    assert!(page.contains("signed receipt</a> by test-1"), "run page");

    assert!(page.contains("receipt not kept"), "run page");

    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/change/{}", heads[0].to_hex()), None, None).await;

    assert!(page.contains("<dt>Receipts</dt>") && page.contains(&format!("/andy/demo/receipts/{attest}")), "change page");

}

- **Signed receipts** (Oxygen spec §14.6): `finish` may carry `signed_receipt`, the runner's signed receipt as hex. The hub keeps it in the repository when four things hold:

  - it verifies;

  - it signs the job's receipt;

  - it is for the run's state (computed from the run's heads) and this repository;

  - its key is the one the runner's token first signed with. The first signed receipt under a token binds that token to its key in `flow_runner_keys`.

  Otherwise the job's result still stands, and the run page says why the receipt was not kept. The run page links each kept receipt, with its signer. The change page lists the receipts for the state whose only head is that change. `/{owner}/{name}/receipts/{attest}` downloads a signed receipt for `oxy receipt verify` or `import`.

- Later, stamps also say which runner key signed each receipt, and which jobs reused an earlier receipt.

    {% if !receipts.is_empty() %}<dt>Receipts</dt><dd class="receipts">{% for r in receipts %}<div><span class="flow-glyph st-{% if r.passed() %}passed{% else %}failed{% endif %}">{% if r.passed() %}✔{% else %}✘{% endif %}</span> {% if r.run.is_empty() %}<strong>{{ r.job }}</strong>{% else %}<a href="{{ r.run }}"><strong>{{ r.job }}</strong></a>{% endif %} <span class="small mute">{% if r.passed() %}passed{% else %}exit {{ r.exit_code }}{% endif %} in {{ r.duration }} · signed by {{ r.runner }} · <time title="{{ r.made_ms|when }}">{{ r.made_ms|ago }}</time></span>{% if !r.problem.is_empty() %} <span class="small warn">{{ r.problem }}</span>{% endif %} <a class="small" href="{{ repo.url }}/receipts/{{ r.attest }}" title="The signed receipt, for oxy receipt verify">download</a></div>{% endfor %}</dd>{% endif %}

  <summary><span class="flow-glyph st-{{ l.j.status }}" data-glyph>{{ crate::web::flow::glyph(l.j.status) }}</span> <strong>{{ l.j.name }}</strong> <span class="small mute" data-meta>{{ l.j.status }}{% if let Some(c) = l.j.exit_code %} · exit {{ c }}{% endif %} · {{ l.duration }}{% if !l.j.runner.is_empty() %} · on {{ l.j.runner }}{% endif %}</span>{% if !l.j.receipt_attest.is_empty() %} <span class="small mute">· <a href="{{ repo.url }}/receipts/{{ l.j.receipt_attest }}" title="Receipt {{ l.j.receipt_hash }}, kept in the repository; check it with oxy receipt verify">signed receipt</a> by {{ l.j.receipt_signer }}</span>{% else if !l.j.receipt_note.is_empty() %} <span class="small warn" title="{{ l.j.receipt_note }}">· receipt not kept: {{ l.j.receipt_note }}</span>{% else if !l.j.receipt_hash.is_empty() %} <span class="small mute">· unsigned receipt <code title="{{ l.j.receipt_hash }}">{{ l.j.receipt_hash|short }}</code></span>{% endif %}</summary>