diff --git a/Cargo.lock b/Cargo.lock index e35b4f4..8b28f4d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "aead" version = "0.6.1" @@ -528,6 +534,15 @@ dependencies = [ "libc", ] +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if", +] + [[package]] name = "criterion" version = "0.5.1" @@ -996,6 +1011,16 @@ version = "0.4.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7ac824320a75a52197e8f2d787f6a38b6718bb6897a35142d749af3c0e8f4fe" +[[package]] +name = "flate2" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" +dependencies = [ + "crc32fast", + "miniz_oxide", +] + [[package]] name = "fluent-uri" version = "0.3.2" @@ -1840,6 +1865,16 @@ version = "0.3.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" +[[package]] +name = "miniz_oxide" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.1.0" @@ -1860,11 +1895,14 @@ dependencies = [ "console_error_panic_hook", "console_log", "ed25519-dalek", + "flate2", + "futures-util", "generic_log_worker", "getrandom 0.3.4", "getrandom 0.4.2", "hex", "jsonschema", + "length_prefixed", "log", "mirror_worker_config", "ml-dsa", @@ -1872,11 +1910,15 @@ dependencies = [ "serde", "serde_json", "serde_with", + "sha2", "signed_note", "tlog_checkpoint", "tlog_core", "tlog_cosignature", + "tlog_mirror", + "tlog_tiles", "tlog_witness", + "tokio", "tower-service", "worker", ] @@ -3029,6 +3071,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + [[package]] name = "slab" version = "0.4.11" diff --git a/Cargo.toml b/Cargo.toml index 79594e3..8633d70 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -121,6 +121,7 @@ der = { version = "0.8", features = ["oid"] } # (crypto-common 0.2 / signature 3.0), so it remains pinned to a pre-release. # The rest of the batch is now stable. ed25519-dalek = { version = "3.0.0", features = ["pem"] } +flate2 = { version = "1", default-features = false, features = ["rust_backend"] } futures-executor = "0.3.31" futures-util = "0.3.31" getrandom = { version = "0.4", features = ["wasm_js"] } @@ -157,6 +158,7 @@ tlog_checkpoint = { path = "crates/tlog_checkpoint", version = "0.2.0" } tlog_core = { path = "crates/tlog_core", version = "0.2.0" } tlog_cosignature = { path = "crates/tlog_cosignature", version = "0.2.0" } tlog_entry = { path = "crates/tlog_entry", version = "0.2.0" } +tlog_mirror = { path = "crates/tlog_mirror", version = "0.2.0" } tlog_tiles = { path = "crates/tlog_tiles", version = "0.2.0" } tlog_witness = { path = "crates/tlog_witness", version = "0.2.0" } tokio = { version = "1", features = ["sync"] } diff --git a/crates/mirror_worker/.dev.vars b/crates/mirror_worker/.dev.vars index c0d8d88..2150b9d 100644 --- a/crates/mirror_worker/.dev.vars +++ b/crates/mirror_worker/.dev.vars @@ -1 +1,2 @@ MIRROR_SIGNING_KEY="-----BEGIN PRIVATE KEY-----\nMDQCAQAwCwYJYIZIAWUDBAMRBCKAIEJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJC\nQkJCQkJC\n-----END PRIVATE KEY-----\n" +MIRROR_TICKET_KEY="Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc=" diff --git a/crates/mirror_worker/Cargo.toml b/crates/mirror_worker/Cargo.toml index 23f53a8..74d8b39 100644 --- a/crates/mirror_worker/Cargo.toml +++ b/crates/mirror_worker/Cargo.toml @@ -32,33 +32,40 @@ serde_json.workspace = true [dependencies] axum.workspace = true +base64.workspace = true config = { path = "./config", package = "mirror_worker_config" } console_error_panic_hook.workspace = true console_log.workspace = true +flate2.workspace = true +futures-util.workspace = true generic_log_worker.workspace = true getrandom.workspace = true getrandom_03.workspace = true hex.workspace = true +length_prefixed.workspace = true log.workspace = true ml-dsa.workspace = true pkcs8.workspace = true serde.workspace = true serde_json.workspace = true serde_with.workspace = true +sha2.workspace = true signed_note.workspace = true tlog_checkpoint.workspace = true tlog_core.workspace = true tlog_cosignature.workspace = true +tlog_mirror.workspace = true +tlog_tiles.workspace = true tlog_witness.workspace = true +tokio.workspace = true tower-service.workspace = true worker = { workspace = true, features = ["http", "axum"] } [dev-dependencies] -# base64 is used only by the dev-config pin tests; ed25519-dalek only to -# check that a non-ML-DSA-44 MIRROR_SIGNING_KEY is rejected. The shipped -# worker is ML-DSA-44 only. -base64.workspace = true +# Ed25519 is used only by unit tests, to check that a non-ML-DSA-44 +# MIRROR_SIGNING_KEY is rejected. The shipped worker is ML-DSA-44 only. ed25519-dalek.workspace = true +tokio = { workspace = true, features = ["macros", "rt"] } [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = [ diff --git a/crates/mirror_worker/config/src/lib.rs b/crates/mirror_worker/config/src/lib.rs index 1998dcf..5197b50 100644 --- a/crates/mirror_worker/config/src/lib.rs +++ b/crates/mirror_worker/config/src/lib.rs @@ -141,6 +141,10 @@ impl AppConfig { /// signed-note key name length cap, since each origin is itself /// used as a checkpoint origin. /// + /// Simple single-field bounds (e.g. the log-number ranges) are + /// expressed in `config.schema.json` and enforced by the build + /// script, so they are not re-checked here. + /// /// `log_key_name` uniqueness across log entries is not checked here; /// it is enforced earlier, during deserialization (see /// [`deserialize_logs`]). A plain `serde_json` object silently keeps diff --git a/crates/mirror_worker/src/add_entries.rs b/crates/mirror_worker/src/add_entries.rs new file mode 100644 index 0000000..4731066 --- /dev/null +++ b/crates/mirror_worker/src/add_entries.rs @@ -0,0 +1,1125 @@ +// Copyright (c) 2025-2026 Cloudflare, Inc. All rights reserved. +// SPDX-License-Identifier: BSD-3-Clause + +//! `POST /add-entries` handler. +//! +//! Implements the [c2sp.org/tlog-mirror `add-entries`][add-e] endpoint: +//! read the (optionally gzip) request body, verify each [`EntryPackage`] +//! against the target pending checkpoint with a subtree consistency proof, +//! persist the verified entries as bundles and hash tiles (see +//! [`crate::commit`]), and advance the persisted-entry frontier. +//! +//! The whole body is buffered, then all its packages are committed in a +//! single pass; the mirror checkpoint is cosigned only once the upload is +//! durably committed. A follow-up commit adds incremental streaming so a +//! large upload does not have to buffer the entire body in memory. +//! +//! A complete upload writes the cosigned checkpoint and returns 200 with +//! the mirror's [cosignature][cosig] line(s); a client-truncated upload +//! keeps the persisted prefix and returns 202 with the advanced next entry +//! so the client can resume (see [Processing][proc]). +//! +//! [add-e]: https://c2sp.org/tlog-mirror#add-entries +//! [proc]: https://c2sp.org/tlog-mirror#processing +//! [cosig]: https://c2sp.org/tlog-cosignature +//! [`EntryPackage`]: tlog_mirror::EntryPackage + +use std::collections::HashMap; +use std::io::{Cursor, ErrorKind}; + +use axum::extract::State; +use axum::http::{StatusCode, header::CONTENT_TYPE}; +use axum::response::IntoResponse as _; +use signed_note::{Note, NoteError}; +use tlog_checkpoint::CheckpointText; +use tlog_core::{ + Hash, HashReader, Subtree, TlogError, stored_hash_index, stored_hashes_for_record_hash, + tree_hash, verify_subtree_consistency_proof, +}; +use tlog_mirror::{ + AddEntriesRequestHeader, EntryPackage, MIRROR_INFO_CONTENT_TYPE, MirrorInfo, PACKAGE_ALIGNMENT, + ParseError, package_ranges, +}; +#[allow(clippy::wildcard_imports)] +use worker::*; + +use generic_log_worker::util::now_millis; + +use crate::{ + body, commit, + frontend_worker::{ApiResult, AppError}, + load_mirror_signer, load_ticket_sealer, log_verifiers, + mirror_state_do::{ + AdvanceNextEntryRequest, CommitRequest, CommittedCheckpoint, MirrorStateSnapshot, + NextEntry, PendingCheckpoint, state_stub, + }, + storage::load_origin_bucket, +}; + +/// Handle `POST /add-entries`. +/// +/// See the module-level comment for the full flow: read and verify entry +/// packages from the request body, persist the verified entries and +/// advance the persisted-entry frontier, and either cosign the mirror +/// checkpoint (200) or, for a truncated upload, persist the verified +/// prefix and return 202. +#[worker::send] +pub(crate) async fn add_entries( + State(env): State, + req: axum::extract::Request, +) -> ApiResult { + // Spec: the add-entries request body MUST have Content-Type + // application/octet-stream. Reject anything else up front (before + // spending time reading or decoding the body). + let (parts, body) = req.into_parts(); + if !content_type_is_octet_stream(&parts.headers) { + return Err(AppError::UnsupportedMediaType( + "add-entries requires Content-Type: application/octet-stream".to_owned(), + )); + } + + // No DefaultBodyLimit: Cloudflare enforces a request-body cap at the + // edge (100 MB, higher on paid plans) and 413s oversized bodies there. + // Clients on body-limited platforms truncate at a package boundary and + // resume via the 202 + advanced next_entry (tlog-mirror "Implementation + // Considerations"). The Workers runtime does not decompress request + // bodies, so gzip-encoded bodies are gunzipped here; unknown encodings + // are 415'd (see `crate::body`). + // + // The whole (decoded) body is buffered before processing; a follow-up + // commit streams it instead of buffering it all. + let raw = body::read_decoded_body(&parts.headers, body).await?; + let mut cursor = Cursor::new(raw.as_slice()); + + let header = match AddEntriesRequestHeader::read_from(&mut cursor) { + Ok(header) => header, + Err(e) => { + log::warn!("add-entries: malformed header: {e:?}"); + return Err(AppError::BadRequest(e.to_string())); + } + }; + + let Some(verifiers) = log_verifiers(&header.log_origin) else { + return Err(AppError::UnknownLogOrigin); + }; + + let snapshot = fetch_snapshot(&env, &header.log_origin).await?; + + // No pending checkpoint accepted yet means there is nothing to + // authenticate the entries against, so this MUST be 422, not 409: the + // client can't make progress by retrying and must first drive an + // add-checkpoint (tlog-mirror "Processing"). Empty pending signed-note bytes + // reliably mean pristine state; once accepted, the DO retains the + // latest pending forever. + if snapshot.pending.signed_note_bytes.is_empty() { + log::info!( + "add-entries: no pending checkpoint for origin {:?}; returning 422", + header.log_origin, + ); + return Err(AppError::UnprocessableEntity( + "mirror has no pending checkpoint for this log".to_owned(), + )); + } + + let target = match resolve_target_pending(&env, &header, &snapshot, &verifiers) { + Ok(t) => t, + Err(reason) => { + log::info!( + "add-entries: rejecting target pending: {reason} \ + (origin={origin:?}, upload_end={ue}, pending_size={ps}, committed_size={cs})", + origin = header.log_origin, + ue = header.upload_end, + ps = snapshot.pending.size, + cs = snapshot.committed.size, + ); + return Ok(mirror_info_409(&env, &snapshot, &header.log_origin)); + } + }; + + // upload_start must be <= next_entry (the first index not yet + // persisted); a client resuming after a 202 sets it to the advertised + // next entry. A non-256-aligned value is accepted; see + // `first_package_prefix`. + if header.upload_start > snapshot.next_entry.size { + log::info!( + "add-entries: rejecting upload_start={us} > next_entry={ne}", + us = header.upload_start, + ne = snapshot.next_entry.size, + ); + return Ok(mirror_info_409(&env, &snapshot, &header.log_origin)); + } + + // Spec (add-entries "Processing"): reject when excess_entries is too + // large. These are entries in [upload_start, next_entry) that are + // already persisted, so they are re-verified (subtree consistency) but + // not re-saved. Without a bound a client could set upload_start=0 and + // force the mirror to re-verify the entire persisted prefix on every + // request (a cheap DoS). A legitimate resume sets upload_start to the + // persisted frontier, or, when the frontier is mid-tile, to the + // frontier rounded down to a 256 boundary; excess_entries is then at + // most one package (256), which is our threshold. + let excess = excess_entries( + header.upload_start, + header.upload_end, + snapshot.next_entry.size, + ); + if excess > PACKAGE_ALIGNMENT { + log::info!( + "add-entries: rejecting upload_start={us} with excess_entries {excess} > {PACKAGE_ALIGNMENT} (next_entry={ne})", + us = header.upload_start, + ne = snapshot.next_entry.size, + ); + return Ok(mirror_info_409(&env, &snapshot, &header.log_origin)); + } + + let first_prefix = first_package_prefix( + &env, + &header, + snapshot.next_entry.size, + snapshot.next_entry.hash, + ) + .await?; + + verify_and_persist( + &env, + &header, + &snapshot, + &target, + &mut cursor, + &first_prefix, + ) + .await +} + +/// Return true iff the request's `Content-Type` is +/// `application/octet-stream`, ignoring any parameters (e.g. charset). +fn content_type_is_octet_stream(headers: &axum::http::HeaderMap) -> bool { + headers + .get(CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .unwrap_or_default() + .split(';') + .next() + .unwrap_or_default() + .trim() + == "application/octet-stream" +} + +/// Read, verify, and persist the entry packages for `[upload_start, +/// upload_end)`, then produce the HTTP response. +/// +/// Packages are read and verified in canonical order from the buffered +/// body. Entries below the frontier at request start are already +/// persisted, so they are verified but not re-saved (spec: "skip saving +/// already-written entries"); the remaining entries are committed in a +/// single [`commit::persist_entries`] call and the persisted-entry +/// frontier is advanced once in the DO. A follow-up commit flushes +/// incrementally instead of once at the end. +/// +/// Response cases (see [Processing][proc]): +/// +/// * every package received and the recomputed tree matches the target: +/// cosign the mirror checkpoint, advance it, and return 200 with the +/// cosignature line(s). +/// * client truncation after at least one complete package: persist the +/// verified prefix and return 202 with the advanced next entry. +/// * truncation before the first complete package, or a malformed body: +/// 400. A package that fails subtree-consistency verification: 422. +/// +/// # Errors +/// +/// Returns an error on a storage failure while persisting, or (500) if the +/// recomputed root of a complete upload disagrees with the target +/// checkpoint. +/// +/// [proc]: https://c2sp.org/tlog-mirror#processing +async fn verify_and_persist( + env: &Env, + header: &AddEntriesRequestHeader, + snapshot: &MirrorStateSnapshot, + target: &PendingCheckpoint, + cursor: &mut Cursor<&[u8]>, + first_prefix: &[Vec], +) -> ApiResult { + let bucket = load_origin_bucket(env, &header.log_origin)?; + + // Entries below the request-start frontier are already persisted; new + // persistence begins at this fixed boundary. + let initial_next = snapshot.next_entry.size; + + // Newly-received entries collected across all packages for a single + // commit, and the leaf index one past the last collected entry. + let mut entries: Vec> = Vec::new(); + let mut collected_end = initial_next; + let mut packages_received: u64 = 0; + let mut truncated = false; + + for (pkg_start, pkg_end) in package_ranges(header.upload_start, header.upload_end) { + let num_entries = pkg_end - pkg_start; + let pkg = match read_package(cursor, num_entries) { + PackageOutcome::Ok(pkg) => pkg, + // The body ended between or partway through a package: client + // truncation. Keep whatever complete packages were verified. + PackageOutcome::Eof => { + truncated = true; + break; + } + PackageOutcome::Err(e) => { + log::warn!("add-entries: malformed package [{pkg_start}, {pkg_end}): {e:?}"); + return Err(AppError::BadRequest(e.to_string())); + } + }; + + // Only the first package can be non-256-aligned, carrying the + // persisted-leaf prefix; later packages start on a 256 boundary + // (subtree start == pkg_start) and need no prefix. + let subtree_start = (pkg_start / PACKAGE_ALIGNMENT) * PACKAGE_ALIGNMENT; + let prefix: &[Vec] = if packages_received == 0 { + first_prefix + } else { + &[] + }; + if let Err(reason) = verify_package(prefix, &pkg, subtree_start, pkg_end, target) { + log::info!( + "add-entries: package [{pkg_start}, {pkg_end}) failed verification: {reason}" + ); + return Err(AppError::UnprocessableEntity(reason.to_owned())); + } + packages_received += 1; + + // Collect only the not-yet-persisted tail of this package. + if pkg_end > initial_next { + let skip = usize::try_from(initial_next.saturating_sub(pkg_start)) + .map_err(|_| Error::from("skip count overflows usize"))?; + entries.extend(pkg.entries.into_iter().skip(skip)); + collected_end = pkg_end; + } + } + + // Truncation before the first complete package is a hard 400. + if truncated && packages_received == 0 { + log::warn!("add-entries: body truncated before the first complete package"); + return Err(AppError::BadRequest( + "no complete entry package received".to_owned(), + )); + } + + // Persist all newly-received entries in a single commit, advancing the + // persisted-entry frontier once. + let mut frontier_size = initial_next; + let mut frontier_hash = snapshot.next_entry.hash; + if !entries.is_empty() { + let root = commit::persist_entries( + &bucket, + frontier_size, + frontier_hash, + collected_end, + &entries, + ) + .await?; + advance_next_entry(env, &header.log_origin, collected_end, root).await?; + frontier_size = collected_end; + frontier_hash = root; + } + let persisted_new = frontier_size > initial_next; + + if truncated { + log::info!( + "add-entries: client-truncated after {packages_received} complete packages; \ + persisted through {frontier_size}", + ); + return Ok(mirror_info_202( + env, + snapshot, + &header.log_origin, + frontier_size, + )); + } + + // Every canonical package was received. Any bytes past the last one + // are discarded, not rejected: the spec says "the mirror discards any + // partial bytes after the last successfully authenticated entry + // package", and the only defined 400 is when no package authenticated + // at all (handled above). Rejecting here would also be dishonest, since + // the entries were already persisted and the frontier advanced. + + // When we persisted new entries, the recomputed tree MUST match the + // pending checkpoint the packages were proven against. A mismatch means + // proof verification and tile computation disagree: an internal error, + // never a client fault. When nothing new was persisted (a re-upload of + // an already-persisted range), the log-signed target is trusted + // directly. + if persisted_new && (frontier_size != header.upload_end || frontier_hash != target.hash) { + log::error!( + "add-entries: recomputed frontier ({frontier_size}, {frontier_hash}) != target ({}, {})", + target.size, + target.hash, + ); + return Err(AppError::InternalServerError( + "recomputed root mismatch".to_owned(), + )); + } + + cosign_and_serve(env, header, target, snapshot).await +} + +/// The spec's `excess_entries = min(upload_end, next_entry) - +/// upload_start`: the count of already-persisted entries a request +/// re-verifies without re-saving (see add-entries "Processing"). +/// +/// Saturates at 0 so an `upload_start` above the frontier (rejected +/// separately) can't underflow. +fn excess_entries(upload_start: u64, upload_end: u64, next_entry: u64) -> u64 { + next_entry.min(upload_end).saturating_sub(upload_start) +} + +/// Outcome of reading the next entry package from the buffered body. +enum PackageOutcome { + Ok(EntryPackage), + /// The body ended between packages or partway through one: a client + /// truncation. Complete packages already read are persisted (partial + /// progress); truncation before the first complete package is a 400. + Eof, + Err(ParseError), +} + +/// Read the next entry package from `cursor`, returning +/// [`PackageOutcome::Eof`] when the buffered body is exhausted (cleanly +/// between packages or mid-package) so the caller treats it as a client +/// truncation. +fn read_package(cursor: &mut Cursor<&[u8]>, num_entries: u64) -> PackageOutcome { + if usize::try_from(cursor.position()).unwrap_or(usize::MAX) >= cursor.get_ref().len() { + return PackageOutcome::Eof; + } + match EntryPackage::read_from(cursor, num_entries) { + Ok(pkg) => PackageOutcome::Ok(pkg), + Err(ParseError::Io(ref e)) if e.kind() == ErrorKind::UnexpectedEof => PackageOutcome::Eof, + Err(e) => PackageOutcome::Err(e), + } +} + +/// Cosign the target checkpoint with the mirror key, advance the durable +/// mirror checkpoint via the DO, and return the 200 response carrying the +/// mirror cosignature line(s). +/// +/// Called only once the whole upload is durably committed (spec: the +/// mirror MUST NOT sign until all entries are committed). +/// +/// The DO's `/commit` both advances the durable checkpoint and writes the +/// served checkpoint object to R2, serialized under its commit lock, so +/// concurrent commits cannot rewind the served checkpoint. The frontend +/// therefore does not write R2 itself. +/// +/// If the DO reports a committed size ahead of `upload_end`, a concurrent +/// `add-entries` already advanced the mirror checkpoint past this upload. +/// Per the spec's final step ("If `upload_end` was too small, the mirror +/// MUST respond with a 409 Conflict") this returns a 409 with mirror-info +/// so the client resyncs, rather than a 200 that would falsely claim the +/// client's smaller checkpoint is being served. +/// +/// # Errors +/// +/// Returns an error if the target note/checkpoint fails to parse, signing +/// fails, or the DO commit fails. +async fn cosign_and_serve( + env: &Env, + header: &AddEntriesRequestHeader, + target: &PendingCheckpoint, + snapshot: &MirrorStateSnapshot, +) -> ApiResult { + // The checkpoint text comes from the log-signed pending note; the + // response is the bare cosignature line(s), identical to a witness's + // add-checkpoint response. + let note = Note::from_bytes(&target.signed_note_bytes) + .map_err(|e| Error::from(format!("target note parse: {e:?}")))?; + let cp_text = CheckpointText::from_bytes(note.text()) + .map_err(|e| Error::from(format!("target checkpoint parse: {e:?}")))?; + let note_sig = load_mirror_signer(env)? + .as_checkpoint_signer() + .sign(now_millis(), &cp_text) + .map_err(|e| Error::from(format!("mirror cosign: {e:?}")))?; + let cosig_body = + tlog_witness::serialize_add_checkpoint_response(std::slice::from_ref(¬e_sig)); + + // The served checkpoint is the log's signed note with the mirror's + // cosignature appended. Build it once so both the early-return and the + // `/commit` paths use the same bytes. + let checkpoint_obj = [target.signed_note_bytes.as_slice(), &cosig_body].concat(); + + // When the upload only reaches the mirror checkpoint's current size, + // the checkpoint at that size is already committed and served. There + // is nothing to advance, and re-committing would redundantly rewrite + // R2 and append a duplicate cosignature line to the served note, so + // return a fresh cosignature without dispatching `/commit`. (Committed + // is monotonic, so a stale-low snapshot only skips this optimization, + // never the other way.) + // + // Note: this also means a previously-failed R2 checkpoint write won't + // self-heal at the same size; it heals on the next larger commit. That + // is acceptable because the durable committed state advanced before the + // R2 write, and duplicate cosignatures are worse than a lagging object. + if header.upload_end <= snapshot.committed.size { + return Ok(( + StatusCode::OK, + [(CONTENT_TYPE, "text/plain; charset=utf-8")], + cosig_body, + ) + .into_response()); + } + + // The DO writes the cosigned checkpoint to R2 while advancing the + // durable checkpoint under its commit lock. + let committed = dispatch_commit( + env, + &header.log_origin, + &CommitRequest { + size: header.upload_end, + hash: target.hash, + signed_note_bytes: checkpoint_obj.clone(), + }, + ) + .await?; + + debug_assert_eq!( + committed.signed_note_bytes, checkpoint_obj, + "DO returned checkpoint bytes that do not match the cosigned note we sent" + ); + + if committed.size != header.upload_end { + // The DO refused to rewind: a concurrent commit already advanced + // the mirror checkpoint past upload_end, so ours was skipped. + // Fetch the latest state so the 409 mirror-info body advertises + // the current pending size and next entry, not the stale snapshot + // from the start of this request. + log::info!( + "add-entries: commit skipped, mirror checkpoint {} already past upload_end {}; \ + returning 409", + committed.size, + header.upload_end, + ); + let fresh_snapshot = match fetch_snapshot(env, &header.log_origin).await { + Ok(s) => s, + Err(e) => { + log::warn!( + "add-entries: failed to fetch fresh snapshot for 409; using stale: {e:?}" + ); + snapshot.clone() + } + }; + return Ok(mirror_info_409(env, &fresh_snapshot, &header.log_origin)); + } + + Ok(( + StatusCode::OK, + [(CONTENT_TYPE, "text/plain; charset=utf-8")], + cosig_body, + ) + .into_response()) +} + +/// Read the persisted-leaf prefix required to verify a non-256-aligned +/// first package: the leaves `[subtree_start, upload_start)` where +/// `subtree_start` is `upload_start` rounded down to a 256 boundary. +/// +/// Returns an empty vec when `upload_start` is already 256-aligned (the +/// common case and every non-first package). Because `upload_start <= +/// next_entry.size` is enforced upstream, the requested leaves are always +/// present in storage. +/// +/// The prefix is authenticated against the frontier hash tiles inside +/// [`commit::read_persisted_leaves`]; `persisted_hash` is the frontier +/// root at `persisted_size`. +/// +/// # Errors +/// +/// Returns an error if opening the origin bucket or reading the persisted +/// entry bundle fails, or the reloaded leaves fail authentication. +async fn first_package_prefix( + env: &Env, + header: &AddEntriesRequestHeader, + persisted_size: u64, + persisted_hash: Hash, +) -> Result>> { + let subtree_start = (header.upload_start / PACKAGE_ALIGNMENT) * PACKAGE_ALIGNMENT; + if header.upload_start == subtree_start { + return Ok(Vec::new()); + } + let bucket = load_origin_bucket(env, &header.log_origin)?; + commit::read_persisted_leaves( + &bucket, + subtree_start, + header.upload_start - subtree_start, + persisted_size, + persisted_hash, + ) + .await +} + +/// POST the [`CommitRequest`] to the per-origin DO, advancing the mirror +/// checkpoint, and return the DO's resulting [`CommittedCheckpoint`]. +/// +/// The returned checkpoint may be ahead of `commit_req` when a concurrent +/// `add-entries` already advanced past it (the DO refuses to rewind); the +/// caller compares sizes to detect that skip. A non-200 status (a +/// `/commit` beyond the persisted frontier) is a frontend/mirror bug, +/// surfaced as a transport error that the handler maps to 500. +async fn dispatch_commit( + env: &Env, + origin: &str, + commit_req: &CommitRequest, +) -> Result { + let stub = state_stub(env, origin)?; + let mut resp = stub + .fetch_with_request(Request::new_with_init( + "http://do/commit", + &RequestInit { + method: Method::Post, + body: Some(serde_json::to_string(commit_req)?.into()), + headers: { + let h = Headers::new(); + h.set("content-type", "application/json")?; + h + }, + ..Default::default() + }, + )?) + .await?; + match resp.status_code() { + 200 => Ok(resp.json().await?), + status => { + let msg = resp.text().await.unwrap_or_default(); + log::error!("add-entries: DO /commit returned {status}: {msg}"); + Err(Error::from(format!("commit failed ({status})"))) + } + } +} + +/// POST an [`AdvanceNextEntryRequest`] to the per-origin DO, advancing +/// the persisted-entry frontier to `(size, hash)`. Returns the effective +/// frontier (which reflects any concurrent advance). A DO rejection +/// (`size > pending`, a mirror bug) or RPC failure is surfaced as a +/// transport error that the handler maps to 500. +async fn advance_next_entry(env: &Env, origin: &str, size: u64, hash: Hash) -> Result { + let stub = state_stub(env, origin)?; + let req = AdvanceNextEntryRequest { size, hash }; + let mut resp = stub + .fetch_with_request(Request::new_with_init( + "http://do/advance-next-entry", + &RequestInit { + method: Method::Post, + body: Some(serde_json::to_string(&req)?.into()), + headers: { + let h = Headers::new(); + h.set("content-type", "application/json")?; + h + }, + ..Default::default() + }, + )?) + .await?; + match resp.status_code() { + 200 => Ok(resp.json().await?), + status => { + let msg = resp.text().await.unwrap_or_default(); + log::error!("add-entries: DO /advance-next-entry returned {status}: {msg}"); + Err(Error::from(format!("advance-next-entry failed ({status})"))) + } + } +} + +/// Read the per-origin DO state snapshot. A non-200 status or RPC failure +/// is a transport-level error the handler maps to 500. +async fn fetch_snapshot(env: &Env, origin: &str) -> Result { + let stub = state_stub(env, origin)?; + let mut resp = stub + .fetch_with_request(Request::new_with_init( + "http://do/get-state", + &RequestInit { + method: Method::Post, + body: None, + headers: Headers::new(), + ..Default::default() + }, + )?) + .await?; + if resp.status_code() != 200 { + return Err(Error::from(format!( + "DO /get-state returned {}", + resp.status_code() + ))); + } + let snapshot: MirrorStateSnapshot = resp.json().await?; + Ok(snapshot) +} + +/// Resolve the target pending checkpoint that this `add-entries` +/// request is uploading toward. Any of: +/// +/// * `upload_end == snapshot.pending.size`: use the current pending. +/// * `upload_end == snapshot.committed.size`: use the mirror +/// checkpoint, which the spec requires the mirror to accept as a +/// valid `upload_end` independent of any ticket. +/// * The ticket round-trips and yields a past pending whose +/// embedded checkpoint has size `upload_end` and verifies against +/// the trusted log keys: use that. +/// +/// Returns `Err(reason)` (a static `&str` describing why) when +/// none of these produce a target. The frontend turns the error into +/// a 409 with `text/x.tlog.mirror-info`. +fn resolve_target_pending( + env: &Env, + header: &AddEntriesRequestHeader, + snapshot: &MirrorStateSnapshot, + verifiers: &signed_note::VerifierList, +) -> std::result::Result { + if header.upload_end == snapshot.pending.size { + // Per spec: `upload_end` must be at-or-above the mirror's + // committed checkpoint. The DO state guarantees pending >= + // committed (the `/commit` RPC enforces it on the write + // side), so checking against pending is sufficient. + if header.upload_end < snapshot.committed.size { + return Err("upload_end below committed checkpoint"); + } + return Ok(snapshot.pending.clone()); + } + + // Spec: the mirror MUST also accept `upload_end` equal to the mirror + // checkpoint's tree size, whatever the ticket says. Everything up to + // it is already committed and served, so the committed checkpoint is + // the target and nothing new is persisted; `cosign_and_serve` returns + // a fresh cosignature without re-advancing. + if snapshot.committed.size > 0 && header.upload_end == snapshot.committed.size { + return Ok(PendingCheckpoint { + size: snapshot.committed.size, + hash: snapshot.committed.hash, + signed_note_bytes: snapshot.committed.signed_note_bytes.clone(), + }); + } + + // Try the ticket. An empty ticket can't carry a past pending, so + // there's nothing to fall back to. + if header.ticket.is_empty() { + return Err("upload_end does not match current pending and no ticket provided"); + } + let sealer = match load_ticket_sealer(env) { + Ok(m) => m, + Err(e) => { + // A missing/malformed MIRROR_TICKET_KEY is an operator + // misconfiguration, not a client error. Surface as 409 + // (which we'd return anyway) and log so an operator + // notices. + log::error!("add-entries: ticket sealer unavailable: {e:?}"); + return Err("ticket key unavailable"); + } + }; + // The log origin is bound as associated data, so a ticket minted + // for one log cannot be opened against another. + let Ok(plaintext) = sealer.open(&header.ticket, header.log_origin.as_bytes()) else { + return Err("ticket authentication failed"); + }; + // The ticket plaintext is the full signed-note bytes of a + // previously accepted pending checkpoint. Re-parse and re-verify + // against the trusted log keys; tickets are mirror-keyed so we + // know they came from us, but the *embedded* signature is the + // log's, and we've already established the log key isn't + // self-signed by the ticket key. + let Ok(note) = Note::from_bytes(&plaintext) else { + return Err("ticket plaintext is not a valid signed note"); + }; + if let Err(e) = note.verify(verifiers) { + match e { + NoteError::UnverifiedNote | NoteError::InvalidSignature { .. } => { + return Err("ticket-bound note has no valid signatures from trusted log keys"); + } + _ => return Err("ticket-bound note failed structural verification"), + } + } + let Ok(cp_text) = CheckpointText::from_bytes(note.text()) else { + return Err("ticket-bound note text is not a valid checkpoint"); + }; + if cp_text.origin() != header.log_origin { + return Err("ticket-bound checkpoint has a different origin"); + } + if cp_text.size() != header.upload_end { + return Err("ticket-bound checkpoint size != upload_end"); + } + if cp_text.size() < snapshot.committed.size { + return Err("ticket-bound checkpoint size < committed checkpoint size"); + } + Ok(PendingCheckpoint { + size: cp_text.size(), + hash: *cp_text.hash(), + signed_note_bytes: plaintext, + }) +} + +/// Verify a single [`EntryPackage`] against the target pending +/// checkpoint at `upload_end`. +/// +/// A package's subtree is `[subtree_start, pkg_end)` where `subtree_start` +/// is `pkg_start` rounded down to a [`PACKAGE_ALIGNMENT`] (256) boundary. +/// Only the *first* package of a request can have `pkg_start > +/// subtree_start`, when the client's `upload_start` is not +/// 256-aligned; the leading leaves `[subtree_start, pkg_start)` are then +/// already in the log and supplied here as `prefix_leaves` (read from +/// storage by the caller). For every other package `prefix_leaves` is +/// empty and `subtree_start == pkg_start`. +/// +/// `prefix_leaves` and `pkg.entries` are the raw entry bytes for the +/// contiguous leaves `[subtree_start, pkg_end)`. +/// +/// Returns `Err(reason)` if proof verification fails. +fn verify_package( + prefix_leaves: &[Vec], + pkg: &EntryPackage, + subtree_start: u64, + pkg_end: u64, + target: &PendingCheckpoint, +) -> std::result::Result<(), &'static str> { + debug_assert!( + subtree_start.is_multiple_of(PACKAGE_ALIGNMENT), + "subtree_start must be 256-aligned" + ); + let prefix_len = u64::try_from(prefix_leaves.len()).map_err(|_| "prefix too large")?; + let pkg_start = subtree_start + prefix_len; + let received = u64::try_from(pkg.entries.len()).map_err(|_| "package has too many entries")?; + if received != pkg_end - pkg_start { + return Err("package entry count != range size"); + } + + // Reconstruct the package's subtree hash as a standalone Merkle tree + // over its `count = prefix_len + received` leaves (persisted + // `prefix_leaves` first, then uploaded `pkg.entries`). Leaves are + // replayed with *local* 0-based indices so a subtree-completing leaf + // merges only within the subtree; replaying with absolute indices + // would, at a subtree boundary, reach for a left sibling outside it + // (e.g. leaf 511 of `[256,512)` would reach for `[0,256)`). + let count = prefix_len + received; + let mut store: HashMap = + HashMap::with_capacity(usize::try_from(2 * count).unwrap_or(usize::MAX)); + let mut next_idx = stored_hash_index(0, 0); + for (local_index, entry) in prefix_leaves.iter().chain(pkg.entries.iter()).enumerate() { + let hashes = { + let reader = MapReader { store: &store }; + stored_hashes_for_record_hash( + local_index as u64, + tlog_core::record_hash(entry), + &reader, + ) + .map_err(|_| "failed to compute stored hashes for leaf")? + }; + for h in hashes { + store.insert(next_idx, h); + next_idx += 1; + } + } + let reader = MapReader { store: &store }; + let Ok(pkg_hash) = tree_hash(count, &reader) else { + return Err("failed to compute package subtree hash"); + }; + + // The package's subtree is `[subtree_start, pkg_end)`. `Subtree::new` + // requires `subtree_start` to be aligned to the next-power-of-2 >= + // `pkg_end - subtree_start`; the 256-aligned `subtree_start` and a + // span of at most 256 leaves guarantee that. + let subtree = + Subtree::new(subtree_start, pkg_end).map_err(|_| "package range is not a valid subtree")?; + + // Verify the consistency proof against the target tree size. + if verify_subtree_consistency_proof(&pkg.proof, target.size, target.hash, &subtree, pkg_hash) + .is_err() + { + return Err("subtree consistency proof failed"); + } + + Ok(()) +} + +/// Build a `text/x.tlog.mirror-info` response carrying the mirror's +/// current pending tree size, the advertised `next_entry`, and a sealed +/// ticket, at the given HTTP `status`. +/// +/// The ticket is sealed via [`tlog_mirror::TicketSealer`] (AES-256-GCM-SIV, log +/// origin bound as associated data) so the client can present it on +/// retry to recover the pending state without keeping it in DO storage. +/// If sealing fails (operator misconfigured `MIRROR_TICKET_KEY`), the +/// response still carries an empty ticket; the client falls back to a +/// `(0, 0)` initial query. +/// +/// Two status codes use this shape (see [`mirror_info_409`] / +/// [`mirror_info_202`]): +/// +/// * `409 Conflict`: the request could not be applied (stale +/// `upload_start`/`upload_end`, no matching pending). `next_entry` +/// reports the persisted frontier so the client can resume. +/// * `202 Accepted`: a partial run of packages was persisted; +/// `next_entry` reports the *advanced* frontier so the client +/// continues from there. +fn mirror_info_response( + env: &Env, + snapshot: &MirrorStateSnapshot, + origin: &str, + status: StatusCode, + next_entry: u64, +) -> axum::response::Response { + let ticket = if snapshot.pending.signed_note_bytes.is_empty() { + Vec::new() + } else { + match load_ticket_sealer(env) { + // Bind the log origin as associated data so the ticket can + // only be reopened for the same log. + Ok(m) => m.seal(&snapshot.pending.signed_note_bytes, origin.as_bytes()), + Err(e) => { + log::error!("add-entries: cannot seal ticket: {e:?}"); + Vec::new() + } + } + }; + let info = MirrorInfo { + tree_size: snapshot.pending.size, + next_entry, + ticket, + }; + ( + status, + [(CONTENT_TYPE, MIRROR_INFO_CONTENT_TYPE)], + info.to_body(), + ) + .into_response() +} + +/// 409 Conflict carrying the mirror's current state. Advertises the +/// persisted frontier (`next_entry.size`) as the resume point. +fn mirror_info_409( + env: &Env, + snapshot: &MirrorStateSnapshot, + origin: &str, +) -> axum::response::Response { + mirror_info_response( + env, + snapshot, + origin, + StatusCode::CONFLICT, + snapshot.next_entry.size, + ) +} + +/// 202 Accepted after a partial persist. Advertises the freshly-advanced +/// persisted frontier so the client resumes appending from there. +fn mirror_info_202( + env: &Env, + snapshot: &MirrorStateSnapshot, + origin: &str, + next_entry: u64, +) -> axum::response::Response { + mirror_info_response(env, snapshot, origin, StatusCode::ACCEPTED, next_entry) +} + +/// A [`HashReader`] backed by a sparse `HashMap` indexed +/// by absolute stored-hash index. Used during package verification +/// to reconstruct the subtree hash from leaves we just received, +/// without needing access to the mirror's full storage backend. +struct MapReader<'a> { + store: &'a HashMap, +} + +impl HashReader for MapReader<'_> { + fn read_hashes(&self, indexes: &[u64]) -> std::result::Result, TlogError> { + indexes + .iter() + .map(|i| { + self.store + .get(i) + .copied() + .ok_or(TlogError::IndexesNotInTree) + }) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::{ + CONTENT_TYPE, MapReader, content_type_is_octet_stream, excess_entries, verify_package, + }; + use crate::mirror_state_do::PendingCheckpoint; + use std::collections::HashMap; + use tlog_core::{Hash, Subtree, stored_hash_index, stored_hashes, tree_hash}; + use tlog_mirror::EntryPackage; + use tlog_mirror::PACKAGE_ALIGNMENT; + + /// Deterministic distinct entry bytes for leaf `i`. + fn entry(i: u64) -> Vec { + format!("entry-{i}").into_bytes() + } + + /// Build a full in-memory stored-hash store for the first `n` leaves. + fn build_store(n: u64) -> HashMap { + let mut store: HashMap = HashMap::new(); + for i in 0..n { + let hashes = stored_hashes(i, &entry(i), &MapReader { store: &store }).unwrap(); + for (j, h) in hashes.iter().enumerate() { + store.insert(stored_hash_index(0, i) + j as u64, *h); + } + } + store + } + + /// Construct the target pending checkpoint and a verified package for + /// the subtree `[subtree_start, pkg_end)` of a tree of size `target`. + /// `pkg_start` is where the *uploaded* entries begin, so leaves + /// `[subtree_start, pkg_start)` become the persisted prefix. + fn fixture( + target: u64, + subtree_start: u64, + pkg_start: u64, + pkg_end: u64, + ) -> (Vec>, EntryPackage, PendingCheckpoint) { + let store = build_store(target); + let reader = MapReader { store: &store }; + let target_hash = tree_hash(target, &reader).unwrap(); + let subtree = Subtree::new(subtree_start, pkg_end).unwrap(); + let proof = tlog_core::subtree_consistency_proof(target, &subtree, &reader).unwrap(); + + let prefix: Vec> = (subtree_start..pkg_start).map(entry).collect(); + let pkg = EntryPackage { + entries: (pkg_start..pkg_end).map(entry).collect(), + proof, + }; + let cp = PendingCheckpoint { + size: target, + hash: target_hash, + signed_note_bytes: Vec::new(), + }; + (prefix, pkg, cp) + } + + #[test] + fn verify_aligned_package_ok() { + // Aligned first package: no persisted prefix. + let (prefix, pkg, cp) = fixture(1000, 256, 256, 512); + assert!(prefix.is_empty()); + verify_package(&prefix, &pkg, 256, 512, &cp).expect("aligned package verifies"); + } + + #[test] + fn verify_nonaligned_first_package_with_prefix_ok() { + // upload_start = 300 (non-aligned): subtree starts at 256, and the + // persisted leaves [256, 300) are supplied as the prefix. + let (prefix, pkg, cp) = fixture(1000, 256, 300, 512); + assert_eq!(prefix.len(), 44); + verify_package(&prefix, &pkg, 256, 512, &cp).expect("non-aligned package verifies"); + } + + #[test] + fn verify_first_ever_package_from_zero_ok() { + // Subtree rooted at 0 (first bundle), partial last package. + let (prefix, pkg, cp) = fixture(300, 0, 0, 256); + verify_package(&prefix, &pkg, 0, 256, &cp).expect("first bundle verifies"); + } + + #[test] + fn verify_rejects_wrong_prefix() { + // A prefix leaf that doesn't match the persisted entry yields the + // wrong subtree hash, so proof verification must fail. + let (mut prefix, pkg, cp) = fixture(1000, 256, 300, 512); + prefix[0] = b"tampered".to_vec(); + assert!(verify_package(&prefix, &pkg, 256, 512, &cp).is_err()); + } + + #[test] + fn verify_rejects_wrong_target_hash() { + let (prefix, pkg, mut cp) = fixture(1000, 256, 300, 512); + cp.hash = Hash([0xab; tlog_core::HASH_SIZE]); + assert!(verify_package(&prefix, &pkg, 256, 512, &cp).is_err()); + } + + #[test] + fn verify_rejects_entry_count_mismatch() { + let (prefix, mut pkg, cp) = fixture(1000, 256, 300, 512); + pkg.entries.push(entry(999)); // one too many + assert_eq!( + verify_package(&prefix, &pkg, 256, 512, &cp), + Err("package entry count != range size") + ); + } + + #[test] + fn verify_rejects_tampered_entry() { + let (prefix, mut pkg, cp) = fixture(1000, 256, 300, 512); + pkg.entries[0] = b"not the real entry".to_vec(); + assert!(verify_package(&prefix, &pkg, 256, 512, &cp).is_err()); + } + + #[test] + fn excess_entries_resume_at_frontier_is_zero() { + // The common resume: upload_start == next_entry, no overlap. + assert_eq!(excess_entries(2816, 3000, 2816), 0); + } + + #[test] + fn excess_entries_mid_tile_resume_within_one_package() { + // Mid-tile frontier: client rounds upload_start down to the 256 + // boundary, so overlap is the sub-256 tail and is accepted. + let next_entry = 600; + let upload_start = 512; // 600 rounded down to a 256 boundary + assert_eq!( + excess_entries(upload_start, 1000, next_entry), + next_entry - upload_start + ); + assert!(excess_entries(upload_start, 1000, next_entry) <= PACKAGE_ALIGNMENT); + } + + #[test] + fn excess_entries_from_zero_reverifies_whole_prefix() { + // The DoS case: upload_start=0 against a large frontier forces + // re-verification of the entire persisted prefix, well over the + // one-package threshold. + let excess = excess_entries(0, 10_000, 5_000); + assert_eq!(excess, 5_000); + assert!(excess > PACKAGE_ALIGNMENT); + } + + #[test] + fn excess_entries_bounded_by_upload_end() { + // Only entries below upload_end count as already-persisted overlap. + assert_eq!(excess_entries(100, 300, 5_000), 200); + } + + #[test] + fn excess_entries_saturates_above_frontier() { + // upload_start past the frontier (rejected separately) must not + // underflow. + assert_eq!(excess_entries(5_000, 6_000, 4_000), 0); + } + + #[test] + fn content_type_octet_stream_accepted() { + let mut headers = axum::http::HeaderMap::new(); + headers.insert(CONTENT_TYPE, "application/octet-stream".parse().unwrap()); + assert!(content_type_is_octet_stream(&headers)); + } + + #[test] + fn content_type_octet_stream_with_params_accepted() { + let mut headers = axum::http::HeaderMap::new(); + headers.insert( + CONTENT_TYPE, + "application/octet-stream; charset=binary".parse().unwrap(), + ); + assert!(content_type_is_octet_stream(&headers)); + } + + #[test] + fn content_type_missing_rejected() { + let headers = axum::http::HeaderMap::new(); + assert!(!content_type_is_octet_stream(&headers)); + } + + #[test] + fn content_type_non_octet_stream_rejected() { + let mut headers = axum::http::HeaderMap::new(); + headers.insert(CONTENT_TYPE, "application/json".parse().unwrap()); + assert!(!content_type_is_octet_stream(&headers)); + } +} diff --git a/crates/mirror_worker/src/body.rs b/crates/mirror_worker/src/body.rs new file mode 100644 index 0000000..298e50e --- /dev/null +++ b/crates/mirror_worker/src/body.rs @@ -0,0 +1,127 @@ +// Copyright (c) 2025-2026 Cloudflare, Inc. All rights reserved. +// SPDX-License-Identifier: BSD-3-Clause + +//! Request-body decoding for `add-entries`. +//! +//! [c2sp.org/tlog-mirror][spec] requires mirrors to accept +//! `Content-Encoding: gzip` request bodies (see [Request Body][reqbody]); +//! clients MAY send gzip without negotiating first. The Cloudflare Workers +//! runtime does not transparently decompress request bodies (unlike +//! responses, which it (de)compresses based on `Accept-Encoding`), so the +//! mirror must gunzip the body itself. +//! +//! This module reads the whole body into memory and inflates it in one +//! pass. A follow-up commit replaces this with incremental streaming so a +//! large upload does not have to buffer the entire body (Workers isolates +//! have a ~128 MB memory ceiling). +//! +//! [spec]: https://c2sp.org/tlog-mirror#add-entries +//! [reqbody]: https://c2sp.org/tlog-mirror#request-body + +use std::io::Read as _; + +use flate2::read::GzDecoder; +use futures_util::StreamExt as _; +#[allow(clippy::wildcard_imports)] +use worker::*; + +use crate::frontend_worker::{ApiResult, AppError}; + +/// Read the whole request body and decode it per `Content-Encoding`. +/// +/// `identity` (or an absent header) returns the body unchanged; +/// `gzip`/`x-gzip` is inflated. Any other encoding is unsupported: the +/// mirror can't authenticate a body it can't read, so this returns 415. +/// +/// # Errors +/// +/// Returns [`AppError::UnsupportedMediaType`] for an unrecognized +/// `Content-Encoding`, [`AppError::BadRequest`] for a malformed/truncated +/// gzip body, or a transport error while reading the body stream. +pub(crate) async fn read_decoded_body( + headers: &axum::http::HeaderMap, + body: axum::body::Body, +) -> ApiResult> { + let encoding = headers + .get(axum::http::header::CONTENT_ENCODING) + .and_then(|v| v.to_str().ok()) + .unwrap_or_default() + .trim() + .to_ascii_lowercase(); + + let mut raw = Vec::new(); + let mut stream = body.into_data_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|e| Error::from(e.to_string()))?; + raw.extend_from_slice(&chunk); + } + + match encoding.as_str() { + "" | "identity" => Ok(raw), + "gzip" | "x-gzip" => { + let mut out = Vec::new(); + GzDecoder::new(raw.as_slice()) + .read_to_end(&mut out) + .map_err(|e| AppError::BadRequest(format!("gzip decode failed: {e}")))?; + Ok(out) + } + other => Err(AppError::UnsupportedMediaType(format!( + "Unsupported Content-Encoding: {other}" + ))), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use flate2::Compression; + use flate2::write::GzEncoder; + use std::io::Write as _; + + /// gzip-compress `data` into a single buffer for test input. + fn gzip(data: &[u8]) -> Vec { + let mut enc = GzEncoder::new(Vec::new(), Compression::default()); + enc.write_all(data).unwrap(); + enc.finish().unwrap() + } + + /// Inflate `compressed` the same way [`read_decoded_body`] does. + fn gunzip(compressed: &[u8]) -> Result> { + let mut out = Vec::new(); + GzDecoder::new(compressed) + .read_to_end(&mut out) + .map_err(|e| Error::from(format!("gzip decode failed: {e}")))?; + Ok(out) + } + + #[test] + fn gzip_roundtrips() { + let mut plain: Vec = Vec::new(); + for i in 0..20_000u32 { + plain.extend_from_slice(format!("entry-{i};").as_bytes()); + } + let decoded = gunzip(&gzip(&plain)).expect("roundtrip"); + assert_eq!(decoded, plain); + } + + #[test] + fn empty_payload_roundtrips() { + let decoded = gunzip(&gzip(b"")).expect("roundtrip"); + assert!(decoded.is_empty()); + } + + #[test] + fn truncated_gzip_errors() { + let mut compressed = gzip(b"the quick brown fox jumps over the lazy dog"); + compressed.truncate(compressed.len() - 6); + assert!(gunzip(&compressed).is_err(), "truncated gzip must error"); + } + + #[test] + fn corrupt_gzip_errors() { + let mut compressed = gzip(b"hello world, this is a test payload for corruption"); + let mid = compressed.len() / 2; + compressed[mid] ^= 0xff; + assert!(gunzip(&compressed).is_err(), "corrupt gzip must error"); + } +} diff --git a/crates/mirror_worker/src/commit.rs b/crates/mirror_worker/src/commit.rs new file mode 100644 index 0000000..6f0bbf0 --- /dev/null +++ b/crates/mirror_worker/src/commit.rs @@ -0,0 +1,735 @@ +// Copyright (c) 2025-2026 Cloudflare, Inc. All rights reserved. +// SPDX-License-Identifier: BSD-3-Clause + +//! Commit path for `add-entries`: persist verified entries as +//! [tlog-tiles][tiles] entry bundles and (re)compute the Merkle hash +//! tiles, growing the mirrored copy of a log from its persisted frontier +//! up to a pending checkpoint. +//! +//! One commit maps to one `add-entries` request: [`persist_entries`] runs +//! once per request over the full batch of entries received across that +//! request's packages, so the persisted frontier is read from R2 once per +//! request (see [`read_edge_tiles`]), not once per package or entry. +//! +//! This mirrors the sequencer's append path +//! ([`generic_log_worker::log_ops`]'s `sequence_entries`), with two +//! differences: the mirror is stateless between requests, so +//! [`persist_entries`] re-reads and authenticates the persisted frontier +//! from R2 on every commit rather than caching it; and it stores the +//! tlog-tiles entry-bundle framing directly (each leaf `uint16_be(len) || +//! entry`, leaf hash `record_hash(entry)`) instead of owning an +//! application entry type. +//! +//! Commits are safe to interrupt and repeat: entry bundles and hash tiles +//! are immutable and content-addressed, so re-uploading one is a harmless +//! identical overwrite. The caller advances the durable mirror checkpoint +//! only after [`persist_entries`] returns and the recomputed root matches. +//! +//! [tiles]: https://c2sp.org/tlog-tiles + +use std::collections::HashMap; + +use futures_util::{ + future::try_join_all, + stream::{StreamExt as _, TryStreamExt as _, iter as stream_iter}, +}; +use generic_log_worker::{ + ObjectBackend, + log_ops::{HashReaderWithOverlay, TileWithBytes, UploadOptions, read_edge_tiles}, +}; +use length_prefixed::{ReadLengthPrefixedBytesExt as _, WriteLengthPrefixedBytesExt as _}; +use tlog_core::{ + Hash, HashReader as _, TlogError, record_hash, stored_hash_index, stored_hashes_for_record_hash, +}; +use tlog_tiles::{PathElem, PreloadedTlogTileReader, TileHashReader, TlogTile, TlogTileRecorder}; +#[allow(clippy::wildcard_imports)] +use worker::*; + +/// tlog-tiles fixes a tile height of 8, i.e. 256 entries per full tile. +const TILE_WIDTH: u64 = TlogTile::FULL_WIDTH as u64; + +/// Max in-flight R2 uploads per commit. A single `add-entries` request is +/// bounded only by the request-body size, so uploading every entry bundle +/// and hash tile at once could open thousands of connections and hold all +/// their bytes in memory. Workers allows only six connections to be waiting +/// on response headers at a time, so a small bound captures the concurrency +/// win without the unbounded fan-out. +const UPLOAD_CONCURRENCY: usize = 6; + +/// Object key for the (cosigned) checkpoint the mirror serves at +/// `//checkpoint`. Matches +/// [`generic_log_worker::log_ops::CHECKPOINT_KEY`]. +pub(crate) const CHECKPOINT_KEY: &str = "checkpoint"; + +/// Read `count` already-persisted log entries starting at leaf index +/// `start` back out of their entry bundle in object storage. +/// +/// `start` MUST be aligned to a 256-entry bundle boundary and `start + +/// count` MUST NOT exceed `persisted_size`, so the requested leaves all +/// live in the single entry bundle beginning at `start` (a package's +/// subtree spans at most 256 leaves). Returns the entries in order, +/// stripped of their uint16 length prefixes. +/// +/// Used by [`crate::add_entries`] to reconstruct the subtree hash of a +/// non-256-aligned first package, whose leading leaves `[subtree_start, +/// upload_start)` are already in the log and therefore absent from the +/// uploaded package. +/// +/// The reloaded bytes are untrusted storage output, so each decoded entry +/// is authenticated against the hash tiles committed at the persisted +/// frontier (`persisted_hash` is the frontier root, used to authenticate +/// the edge tiles via [`read_edge_tiles`]). Without this a corrupted or +/// stale bundle would flow into package verification and surface as a +/// client "422 Unprocessable Entity" for what is actually a mirror +/// storage fault. +/// +/// # Errors +/// +/// Returns an error if the bundle is missing from storage, is shorter than +/// `count` entries, or a decoded entry does not match its authenticated +/// leaf hash. +pub(crate) async fn read_persisted_leaves( + object: &impl ObjectBackend, + start: u64, + count: u64, + persisted_size: u64, + persisted_hash: Hash, +) -> Result>> { + debug_assert!( + start.is_multiple_of(TILE_WIDTH), + "start must be bundle-aligned" + ); + debug_assert!(start + count <= persisted_size, "leaves must be persisted"); + + // The bundle's stored width is 256 if the whole tile is persisted, + // otherwise the trailing partial width `persisted_size - start`. The + // data-tile path encodes that width (`from_index` derives it from the + // last stored leaf), so pick the last persisted leaf in this tile. + let stored_width = TILE_WIDTH.min(persisted_size - start); + let last_leaf = start + stored_width - 1; + let tile = + TlogTile::from_index(stored_hash_index(0, last_leaf)).with_data_path(PathElem::Entries); + let bytes = object + .fetch(tile.path()) + .await? + .ok_or_else(|| Error::from(format!("persisted entry bundle missing: {}", tile.path())))?; + + // Authenticate the decoded leaves against the frontier hash tiles. + // Leaves below the frontier's edge tile are not covered by + // `read_edge_tiles`, and the `excess_entries` bound lets a resume land + // in the tile just before the edge, so authenticate against the tree + // root via the full tile-hash reader instead. + let indexes: Vec = (start..start + count) + .map(|leaf| stored_hash_index(0, leaf)) + .collect(); + let want = authenticated_leaf_hashes(object, persisted_size, persisted_hash, &indexes).await?; + + let mut cur: &[u8] = &bytes; + let mut out = Vec::with_capacity(usize::try_from(count).unwrap_or(0)); + for i in 0..count { + let leaf = start + i; + let entry = cur + .read_length_prefixed(2) + .map_err(|e| Error::from(format!("persisted bundle leaf {leaf} truncated: {e}")))?; + let idx = usize::try_from(i).unwrap_or(usize::MAX); + if record_hash(&entry) != want[idx] { + return Err(Error::from(format!( + "persisted bundle leaf {leaf} does not match its authenticated hash" + ))); + } + out.push(entry); + } + Ok(out) +} + +/// Fetch and authenticate the record hashes for `indexes` (level-0 leaf +/// stored-hash indexes) against the frontier `(tree_size, tree_hash)`. +/// +/// Runs the standard tlog-tiles two-pass [`TileHashReader`] protocol: a +/// recording pass discovers the hash tiles needed to prove the requested +/// leaves, they are fetched from storage, then a verifying pass +/// authenticates them against `tree_hash`. A storage fault (missing, +/// stale, or tampered tile) therefore surfaces as an error here rather +/// than as a spurious client rejection downstream. +/// +/// # Errors +/// +/// Returns an error if a required hash tile is missing from storage or the +/// fetched tiles do not authenticate against `tree_hash`. +async fn authenticated_leaf_hashes( + object: &impl ObjectBackend, + tree_size: u64, + tree_hash: Hash, + indexes: &[u64], +) -> Result> { + // Recording pass: `TlogTileRecorder` short-circuits `read_hashes` with + // `RecordedTilesOnly` after collecting the tiles it would need. + let recorder = TlogTileRecorder::default(); + match TileHashReader::new(tree_size, Hash::default(), &recorder).read_hashes(indexes) { + Err(TlogError::RecordedTilesOnly) => {} + other => { + return Err(Error::from(format!( + "expected RecordedTilesOnly while recording hash tiles, got {other:?}" + ))); + } + } + + let tile_futures = recorder.0.into_inner().into_iter().map(|tile| { + let path = tile.path(); + async move { + let bytes = object + .fetch(path.clone()) + .await? + .ok_or_else(|| Error::from(format!("persisted hash tile missing: {path}")))?; + Ok::<(TlogTile, Vec), Error>((tile, bytes)) + } + }); + let tiles: HashMap> = try_join_all(tile_futures).await?.into_iter().collect(); + + let reader = PreloadedTlogTileReader(tiles); + TileHashReader::new(tree_size, tree_hash, &reader) + .read_hashes(indexes) + .map_err(|e| Error::from(format!("authenticate persisted leaf hashes: {e:?}"))) +} + +/// Authenticate a reloaded partial entry bundle before it is extended and +/// re-served. +/// +/// The bytes come back from object storage untrusted. New-leaf hashes are +/// computed only from the freshly-uploaded entries and the authenticated +/// edge tiles, and the recomputed root never reads the reloaded bundle +/// bytes, so a corrupted or tampered partial bundle would otherwise be +/// silently re-uploaded and served. +/// +/// Confirms the bundle decodes to exactly the `[subtree_start, +/// persisted_size)` leaves as length-prefixed entries with no trailing +/// bytes, and that each decoded entry's `record_hash` matches the +/// authenticated leaf hash from `edge_tiles`. +/// +/// # Errors +/// +/// Returns an error if the bundle is truncated, carries trailing bytes, a +/// leaf hash is unavailable, or a decoded entry does not match its +/// authenticated leaf hash. +fn verify_partial_bundle( + bytes: &[u8], + subtree_start: u64, + persisted_size: u64, + edge_tiles: &HashMap, +) -> Result<()> { + let overlay = HashMap::new(); + let reader = HashReaderWithOverlay { + edge_tiles, + overlay: &overlay, + }; + let mut cur: &[u8] = bytes; + for leaf in subtree_start..persisted_size { + let entry = cur + .read_length_prefixed(2) + .map_err(|e| Error::from(format!("partial entry bundle leaf {leaf} truncated: {e}")))?; + let want = reader + .read_hashes(&[stored_hash_index(0, leaf)]) + .map_err(|e| Error::from(format!("missing persisted leaf hash {leaf}: {e:?}")))?; + if record_hash(&entry) != want[0] { + return Err(Error::from(format!( + "partial entry bundle leaf {leaf} does not match its authenticated hash" + ))); + } + } + if !cur.is_empty() { + return Err(Error::from( + "partial entry bundle has trailing bytes past the persisted frontier", + )); + } + Ok(()) +} + +/// Serialize one entry into its tlog-tiles entry-bundle framing (a +/// big-endian uint16 length prefix followed by the entry bytes) and +/// append it to `buf`. Matches `tlog_entry`'s `to_data_tile_entry` so +/// the mirror's bundles are byte-identical to a native tlog-tiles log's. +/// +/// # Errors +/// +/// Returns an error if `entry` exceeds 65535 bytes (the u16 length +/// prefix limit). Entries were parsed off the wire with the same uint16 +/// framing, so in practice this never fires. +fn push_tile_leaf(buf: &mut Vec, entry: &[u8]) -> Result<()> { + buf.write_length_prefixed(entry, 2) + .map_err(|e| Error::from(format!("entry too large for tile bundle: {e}"))) +} + +/// The `UploadOptions` for an immutable, content-addressed tile (entry +/// bundle or hash tile). +fn immutable_tile_opts() -> UploadOptions { + UploadOptions { + content_type: Some("application/octet-stream".to_owned()), + immutable: true, + } +} + +/// Persist entries `[persisted_size, target_size)` into object storage, +/// growing the mirrored tree and returning the recomputed root hash. +/// +/// Called once per `add-entries` request; `entries` spans all packages +/// received in that request. `entries[i]` MUST be the raw log-entry bytes +/// (no length prefix) for leaf `persisted_size + i`, covering exactly +/// `[persisted_size, target_size)`. +/// +/// Reads the persisted frontier from R2 (skipped at size 0), replays each +/// leaf through [`stored_hashes_for_record_hash`] while flushing full and +/// trailing-partial entry bundles, then (re)computes and uploads every +/// hash tile in [`TlogTile::new_tiles`]`(persisted_size, target_size)`. +/// +/// It does not write the checkpoint object or advance the mirror +/// checkpoint; the caller does that after verifying the returned root. +/// +/// # Errors +/// +/// Returns an error on any storage failure, if a persisted tile is +/// missing or fails authentication, or if `entries.len()` does not equal +/// `target_size - persisted_size`. +pub(crate) async fn persist_entries( + object: &impl ObjectBackend, + persisted_size: u64, + persisted_hash: Hash, + target_size: u64, + entries: &[Vec], +) -> Result { + let expected = target_size + .checked_sub(persisted_size) + .ok_or_else(|| Error::from("target_size < persisted_size"))?; + if entries.len() as u64 != expected { + return Err(Error::from(format!( + "commit entry count {} != range {persisted_size}..{target_size}", + entries.len() + ))); + } + if expected == 0 { + return Ok(persisted_hash); + } + + // Genesis (persisted_size 0): no persisted frontier exists yet, so there + // are no edge tiles to read; start from an empty overlay. + let mut edge_tiles = if persisted_size == 0 { + HashMap::new() + } else { + read_edge_tiles(object, persisted_size, &persisted_hash).await? + }; + + // Load the current partial entry bundle so we extend rather than + // overwrite it. Only exists when the frontier is mid-tile. The reloaded + // bytes are untrusted (storage could return corrupted or tampered + // data), so authenticate them before extending and re-serving them. + let mut data_tile = Vec::new(); + if persisted_size > 0 && !persisted_size.is_multiple_of(TILE_WIDTH) { + let subtree_start = (persisted_size / TILE_WIDTH) * TILE_WIDTH; + let partial = TlogTile::from_index(stored_hash_index(0, persisted_size - 1)) + .with_data_path(PathElem::Entries); + data_tile = object.fetch(partial.path()).await?.ok_or_else(|| { + Error::from(format!("partial entry bundle missing: {}", partial.path())) + })?; + verify_partial_bundle(&data_tile, subtree_start, persisted_size, &edge_tiles)?; + } + + // Replay leaves, buffering entry-bundle uploads until the end so they + // can be issued with bounded concurrency. Bundles are immutable and + // idempotent, so overlapping them is safe. + let mut overlay: HashMap = HashMap::new(); + let mut n = persisted_size; + let mut bundle_uploads = Vec::new(); + for entry in entries { + push_tile_leaf(&mut data_tile, entry)?; + let hashes = stored_hashes_for_record_hash( + n, + record_hash(entry), + &HashReaderWithOverlay { + edge_tiles: &edge_tiles, + overlay: &overlay, + }, + ) + .map_err(|e| Error::from(format!("couldn't compute hashes for leaf {n}: {e}")))?; + for (i, h) in hashes.iter().enumerate() { + overlay.insert(stored_hash_index(0, n) + i as u64, *h); + } + n += 1; + if n.is_multiple_of(TILE_WIDTH) { + bundle_uploads.push(upload_entry_bundle( + object, + n, + std::mem::take(&mut data_tile), + )); + } + } + debug_assert_eq!(n, target_size); + // Trailing partial entry bundle. + if !target_size.is_multiple_of(TILE_WIDTH) { + bundle_uploads.push(upload_entry_bundle( + object, + target_size, + std::mem::take(&mut data_tile), + )); + } + stream_iter(bundle_uploads) + .buffer_unordered(UPLOAD_CONCURRENCY) + .try_collect::<()>() + .await?; + + // (Re)compute hash tiles, then upload them with bounded concurrency + // while keeping edge_tiles current for the final root hash. + let tile_opts = immutable_tile_opts(); + let mut hash_uploads = Vec::new(); + for tile in TlogTile::new_tiles(persisted_size, target_size) { + let bytes = tile + .read_data(&HashReaderWithOverlay { + edge_tiles: &edge_tiles, + overlay: &overlay, + }) + .map_err(|e| Error::from(format!("couldn't build hash tile {tile:?}: {e}")))?; + edge_tiles.insert( + tile.level(), + TileWithBytes { + tile, + b: bytes.clone(), + }, + ); + hash_uploads.push(object.upload(tile.path(), bytes, &tile_opts)); + } + stream_iter(hash_uploads) + .buffer_unordered(UPLOAD_CONCURRENCY) + .try_collect::<()>() + .await?; + + // Recompute the root hash from the frontier we just built. + tlog_core::tree_hash( + target_size, + &HashReaderWithOverlay { + edge_tiles: &edge_tiles, + overlay: &overlay, + }, + ) + .map_err(|e| Error::from(format!("couldn't compute root hash: {e}"))) +} + +/// Write the mirror's served checkpoint object. Unlike tiles the +/// checkpoint is mutable (it advances as the mirror commits), so it is +/// stored with `no-store` caching (`immutable: false`). +/// +/// `bytes` MUST be the checkpoint note the mirror serves at +/// `//checkpoint`: the origin log's signed +/// checkpoint with the mirror's own cosignature line(s) appended. +/// +/// # Errors +/// +/// Returns an error if the storage write fails. +pub(crate) async fn write_checkpoint(object: &impl ObjectBackend, bytes: Vec) -> Result<()> { + object + .upload( + CHECKPOINT_KEY, + bytes, + &UploadOptions { + content_type: Some("text/plain; charset=utf-8".to_owned()), + immutable: false, + }, + ) + .await +} + +/// Upload one entry bundle (data tile). `n` is the tree size after the +/// bundle's last entry, so the bundle covers leaves ending at `n - 1`. +async fn upload_entry_bundle(object: &impl ObjectBackend, n: u64, bytes: Vec) -> Result<()> { + let tile = TlogTile::from_index(stored_hash_index(0, n - 1)).with_data_path(PathElem::Entries); + object + .upload(tile.path(), bytes, &immutable_tile_opts()) + .await +} + +#[cfg(test)] +mod tests { + use super::*; + use std::cell::RefCell; + use std::collections::HashMap; + use tlog_core::{EMPTY_HASH, HashReader, TlogError, stored_hashes}; + use tlog_tiles::{PreloadedTlogTileReader, TileHashReader, TlogTileRecorder}; + + /// In-memory [`ObjectBackend`] for exercising the commit path without + /// a live R2 bucket. + #[derive(Default)] + struct MemBackend { + store: RefCell>>, + } + + impl ObjectBackend for MemBackend { + async fn upload, D: Into>>( + &self, + key: S, + data: D, + _opts: &UploadOptions, + ) -> Result<()> { + self.store + .borrow_mut() + .insert(key.as_ref().to_owned(), data.into()); + Ok(()) + } + + async fn fetch>(&self, key: S) -> Result>> { + Ok(self.store.borrow().get(key.as_ref()).cloned()) + } + } + + /// Deterministic distinct entry bytes for leaf `i`. + fn entry(i: u64) -> Vec { + format!("entry-{i}").into_bytes() + } + + /// A [`HashReader`] over an in-memory stored-hash map, used to build + /// the reference tree hash independently of the commit path. + struct MapReader<'a>(&'a HashMap); + impl HashReader for MapReader<'_> { + fn read_hashes(&self, idx: &[u64]) -> std::result::Result, TlogError> { + idx.iter() + .map(|i| self.0.get(i).copied().ok_or(TlogError::IndexesNotInTree)) + .collect() + } + } + + /// Compute the reference tree hash for the first `n` leaves by + /// building a full in-memory stored-hash map with [`tlog_core`], so + /// we can check that what the commit path stored is correct. + fn reference_root(n: u64) -> Hash { + let mut store: HashMap = HashMap::new(); + for i in 0..n { + let hashes = stored_hashes(i, &entry(i), &MapReader(&store)).unwrap(); + for (j, h) in hashes.iter().enumerate() { + store.insert(stored_hash_index(0, i) + j as u64, *h); + } + } + tlog_core::tree_hash(n, &MapReader(&store)).unwrap() + } + + fn leaves(range: std::ops::Range) -> Vec> { + range.map(entry).collect() + } + + #[tokio::test] + async fn commit_from_empty_matches_reference() { + let obj = MemBackend::default(); + // 300 leaves crosses one full tile (256) + a 44-wide partial. + let root = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + assert_eq!(root, reference_root(300)); + } + + #[tokio::test] + async fn incremental_commit_reads_persisted_frontier() { + let obj = MemBackend::default(); + // First commit to a mid-tile size (300). + let root0 = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + assert_eq!(root0, reference_root(300)); + + // Second commit continues from the mid-tile frontier, exercising + // read_edge_tiles + partial-entry-bundle reload from storage. + let root1 = persist_entries(&obj, 300, root0, 800, &leaves(300..800)) + .await + .unwrap(); + assert_eq!(root1, reference_root(800)); + } + + #[tokio::test] + async fn incremental_commit_rejects_tampered_partial_bundle() { + let obj = MemBackend::default(); + // First commit to a mid-tile size (300), leaving a partial bundle + // [256, 300) that the next commit reloads and extends. + let root0 = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + + // Corrupt a byte inside an entry of the persisted partial bundle + // while keeping its length framing valid, so it still decodes but no + // longer matches its authenticated leaf hash. + let key = TlogTile::from_index(stored_hash_index(0, 299)) + .with_data_path(PathElem::Entries) + .path(); + let mut bytes = obj + .fetch(&key) + .await + .unwrap() + .expect("partial bundle stored"); + let last = bytes.len() - 1; + bytes[last] ^= 0xff; + obj.upload(&key, bytes, &immutable_tile_opts()) + .await + .unwrap(); + + // The next incremental commit reloads the tampered bundle and must + // reject it rather than re-serve unverified bytes. + assert!( + persist_entries(&obj, 300, root0, 800, &leaves(300..800)) + .await + .is_err(), + "tampered partial bundle must be rejected" + ); + } + + #[tokio::test] + async fn incremental_commit_rejects_truncated_partial_bundle() { + let obj = MemBackend::default(); + let root0 = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + + // Drop the trailing bytes of the partial bundle: it no longer + // decodes to the full [256, 300) range. + let key = TlogTile::from_index(stored_hash_index(0, 299)) + .with_data_path(PathElem::Entries) + .path(); + let mut bytes = obj + .fetch(&key) + .await + .unwrap() + .expect("partial bundle stored"); + bytes.truncate(bytes.len() - 3); + obj.upload(&key, bytes, &immutable_tile_opts()) + .await + .unwrap(); + + assert!( + persist_entries(&obj, 300, root0, 800, &leaves(300..800)) + .await + .is_err(), + "truncated partial bundle must be rejected" + ); + } + + #[tokio::test] + async fn entry_bundles_roundtrip() { + use length_prefixed::ReadLengthPrefixedBytesExt as _; + let obj = MemBackend::default(); + persist_entries(&obj, 0, EMPTY_HASH, 260, &leaves(0..260)) + .await + .unwrap(); + + // First full bundle: leaves [0, 256). + let full = TlogTile::from_index(stored_hash_index(0, 255)) + .with_data_path(PathElem::Entries) + .path(); + let bytes = obj.fetch(&full).await.unwrap().expect("full bundle stored"); + let mut cur: &[u8] = &bytes; + for i in 0..256u64 { + let got = cur.read_length_prefixed(2).unwrap(); + assert_eq!(got, entry(i), "leaf {i} mismatch in full bundle"); + } + assert!(cur.is_empty(), "full bundle has trailing bytes"); + + // Trailing partial bundle: leaves [256, 260). + let partial = TlogTile::from_index(stored_hash_index(0, 259)) + .with_data_path(PathElem::Entries) + .path(); + let bytes = obj + .fetch(&partial) + .await + .unwrap() + .expect("partial bundle stored"); + let mut cur: &[u8] = &bytes; + for i in 256..260u64 { + let got = cur.read_length_prefixed(2).unwrap(); + assert_eq!(got, entry(i), "leaf {i} mismatch in partial bundle"); + } + assert!(cur.is_empty(), "partial bundle has trailing bytes"); + } + + #[tokio::test] + async fn read_persisted_leaves_from_full_and_partial_bundles() { + let obj = MemBackend::default(); + // 300 leaves: one full bundle [0,256) + a partial bundle [256,300). + let root = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + + // Prefix within the full bundle (persisted_size 300 -> stored + // width 256 for tile 0). + let got = read_persisted_leaves(&obj, 0, 44, 300, root).await.unwrap(); + assert_eq!(got, leaves(0..44)); + + // Whole full bundle. + let got = read_persisted_leaves(&obj, 0, 256, 300, root) + .await + .unwrap(); + assert_eq!(got, leaves(0..256)); + + // Prefix within the trailing partial bundle (tile 1, stored width + // 300 - 256 = 44). + let got = read_persisted_leaves(&obj, 256, 20, 300, root) + .await + .unwrap(); + assert_eq!(got, leaves(256..276)); + } + + #[tokio::test] + async fn read_persisted_leaves_missing_bundle_errors() { + let obj = MemBackend::default(); + assert!( + read_persisted_leaves(&obj, 0, 10, 300, EMPTY_HASH) + .await + .is_err() + ); + } + + #[tokio::test] + async fn read_persisted_leaves_rejects_tampered_bundle() { + let obj = MemBackend::default(); + let root = persist_entries(&obj, 0, EMPTY_HASH, 300, &leaves(0..300)) + .await + .unwrap(); + + // Corrupt a byte inside the full bundle [0,256) while keeping the + // length framing valid, so it still decodes but no longer matches + // its authenticated leaf hash. + let key = TlogTile::from_index(stored_hash_index(0, 255)) + .with_data_path(PathElem::Entries) + .path(); + let mut bytes = obj.fetch(&key).await.unwrap().expect("full bundle stored"); + let last = bytes.len() - 1; + bytes[last] ^= 0xff; + obj.upload(&key, bytes, &immutable_tile_opts()) + .await + .unwrap(); + + assert!( + read_persisted_leaves(&obj, 0, 256, 300, root) + .await + .is_err(), + "tampered persisted bundle must be rejected" + ); + } + + #[tokio::test] + async fn hash_tiles_authenticate_against_root() { + let obj = MemBackend::default(); + let root = persist_entries(&obj, 0, EMPTY_HASH, 500, &leaves(0..500)) + .await + .unwrap(); + + // A TileHashReader over the stored tiles must authenticate an + // arbitrary leaf hash against the recomputed root. + let idx = [stored_hash_index(0, 499)]; + let recorder = TlogTileRecorder::default(); + let probe = TileHashReader::new(500, root, &recorder); + assert!(matches!( + probe.read_hashes(&idx), + Err(TlogError::RecordedTilesOnly) + )); + let mut fetched: HashMap> = HashMap::new(); + for tile in recorder.0.into_inner() { + let bytes = obj.fetch(tile.path()).await.unwrap().expect("tile stored"); + fetched.insert(tile, bytes); + } + let reader = PreloadedTlogTileReader(fetched); + let hash_reader = TileHashReader::new(500, root, &reader); + let got = hash_reader.read_hashes(&idx).expect("authenticates"); + assert_eq!(got[0], record_hash(&entry(499))); + } +} diff --git a/crates/mirror_worker/src/frontend_worker.rs b/crates/mirror_worker/src/frontend_worker.rs index a6fb6f7..c6bea37 100644 --- a/crates/mirror_worker/src/frontend_worker.rs +++ b/crates/mirror_worker/src/frontend_worker.rs @@ -29,7 +29,7 @@ use axum::{ Json, Router, body::Bytes, extract::{DefaultBodyLimit, State}, - http::{StatusCode, header}, + http::{HeaderValue, StatusCode, header}, response::IntoResponse, routing::{get, post}, }; @@ -56,6 +56,22 @@ fn start() { let _ = console_log::init_with_level(level); } +/// Middleware that adds `Accept-Encoding: gzip` to every response. +/// +/// [c2sp.org/tlog-mirror][spec] says mirrors SHOULD advertise supported +/// compression algorithms in responses so clients can compress future +/// `add-entries` request bodies. +/// +/// [spec]: https://c2sp.org/tlog-mirror#add-entries +async fn add_accept_encoding( + mut response: axum::http::Response, +) -> axum::http::Response { + response + .headers_mut() + .insert(header::ACCEPT_ENCODING, HeaderValue::from_static("gzip")); + response +} + /// Top-level `#[event(fetch)]` handler. Delegates to the axum router; /// unmatched routes return 404. #[event(fetch, respond_with_errors)] @@ -69,14 +85,20 @@ async fn fetch( // handler is captured and shipped before the WASM isolate is torn // down. `Router`'s `Service::Error` is `Infallible`; the `?` below // performs the trivial conversion into `worker::Error`. + // + // `/add-entries` streams a potentially large (optionally gzip) body, + // so it uses the raw `Request` extractor with no `DefaultBodyLimit`; + // the buffered endpoints cap their bodies via the layer. let response = generic_log_worker::obs::sentry::catch_unwind_and_flush(async { Router::new() .route( "/add-checkpoint", post(add_checkpoint).layer(DefaultBodyLimit::max(MAX_ADD_CHECKPOINT_BODY_SIZE)), ) + .route("/add-entries", post(crate::add_entries::add_entries)) .route("/metadata", get(metadata)) .route("/", get(root)) + .layer(axum::middleware::map_response(add_accept_encoding)) .with_state(env) .call(req) .await @@ -97,16 +119,20 @@ async fn root() -> impl IntoResponse { } /// Error type for the mirror's axum handlers, mapped to an HTTP status by -/// [`IntoResponse`]. -enum AppError { +/// [`IntoResponse`]. Success and special-body responses (the 200 +/// cosignature, the 409/202 `text/x.tlog.mirror-info` and `text/x.tlog.size` +/// bodies) are built as axum responses directly, not via this enum. +pub(crate) enum AppError { InternalServerError(String), BadRequest(String), + UnsupportedMediaType(String), + UnprocessableEntity(String), UnknownLogOrigin, NoValidSignatures, } /// Result type for the mirror's axum handlers. -type ApiResult = std::result::Result; +pub(crate) type ApiResult = std::result::Result; impl From for AppError { fn from(err: worker::Error) -> Self { @@ -124,6 +150,14 @@ impl IntoResponse for AppError { AppError::BadRequest(e) => { (StatusCode::BAD_REQUEST, format!("Bad request: {e}")).into_response() } + AppError::UnsupportedMediaType(e) => { + (StatusCode::UNSUPPORTED_MEDIA_TYPE, e).into_response() + } + AppError::UnprocessableEntity(e) => ( + StatusCode::UNPROCESSABLE_ENTITY, + format!("Unprocessable Entity: {e}"), + ) + .into_response(), AppError::UnknownLogOrigin => { (StatusCode::NOT_FOUND, "Unknown log origin").into_response() } @@ -397,3 +431,19 @@ fn tlog_size_conflict(current: &PendingCheckpoint) -> axum::response::Response { ) .into_response() } + +#[cfg(test)] +mod tests { + use super::add_accept_encoding; + use axum::http::header::ACCEPT_ENCODING; + + #[tokio::test] + async fn accept_encoding_middleware_adds_gzip() { + let response = axum::http::Response::new(axum::body::Body::empty()); + let response = add_accept_encoding(response).await; + assert_eq!( + response.headers().get(ACCEPT_ENCODING).unwrap().as_bytes(), + b"gzip" + ); + } +} diff --git a/crates/mirror_worker/src/lib.rs b/crates/mirror_worker/src/lib.rs index a839e00..54af74a 100644 --- a/crates/mirror_worker/src/lib.rs +++ b/crates/mirror_worker/src/lib.rs @@ -4,9 +4,11 @@ //! A transparency-log mirror implementing [c2sp.org/tlog-mirror][mirror] on //! Cloudflare Workers, specialized for MTC issuance logs. //! -//! This worker handles the [`add-checkpoint`][add-cp] submission endpoint, -//! which updates the pending checkpoint for a log origin, and publishes -//! the mirror's identity and per-log configuration at `/metadata`. +//! This worker handles the [`add-checkpoint`][add-cp] and +//! [`add-entries`][add-e] submission endpoints, and publishes the mirror's +//! identity and per-log configuration at `/metadata`. The +//! [tlog-tiles][tiles] read interface is served directly from object +//! storage (see the `storage` module). //! //! Per-origin persistent state lives in a `MirrorState` Durable Object, //! one per log origin. Its single-threaded execution model provides the @@ -15,7 +17,10 @@ //! //! [mirror]: https://c2sp.org/tlog-mirror //! [add-cp]: https://c2sp.org/tlog-mirror#add-checkpoint +//! [add-e]: https://c2sp.org/tlog-mirror#add-entries +//! [tiles]: https://c2sp.org/tlog-tiles +use base64::Engine as _; use config::AppConfig; use ml_dsa::pkcs8::{DecodePrivateKey as _, EncodePublicKey as _}; use ml_dsa::{MlDsa44, VerifyingKey as MlDsaVerifyingKey}; @@ -23,12 +28,17 @@ use pkcs8::{PrivateKeyInfoRef, SecretDocument, der::oid::db::fips204::ID_ML_DSA_ use signed_note::{KeyName, NoteVerifier, VerifierList}; use std::collections::HashMap; use std::sync::{Arc, LazyLock, OnceLock}; -use tlog_cosignature::SubtreeV1NoteVerifier; +use tlog_cosignature::{SubtreeV1CheckpointSigner, SubtreeV1NoteVerifier}; +use tlog_mirror::TicketSealer; #[allow(clippy::wildcard_imports)] use worker::*; +mod add_entries; +mod body; +mod commit; mod frontend_worker; mod mirror_state_do; +mod storage; /// The binding name used in `wrangler.jsonc` for the `MirrorState` DO. pub(crate) const MIRROR_STATE_BINDING: &str = "MIRROR_STATE"; @@ -124,12 +134,15 @@ pub(crate) fn log_verifiers(origin: &str) -> Option { /// /// The mirror is an MTC cosigner, which per [c2sp.org/mtc-tlog][mtc] MUST /// use an ML-DSA-44 key and produce [`subtree/v1`][cosig] messages, so -/// this worker supports only that algorithm. Holds the DER-encoded -/// `SubjectPublicKeyInfo` computed once at load and served by `/metadata`. +/// this worker supports only that algorithm. Holds the signer plus the +/// DER-encoded `SubjectPublicKeyInfo` computed once at load and served by +/// `/metadata`. The signer is boxed because the expanded ML-DSA-44 key is +/// large (~64 KiB). /// /// [mtc]: https://c2sp.org/mtc-tlog /// [cosig]: https://c2sp.org/tlog-cosignature pub(crate) struct MirrorSigner { + signer: Box, public_key_der: Vec, } @@ -147,6 +160,15 @@ impl MirrorSigner { pub(crate) fn algorithm(&self) -> &'static str { "subtree/v1" } + + /// The inner [`CheckpointSigner`] trait object, used by the + /// `add-entries` handler to emit the mirror's checkpoint cosignature + /// on a successful upload. + /// + /// [`CheckpointSigner`]: tlog_checkpoint::CheckpointSigner + pub(crate) fn as_checkpoint_signer(&self) -> &dyn tlog_checkpoint::CheckpointSigner { + &*self.signer + } } /// Cached mirror signer, so the PKCS#8 parse happens at most once per @@ -178,6 +200,8 @@ pub(crate) fn load_mirror_signer(env: &Env) -> Result<&'static MirrorSigner> { /// any other algorithm is rejected (the mirror's cosigner must be an MTC /// cosigner, see [`MirrorSigner`]). fn build_mirror_signer(pem: &str) -> Result { + let name = KeyName::new(CONFIG.mirror_name.clone()) + .map_err(|e| Error::from(format!("invalid mirror_name: {e:?}")))?; let (_label, doc) = SecretDocument::from_pem(pem).map_err(|e| Error::from(format!("PEM parse: {e}")))?; let pk_info = PrivateKeyInfoRef::try_from(doc.as_bytes()) @@ -185,7 +209,8 @@ fn build_mirror_signer(pem: &str) -> Result { match pk_info.algorithm.oid { ID_ML_DSA_44 => { // ml-dsa's PKCS#8 stores only the 32-byte seed; `from_pkcs8_der` - // expands it on the way in. + // expands it on the way in. The expanded key never leaves this + // worker. let expanded = ml_dsa::ExpandedSigningKey::::from_pkcs8_der(doc.as_bytes()) .map_err(|e| Error::from(format!("ML-DSA-44 PKCS#8 parse: {e}")))?; let public_key_der = expanded @@ -193,7 +218,10 @@ fn build_mirror_signer(pem: &str) -> Result { .to_public_key_der() .map_err(|e| Error::from(format!("ML-DSA-44 SPKI encode: {e}")))? .to_vec(); - Ok(MirrorSigner { public_key_der }) + Ok(MirrorSigner { + signer: Box::new(SubtreeV1CheckpointSigner::new(name, expanded)), + public_key_der, + }) } oid => Err(Error::from(format!( "unsupported MIRROR_SIGNING_KEY algorithm OID {oid}: expected id-ml-dsa-44 \ @@ -213,6 +241,48 @@ pub(crate) fn load_mirror_public_key_der(env: &Env) -> Result<&'static [u8]> { Ok(load_mirror_signer(env)?.public_key_der()) } +// --------------------------------------------------------------------------- +// Ticket key +// --------------------------------------------------------------------------- + +/// Cached ticket authenticator, built lazily on first request. +/// +/// The mirror's ticket scheme (base64 blobs returned in the +/// `text/x.tlog.mirror-info` 409 response body and round-tripped via the +/// `add-entries` request) is sealed with AES-256-GCM-SIV, a fresh random +/// nonce per ticket, with the log origin bound as associated data. See +/// [`tlog_mirror::TicketSealer`]; this static holds a single instance +/// keyed off the `MIRROR_TICKET_KEY` secret. +static TICKET_SEALER: OnceLock = OnceLock::new(); + +/// Load (or return the already-cached) ticket authenticator. +/// +/// The `MIRROR_TICKET_KEY` secret is 32 raw bytes encoded as standard +/// base64 (RFC 4648 section 4, no URL-safe variant). Operators can +/// generate one with `head -c 32 /dev/urandom | base64` and load it via +/// `wrangler secret put MIRROR_TICKET_KEY`. +/// +/// # Errors +/// +/// Returns an error if the `MIRROR_TICKET_KEY` secret is missing, is not +/// valid base64, or does not decode to exactly 32 bytes. +pub(crate) fn load_ticket_sealer(env: &Env) -> Result<&'static TicketSealer> { + if let Some(t) = TICKET_SEALER.get() { + return Ok(t); + } + let b64 = env.secret("MIRROR_TICKET_KEY")?.to_string(); + let raw = base64::engine::general_purpose::STANDARD + .decode(b64.trim()) + .map_err(|e| Error::from(format!("MIRROR_TICKET_KEY base64 decode: {e}")))?; + let key: [u8; 32] = raw.try_into().map_err(|v: Vec| { + Error::from(format!( + "MIRROR_TICKET_KEY must decode to exactly 32 bytes; got {}", + v.len() + )) + })?; + Ok(TICKET_SEALER.get_or_init(|| TicketSealer::new(&key))) +} + /// Initialize sentry from the `SENTRY_DSN` environment variable. /// /// Does nothing when the variable is absent or empty, allowing @@ -393,4 +463,20 @@ mod dev_config_tests { "dev MIRROR_SIGNING_KEY must load as a subtree/v1 signer", ); } + + /// `MIRROR_TICKET_KEY` in `.dev.vars` is base64 of exactly 32 bytes, + /// the precondition for [`crate::load_ticket_sealer`]. + #[test] + fn dev_vars_ticket_key_is_32_bytes_base64() { + let b64 = dev_var("MIRROR_TICKET_KEY"); + let raw = BASE64_STANDARD + .decode(b64.trim()) + .expect("MIRROR_TICKET_KEY must be valid base64"); + assert_eq!( + raw.len(), + 32, + "MIRROR_TICKET_KEY must decode to exactly 32 bytes; got {}", + raw.len() + ); + } } diff --git a/crates/mirror_worker/src/mirror_state_do.rs b/crates/mirror_worker/src/mirror_state_do.rs index eace124..1d88741 100644 --- a/crates/mirror_worker/src/mirror_state_do.rs +++ b/crates/mirror_worker/src/mirror_state_do.rs @@ -28,10 +28,11 @@ use serde::{Deserialize, Serialize}; use serde_with::{base64::Base64 as Base64As, serde_as}; use tlog_core::{Hash, verify_consistency_proof}; +use tokio::sync::Mutex; #[allow(clippy::wildcard_imports)] use worker::*; -use crate::MIRROR_STATE_BINDING; +use crate::{MIRROR_STATE_BINDING, commit, storage::load_origin_bucket}; const PENDING_KEY: &str = "pending"; const COMMITTED_KEY: &str = "committed"; @@ -74,8 +75,9 @@ pub struct CommittedCheckpoint { /// Root hash. All-zero if `size` is 0. #[serde(with = "hash_hex")] pub hash: Hash, - /// Full signed-note bytes as the log signed them, empty if `size` is - /// 0. Served with the mirror's cosignature lines appended. + /// The served checkpoint bytes: the log's signed note with the + /// mirror's cosignature line(s) appended, exactly as written to R2. + /// Empty if `size` is 0. #[serde_as(as = "Base64As")] pub signed_note_bytes: Vec, } @@ -169,6 +171,16 @@ pub struct UpdatePendingRequest { #[durable_object(fetch)] struct MirrorState { state: State, + /// Held to resolve the origin's R2 bucket when the DO writes the + /// served checkpoint on `/commit`. + env: Env, + /// Serializes `/commit` so the durable-storage advance and the R2 + /// served-checkpoint write happen atomically with respect to + /// concurrent commits, even across the external R2 write's await. + /// Without it a slower commit could overwrite the served checkpoint + /// with an older one after a concurrent commit already advanced it. + /// Mirrors the sequencer's `init_mux` (see `generic_log_worker`). + commit_mux: Mutex<()>, } // SAFETY: Durable Objects are single-threaded; the `RefUnwindSafe` bound @@ -179,7 +191,11 @@ impl std::panic::RefUnwindSafe for MirrorState {} impl DurableObject for MirrorState { fn new(state: State, env: Env) -> Self { crate::init_sentry(&env); - Self { state } + Self { + state, + env, + commit_mux: Mutex::new(()), + } } async fn fetch(&self, req: Request) -> Result { @@ -200,42 +216,8 @@ impl MirrorState { Response::from_json(&snapshot) } (Method::Post, "/commit") => { - // Compare-and-swap advance of the mirror checkpoint: - // * size > next_entry.size: commit beyond the - // persisted-entry frontier, i.e. cosigning entries - // that have not been durably written yet; 400. This - // preserves `committed.size <= next_entry.size`. Since - // `next_entry.size <= pending.size`, it also rejects - // commits beyond the accepted pending checkpoint. - // * size < committed.size: a concurrent add-entries - // already advanced past us. Spec forbids rolling - // back, so no-op and return the current state. - // * otherwise: advance committed. let body: CommitRequest = req.json().await?; - let snapshot = self.read_snapshot().await?; - if body.size > snapshot.next_entry.size { - return Response::error( - format!( - "commit beyond persisted-entry frontier: requested size {} > next_entry size {}", - body.size, snapshot.next_entry.size - ), - 400, - ); - } - if body.size < snapshot.committed.size { - // Already ahead; no-op success. - return Response::from_json(&snapshot.committed); - } - let new_committed = CommittedCheckpoint { - size: body.size, - hash: body.hash, - signed_note_bytes: body.signed_note_bytes, - }; - self.state - .storage() - .put(COMMITTED_KEY, &new_committed) - .await?; - Response::from_json(&new_committed) + self.commit(body).await } (Method::Post, "/advance-next-entry") => { let body: AdvanceNextEntryRequest = req.json().await?; @@ -308,6 +290,73 @@ impl MirrorState { } impl MirrorState { + /// Handle `/commit`: monotonically advance the mirror checkpoint and + /// write the served checkpoint object to R2. + /// + /// `commit_mux` serializes the whole read-check-advance-write + /// sequence, including the external R2 write, so concurrent commits + /// cannot interleave and rewind the served checkpoint. Compare-and-swap + /// semantics: + /// * `size > next_entry.size`: commit beyond the persisted-entry + /// frontier (cosigning entries not yet durably written); 400. This + /// preserves `committed.size <= next_entry.size`, and since + /// `next_entry.size <= pending.size` it also rejects commits beyond + /// the accepted pending checkpoint. + /// * `size < committed.size`: a concurrent add-entries already + /// advanced past us. The spec forbids rolling back, so no-op and + /// return the current committed checkpoint (whose served object the + /// concurrent commit already wrote). + /// * otherwise: advance committed in durable storage, then write the + /// served checkpoint to R2. + /// + /// Durable storage is advanced before the R2 write so the served + /// checkpoint is never ahead of committed; a failed R2 write leaves R2 + /// lagging and is rewritten by the next commit. + async fn commit(&self, body: CommitRequest) -> Result { + let _guard = self.commit_mux.lock().await; + + let snapshot = self.read_snapshot().await?; + if body.size > snapshot.next_entry.size { + return Response::error( + format!( + "commit beyond persisted-entry frontier: requested size {} > next_entry size {}", + body.size, snapshot.next_entry.size + ), + 400, + ); + } + if body.size < snapshot.committed.size { + // Already ahead; no-op success. The concurrent commit that + // advanced past us already wrote the newer served checkpoint. + return Response::from_json(&snapshot.committed); + } + + let new_committed = CommittedCheckpoint { + size: body.size, + hash: body.hash, + signed_note_bytes: body.signed_note_bytes, + }; + self.state + .storage() + .put(COMMITTED_KEY, &new_committed) + .await?; + + // Write the served checkpoint (the log's signed note plus the + // mirror's cosignature) to R2 at + // //checkpoint. The DO owns this write so + // it stays serialized with the durable advance above; the origin is + // the DO's own name. + let origin = self + .state + .id() + .name() + .ok_or_else(|| Error::from("mirror state DO missing origin name"))?; + let bucket = load_origin_bucket(&self.env, &origin)?; + commit::write_checkpoint(&bucket, new_committed.signed_note_bytes.clone()).await?; + + Response::from_json(&new_committed) + } + /// Handle `/advance-next-entry`: monotonically advance the /// persisted-entry frontier. Compare-and-swap, like `/commit`: /// diff --git a/crates/mirror_worker/src/storage.rs b/crates/mirror_worker/src/storage.rs new file mode 100644 index 0000000..4105eb8 --- /dev/null +++ b/crates/mirror_worker/src/storage.rs @@ -0,0 +1,108 @@ +// Copyright (c) 2025-2026 Cloudflare, Inc. All rights reserved. +// SPDX-License-Identifier: BSD-3-Clause + +//! Object storage for the mirrored copy of each origin log. +//! +//! A single R2 bucket (bound as [`PUBLIC_BUCKET_BINDING`]) backs every +//! origin the mirror serves. Objects for a given origin are stored under +//! a distinct `/` key prefix, where `origin hash` is the +//! lowercase hex SHA-256 of the log's origin, the same identifier the +//! [c2sp.org/tlog-mirror][spec] monitoring interface uses in its URL +//! layout (`//...`; see the spec +//! [Introduction][intro]). +//! Keeping the storage prefix identical to the served path means the read +//! interface can proxy R2 keys to URLs without +//! translation. +//! +//! [`OriginBucket`] is a thin [`ObjectBackend`] adapter that prepends the +//! prefix to every key, so the commit path can use bare +//! [tlog-tiles][tiles] paths (`checkpoint`, `tile/...`, `tile/entries/...`) +//! exactly as [`generic_log_worker`] and [`tlog_tiles`] produce them. +//! +//! [spec]: https://c2sp.org/tlog-mirror +//! [intro]: https://c2sp.org/tlog-mirror#introduction +//! [tiles]: https://c2sp.org/tlog-tiles + +use generic_log_worker::{ObjectBackend, ObjectBucket, log_ops::UploadOptions}; +use sha2::{Digest as _, Sha256}; +#[allow(clippy::wildcard_imports)] +use worker::*; + +/// `wrangler.jsonc` binding name for the mirror's public R2 bucket. +pub(crate) const PUBLIC_BUCKET_BINDING: &str = "PUBLIC_BUCKET"; + +/// Compute a log's *origin hash*: the lowercase hex-encoded SHA-256 of +/// the origin string, per [c2sp.org/tlog-mirror][spec]. Used both as the +/// R2 key prefix and (by the read interface) as the monitoring URL path +/// component. +/// +/// [spec]: https://c2sp.org/tlog-mirror +#[must_use] +pub(crate) fn origin_hash(origin: &str) -> String { + hex::encode(Sha256::digest(origin.as_bytes())) +} + +/// An [`ObjectBackend`] that stores every object under a single origin's +/// `/` key prefix in the shared public bucket. +/// +/// Constructed per request via [`load_origin_bucket`]. Callers pass bare +/// tlog-tiles paths; the prefix is applied transparently. +pub(crate) struct OriginBucket { + inner: ObjectBucket, + prefix: String, +} + +impl OriginBucket { + /// Prepend the origin prefix to a bare tlog-tiles key. + fn prefixed(&self, key: &str) -> String { + format!("{}{key}", self.prefix) + } +} + +impl ObjectBackend for OriginBucket { + async fn upload, D: Into>>( + &self, + key: S, + data: D, + opts: &UploadOptions, + ) -> Result<()> { + self.inner + .upload(self.prefixed(key.as_ref()), data, opts) + .await + } + + async fn fetch>(&self, key: S) -> Result>> { + self.inner.fetch(self.prefixed(key.as_ref())).await + } +} + +/// Build an [`OriginBucket`] for `origin`, backed by the shared public R2 +/// bucket bound as [`PUBLIC_BUCKET_BINDING`]. +/// +/// # Errors +/// +/// Returns an error if the `PUBLIC_BUCKET` binding is missing or not an +/// R2 bucket. +pub(crate) fn load_origin_bucket(env: &Env, origin: &str) -> Result { + let bucket = env.bucket(PUBLIC_BUCKET_BINDING)?; + Ok(OriginBucket { + inner: ObjectBucket::new(bucket), + prefix: format!("{}/", origin_hash(origin)), + }) +} + +#[cfg(test)] +mod tests { + use super::origin_hash; + + /// Pin the origin-hash construction against a known SHA-256 vector so + /// the storage prefix (and, later, the monitoring URL path) can't + /// drift. `echo -n "example.com/log1" | sha256sum`. + #[test] + fn origin_hash_matches_known_vector() { + assert_eq!( + origin_hash("example.com/log1"), + "82df480cc8e80fed3584d9ac8520c582266fcefbb4257d4c758a0efa6bad9c95" + ); + } +} diff --git a/crates/mirror_worker/wrangler.jsonc b/crates/mirror_worker/wrangler.jsonc index fe5de69..75f4786 100644 --- a/crates/mirror_worker/wrangler.jsonc +++ b/crates/mirror_worker/wrangler.jsonc @@ -29,6 +29,17 @@ } ] }, + // Public R2 bucket backing the mirrored copy of every origin + // log. The mirror stores each origin under a distinct + // `/` key prefix (see `crate::storage`), so a + // single bucket serves all configured logs. Served publicly + // via the c2sp.org/tlog-tiles read interface. + "r2_buckets": [ + { + "bucket_name": "tlog-mirror-public-dev", + "binding": "PUBLIC_BUCKET" + } + ], "migrations": [ { "tag": "v1",