The hub's transparency log (step 1 of docs/transparency.md) 965b950f9b96

-- The transparency log's index (oxygenhub#238, docs/transparency.md):

-- which entry logged which object, so a proof can be given for it and an

-- object is logged once. The log itself is append-only files in the data

-- directory (tlog/); these rows only point into it.

CREATE TABLE tlog_entries (

    idx         BIGINT PRIMARY KEY,

    object      TEXT NOT NULL,

    kind        TEXT NOT NULL,

    repo_id     BIGINT REFERENCES repositories(id) ON DELETE SET NULL,

    received_ms BIGINT NOT NULL

);

CREATE UNIQUE INDEX tlog_entries_object ON tlog_entries(object, kind);

CREATE INDEX tlog_entries_repo ON tlog_entries(repo_id, idx);

-- The transparency log's index (oxygenhub#238, docs/transparency.md):

-- which entry logged which object, so a proof can be given for it and an

-- object is logged once. The log itself is append-only files in the data

-- directory (tlog/); these rows only point into it.

CREATE TABLE tlog_entries (

    idx         INTEGER PRIMARY KEY,

    object      TEXT NOT NULL,

    kind        TEXT NOT NULL,

    repo_id     INTEGER REFERENCES repositories(id) ON DELETE SET NULL,

    received_ms INTEGER NOT NULL

);

CREATE UNIQUE INDEX tlog_entries_object ON tlog_entries(object, kind);

CREATE INDEX tlog_entries_repo ON tlog_entries(repo_id, idx);

//! The transparency log's index (migration 0076): which entry logged which

//! object. The log itself is files (`crate::tlog`).

use super::Pool;

use sqlx::FromRow;

#[derive(Clone, Debug, FromRow)]

pub struct Logged {

    pub idx: i64,

    pub object: String,

    pub kind: String,

    pub repo_id: Option<i64>,

    pub received_ms: i64,

}

/// The entry that logged `object` as `kind`, if one did.

pub async fn entry_of(pool: &Pool, object: &str, kind: &str) -> sqlx::Result<Option<i64>> {

    sqlx::query_scalar("SELECT idx FROM tlog_entries WHERE object = $1 AND kind = $2").bind(object).bind(kind).fetch_optional(pool).await

}

pub async fn record(pool: &Pool, idx: i64, object: &str, kind: &str, repo_id: Option<i64>, received_ms: i64) -> sqlx::Result<()> {

    sqlx::query("INSERT INTO tlog_entries (idx, object, kind, repo_id, received_ms) VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING")

        .bind(idx)

        .bind(object)

        .bind(kind)

        .bind(repo_id)

        .bind(received_ms)

        .execute(pool)

        .await?;

    Ok(())

}

/// The newest entries, newest first; of one repository when given.

pub async fn recent(pool: &Pool, repo_id: Option<i64>, limit: i64) -> sqlx::Result<Vec<Logged>> {

    match repo_id {

        Some(r) => {

            sqlx::query_as("SELECT * FROM tlog_entries WHERE repo_id = $1 ORDER BY idx DESC LIMIT $2")

                .bind(r)

                .bind(limit)

                .fetch_all(pool)

                .await

        }

        None => sqlx::query_as("SELECT * FROM tlog_entries ORDER BY idx DESC LIMIT $1").bind(limit).fetch_all(pool).await,

    }

}

//! The hub's transparency log (oxygenhub#238, docs/transparency.md, Oxygen

//! spec §14.8): one entry per signed object the hub accepts, in the order

//! it accepts them, kept as append-only files and served as C2SP

//! tlog-tiles with a signed checkpoint, so anyone can check the hub shows

//! everyone the same history.

//!

//! Files in `<data>/tlog/`:

//! - `entries`: each entry as a big-endian u16 length and its bytes;

//! - `offsets`: each entry's offset in `entries`, a u64 (little-endian);

//! - `hashes`: every hash of the tree in stored order (Oxygen spec §14.8);

//! - `checkpoint`: the latest signed checkpoint note (replaced, not

//!   appended, as it is only a signature over the others).

//!

//! Nothing is rewritten: a crash mid-append leaves at most a tail that is

//! cut back to the last whole entry on open, and the hashes are rebuilt

//! from the entries. The database's `tlog_entries` points from objects to

//! entry numbers.

use crate::{db, Shared};

use axum::extract::{Path as UrlPath, State};

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

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

use oxy_core::hash::{KeyId, ObjectRef};

use oxy_core::repo::{PushReport, Repo};

use oxy_core::tlog::{self, Checkpoint, Entry, Hash, HashReader, Tile, TlogError, TILE_WIDTH};

use std::fs::{File, OpenOptions};

use std::io::Write;

use std::os::unix::fs::FileExt;

use std::path::{Path, PathBuf};

use std::sync::Mutex;

/// How often a new checkpoint is signed while entries arrive.

pub const CHECKPOINT_EVERY: std::time::Duration = std::time::Duration::from_secs(5);

struct Files {

    entries: File,

    offsets: File,

    hashes: File,

    /// Entries in the log.

    size: u64,

    /// Bytes in `entries`.

    entries_len: u64,

}

/// The stored hashes in a file, by stored index.

struct FileHashes<'a>(&'a File);

impl HashReader for FileHashes<'_> {

    fn read_hashes(&self, indexes: &[u64]) -> Result<Vec<Hash>, TlogError> {

        indexes

            .iter()

            .map(|&i| {

                let mut h = [0u8; 32];

                self.0.read_exact_at(&mut h, i * 32).map_err(|e| TlogError::Read(format!("stored hash {i}: {e}")))?;

                Ok(h)

            })

            .collect()

    }

}

/// The log on disk.

pub struct Log {

    dir: PathBuf,

    /// The log's name in its checkpoints: the hub's host and `/log`.

    pub origin: String,

    files: Mutex<Files>,

    /// The latest signed checkpoint: its size and its note.

    checkpoint: Mutex<Option<(u64, String)>>,

}

fn read_u64_at(f: &File, at: u64) -> std::io::Result<u64> {

    let mut b = [0u8; 8];

    f.read_exact_at(&mut b, at)?;

    Ok(u64::from_le_bytes(b))

}

/// The origin of a hub's log: its host (and port) and `/log`.

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

    let host = base_url.split_once("://").map(|(_, h)| h).unwrap_or(base_url);

    format!("{}/log", host.trim_end_matches('/'))

}

impl Log {

    /// Open (or start) the log in `dir`, recovering from an interrupted

    /// append.

    pub fn open(dir: &Path, origin: &str) -> anyhow::Result<Log> {

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

        let open = |name: &str| OpenOptions::new().read(true).append(true).create(true).open(dir.join(name));

        let (entries, offsets, hashes) = (open("entries")?, open("offsets")?, open("hashes")?);

        let entries_on_disk = entries.metadata()?.len();

        // The whole entries: those whose offset was written and whose

        // bytes all were.

        let mut size = offsets.metadata()?.len() / 8;

        let mut entries_len = 0;

        while size > 0 {

            let off = read_u64_at(&offsets, (size - 1) * 8)?;

            let mut len = [0u8; 2];

            if off + 2 <= entries_on_disk && entries.read_exact_at(&mut len, off).is_ok() {

                let end = off + 2 + u16::from_be_bytes(len) as u64;

                if end <= entries_on_disk {

                    entries_len = end;

                    break;

                }

            }

            size -= 1;

        }

        offsets.set_len(size * 8)?;

        entries.set_len(entries_len)?;

        // The hashes of every leaf whose hashes were all written; the rest

        // are computed again from the entries.

        let have = hashes.metadata()?.len() / 32;

        let mut whole = size;

        while tlog::stored_hash_count(whole) > have {

            whole -= 1;

        }

        hashes.set_len(tlog::stored_hash_count(whole) * 32)?;

        let mut files = Files { entries, offsets, hashes, size: whole, entries_len };

        for n in whole..size {

            let bytes = Self::entry_bytes(&files, n)?;

            let add = tlog::stored_hashes_for_leaf(n, tlog::leaf_hash(&bytes), &FileHashes(&files.hashes))?;

            for h in add {

                files.hashes.write_all(&h)?;

            }

            files.size = n + 1;

        }

        files.size = size;

        files.hashes.sync_data()?;

        let checkpoint = std::fs::read_to_string(dir.join("checkpoint")).ok().and_then(|note| {

            let n = tlog::Note::parse(&note).ok()?;

            let c = Checkpoint::parse(&n.text).ok()?;

            (c.size <= size).then_some((c.size, note))

        });

        Ok(Log { dir: dir.to_path_buf(), origin: origin.to_string(), files: Mutex::new(files), checkpoint: Mutex::new(checkpoint) })

    }

    fn entry_bytes(f: &Files, n: u64) -> anyhow::Result<Vec<u8>> {

        let off = read_u64_at(&f.offsets, n * 8)?;

        let mut len = [0u8; 2];

        f.entries.read_exact_at(&mut len, off)?;

        let mut b = vec![0u8; u16::from_be_bytes(len) as usize];

        f.entries.read_exact_at(&mut b, off + 2)?;

        Ok(b)

    }

    /// Entries in the log.

    pub fn size(&self) -> u64 {

        self.files.lock().unwrap().size

    }

    /// Append entries, durably; their numbers.

    pub fn append(&self, entries: &[Entry]) -> anyhow::Result<Vec<u64>> {

        let mut f = self.files.lock().unwrap();

        let mut out = Vec::with_capacity(entries.len());

        for e in entries {

            let bytes = e.encode();

            let n = f.size;

            let add = tlog::stored_hashes_for_leaf(n, tlog::leaf_hash(&bytes), &FileHashes(&f.hashes))?;

            let off = f.entries_len;

            let mut rec = (bytes.len() as u16).to_be_bytes().to_vec();

            rec.extend_from_slice(&bytes);

            f.entries.write_all(&rec)?;

            f.entries_len += rec.len() as u64;

            f.offsets.write_all(&off.to_le_bytes())?;

            let mut hs = Vec::with_capacity(add.len() * 32);

            for h in add {

                hs.extend_from_slice(&h);

            }

            f.hashes.write_all(&hs)?;

            f.size = n + 1;

            out.push(n);

        }

        f.entries.sync_data()?;

        f.offsets.sync_data()?;

        f.hashes.sync_data()?;

        Ok(out)

    }

    /// The root of the first `n` entries.

    pub fn tree_hash(&self, n: u64) -> anyhow::Result<Hash> {

        let f = self.files.lock().unwrap();

        Ok(tlog::tree_hash(n, &FileHashes(&f.hashes))?)

    }

    /// Entry `n`'s inclusion proof in the tree of `t` entries.

    pub fn prove_inclusion(&self, t: u64, n: u64) -> anyhow::Result<Vec<Hash>> {

        let f = self.files.lock().unwrap();

        Ok(tlog::prove_inclusion(t, n, &FileHashes(&f.hashes))?)

    }

    /// The consistency proof from `n` entries to `t`.

    pub fn prove_consistency(&self, t: u64, n: u64) -> anyhow::Result<Vec<Hash>> {

        let f = self.files.lock().unwrap();

        Ok(tlog::prove_consistency(t, n, &FileHashes(&f.hashes))?)

    }

    /// A tile's hashes, when the log holds all of them.

    pub fn tile(&self, t: &Tile) -> anyhow::Result<Option<Vec<u8>>> {

        let f = self.files.lock().unwrap();

        let at_level = f.size >> (tlog::TILE_HEIGHT * t.level);

        if at_level < (t.n << tlog::TILE_HEIGHT) + t.width as u64 {

            return Ok(None);

        }

        Ok(Some(t.read_data(&FileHashes(&f.hashes))?))

    }

    /// The entry bundle beside level-0 tile `t`, when the log holds it.

    pub fn bundle(&self, t: &Tile) -> anyhow::Result<Option<Vec<u8>>> {

        let f = self.files.lock().unwrap();

        let first = t.n << tlog::TILE_HEIGHT;

        if f.size < first + t.width as u64 {

            return Ok(None);

        }

        let entries: Vec<Vec<u8>> = (first..first + t.width as u64).map(|n| Self::entry_bytes(&f, n)).collect::<anyhow::Result<_>>()?;

        Ok(Some(tlog::encode_bundle(&entries)?))

    }

    /// The latest signed checkpoint note.

    pub fn checkpoint(&self) -> Option<String> {

        self.checkpoint.lock().unwrap().as_ref().map(|(_, n)| n.clone())

    }

    /// Sign a checkpoint of the log as it is, unless the last one already

    /// covers it. True when a new one was signed.

    pub fn refresh_checkpoint(&self, identity: &oxy_core::identity::Identity, sk: &oxy_core::identity::SecretKey) -> anyhow::Result<bool> {

        let size = self.size();

        if self.checkpoint.lock().unwrap().as_ref().is_some_and(|(s, _)| *s >= size) {

            return Ok(false);

        }

        let c = Checkpoint { origin: self.origin.clone(), size, root: self.tree_hash(size)?, extensions: vec![] };

        let note = c.sign(identity, sk)?.encode();

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

        *self.checkpoint.lock().unwrap() = Some((size, note));

        Ok(true)

    }

}

/// An object to log.

#[derive(Clone, Debug)]

pub struct Pending {

    pub kind: &'static str,

    /// The repository's id (zero for none) and its row on this hub.

    pub repo: [u8; 16],

    pub repo_row: Option<i64>,

    pub object: ObjectRef,

    pub signers: Vec<KeyId>,

}

/// Log what is not logged yet. A failure is reported, never fatal: the

/// object stays accepted (enforcing logging before acceptance comes with

/// inclusion proofs in push responses).

pub async fn log(app: &Shared, items: Vec<Pending>) -> Vec<u64> {

    let mut fresh = Vec::new();

    for p in items {

        let object = p.object.to_hex();

        match db::tlog::entry_of(&app.db, &object, p.kind).await {

            Ok(Some(_)) => continue,

            Ok(None) => {}

            Err(e) => {

                eprintln!("tlog: looking up {} {}: {e}", p.kind, &object[..12]);

                continue;

            }

        }

        if !fresh.iter().any(|q: &Pending| q.kind == p.kind && q.object == p.object) {

            fresh.push(p);

        }

    }

    if fresh.is_empty() {

        return vec![];

    }

    let now = db::now_ms();

    let entries: Vec<Entry> = fresh

        .iter()

        .map(|p| Entry { kind: p.kind.to_string(), repo: p.repo, object: p.object, signers: p.signers.clone(), received: now as u64 })

        .collect();

    let tl = app.tlog.clone();

    let numbers = match tokio::task::spawn_blocking(move || tl.append(&entries)).await {

        Ok(Ok(n)) => n,

        Ok(Err(e)) => {

            eprintln!("tlog: appending {} entries: {e}", fresh.len());

            return vec![];

        }

        Err(e) => {

            eprintln!("tlog: appending: {e}");

            return vec![];

        }

    };

    for (p, n) in fresh.iter().zip(&numbers) {

        if let Err(e) = db::tlog::record(&app.db, *n as i64, &p.object.to_hex(), p.kind, p.repo_row, now).await {

            eprintln!("tlog: indexing entry {n}: {e}");

        }

    }

    numbers

}

fn sig_keys(sigs: &[oxy_core::identity::Signature]) -> Vec<KeyId> {

    let mut keys: Vec<KeyId> = sigs.iter().map(|s| s.key).collect();

    keys.sort();

    keys.dedup();

    keys

}

/// The certificates of short-lived keys among `signers`, to log beside

/// what they signed.

fn certificates(repo: &Repo, rid: [u8; 16], row: Option<i64>, signers: &[KeyId], out: &mut Vec<Pending>) {

    for k in signers {

        if let Ok(Some(c)) = repo.store().certificate_of(k) {

            let issuer = repo.store().get_certificate(&c).map(|sc| vec![sc.cert.issuer]).unwrap_or_default();

            out.push(Pending { kind: "certificate", repo: rid, repo_row: row, object: c, signers: issuer });

        }

    }

}

/// What a push brought that the log keeps: the attestations of the

/// references it moved (and the hub's acceptance of them), tags, seals,

/// the repository's policy, and the certificates of short-lived signers.

pub fn push_entries(repo: &Repo, row: Option<i64>, report: &PushReport) -> Vec<Pending> {

    let Ok(rid) = repo.id() else { return vec![] };

    let mut out = Vec::new();

    let add = |out: &mut Vec<Pending>, kind: &'static str, object: ObjectRef, signers: Vec<KeyId>| {

        certificates(repo, rid, row, &signers, out);

        out.push(Pending { kind, repo: rid, repo_row: row, object, signers });

    };

    for u in &report.updated {

        if u.new_state.is_zero() || !report.results.iter().any(|r| r.ok && r.name == u.name) {

            continue;

        }

        if let Some(tag) = u.name.strip_prefix("tags/") {

            if let Ok(Some((r, sa))) = repo.tag_attest(tag) {

                add(&mut out, "tag", r, sig_keys(&sa.sigs));

            }

        } else if let Ok(Some((r, sa))) = repo.latest_ref_attest(&u.name) {

            add(&mut out, "ref", r, sig_keys(&sa.sigs));

            if let Ok(Some(acc)) = repo.acceptance_of(&r) {

                add(&mut out, "accepted", acc.attest.hash(), sig_keys(&acc.sigs));

            }

        }

    }

    let mut seals = std::collections::BTreeSet::new();

    for h in &report.imported {

        for s in repo.store().seals_for(h).unwrap_or_default() {

            seals.insert(s);

        }

    }

    for s in seals {

        let signers = sig_keys(&repo.store().signatures(&s).unwrap_or_default());

        add(&mut out, "seal", s, signers);

    }

    if let Ok(Some((r, sp))) = repo.current_policy() {

        add(&mut out, "policy", r, sig_keys(&sp.sigs));

    }

    out

}

/// Sign a new checkpoint every few seconds while entries arrive.

pub async fn checkpointer(app: Shared) {

    let mut tick = tokio::time::interval(CHECKPOINT_EVERY);

    loop {

        tick.tick().await;

        let (identity, keys) = (app.hub_key.identity.clone(), app.config.keys_dir());

        let tl = app.tlog.clone();

        let done = tokio::task::spawn_blocking(move || -> anyhow::Result<bool> {

            let (_, sk) = crate::keys::load_secret(&keys)?;

            tl.refresh_checkpoint(&identity, &sk)

        })

        .await;

        if let Ok(Err(e)) = done {

            eprintln!("tlog: signing a checkpoint: {e}");

        }

    }

}

fn plain(status: StatusCode, why: &str) -> Response {

    (status, [(header::CONTENT_TYPE, "text/plain; charset=utf-8")], format!("{why}\n")).into_response()

}

/// `GET /log/checkpoint`: the latest signed checkpoint (C2SP

/// tlog-checkpoint, a signed note).

pub async fn checkpoint(State(app): State<Shared>) -> Response {

    match app.tlog.checkpoint() {

        Some(note) => ([(header::CONTENT_TYPE, "text/plain; charset=utf-8"), (header::CACHE_CONTROL, "no-cache")], note).into_response(),

        None => plain(StatusCode::SERVICE_UNAVAILABLE, "no checkpoint yet: the log signs one within seconds"),

    }

}

/// `GET /log/tile/{L}/{N}[.p/{W}]` and `/log/tile/entries/{N}[.p/{W}]`:

/// a tile of hashes or an entry bundle (C2SP tlog-tiles). Full tiles never

/// change and are cached for good.

pub async fn tile(State(app): State<Shared>, UrlPath(path): UrlPath<String>) -> Response {

    let (entries, rest) = match path.strip_prefix("entries/") {

        Some(r) => (true, format!("0/{r}")),

        None => (false, path.clone()),

    };

    let Some(t) = Tile::parse(&rest) else {

        return plain(StatusCode::NOT_FOUND, "no such tile");

    };

    let tl = app.tlog.clone();

    let got = tokio::task::spawn_blocking(move || if entries { tl.bundle(&t) } else { tl.tile(&t) }).await;

    match got {

        Ok(Ok(Some(data))) => {

            let cache = if t.width as u64 == TILE_WIDTH { "public, max-age=31536000, immutable" } else { "public, max-age=5" };

            ([(header::CONTENT_TYPE, "application/octet-stream"), (header::CACHE_CONTROL, cache)], data).into_response()

        }

        Ok(Ok(None)) => plain(StatusCode::NOT_FOUND, "the log does not hold that tile yet"),

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

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

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    fn entry(i: u64) -> Entry {

        Entry { kind: "ref".into(), repo: [1; 16], object: ObjectRef([i as u8; 32]), signers: vec![], received: i }

    }

    #[test]

    fn the_log_survives_a_torn_append_and_keeps_its_root() {

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

        let log = Log::open(dir.path(), "hub.example/log").unwrap();

        let first: Vec<Entry> = (0..300).map(entry).collect();

        assert_eq!(log.append(&first).unwrap(), (0..300).collect::<Vec<_>>());

        let root = log.tree_hash(300).unwrap();

        // Leaves alone, as RFC 6962 roots them.

        let mut stored: Vec<Hash> = Vec::new();

        for (n, e) in first.iter().enumerate() {

            stored.extend(tlog::stored_hashes_for_leaf(n as u64, e.leaf_hash(), &stored).unwrap());

        }

        assert_eq!(root, tlog::tree_hash(300, &stored).unwrap());

        drop(log);

        // A crash after the entry and part of the hashes were written.

        let torn = dir.path().join("hashes");

        let len = std::fs::metadata(&torn).unwrap().len();

        OpenOptions::new().write(true).open(&torn).unwrap().set_len(len - 40).unwrap();

        let mut f = OpenOptions::new().append(true).open(dir.path().join("entries")).unwrap();

        f.write_all(&[0, 200, 1, 2]).unwrap(); // a partial entry

        let log = Log::open(dir.path(), "hub.example/log").unwrap();

        assert_eq!(log.size(), 300);

        assert_eq!(log.tree_hash(300).unwrap(), root);

        log.append(&[entry(300)]).unwrap();

        assert_eq!(log.size(), 301);

        let p = log.prove_inclusion(301, 7).unwrap();

        tlog::check_inclusion(&p, 301, &log.tree_hash(301).unwrap(), 7, &first[7].leaf_hash()).unwrap();

        let c = log.prove_consistency(301, 300).unwrap();

        tlog::check_consistency(&c, 301, &log.tree_hash(301).unwrap(), 300, &root).unwrap();

        // Tiles and bundles: the full level-0 tile and the partial one.

        let full = log.tile(&Tile { level: 0, n: 0, width: 256 }).unwrap().unwrap();

        assert_eq!(full.len(), 256 * 32);

        assert!(log.tile(&Tile { level: 0, n: 1, width: 45 }).unwrap().is_some());

        assert!(log.tile(&Tile { level: 0, n: 1, width: 46 }).unwrap().is_none(), "not yet");

        let bundle = log.bundle(&Tile { level: 0, n: 1, width: 45 }).unwrap().unwrap();

        assert_eq!(tlog::decode_bundle(&bundle).unwrap()[0], first[256].encode());

    }

    #[test]

    fn checkpoints_are_signed_when_the_log_grows() {

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

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

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

        let log = Log::open(dir.path(), &origin_of("https://hub.example:7343/")).unwrap();

        assert_eq!(log.origin, "hub.example:7343/log");

        assert!(log.refresh_checkpoint(&hub, &sk).unwrap(), "the empty log's");

        assert!(!log.refresh_checkpoint(&hub, &sk).unwrap(), "nothing new");

        log.append(&[entry(1), entry(2)]).unwrap();

        assert!(log.refresh_checkpoint(&hub, &sk).unwrap());

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

        let c = Checkpoint::verify(&note, "hub.example:7343/log", &hub).unwrap();

        assert_eq!((c.size, c.root), (2, log.tree_hash(2).unwrap()));

        // Kept across a restart.

        drop(log);

        let log = Log::open(dir.path(), "hub.example:7343/log").unwrap();

        assert_eq!(log.checkpoint().map(|n| n.contains("\n2\n")), Some(true));

    }

}

//! `/log` (oxygenhub#238): the transparency log's latest checkpoint and

//! its newest entries, with where to fetch the rest (C2SP tlog-tiles).

use super::{filters, Page, Shell};

use crate::auth::Viewer;

use crate::{db, Shared};

use askama::Template;

use axum::extract::State;

pub struct Row {

    pub idx: i64,

    pub kind: String,

    pub object: String,

    /// `owner/name`, when the entry is a repository's this hub still has.

    pub repo: Option<String>,

    pub received_ms: i64,

}

#[derive(Template)]

#[template(path = "log.html")]

pub struct LogPage {

    shell: Shell,

    origin: String,

    size: u64,

    note: Option<String>,

    base: String,

    rows: Vec<Row>,

}

pub async fn page(State(app): State<Shared>, viewer: Viewer) -> Page<LogPage> {

    let shell = Shell::new(&app, "/log", &viewer);

    let recent = db::tlog::recent(&app.db, None, 50).await.unwrap_or_default();

    let mut rows = Vec::with_capacity(recent.len());

    for r in recent {

        // Entries of private repositories show only their hashes.

        let repo = match r.repo_id {

            Some(id) => match db::repos::get_by_id(&app.db, id).await {

                Ok(Some(row))

                    if crate::auth::roles::allows(

                        crate::auth::roles::role_for(&app.db, viewer.account(), &row).await.unwrap_or(None),

                        crate::auth::roles::Access::Read,

                    ) =>

                {

                    Some(row.slug())

                }

                _ => None,

            },

            None => None,

        };

        rows.push(Row { idx: r.idx, kind: r.kind, object: r.object, repo, received_ms: r.received_ms });

    }

    Page(LogPage {

        shell,

        origin: app.tlog.origin.clone(),

        size: app.tlog.size(),

        note: app.tlog.checkpoint(),

        base: app.config.base_url().trim_end_matches('/').to_string(),

        rows,

    })

}

{% extends "base.html" %}

{% block title %}Transparency log{% endblock %}

{% block content %}

<div class="head"><h1>Transparency log <small>{{ size }} entries, append-only</small></h1></div>

<section class="panel">

  <p>Every signed object this hub accepts (pushed references and tags, seals, policies, receipts, approvals, certificates and the hub's acceptances) is an entry in one Merkle tree, in the order it was accepted. Anyone can fetch the tree and check that the hub shows everyone the same history (Oxygen spec §14.8).</p>

  <ul class="small">

    <li>Checkpoint: <a href="/log/checkpoint"><code>{{ base }}/log/checkpoint</code></a> (a C2SP signed note, origin <code>{{ origin }}</code>, signed with <a href="/.well-known/oxyhub-key">the hub's key</a>)</li>

    <li>Tiles and entries: <code>{{ base }}/log/tile/…</code> (C2SP tlog-tiles, height 8)</li>

  </ul>

  {% if let Some(n) = note %}<pre class="small">{{ n }}</pre>{% else %}<p class="empty">No checkpoint yet: the hub signs one within seconds.</p>{% endif %}

</section>

{% if rows.is_empty() %}<p class="empty">Nothing has been logged yet.</p>{% else %}

<div class="table"><table>

<tr><th>#</th><th>When</th><th>Kind</th><th>Repository</th><th>Object</th></tr>

{% for r in rows %}

<tr><td class="px">{{ r.idx }}</td><td><time title="{{ r.received_ms|when_i }}">{{ r.received_ms|ago_i }}</time></td><td>{{ r.kind }}</td><td>{% if let Some(s) = r.repo %}<a href="/{{ s }}">{{ s }}</a>{% else %}<span class="mute">—</span>{% endif %}</td><td class="small"><code title="{{ r.object }}">{{ r.object|short }}</code></td></tr>

{% endfor %}

</table></div>

{% endif %}

{% endblock %}

 "const-oid 0.10.2",

version = "0.4.6"

 "sha2 0.11.0",

version = "0.4.6"

version = "0.4.6"

pub mod tlog;

    let (rid, signers) = (sa.attest.repo, sa.sigs.iter().map(|s| s.key).collect::<Vec<_>>());

    crate::tlog::log(app, vec![crate::tlog::Pending { kind: "approval", repo: rid, repo_row: Some(row.id), object: kept, signers }]).await;

    let (rid, signers) = (sr.attest.repo, sr.sigs.iter().map(|s| s.key).collect::<Vec<_>>());

    let cert = sr.certificate.as_ref().map(|c| (c.cert.cert.hash(), c.cert.cert.issuer));

    // The transparency log (#238): the receipt, and a short-lived key's

    // certificate beside it.

    let mut entries = vec![crate::tlog::Pending { kind: "receipt", repo: rid, repo_row: Some(row.id), object: attest, signers }];

    if let Some((c, issuer)) = cert {

        entries.push(crate::tlog::Pending { kind: "certificate", repo: rid, repo_row: Some(row.id), object: c, signers: vec![issuer] });

    }

    crate::tlog::log(app, entries).await;

    // Every certificate is logged (#238), as Fulcio's are in Rekor.

    crate::tlog::log(

        &app,

        vec![crate::tlog::Pending {

            kind: "certificate",

            repo: [0; 16],

            repo_row: None,

            object: signed.cert.hash(),

            signers: vec![app.hub_key.key],

        }],

    )

    .await;

pub mod tlog;

    /// The transparency log (#238).

    pub tlog: Arc<tlog::Log>,

    let tlog = Arc::new(tlog::Log::open(&config.data.join("tlog"), &tlog::origin_of(config.base_url()))?);

        tlog,

    tokio::spawn(tlog::checkpointer(app.clone()));

        // The transparency log (#238): what the push brought, logged before

        // the push is answered.

        let app = self.app_hint.lock().unwrap().as_ref().and_then(|w| w.upgrade());

        if let Some(app) = app {

            let to_log = self.repos.open(&parsed).map(|r| crate::tlog::push_entries(&r, Some(row.id), report)).unwrap_or_default();

            if !to_log.is_empty() {

                self.block(crate::tlog::log(&app, to_log));

            }

        }

mod transparency;

        .route("/log", get(transparency::page))

        .route("/log/checkpoint", get(crate::tlog::checkpoint))

        .route("/log/tile/{*path}", get(crate::tlog::tile))

/// The transparency log (oxygenhub#238): a push's reference attestation,

/// seal and policy become entries; the checkpoint is signed by the hub's

/// key; the tiles and entry bundles served rebuild the checkpoint's root;

/// and the `/log` page lists them.

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

async fn a_push_is_logged_and_the_log_checks_out() {

    use oxy_core::tlog::{self, Checkpoint, Entry, Hash, Note, Tile};

    let h = hub().await;

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

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

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

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

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

    write(work.path(), "a.txt", "a\n");

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

    c.policy_init().unwrap();

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

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

    let (attest, _) = host.latest_ref_attest("main").unwrap().unwrap();

    // Sign a checkpoint now rather than wait for the next tick.

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

    h.app.tlog.refresh_checkpoint(&h.app.hub_key.identity, &sk).unwrap();

    let (st, note) = call(&h.app, "GET", "/log/checkpoint", None, None).await;

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

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

    let cp = Checkpoint::verify(&Note::parse(&note).unwrap(), &origin, &h.app.hub_key.identity).unwrap();

    assert!(cp.size >= 3, "ref, seal and policy: {}", cp.size);

    // The entries, from the bundle beside the partial level-0 tile.

    let bundle_path = Tile { level: 0, n: 0, width: cp.size as u32 }.entries_path();

    let (st, _) = call(&h.app, "GET", &format!("/log/{bundle_path}"), None, None).await;

    assert_eq!(st, StatusCode::OK);

    let res =

        oxygenhub::router(h.app.clone()).oneshot(Request::get(format!("/log/{bundle_path}")).body(Body::empty()).unwrap()).await.unwrap();

    assert_eq!(res.headers()[header::CACHE_CONTROL], "public, max-age=5", "a partial tile changes");

    let bytes = res.into_body().collect().await.unwrap().to_bytes();

    let entries: Vec<Entry> = tlog::decode_bundle(&bytes).unwrap().iter().map(|b| Entry::decode(b).unwrap()).collect();

    let kinds: Vec<&str> = entries.iter().map(|e| e.kind.as_str()).collect();

    assert!(kinds.contains(&"ref") && kinds.contains(&"seal") && kinds.contains(&"policy"), "{kinds:?}");

    let r = entries.iter().find(|e| e.kind == "ref").unwrap();

    assert_eq!(r.object, attest);

    assert_eq!(r.repo, c.id().unwrap());

    // The leaves rebuild the checkpoint's root, and the hash tile served

    // is what they make.

    let mut stored: Vec<Hash> = Vec::new();

    for (n, e) in entries.iter().enumerate() {

        stored.extend(tlog::stored_hashes_for_leaf(n as u64, e.leaf_hash(), &stored).unwrap());

    }

    assert_eq!(tlog::tree_hash(cp.size, &stored).unwrap(), cp.root);

    let t = Tile { level: 0, n: 0, width: cp.size as u32 };

    let res =

        oxygenhub::router(h.app.clone()).oneshot(Request::get(format!("/log/{}", t.path())).body(Body::empty()).unwrap()).await.unwrap();

    assert_eq!(res.status(), StatusCode::OK);

    let data = res.into_body().collect().await.unwrap().to_bytes();

    assert_eq!(data.to_vec(), t.read_data(&stored).unwrap());

    let (st, _) = call(&h.app, "GET", "/log/tile/0/000.p/200", None, None).await;

    assert_eq!(st, StatusCode::NOT_FOUND, "not held yet");

    // Pushing the same again logs nothing new.

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

    assert_eq!(h.app.tlog.size(), cp.size);

    let (st, page) = call(&h.app, "GET", "/log", None, None).await;

    assert_eq!(st, StatusCode::OK);

    assert!(page.contains("Transparency log") && page.contains("andy/demo") && page.contains(&origin), "{page}");

}

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

The hub logs every signed object it accepts, in the order it accepts them, as entries of one Merkle tree (Oxygen spec §14.8):

- **What it logs:**

  - from pushes: reference and tag attestations, seals, the repository's policy, the hub's own acceptances, and the certificates of short-lived signers;

  - receipts as they are kept, and sealed approvals;

  - every certificate `keyless/certify` issues.

  An object is logged once (`tlog_entries`, migration 0076, maps objects to entry numbers), and a push is logged before it is answered.

- **Storage:** the log is append-only files in `<data>/tlog/`: `entries` (each a u16 length and its bytes), `offsets`, `hashes` (every tree hash in stored order) and `checkpoint`. A torn append is cut back to the last whole entry on start and its hashes rebuilt; nothing is ever rewritten.

- **Checkpoints:** every 5 seconds while entries arrive, the hub signs a checkpoint with its key. The origin is the base URL's host and `/log`, and the note carries an Ed25519 line and a composite line.

- **Serving:** it serves `GET /log/checkpoint` (no-cache), `GET /log/tile/<L>/<N>[.p/<W>]` and `GET /log/tile/entries/<N>[.p/<W>]` (C2SP tlog-tiles: full tiles cached for a year as immutable, partial ones for 5 s; a tile the log does not hold yet is 404), and `/log`, a page with the checkpoint and the newest entries (a private repository's entries show only their hashes).

- **Not yet:** inclusion proofs in push and refs responses, oxy's checks, witnesses and trusted time are the next steps (docs/transparency.md). Backups do not copy the log yet.

decisions below as recommended). Built 2026-10-18: step 1's log (oxy-core

`tlog`, Oxygen spec §14.8; the hub's log, served as tiles with signed

checkpoints, hub spec §13.5.3); inclusion proofs in responses and the rest

follow. Hashes are BLAKE3, like the rest of Oxygen: tools for the tile and

note formats read the log, but only Oxygen's code recomputes its roots.