Keyless step 2: the hub certifies fresh keys for ten minutes 026a76d251ce

//! Certificates for short-lived keys (oxygenhub#241, docs/keyless.md,

//! Oxygen spec §14.7): the hub vouches, for ten minutes, for a fresh key

//! whose holder proves who it is with a credential the hub issued. A Flow

//! job asks with its job token; anything else with one of the hub's OIDC

//! tokens for the keyless audience. The key signs, then is thrown away:

//! nothing long-lived sits on a runner.

use crate::api::ApiError;

use crate::{db, Shared};

use axum::extract::State;

use axum::http::{HeaderMap, StatusCode};

use axum::Json;

use oxy_core::cert::{certify_challenge, Certificate, SignedCertificate};

use oxy_core::identity::{verify_challenge, Identity, Scope, Signature};

use serde::{Deserialize, Serialize};

/// How long a certificate lasts.

pub const LIFETIME_MS: i64 = 10 * 60 * 1000;

/// How far back a certificate starts, for clocks a little behind.

const SKEW_MS: i64 = 30 * 1000;

/// The OIDC audience a token must have to be exchanged here.

pub fn audience(base_url: &str) -> String {

    format!("{}/keyless", base_url.trim_end_matches('/'))

}

#[derive(Deserialize)]

pub struct CertifyRequest {

    /// The fresh key's identity object, base64.

    pub identity: String,

    /// The key's signature over `certify_challenge(credential)`, base64 CBOR.

    pub proof: String,

    /// `changes,refs` and so on.

    pub scope: String,

}

#[derive(Serialize)]

pub struct CertifyResponse {

    /// The signed certificate, base64 (`[certificate, signature]`).

    pub certificate: String,

    pub subject: String,

    pub expires_ms: i64,

}

/// Who a credential says its holder is.

struct Proven {

    subject: String,

    claims: Vec<(String, String)>,

    /// The credential's own end, which a certificate never outlives.

    until_ms: i64,

    target: String,

}

fn bad(status: StatusCode, why: impl Into<String>) -> ApiError {

    ApiError(status, why.into())

}

/// A running job's token: its repository, bookmark, run and job.

async fn job_proof(app: &Shared, token: &str) -> Result<Proven, ApiError> {

    let invalid = || bad(StatusCode::UNAUTHORIZED, "this job token is not valid (the job must be running)");

    let hash = crate::auth::token_hash(token);

    let job = db::flow::job_by_token(&app.db, &hash, crate::flow::JOB_TOKEN_MAX_AGE_MS)

        .await

        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .ok_or_else(invalid)?;

    let run = db::flow::run_by_id(&app.db, job.run_id)

        .await

        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .ok_or_else(invalid)?;

    let repo = db::repos::get_by_id(&app.db, run.repo_id)

        .await

        .map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?

        .ok_or_else(invalid)?;

    let mut claims = vec![

        ("repository".to_string(), format!("{}/{}", repo.owner, repo.name)),

        ("bookmark".to_string(), run.bookmark.clone()),

        ("run".to_string(), run.number.to_string()),

        ("job".to_string(), job.name.clone()),

        ("runner".to_string(), job.runner.clone()),

    ];

    if !job.environment.is_empty() && !job.approved_by.is_empty() {

        claims.push(("environment".to_string(), job.environment.clone()));

        claims.push(("approved_by".to_string(), job.approved_by.clone()));

    }

    Ok(Proven {

        subject: format!("repo:{}/{}:bookmark:{}", repo.owner, repo.name, run.bookmark),

        claims,

        until_ms: job.started_ms.unwrap_or(0) + crate::flow::JOB_TOKEN_MAX_AGE_MS,

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

    })

}

/// One of this hub's OIDC tokens for the keyless audience.

async fn oidc_proof(app: &Shared, token: &str, now: i64) -> Result<Proven, ApiError> {

    let issuer = app.oidc().await.map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    let aud = audience(app.config.base_url());

    let claims = issuer

        .verify(token, &aud, now)

        .ok_or_else(|| bad(StatusCode::UNAUTHORIZED, format!("the token is not one of this hub's, for audience {aud}, still valid")))?;

    let subject = claims["sub"].as_str().unwrap_or("").to_string();

    if subject.is_empty() {

        return Err(bad(StatusCode::UNAUTHORIZED, "the token names no subject"));

    }

    let keep = ["repository", "bookmark", "run_number", "job", "runner", "environment", "approved_by"];

    let mut out = Vec::new();

    for k in keep {

        match &claims[k] {

            serde_json::Value::String(s) => out.push((k.to_string(), s.clone())),

            serde_json::Value::Number(n) => out.push((k.to_string(), n.to_string())),

            _ => {}

        }

    }

    Ok(Proven {

        target: claims["repository"].as_str().unwrap_or(&subject).to_string(),

        subject,

        claims: out,

        until_ms: claims["exp"].as_i64().unwrap_or(0) * 1000,

    })

}

/// `POST /api/v1/keyless/certify`: a ten-minute certificate for a fresh

/// key, for whoever the bearer credential proves.

pub async fn certify(

    State(app): State<Shared>,

    headers: HeaderMap,

    Json(req): Json<CertifyRequest>,

) -> Result<Json<CertifyResponse>, ApiError> {

    let token = oxy_server::bearer_token(&headers)

        .ok_or_else(|| bad(StatusCode::UNAUTHORIZED, "ask with a job token or one of this hub's OIDC tokens as the bearer"))?

        .to_string();

    let now = db::now_ms();

    let proven = if token.starts_with(crate::flow::JOB_TOKEN_PREFIX) {

        job_proof(&app, &token).await?

    } else {

        oidc_proof(&app, &token, now).await?

    };

    let decode = |what: &str, b64: &str| {

        oxy_core::dsse::base64_decode(b64.trim()).ok_or_else(|| bad(StatusCode::BAD_REQUEST, format!("{what} is not base64")))

    };

    let identity =

        Identity::decode(&decode("identity", &req.identity)?).map_err(|e| bad(StatusCode::BAD_REQUEST, format!("identity: {e}")))?;

    let proof = oxy_core::cbor::decode(&decode("proof", &req.proof)?)

        .map_err(|e| e.to_string())

        .and_then(|v| Signature::from_value(&v).map_err(|e| e.to_string()))

        .map_err(|e| bad(StatusCode::BAD_REQUEST, format!("proof: {e}")))?;

    verify_challenge(&identity, &proof, &certify_challenge(&token))

        .map_err(|e| bad(StatusCode::UNAUTHORIZED, format!("the proof does not show this key's holder asked: {e}")))?;

    let scope = Scope::parse(&req.scope).map_err(|e| bad(StatusCode::BAD_REQUEST, e))?;

    let expires = (now + LIFETIME_MS).min(proven.until_ms);

    if expires <= now {

        return Err(bad(StatusCode::UNAUTHORIZED, "the credential is about to expire: ask with a fresh one"));

    }

    let (key, secret) =

        crate::keys::load_secret(&app.config.keys_dir()).map_err(|e| bad(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;

    let cert = Certificate {

        key: identity,

        issuer: key,

        subject: proven.subject.clone(),

        claims: proven.claims,

        scope,

        not_before: (now - SKEW_MS) as u64,

        expires: expires as u64,

    };

    let signed = SignedCertificate::sign(cert, &app.hub_key.identity, &secret);

    let detail = format!("{} for {} until {}", signed.cert.key_id().short(12), proven.subject, expires);

    db::accounts::audit(&app.db, None, "keyless.certified", &proven.target, &detail).await.ok();

    Ok(Json(CertifyResponse { certificate: oxy_core::dsse::base64(&signed.encode()), subject: proven.subject, expires_ms: expires }))

}

        .route("/keyless/certify", post(crate::keyless::certify))

pub mod keyless;

/// Keyless (oxygenhub#241): a running job's token gets a ten-minute

/// certificate for a fresh key from the hub, signed with the hub's key and

/// naming the job's repository and bookmark; a proof made for another

/// credential, an OIDC token for another audience, and a finished job's

/// token get nothing.

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

async fn a_running_job_gets_a_short_certificate_for_a_fresh_key() {

    use oxy_core::cert::{certify_challenge, SignedCertificate};

    use oxy_core::identity::{Scope, SecretKey};

    let h = hub().await;

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

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

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

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

    write(work.path(), "test.sh", "echo ok\n");

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

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

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

    assert_eq!(runs_of(&h.app, "andy", "demo").await.len(), 1);

    let job = claim(&h.app).await.expect("a job");

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

    let job_id = job["job_id"].as_i64().unwrap();

    let sk = SecretKey::generate();

    let id = sk.identity("job", "[email protected]", 1);

    let ask = |credential: &str, signed_for: &str| {

        let proof = sk.sign_challenge(&id.hash(), &certify_challenge(signed_for)).unwrap();

        let body = serde_json::json!({

            "identity": oxy_core::dsse::base64(&id.encode()),

            "proof": oxy_core::dsse::base64(&oxy_core::cbor::encode(&proof.to_value())),

            "scope": "changes,refs",

        });

        let (credential, app) = (credential.to_string(), h.app.clone());

        async move { call(&app, "POST", "/api/v1/keyless/certify", Some(&credential), Some(body.to_string())).await }

    };

    let (st, body) = ask(&token, &token).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    let v: serde_json::Value = serde_json::from_str(&body).unwrap();

    let sc = SignedCertificate::decode(&oxy_core::dsse::base64_decode(v["certificate"].as_str().unwrap()).unwrap()).unwrap();

    assert!(sc.verify(&h.app.hub_key.identity).is_ok(), "signed with the hub's key");

    assert_eq!(sc.cert.key_id(), id.hash());

    assert_eq!(sc.cert.subject, "repo:andy/demo:bookmark:main");

    assert_eq!(sc.cert.claim("repository"), Some("andy/demo"));

    assert!(sc.cert.claim("job").is_some() && sc.cert.claim("run") == Some("1"));

    assert_eq!(sc.cert.scope, Scope::CHANGES | Scope::REFS);

    assert!(sc.cert.expires - sc.cert.not_before <= 10 * 60_000 + 30_000, "ten minutes");

    // A proof made for another credential is not this one's.

    let (st, body) = ask(&token, "flowjob_someone_else").await;

    assert_eq!(st, StatusCode::UNAUTHORIZED, "{body}");

    assert!(body.contains("the proof does not show"), "{body}");

    // The hub's own OIDC token, for the keyless audience, works too; for

    // another audience it does not.

    let issuer = h.app.oidc().await.unwrap();

    let now = oxygenhub::db::now_ms();

    let aud = oxygenhub::keyless::audience(h.app.config.base_url());

    let good = issuer.mint_for(h.app.config.base_url(), "agent:claude@andy", &aud, now, 600);

    let (st, body) = ask(&good, &good).await;

    assert_eq!(st, StatusCode::OK, "{body}");

    assert!(body.contains("\"subject\":\"agent:claude@andy\""), "{body}");

    let other = issuer.mint_for(h.app.config.base_url(), "agent:claude@andy", "sts.amazonaws.com", now, 600);

    let (st, _) = ask(&other, &other).await;

    assert_eq!(st, StatusCode::UNAUTHORIZED);

    // A job that has finished asks with a token that no longer counts.

    sqlx::query("UPDATE flow_jobs SET status = 'passed' WHERE id = $1").bind(job_id).execute(&h.app.db).await.unwrap();

    let (st, body) = ask(&token, &token).await;

    assert_eq!(st, StatusCode::UNAUTHORIZED, "{body}");

}

### 13.5.2 Certificates for short-lived keys (oxygenhub#241, docs/keyless.md, step 2 built 2026-10-17)

`POST /api/v1/keyless/certify` gives a fresh key a certificate (Oxygen spec §14.7) for ten minutes, signed with the hub's key. The bearer is the credential: a running job's token (subject `repo:OWNER/NAME:bookmark:B`, claims `repository`, `bookmark`, `run`, `job`, `runner`, and `environment` with `approved_by` once an approved job may deploy), or one of this hub's OIDC tokens for the audience `{base_url}/keyless` (its `sub` the subject, and the same claims it carries). The body is `{"identity": base64 identity object, "proof": base64 CBOR Signature, "scope": "changes,refs"}`, the proof the fresh key's challenge signature (§14.3) over `certify_challenge(credential)` (`"oxy1 certify "` and the BLAKE3 of the credential), so it cannot be replayed with another credential. The answer is `{"certificate": base64 [certificate, signature], "subject", "expires_ms"}`; a certificate never outlives its credential, and it starts 30 s back for clocks a little behind. Each one is audited (`keyless.certified`). Which issuers and subjects a repository trusts is its policy's, and signing with these keys is steps 3 to 5 of docs/keyless.md.