//! 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 `/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::identity::Identity; 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, 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, /// The latest signed checkpoint: its size and its note. checkpoint: Mutex>, /// Witnesses this log asks to cosign its checkpoints. witnesses: std::sync::RwLock>, /// Logs this hub witnesses: their origins and keys. witness_for: std::sync::RwLock>, } /// 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, } fn read_u64_at(f: &File, at: u64) -> std::io::Result { 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 { 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(¬e).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), witnesses: Default::default(), witness_for: Default::default(), }) } fn entry_bytes(f: &Files, n: u64) -> anyhow::Result> { 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) } /// Entry `n`'s bytes. pub fn entry(&self, n: u64) -> anyhow::Result> { let f = self.files.lock().unwrap(); anyhow::ensure!(n < f.size, "entry {n} is not in the log"); Self::entry_bytes(&f, n) } /// 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> { 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 { 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> { 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> { 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>> { 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>> { 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> = (first..first + t.width as u64).map(|n| Self::entry_bytes(&f, n)).collect::>()?; Ok(Some(tlog::encode_bundle(&entries)?)) } /// 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 { 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) { *self.witnesses.write().unwrap() = peers; } pub fn witnesses(&self) -> Vec { 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 { 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, ¬e)?; File::open(&tmp)?.sync_all()?; std::fs::rename(&tmp, self.dir.join("checkpoint"))?; *cur = Some((size, note)); Ok(()) } /// The latest signed checkpoint note. pub fn checkpoint(&self) -> Option { 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 { 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, ¬e)?; 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, pub object: ObjectRef, pub signers: Vec, } /// 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) -> Vec { 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 = 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 { let mut keys: Vec = 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, signers: &[KeyId], out: &mut Vec) { 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, report: &PushReport) -> Vec { let Ok(rid) = repo.id() else { return vec![] }; let mut out = Vec::new(); let add = |out: &mut Vec, kind: &'static str, object: ObjectRef, signers: Vec| { 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 } /// Add the log to a refs answer (Oxygen spec §14.8): the latest /// checkpoint, and for each bookmark's attestation its entry and inclusion /// proof against it. An attestation the hub serves but never logged (one /// pushed before the log began) is logged now, and proved on a later fetch. pub fn add_log(app: &Shared, row: Option, refs: &mut oxy_core::protocol::RefsResponse) -> Vec { let mut backfill = Vec::new(); let Some(note) = app.tlog.checkpoint() else { return backfill }; let Some(size) = tlog::Note::parse(¬e).ok().and_then(|n| Checkpoint::parse(&n.text).ok()).map(|c| c.size) else { return backfill; }; refs.log = Some(note); // The witnesses a client may pin, for integrated times (#240). refs.witnesses = app.tlog.witnesses().iter().map(|p| (p.name.clone(), p.identity.encode())).collect(); for r in refs.refs.iter_mut() { let Some(a) = &r.attestation else { continue }; let object = a.attest.hash(); let idx = tokio::task::block_in_place(|| { tokio::runtime::Handle::current().block_on(db::tlog::entry_of(&app.db, &object.to_hex(), "ref")) }); match idx { Ok(Some(i)) if (i as u64) < size => { let (Ok(entry), Ok(proof)) = (app.tlog.entry(i as u64), app.tlog.prove_inclusion(size, i as u64)) else { continue }; r.logged = Some(oxy_core::protocol::Logged { index: i as u64, entry, proof }); // When witnesses first saw it (#240): the first cosigned // checkpoints that hold it, each with the entry's proof. let notes = tokio::task::block_in_place(|| { tokio::runtime::Handle::current().block_on(db::tlog::cosigned_notes_after(&app.db, i, 2)) }) .unwrap_or_default(); for (at, text) in notes { if let Ok(proof) = app.tlog.prove_inclusion(at as u64, i as u64) { r.integration.push(oxy_core::protocol::Integration { note: text, proof }); } } } Ok(Some(_)) => {} Ok(None) => backfill.push(Pending { kind: "ref", repo: a.attest.repo, repo_row: row, object, signers: sig_keys(&a.sigs) }), Err(e) => eprintln!("tlog: looking up {}: {e}", &object.to_hex()[..12]), } } backfill } /// 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(¬e.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 { // Kept by size too, for integrated times (#240). db::tlog::keep_cosigned_note(&app.db, cp.size as i64, ¬e.encode()).await.ok(); 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, 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), } } /// 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 { let (_, sk) = crate::keys::load_secret(&keys)?; tl.refresh_checkpoint(&identity, &sk) }) .await; 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, } } } 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) -> 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/witnesses` (#240): this log's witnesses, one `NAME BASE64` /// line each (the identity object), which `oxy tlog verify` pins. pub async fn witnesses(State(app): State) -> Response { let body: String = app.tlog.witnesses().iter().map(|p| format!("{} {}\n", p.name, oxy_core::dsse::base64(&p.identity.encode()))).collect(); ([(header::CONTENT_TYPE, "text/plain; charset=utf-8"), (header::CACHE_CONTROL, "no-cache")], body).into_response() } /// `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, UrlPath(path): UrlPath) -> 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 = (0..300).map(entry).collect(); assert_eq!(log.append(&first).unwrap(), (0..300).collect::>()); let root = log.tree_hash(300).unwrap(); // Leaves alone, as RFC 6962 roots them. let mut stored: Vec = 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", "hub@example.org", 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(¬e, "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)); } }