A job's identity: the hub is an OIDC issuer d1727c573f30

//! The hub as an OpenID Connect issuer for jobs (hub spec §13.5,

//! oxygenhub#103): a job that declares `identity "AUDIENCE" { mount-path … }`

//! gets a signed JWT naming the repository, bookmark, run and job, so a

//! cloud or a vault that trusts this hub can let the job in without a

//! long-lived secret. The key is RSA 2048 in `keys/oidc.pem`, made at

//! first use; `/.well-known/openid-configuration` and

//! `/.well-known/jwks.json` publish what verifiers need.

use rsa::pkcs1v15::{SigningKey, VerifyingKey};

use rsa::pkcs8::{DecodePrivateKey, EncodePrivateKey};

use rsa::signature::{SignatureEncoding, Signer, Verifier};

use rsa::traits::PublicKeyParts;

use rsa::RsaPrivateKey;

use sha2::Sha256;

use std::path::Path;

use crate::api::ApiError;

use crate::Shared;

use axum::extract::State;

use axum::http::StatusCode;

use axum::response::{IntoResponse, Response};

use axum::Json;

/// The longest a job token lives, whatever the job's limit.

pub const MAX_TTL_S: u64 = 6 * 3600;

pub struct Issuer {

    signing: SigningKey<Sha256>,

    verifying: VerifyingKey<Sha256>,

    /// The key id: the first 16 hex of BLAKE3 over the public key's

    /// modulus, as the JWKS names it.

    pub kid: String,

    n: String,

    e: String,

}

/// What a token says about its job.

pub struct JobClaims<'a> {

    pub issuer: &'a str,

    pub audience: &'a str,

    pub owner: &'a str,

    pub name: &'a str,

    pub visibility: &'a str,

    pub bookmark: &'a str,

    pub run_number: i64,

    pub job: &'a str,

    pub state: &'a str,

    pub runner: &'a str,

    pub actor: &'a str,

    pub now_ms: i64,

    pub ttl_s: u64,

}

impl Issuer {

    /// The key from `<keys>/oidc.pem`, or a new 2048-bit one written

    /// there (mode 0600).

    pub fn load_or_create(keys: &Path) -> anyhow::Result<Issuer> {

        std::fs::create_dir_all(keys)?;

        let path = keys.join("oidc.pem");

        let key = match std::fs::read_to_string(&path) {

            Ok(pem) => RsaPrivateKey::from_pkcs8_pem(&pem).map_err(|e| anyhow::anyhow!("{}: {e}", path.display()))?,

            Err(_) => {

                let key = RsaPrivateKey::new(&mut rand::rngs::OsRng, 2048)?;

                let pem = key.to_pkcs8_pem(rsa::pkcs8::LineEnding::LF)?;

                std::fs::write(&path, pem.as_bytes())?;

                #[cfg(unix)]

                {

                    use std::os::unix::fs::PermissionsExt;

                    std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600))?;

                }

                key

            }

        };

        Ok(Issuer::from_key(key))

    }

    fn from_key(key: RsaPrivateKey) -> Issuer {

        let public = key.to_public_key();

        let n_bytes = public.n().to_bytes_be();

        let kid = blake3::hash(&n_bytes).to_hex()[..16].to_string();

        Issuer {

            n: b64url(&n_bytes),

            e: b64url(&public.e().to_bytes_be()),

            verifying: VerifyingKey::new(public),

            signing: SigningKey::new(key),

            kid,

        }

    }

    /// `/.well-known/openid-configuration`.

    pub fn discovery(&self, issuer: &str) -> serde_json::Value {

        serde_json::json!({

            "issuer": issuer,

            "jwks_uri": format!("{issuer}/.well-known/jwks.json"),

            "response_types_supported": ["id_token"],

            "subject_types_supported": ["public"],

            "id_token_signing_alg_values_supported": ["RS256"],

            "claims_supported": ["iss", "sub", "aud", "exp", "iat", "nbf", "jti", "repository", "repository_owner",

                "repository_visibility", "bookmark", "run_number", "job", "state", "runner", "actor"],

            "scopes_supported": ["openid"],

        })

    }

    /// `/.well-known/jwks.json`.

    pub fn jwks(&self) -> serde_json::Value {

        serde_json::json!({ "keys": [{ "kty": "RSA", "use": "sig", "alg": "RS256", "kid": self.kid, "n": self.n, "e": self.e }] })

    }

    /// A signed token for a job: `sub` is `repo:OWNER/NAME:bookmark:B`,

    /// so a trust policy can match a repository and a bookmark.

    pub fn mint(&self, c: &JobClaims<'_>) -> String {

        let iat = c.now_ms / 1000;

        let exp = iat + c.ttl_s.clamp(60, MAX_TTL_S) as i64;

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

        let payload = serde_json::json!({

            "iss": c.issuer,

            "sub": format!("repo:{}/{}:bookmark:{}", c.owner, c.name, c.bookmark),

            "aud": c.audience,

            "iat": iat,

            "nbf": iat - 30,

            "exp": exp,

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

            "repository": format!("{}/{}", c.owner, c.name),

            "repository_owner": c.owner,

            "repository_visibility": c.visibility,

            "bookmark": c.bookmark,

            "run_number": c.run_number,

            "job": c.job,

            "state": c.state,

            "runner": c.runner,

            "actor": c.actor,

        });

        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()))

    }

    /// The claims of a token this hub signed, when its signature holds,

    /// it is for `audience` and it has not expired at `now_ms`.

    pub fn verify(&self, token: &str, audience: &str, now_ms: i64) -> Option<serde_json::Value> {

        let mut parts = token.split('.');

        let (h, p, s) = (parts.next()?, parts.next()?, parts.next()?);

        if parts.next().is_some() {

            return None;

        }

        let header: serde_json::Value = serde_json::from_slice(&b64url_decode(h)?).ok()?;

        if header["alg"] != "RS256" || header["kid"] != self.kid.as_str() {

            return None;

        }

        let sig = rsa::pkcs1v15::Signature::try_from(b64url_decode(s)?.as_slice()).ok()?;

        self.verifying.verify(format!("{h}.{p}").as_bytes(), &sig).ok()?;

        let claims: serde_json::Value = serde_json::from_slice(&b64url_decode(p)?).ok()?;

        let now = now_ms / 1000;

        (claims["aud"] == audience && claims["exp"].as_i64()? > now && claims["nbf"].as_i64()? <= now).then_some(claims)

    }

}

/// `text` with every token this hub signed replaced by `***`: a job

/// that prints its identity does not leave it in the log.

pub fn mask(text: &str, kid: &str) -> String {

    if !text.contains("eyJ") {

        return text.to_string();

    }

    let is_tok = |c: char| c.is_ascii_alphanumeric() || c == '-' || c == '_' || c == '.';

    let mut out = String::with_capacity(text.len());

    let mut rest = text;

    while let Some(i) = rest.find("eyJ") {

        out.push_str(&rest[..i]);

        let end = rest[i..].find(|c: char| !is_tok(c)).map(|e| i + e).unwrap_or(rest.len());

        let cand = &rest[i..end];

        let ours = cand

            .split('.')

            .next()

            .and_then(b64url_decode)

            .and_then(|h| serde_json::from_slice::<serde_json::Value>(&h).ok())

            .is_some_and(|h| h["kid"] == kid);

        if ours && cand.matches('.').count() == 2 {

            out.push_str("***");

        } else {

            out.push_str(cand);

        }

        rest = &rest[end..];

    }

    out.push_str(rest);

    out

}

fn b64url(bytes: &[u8]) -> String {

    const T: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";

    let mut out = String::with_capacity(bytes.len().div_ceil(3) * 4);

    for chunk in bytes.chunks(3) {

        let n = chunk.iter().fold(0u32, |acc, b| (acc << 8) | *b as u32) << (8 * (3 - chunk.len()));

        for i in 0..=chunk.len() {

            out.push(T[((n >> (18 - 6 * i)) & 63) as usize] as char);

        }

    }

    out

}

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

    let val = |c: u8| -> Option<u32> {

        Some(match c {

            b'A'..=b'Z' => (c - b'A') as u32,

            b'a'..=b'z' => (c - b'a') as u32 + 26,

            b'0'..=b'9' => (c - b'0') as u32 + 52,

            b'-' => 62,

            b'_' => 63,

            _ => return None,

        })

    };

    let bytes = s.trim_end_matches('=').as_bytes();

    let mut out = Vec::with_capacity(bytes.len() * 3 / 4);

    for chunk in bytes.chunks(4) {

        if chunk.len() == 1 {

            return None;

        }

        let mut n = 0u32;

        for (i, c) in chunk.iter().enumerate() {

            n |= val(*c)? << (18 - 6 * i);

        }

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

        if chunk.len() > 2 {

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

        }

        if chunk.len() > 3 {

            out.push(n as u8);

        }

    }

    Some(out)

}

/// `GET /.well-known/openid-configuration`.

pub async fn discovery(State(app): State<Shared>) -> Result<Response, ApiError> {

    let issuer = app.oidc().await.map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("oidc: {e}")))?;

    Ok(([(axum::http::header::CACHE_CONTROL, "public, max-age=3600")], Json(issuer.discovery(app.config.base_url()))).into_response())

}

/// `GET /.well-known/jwks.json`.

pub async fn jwks(State(app): State<Shared>) -> Result<Response, ApiError> {

    let issuer = app.oidc().await.map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("oidc: {e}")))?;

    Ok(([(axum::http::header::CACHE_CONTROL, "public, max-age=3600")], Json(issuer.jwks())).into_response())

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn tokens_verify_against_the_published_key_and_are_masked() {

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

        let issuer = Issuer::load_or_create(dir.path()).unwrap();

        let again = Issuer::load_or_create(dir.path()).unwrap();

        assert_eq!(issuer.kid, again.kid, "the key is kept");

        let claims = JobClaims {

            issuer: "https://hub.example",

            audience: "sts.amazonaws.com",

            owner: "andy",

            name: "demo",

            visibility: "public",

            bookmark: "main",

            run_number: 7,

            job: "deploy",

            state: "ab".repeat(32).leak(),

            runner: "earth-1",

            actor: "andy",

            now_ms: 1_800_000_000_000,

            ttl_s: 900,

        };

        let token = issuer.mint(&claims);

        assert_eq!(token.matches('.').count(), 2);

        let got = again.verify(&token, "sts.amazonaws.com", 1_800_000_500_000).expect("verifies");

        assert_eq!(got["sub"], "repo:andy/demo:bookmark:main");

        assert_eq!(got["run_number"], 7);

        assert_eq!(got["exp"].as_i64().unwrap() - got["iat"].as_i64().unwrap(), 900);

        assert!(again.verify(&token, "other", 1_800_000_500_000).is_none(), "another audience");

        assert!(again.verify(&token, "sts.amazonaws.com", 1_800_001_000_000).is_none(), "expired");

        let other = Issuer::load_or_create(tempfile::tempdir().unwrap().path()).unwrap();

        assert!(other.verify(&token, "sts.amazonaws.com", 1_800_000_500_000).is_none(), "another key");

        let tampered = format!("{}x", token);

        assert!(again.verify(&tampered, "sts.amazonaws.com", 1_800_000_500_000).is_none());

        let jwks = issuer.jwks();

        assert_eq!(jwks["keys"][0]["kid"], issuer.kid.as_str());

        assert_eq!(b64url_decode(jwks["keys"][0]["e"].as_str().unwrap()).unwrap(), vec![1, 0, 1]);

        assert_eq!(mask(&format!("token={token} done eyJnot.a.token"), &issuer.kid), "token=*** done eyJnot.a.token");

        assert_eq!(b64url_decode(&b64url(b"hello?")).unwrap(), b"hello?");

    }

}

- **Accounts elsewhere on a profile (#105).** Bluesky, Mastodon and GitHub under Settings › Profile, checked and normalised (a handle, `@user@instance` or a profile URL all work), shown on the profile with the service's icon as `rel="me"` links.

- **A job's identity: the hub is an OIDC issuer (#103).** A job that declares `identity "AUDIENCE" { mount-path … }` gets a signed token naming the repository, bookmark, run and job, delivered as a file like a secret; `/.well-known/openid-configuration` and `/.well-known/jwks.json` publish what AWS, Google Cloud or Vault need to trust it. Tokens are masked in logs.

 "rsa",

 "sha2 0.10.9",

# RS256 for the OIDC job identity (hub spec §13.5); vendored already.

rsa = { version = "0.9", features = ["sha2"] }

    // The identities it asks for (#103): a token each, as a file.

    if !claim.job.identities.is_empty() {

        let issuer = app.oidc().await.map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("oidc: {e}")))?;

        for i in &claim.job.identities {

            let token = issuer.mint(&crate::oidc::JobClaims {

                issuer: app.config.base_url(),

                audience: &i.audience,

                owner: &row.owner,

                name: &row.name,

                visibility: &row.visibility,

                bookmark: &run.bookmark,

                run_number: run.number,

                job: &job.name,

                state: &run.state,

                runner: &runner,

                actor: &run.actor,

                now_ms: db::now_ms(),

                ttl_s: timeout_s,

            });

            claim.secrets.push(flow_core::api::SecretInput {

                name: format!("identity:{}", i.audience),

                path: i.mount_path.clone(),

                value: token,

            });

        }

    }

    let kid = app.oidc.get().map(|i| i.kid.clone());

        }

        if let Some(kid) = &kid {

            r.text = crate::oidc::mask(&r.text, kid);

pub mod oidc;

    /// The OIDC issuer for jobs (hub spec §13.5), made on first use.

    pub oidc: tokio::sync::OnceCell<oidc::Issuer>,

impl App {

    /// The OIDC issuer, its key loaded or made the first time it is asked for.

    pub async fn oidc(&self) -> anyhow::Result<&oidc::Issuer> {

        let keys = self.config.keys_dir();

        self.oidc.get_or_try_init(|| async move { tokio::task::spawn_blocking(move || oidc::Issuer::load_or_create(&keys)).await? }).await

    }

}

        oidc: tokio::sync::OnceCell::new(),

        .route("/.well-known/openid-configuration", get(crate::oidc::discovery))

        .route("/.well-known/jwks.json", get(crate::oidc::jwks))

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

async fn a_job_s_identity_is_a_signed_token_the_discovery_document_lets_anyone_verify() {

    let h = hub().await;

    let app = &h.app;

    let (st, disc) = call(app, "GET", "/.well-known/openid-configuration", None, None).await;

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

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

    assert_eq!(disc["issuer"], h.base.as_str());

    assert_eq!(disc["jwks_uri"], format!("{}/.well-known/jwks.json", h.base));

    let (st, jwks) = call(app, "GET", "/.well-known/jwks.json", None, None).await;

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

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

    assert_eq!(jwks["keys"][0]["alg"], "RS256");

    let kid = jwks["keys"][0]["kid"].as_str().unwrap().to_string();

    create(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("job \"deploy\" {\n    sandbox \"alpine\"\n    identity \"sts.amazonaws.com\" {\n        mount-path \"/run/secrets/aws-token\"\n    }\n    exec \"deploy\"\n}\n"),

    );

    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));

    let mut deploy = None;

    for _ in 0..50 {

        let Some(got) = claim(app).await else {

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

            continue;

        };

        if got["job"]["name"] == "deploy" {

            deploy = Some(got);

            break;

        }

    }

    let got = deploy.expect("the deploy job");

    let secrets = got["secrets"].as_array().expect("secrets");

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

    assert_eq!(

        (secrets[0]["name"].as_str(), secrets[0]["path"].as_str()),

        (Some("identity:sts.amazonaws.com"), Some("/run/secrets/aws-token"))

    );

    let token = secrets[0]["value"].as_str().unwrap().to_string();

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

    assert_eq!(issuer.kid, kid, "the key the JWKS publishes signed it");

    let claims = issuer.verify(&token, "sts.amazonaws.com", db::now_ms()).expect("verifies");

    assert_eq!(claims["sub"], "repo:andy/demo:bookmark:main");

    assert_eq!(claims["repository"], "andy/demo");

    assert_eq!(claims["job"], "deploy");

    assert_eq!(claims["run_number"], 1);

    assert_eq!(claims["runner"], "test-1");

    assert_eq!(claims["iss"], h.base.as_str());

    assert!(claims["state"].as_str().unwrap().len() == 64);

    // A job that prints its token leaves *** in the log.

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

    let records = serde_json::json!({ "records": [{ "t": 1, "stream": "out", "text": format!("TOKEN={token} end") }] });

    let (st, _) = call(app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(records.to_string())).await;

    assert_eq!(st, StatusCode::NO_CONTENT);

    let log = db::flow::log(&app.db, id, 0, 100).await.unwrap();

    assert!(log.iter().any(|l| l.text == "TOKEN=*** end"), "{log:?}");

}

| `/{owner}` | A person: identicon, display name, bio, location, website, accounts elsewhere (a Bluesky handle, a Mastodon address and a GitHub username, each checked and normalised under Settings › Profile and shown with the service's icon as a `rel="me"` link; oxygenhub#105), linked signing keys, organisations, repositories the viewer can see, and their recent activity (never the email). An organisation: identicon, description, members with roles, teams with sizes, repositories, recent activity in them. `/avatar/{name}` serves the account's or organisation's picture: an uploaded photo or logo when there is one, otherwise the identicon (SVG, a mirrored 5×5 pattern and hue from BLAKE3 of the lower-cased name). Pictures are uploaded under Settings › Profile, or an organisation's settings by its owners: PNG, JPEG or WebP recognised by content (not by name), at most 5 MiB and 8 000 pixels a side, decoded, centre-cropped to a square and re-encoded at 400×400 (PNG when it has transparency, JPEG otherwise), so no uploaded bytes are ever served. Removing the picture brings the identicon back. Pictures are cached for five minutes and carry an ETag. |

### 13.5 A job's identity: the hub as an OIDC issuer (done 2026-10-04)

A job that declares `identity "AUDIENCE" { mount-path "/run/secrets/…" }` (OxygenFlow m0 §3) gets a signed JWT so a cloud, a vault or any service that trusts this hub can let it in without a long-lived secret (oxygenhub#103).

- **Issuer.** The hub's `base_url`. `GET /.well-known/openid-configuration` publishes the discovery document (`issuer`, `jwks_uri`, RS256, the claims below) and `GET /.well-known/jwks.json` the public key, both cacheable for an hour. The key is RSA 2048 in `keys/oidc.pem` (PKCS#8, mode 0600), made the first time it is needed; its `kid` is the first 16 hex of BLAKE3 over the modulus. Rotation is by replacing the file and restarting; tokens signed by the old key then fail, which a job limit of hours makes harmless.

- **Token.** Minted at claim time, one per identity, and delivered through the same channel as secrets: the claim's `secrets` carry it as `identity:AUDIENCE` at the mount path, where Flow writes it 0400 on the in-memory mount. Header `{"alg":"RS256","typ":"JWT","kid"}`; claims `iss`, `sub` = `repo:OWNER/NAME:bookmark:BOOKMARK` (what a trust policy matches), `aud`, `iat`, `nbf` (30 s of skew), `exp` = the job's time limit from now, at most six hours, `jti`, and `repository`, `repository_owner`, `repository_visibility`, `bookmark`, `run_number`, `job`, `state` (the state hash), `runner`, `actor` (who started the run). A local `flow run` has no issuer and says the file will be missing.

- **Masking.** Log intake replaces any token this hub signed (recognised by its `kid`) with `***`, as it does secret values.

- **Trust, for example.** AWS: an IAM OIDC provider with the hub's URL and audience `sts.amazonaws.com`, a role whose trust policy matches `sub` `repo:andymccall/metalgear:bookmark:main`; the job then calls `aws sts assume-role-with-web-identity --web-identity-token file:///run/secrets/aws-token`. Google: a workload identity pool provider with the same issuer, mapping `assertion.sub`. HashiCorp Vault: the JWT auth method with `bound_audiences` and `bound_claims` on `repository`.