Hubs witness each other's transparency logs (step A5) 4f40eed0588f

-- Transparency-log witnesses (oxygenhub#238): the last checkpoint this

-- hub cosigned for each log it witnesses, and how far each of this hub's

-- witnesses has cosigned its own log.

CREATE TABLE tlog_witnessed (

    origin     TEXT PRIMARY KEY,

    size       BIGINT NOT NULL,

    root       TEXT NOT NULL,

    updated_ms BIGINT NOT NULL

);

CREATE TABLE tlog_cosigned (

    witness    TEXT PRIMARY KEY,

    size       BIGINT NOT NULL,

    updated_ms BIGINT NOT NULL

);

-- Transparency-log witnesses (oxygenhub#238): the last checkpoint this

-- hub cosigned for each log it witnesses, and how far each of this hub's

-- witnesses has cosigned its own log.

CREATE TABLE tlog_witnessed (

    origin     TEXT PRIMARY KEY,

    size       INTEGER NOT NULL,

    root       TEXT NOT NULL,

    updated_ms INTEGER NOT NULL

);

CREATE TABLE tlog_cosigned (

    witness    TEXT PRIMARY KEY,

    size       INTEGER NOT NULL,

    updated_ms INTEGER NOT NULL

);

    pub tlog: Tlog,

/// The transparency log's witnesses (docs/transparency.md, #238): who

/// cosigns this hub's checkpoints, and whose logs this hub cosigns.

#[derive(Clone, Debug, Default, Deserialize)]

#[serde(default, deny_unknown_fields)]

pub struct Tlog {

    pub witnesses: Vec<TlogWitness>,

    pub witness_for: Vec<TlogWitnessed>,

}

/// A witness this hub's log asks: its name in cosignatures, its

/// add-checkpoint address (`https://HUB/log/witness/add-checkpoint` for

/// another hub), and its identity object in base64 (another hub's

/// `/.well-known/oxyhub-key`).

#[derive(Clone, Debug, Deserialize)]

#[serde(deny_unknown_fields)]

pub struct TlogWitness {

    pub name: String,

    pub url: String,

    pub key: String,

}

/// A log this hub witnesses: its origin (`HOST/log`) and its key, an

/// identity object in base64.

#[derive(Clone, Debug, Deserialize)]

#[serde(deny_unknown_fields)]

pub struct TlogWitnessed {

    pub origin: String,

    pub key: String,

}

            tlog: Tlog::default(),

/// The last checkpoint this hub cosigned for log `origin`: its size and

/// root (hex).

pub async fn witnessed(pool: &Pool, origin: &str) -> sqlx::Result<Option<(i64, String)>> {

    sqlx::query_as("SELECT size, root FROM tlog_witnessed WHERE origin = $1").bind(origin).fetch_optional(pool).await

}

/// Move what this hub cosigned for `origin` from `old` entries (none when

/// zero) to `size` with `root`; false when another request moved it first.

pub async fn witness_advance(pool: &Pool, origin: &str, old: i64, size: i64, root: &str) -> sqlx::Result<bool> {

    let now = super::now_ms();

    let r = if old == 0 {

        sqlx::query(

            "INSERT INTO tlog_witnessed (origin, size, root, updated_ms) VALUES ($1, $2, $3, $4) \

             ON CONFLICT (origin) DO UPDATE SET size = $2, root = $3, updated_ms = $4 WHERE tlog_witnessed.size = 0",

        )

        .bind(origin)

        .bind(size)

        .bind(root)

        .bind(now)

        .execute(pool)

        .await?

    } else {

        sqlx::query("UPDATE tlog_witnessed SET size = $1, root = $2, updated_ms = $3 WHERE origin = $4 AND size = $5")

            .bind(size)

            .bind(root)

            .bind(now)

            .bind(origin)

            .bind(old)

            .execute(pool)

            .await?

    };

    Ok(r.rows_affected() == 1)

}

/// How far witness `name` has cosigned this hub's log.

pub async fn cosigned_size(pool: &Pool, name: &str) -> sqlx::Result<i64> {

    Ok(sqlx::query_scalar("SELECT size FROM tlog_cosigned WHERE witness = $1").bind(name).fetch_optional(pool).await?.unwrap_or(0))

}

pub async fn set_cosigned(pool: &Pool, name: &str, size: i64) -> sqlx::Result<()> {

    sqlx::query(

        "INSERT INTO tlog_cosigned (witness, size, updated_ms) VALUES ($1, $2, $3) \

         ON CONFLICT (witness) DO UPDATE SET size = $2, updated_ms = $3",

    )

    .bind(name)

    .bind(size)

    .bind(super::now_ms())

    .execute(pool)

    .await?;

    Ok(())

}

    tlog.configure(&config.tlog)?;

use oxy_core::identity::Identity;

    /// Witnesses this log asks to cosign its checkpoints.

    witnesses: std::sync::RwLock<Vec<Peer>>,

    /// Logs this hub witnesses: their origins and keys.

    witness_for: std::sync::RwLock<Vec<(String, Identity)>>,

/// A witness of this hub's log.

#[derive(Clone, Debug)]

pub struct Peer {

    pub name: String,

    /// Its add-checkpoint address.

    pub url: String,

    pub identity: Identity,

}

        Ok(Log {

            dir: dir.to_path_buf(),

            origin: origin.to_string(),

            files: Mutex::new(files),

            checkpoint: Mutex::new(checkpoint),

            witnesses: Default::default(),

            witness_for: Default::default(),

        })

    }

    /// Take `[tlog]` from the configuration: witnesses and witnessed logs.

    pub fn configure(&self, c: &crate::config::Tlog) -> anyhow::Result<()> {

        let key = |b64: &str, what: &str| -> anyhow::Result<Identity> {

            let bytes = oxy_core::dsse::base64_decode(b64.trim()).ok_or_else(|| anyhow::anyhow!("[tlog] {what}: the key is not base64"))?;

            Identity::decode(&bytes).map_err(|e| anyhow::anyhow!("[tlog] {what}: {e}"))

        };

        let mut ws = Vec::new();

        for w in &c.witnesses {

            ws.push(Peer { name: w.name.clone(), url: w.url.clone(), identity: key(&w.key, &w.name)? });

        }

        let mut wf = Vec::new();

        for l in &c.witness_for {

            wf.push((l.origin.clone(), key(&l.key, &l.origin)?));

        }

        self.set_witnesses(ws);

        for (o, k) in wf {

            self.add_witness_for(&o, k);

        }

        Ok(())

    }

    pub fn set_witnesses(&self, peers: Vec<Peer>) {

        *self.witnesses.write().unwrap() = peers;

    }

    pub fn witnesses(&self) -> Vec<Peer> {

        self.witnesses.read().unwrap().clone()

    }

    /// Witness the log named `origin`, signed with `key`.

    pub fn add_witness_for(&self, origin: &str, key: Identity) {

        let mut w = self.witness_for.write().unwrap();

        w.retain(|(o, _)| o != origin);

        w.push((origin.to_string(), key));

    }

    fn witnessed_key(&self, origin: &str) -> Option<Identity> {

        self.witness_for.read().unwrap().iter().find(|(o, _)| o == origin).map(|(_, k)| k.clone())

    }

    /// This hub's name as a witness: its log's host.

    pub fn witness_name(&self) -> &str {

        self.origin.trim_end_matches("/log")

    }

    /// Replace the latest checkpoint's note with `note` (the same

    /// checkpoint, with cosignatures added), unless a newer one was signed.

    fn set_note(&self, size: u64, note: String) -> anyhow::Result<()> {

        let mut cur = self.checkpoint.lock().unwrap();

        if cur.as_ref().map(|(s, _)| *s) != Some(size) {

            return Ok(());

        }

        let tmp = self.dir.join("checkpoint.new");

        std::fs::write(&tmp, &note)?;

        File::open(&tmp)?.sync_all()?;

        std::fs::rename(&tmp, self.dir.join("checkpoint"))?;

        *cur = Some((size, note));

        Ok(())

}

/// Ask each witness to cosign the latest checkpoint, and serve their

/// cosignatures with it. A witness that answers 409 is told again from the

/// size it says it has; one that refuses is logged.

pub async fn collect(app: &Shared) {

    let peers = app.tlog.witnesses();

    let Some(text) = app.tlog.checkpoint() else { return };

    if peers.is_empty() {

        return;

    }

    let Ok(mut note) = tlog::Note::parse(&text) else { return };

    let Ok(cp) = Checkpoint::parse(&note.text) else { return };

    // What a witness checks: the note as the log signed it.

    let own = tlog::Note { text: note.text.clone(), sigs: note.sigs.iter().filter(|s| s.name == cp.origin).cloned().collect() };

    let mut changed = false;

    for p in peers {

        if note.cosigned_by(&p.name, &p.identity).is_ok() {

            continue;

        }

        let mut old = db::tlog::cosigned_size(&app.db, &p.name).await.unwrap_or(0) as u64;

        for _ in 0..2 {

            let proof = if old == 0 || old >= cp.size { Vec::new() } else { app.tlog.prove_consistency(cp.size, old).unwrap_or_default() };

            let body = tlog::AddCheckpoint { old_size: old, proof, note: own.encode() }.encode();

            let client = match crate::net::client(&p.url, std::slice::from_ref(&p.url), std::time::Duration::from_secs(20)).await {

                Ok(c) => c,

                Err(e) => {

                    eprintln!("tlog: witness {}: {e}", p.name);

                    break;

                }

            };

            let res = match client.post(&p.url).header(header::CONTENT_TYPE, "text/plain").body(body).send().await {

                Ok(r) => r,

                Err(e) => {

                    eprintln!("tlog: witness {}: {e}", p.name);

                    break;

                }

            };

            let status = res.status();

            let answer = res.text().await.unwrap_or_default();

            if status == StatusCode::CONFLICT {

                match answer.trim().parse() {

                    Ok(n) if n != old => {

                        old = n;

                        continue;

                    }

                    _ => {

                        eprintln!("tlog: witness {}: a conflict without a size: {answer}", p.name);

                        break;

                    }

                }

            }

            if !status.is_success() {

                eprintln!("tlog: witness {} refused the checkpoint of {} entries ({status}): {}", p.name, cp.size, answer.trim());

                break;

            }

            let lines = tlog::Note::parse(&format!("{}\n{answer}", note.text)).map(|n| n.sigs).unwrap_or_default();

            let theirs: Vec<_> = lines.into_iter().filter(|s| s.name == p.name).collect();

            let check = tlog::Note { text: note.text.clone(), sigs: theirs.clone() };

            match check.cosigned_by(&p.name, &p.identity) {

                Ok(_) => {

                    note.sigs.extend(theirs);

                    db::tlog::set_cosigned(&app.db, &p.name, cp.size as i64).await.ok();

                    changed = true;

                }

                Err(e) => eprintln!("tlog: witness {} answered a cosignature that does not verify: {e}", p.name),

            }

            break;

        }

    }

    if changed {

        if let Err(e) = app.tlog.set_note(cp.size, note.encode()) {

            eprintln!("tlog: keeping the cosigned checkpoint: {e}");

        }

    }

}

/// `POST /log/witness/add-checkpoint` (C2SP tlog-witness): cosign a

/// checkpoint of a log this hub witnesses, if it extends the last one

/// cosigned.

pub async fn add_checkpoint(State(app): State<Shared>, body: String) -> Response {

    let Ok(req) = tlog::AddCheckpoint::parse(&body) else {

        return plain(StatusCode::BAD_REQUEST, "not an add-checkpoint request");

    };

    let Some(origin) = tlog::Note::parse(&req.note).ok().and_then(|n| Checkpoint::parse(&n.text).ok()).map(|c| c.origin) else {

        return plain(StatusCode::BAD_REQUEST, "the checkpoint does not parse");

    };

    let Some(key) = app.tlog.witnessed_key(&origin) else {

        return plain(StatusCode::NOT_FOUND, &format!("this hub does not witness {origin}"));

    };

    let latest = match db::tlog::witnessed(&app.db, &origin).await {

        Ok(l) => l.and_then(|(size, root)| Some((size as u64, ObjectRef::from_hex(&root)?.0))),

        Err(e) => return plain(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),

    };

    let sk = match crate::keys::load_secret(&app.config.keys_dir()) {

        Ok((_, sk)) => sk,

        Err(e) => return plain(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),

    };

    let me = tlog::WitnessKey { name: app.tlog.witness_name(), identity: &app.hub_key.identity, sk: &sk };

    let now = (db::now_ms() / 1000) as u64;

    let conflict = |n: u64| ([(header::CONTENT_TYPE, "text/x.tlog.size")], format!("{n}\n"));

    match tlog::witness(&req, &origin, &key, latest, &me, now) {

        Ok(tlog::Witnessed::Cosigned { checkpoint, lines }) => {

            let root = ObjectRef(checkpoint.root).to_hex();

            match db::tlog::witness_advance(&app.db, &origin, req.old_size as i64, checkpoint.size as i64, &root).await {

                Ok(true) => {

                    let body: String = tlog::Note { text: String::new(), sigs: lines }.encode().trim_start_matches('\n').to_string();

                    ([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], body).into_response()

                }

                Ok(false) => {

                    let now = db::tlog::witnessed(&app.db, &origin).await.ok().flatten().map(|(s, _)| s as u64).unwrap_or(0);

                    (StatusCode::CONFLICT, conflict(now)).into_response()

                }

                Err(e) => plain(StatusCode::INTERNAL_SERVER_ERROR, &e.to_string()),

            }

        }

        Ok(tlog::Witnessed::Conflict(n)) => (StatusCode::CONFLICT, conflict(n)).into_response(),

        Err(tlog::WitnessError::Unsigned(e)) => plain(StatusCode::FORBIDDEN, &e),

        Err(tlog::WitnessError::Inconsistent(e)) => plain(StatusCode::UNPROCESSABLE_ENTITY, &e),

        Err(tlog::WitnessError::Bad(e)) => plain(StatusCode::BAD_REQUEST, &e),

    }

        match done {

            Ok(Err(e)) => eprintln!("tlog: signing a checkpoint: {e}"),

            // A new checkpoint, or one some witness has not cosigned yet.

            _ => collect(&app).await,

        .route("/log/witness/add-checkpoint", axum::routing::post(crate::tlog::add_checkpoint))

/// Witnesses (oxygenhub#238, step A5): one hub cosigns another's

/// checkpoints when each extends the last it cosigned, and the log serves

/// the cosignatures with its checkpoint; the witness refuses a fork (422),

/// tells a log that guessed its size wrong (409), and knows only the logs

/// it was told to witness (404).

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

async fn a_hub_witnesses_another_hub_s_log() {

    use oxy_core::tlog::{AddCheckpoint, Checkpoint, Note};

    let log = hub().await;

    let witness = hub().await;

    let origin = oxygenhub::tlog::origin_of(log.app.config.base_url());

    witness.app.tlog.add_witness_for(&origin, log.app.hub_key.identity.clone());

    let name = witness.app.tlog.witness_name().to_string();

    log.app.tlog.set_witnesses(vec![oxygenhub::tlog::Peer {

        name: name.clone(),

        url: format!("{}/log/witness/add-checkpoint", witness.base),

        identity: witness.app.hub_key.identity.clone(),

    }]);

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

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

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

    c.create_identity("Andy", "[email protected]", oxy_core::repo::IdentityScope::Repo).unwrap();

    let (_, sk) = oxygenhub::keys::load_secret(&log.app.config.keys_dir()).unwrap();

    let log_id = log.app.hub_key.identity.clone();

    let wid = witness.app.hub_key.identity.clone();

    for n in 0..2 {

        write(work.path(), "a.txt", &format!("{n}\n"));

        c.commit(&format!("change {n}"), &[]).unwrap();

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

        log.app.tlog.refresh_checkpoint(&log_id, &sk).unwrap();

        oxygenhub::tlog::collect(&log.app).await;

        let note = Note::parse(&log.app.tlog.checkpoint().unwrap()).unwrap();

        assert!(note.cosigned_by(&name, &wid).is_ok(), "round {n}: {}", note.encode());

        // Served as such.

        let (_, served) = call(&log.app, "GET", "/log/checkpoint", None, None).await;

        assert!(Note::parse(&served).unwrap().cosigned_by(&name, &wid).is_ok());

    }

    let cp = Checkpoint::parse(&Note::parse(&log.app.tlog.checkpoint().unwrap()).unwrap().text).unwrap();

    let ask = |req: AddCheckpoint| {

        let app = witness.app.clone();

        async move { call(&app, "POST", "/log/witness/add-checkpoint", None, Some(req.encode())).await }

    };

    // Another log of the same size, signed by the log's own key: a fork.

    let forked = Checkpoint { root: [7; 32], ..cp.clone() }.sign(&log_id, &sk).unwrap().encode();

    let (st, body) = ask(AddCheckpoint { old_size: cp.size, proof: vec![], note: forked.clone() }).await;

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

    // A wrong idea of what the witness saw: its size, to try again from.

    let (st, body) = ask(AddCheckpoint { old_size: 1, proof: vec![], note: forked }).await;

    assert_eq!((st, body.trim()), (StatusCode::CONFLICT, cp.size.to_string().as_str()));

    // A log it was not told to witness.

    let stranger = oxy_core::identity::SecretKey::generate();

    let sid = stranger.identity("elsewhere", "[email protected]", 1);

    let other = Checkpoint { origin: "elsewhere.example/log".into(), ..cp.clone() }.sign(&sid, &stranger).unwrap().encode();

    let (st, _) = ask(AddCheckpoint { old_size: 0, proof: vec![], note: other }).await;

    assert_eq!(st, StatusCode::NOT_FOUND);

}

### 13.5.3 The transparency log (oxygenhub#238, docs/transparency.md, steps 1-3 built 2026-10-18)

- **Witnesses** (step A5, 2026-10-18): a hub can witness other hubs' logs and have its own witnessed (C2SP tlog-witness, Oxygen spec §14.8).

  - As a witness it answers `POST /log/witness/add-checkpoint`: it cosigns, with its key and clock under its host's name, a checkpoint that verifies with the log's key and extends the last it cosigned (`tlog_witnessed`, migration 0077). It answers 409 with its size when the log guessed it wrong, 422 to a fork, 403 to another key's note and 404 to a log it was not told to witness.

  - As a log it sends each new checkpoint to its witnesses, with the consistency proof from what each last cosigned (`tlog_cosigned`), and serves their cosignatures with the checkpoint.

  - Both are set in `hub.toml`; the key is the other hub's `/.well-known/oxyhub-key` in base64:

    ```toml

    [tlog]

    witnesses = [{ name = "gcp.oxygenhub.io", url = "https://gcp.oxygenhub.io/log/witness/add-checkpoint", key = "…" }]

    witness_for = [{ origin = "oxygenhub.io/log", key = "…" }]

    ```

- **Not yet:** proofs in push responses, the `oxy witness` daemon, and trusted time (docs/transparency.md). Backups do not copy the log yet.