diff --git a/Cargo.lock b/Cargo.lock index 35918b5..baa4c6d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1856,13 +1856,25 @@ name = "mirror_worker" version = "0.2.0" dependencies = [ "axum", + "base64", "console_error_panic_hook", "console_log", + "ed25519-dalek", "getrandom 0.4.2", + "hex", "jsonschema", "log", "mirror_worker_config", + "ml-dsa", + "pkcs8", + "serde", "serde_json", + "serde_with", + "signed_note", + "tlog_checkpoint", + "tlog_core", + "tlog_cosignature", + "tlog_witness", "tower-service", "worker", ] diff --git a/crates/mirror_worker/.dev.vars b/crates/mirror_worker/.dev.vars new file mode 100644 index 0000000..c0d8d88 --- /dev/null +++ b/crates/mirror_worker/.dev.vars @@ -0,0 +1 @@ +MIRROR_SIGNING_KEY="-----BEGIN PRIVATE KEY-----\nMDQCAQAwCwYJYIZIAWUDBAMRBCKAIEJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJC\nQkJCQkJC\n-----END PRIVATE KEY-----\n" diff --git a/crates/mirror_worker/Cargo.toml b/crates/mirror_worker/Cargo.toml index 52e507d..5496ee6 100644 --- a/crates/mirror_worker/Cargo.toml +++ b/crates/mirror_worker/Cargo.toml @@ -36,11 +36,28 @@ config = { path = "./config", package = "mirror_worker_config" } console_error_panic_hook.workspace = true console_log.workspace = true getrandom.workspace = true +hex.workspace = true log.workspace = true +ml-dsa.workspace = true +pkcs8.workspace = true +serde.workspace = true serde_json.workspace = true +serde_with.workspace = true +signed_note.workspace = true +tlog_checkpoint.workspace = true +tlog_core.workspace = true +tlog_cosignature.workspace = true +tlog_witness.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-dalek.workspace = true + [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = [ 'cfg(wasm_bindgen_unstable_test_coverage)', diff --git a/crates/mirror_worker/src/frontend_worker.rs b/crates/mirror_worker/src/frontend_worker.rs index e62d2de..0581cf3 100644 --- a/crates/mirror_worker/src/frontend_worker.rs +++ b/crates/mirror_worker/src/frontend_worker.rs @@ -1,17 +1,43 @@ // Copyright (c) 2025-2026 Cloudflare, Inc. All rights reserved. // SPDX-License-Identifier: BSD-3-Clause -//! HTTP entry point for the mirror worker. +//! HTTP entry point + handlers for the mirror worker. //! -//! Serves the root status route; unmatched routes return 404. +//! Routes: +//! +//! - `POST /add-checkpoint`: [c2sp.org/tlog-mirror#add-checkpoint][add-cp]. +//! Updates the pending checkpoint for an origin. Wire format and +//! response semantics are identical to the witness's `add-checkpoint`, +//! with one spec-mandated exception: the mirror MUST NOT cosign in +//! this process. Successful responses have an empty body and HTTP +//! status 200. +//! - `GET /metadata`: mirror identity, ML-DSA-44 SPKI, +//! `mirror_algorithm`, prefixes, and the per-log configuration. +//! - `GET /`: root status string. +//! +//! The mirror's per-origin persistent state lives in a [`MirrorState`] +//! Durable Object; see [`crate::mirror_state_do`] for details. +//! +//! [add-cp]: https://c2sp.org/tlog-mirror#add-checkpoint +//! [`MirrorState`]: crate::mirror_state_do -use crate::CONFIG; +use crate::{ + CONFIG, load_mirror_public_key_der, load_mirror_signer, log_verifiers, + mirror_state_do::{PendingCheckpoint, UpdatePendingRequest, state_stub}, +}; use axum::{ - Router, + Json, Router, + body::Bytes, + extract::{DefaultBodyLimit, State}, http::{StatusCode, header}, response::IntoResponse, - routing::get, + routing::{get, post}, }; +use serde::Serialize; +use serde_with::{base64::Base64 as Base64As, serde_as}; +use signed_note::NoteError; +use tlog_checkpoint::CheckpointText; +use tlog_witness::{AddCheckpointRequest, CONTENT_TYPE_TLOG_SIZE, parse_add_checkpoint_request}; use tower_service::Service as _; #[allow(clippy::wildcard_imports)] use worker::*; @@ -35,12 +61,21 @@ fn start() { #[event(fetch, respond_with_errors)] async fn fetch( req: HttpRequest, - _env: Env, + env: Env, _ctx: Context, ) -> Result> { // `Router`'s `Service::Error` is `Infallible`; `?` performs the // trivial conversion into `worker::Error`. - Ok(Router::new().route("/", get(root)).call(req).await?) + Ok(Router::new() + .route( + "/add-checkpoint", + post(add_checkpoint).layer(DefaultBodyLimit::max(MAX_ADD_CHECKPOINT_BODY_SIZE)), + ) + .route("/metadata", get(metadata)) + .route("/", get(root)) + .with_state(env) + .call(req) + .await?) } /// `GET /` -- mirror identity string. Convenience only; not part of the @@ -52,3 +87,305 @@ async fn root() -> impl IntoResponse { format!("{} - c2sp.org/tlog-mirror mirror\n", CONFIG.mirror_name), ) } + +/// Error type for the mirror's axum handlers, mapped to an HTTP status by +/// [`IntoResponse`]. +enum AppError { + InternalServerError(String), + BadRequest(String), + UnknownLogOrigin, + NoValidSignatures, +} + +/// Result type for the mirror's axum handlers. +type ApiResult = std::result::Result; + +impl From for AppError { + fn from(err: worker::Error) -> Self { + Self::InternalServerError(err.to_string()) + } +} + +impl IntoResponse for AppError { + fn into_response(self) -> axum::response::Response { + match self { + AppError::InternalServerError(error) => { + log::error!("unhandled error: {error}"); + StatusCode::INTERNAL_SERVER_ERROR.into_response() + } + AppError::BadRequest(e) => { + (StatusCode::BAD_REQUEST, format!("Bad request: {e}")).into_response() + } + AppError::UnknownLogOrigin => { + (StatusCode::NOT_FOUND, "Unknown log origin").into_response() + } + AppError::NoValidSignatures => ( + StatusCode::FORBIDDEN, + "No valid signatures from trusted log keys", + ) + .into_response(), + } + } +} + +/// Response body for the `/metadata` endpoint: the mirror's identity, +/// URL prefixes, and per-log configuration. The `mirror_algorithm` field +/// tells clients whether to expect `cosignature/v1` or `subtree/v1` +/// cosignatures. +#[serde_as] +#[derive(Serialize)] +struct MetadataResponse<'a> { + mirror_name: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + description: Option<&'a str>, + /// DER-encoded `SubjectPublicKeyInfo` for the mirror's verifying + /// key. Algorithm is identified by `mirror_algorithm`. + #[serde_as(as = "Base64As")] + mirror_public_key: &'a [u8], + /// Always `"subtree/v1"` (ML-DSA-44); the mirror's cosigner is an MTC + /// cosigner. See + /// [c2sp.org/tlog-cosignature](https://c2sp.org/tlog-cosignature). + mirror_algorithm: &'a str, + submission_prefix: &'a str, + monitoring_prefix: &'a str, + logs: Vec>, +} + +#[serde_as] +#[derive(Serialize)] +struct LogMetadata<'a> { + #[serde(skip_serializing_if = "Option::is_none")] + description: Option<&'a str>, + /// The note-signature name the log's trusted checkpoints carry (for + /// MTC, the CA cosigner ID). The concrete origins served for this + /// key are `.0.` for each `N` in + /// `[min_log_number, max_log_number]`, each addressable at + /// `//`. + log_key_name: &'a str, + /// Lowest MTC log number served for this key (inclusive). For a plain + /// (non-MTC) log this equals `max_log_number`. + min_log_number: u64, + /// Highest MTC log number served for this key (inclusive). + max_log_number: u64, + /// DER-encoded `SubjectPublicKeyInfo` blobs for the log's trusted + /// keys. + #[serde_as(as = "Vec")] + log_public_keys: Vec<&'a [u8]>, +} + +/// Build the per-log metadata entries, one per configured key, sorted by +/// `log_key_name` so the response is deterministic regardless of +/// `HashMap` iteration order (it may be diffed or hashed by monitors). +/// +/// MTC log-number windows are published as `[min_log_number, +/// max_log_number]` bounds; a client derives the concrete origins as +/// `.0.`. +fn metadata_logs( + logs: &std::collections::HashMap, +) -> Vec> { + let mut out: Vec = logs + .iter() + .map(|(log_key_name, p)| LogMetadata { + description: p.description.as_deref(), + log_key_name, + min_log_number: p.min_log_number, + max_log_number: p.max_log_number, + log_public_keys: p.log_public_keys.iter().map(Vec::as_slice).collect(), + }) + .collect(); + out.sort_by(|a, b| a.log_key_name.cmp(b.log_key_name)); + out +} + +/// `GET /metadata` handler. +#[worker::send] +async fn metadata(State(env): State) -> ApiResult { + let mirror_public_key = load_mirror_public_key_der(&env)?; + let mirror_algorithm = load_mirror_signer(&env)?.algorithm(); + let logs = metadata_logs(&CONFIG.logs); + let body = MetadataResponse { + mirror_name: &CONFIG.mirror_name, + description: CONFIG.description.as_deref(), + mirror_public_key, + mirror_algorithm, + submission_prefix: &CONFIG.submission_prefix, + monitoring_prefix: CONFIG + .monitoring_prefix + .as_deref() + .unwrap_or(&CONFIG.submission_prefix), + logs, + }; + Ok((StatusCode::OK, Json(body))) +} + +/// Handle `POST /add-checkpoint`, updating the pending checkpoint for a +/// log. Handled like a witness's `add-checkpoint`, but the mirror does +/// not cosign and returns an empty body ([spec][add-cp]). +/// +/// After validating the request (parse, log lookup, signature, and size +/// range), the check-proof-and-update against persisted pending state is +/// delegated to the per-origin [`MirrorState`] DO so it happens +/// atomically; see [`dispatch_update_pending`] for the status-code +/// mapping. +/// +/// [add-cp]: https://c2sp.org/tlog-mirror#add-checkpoint +/// [`MirrorState`]: crate::mirror_state_do +#[worker::send] +async fn add_checkpoint( + State(env): State, + body: Bytes, +) -> ApiResult { + // The body size is capped by the `DefaultBodyLimit` layer on this + // route, which rejects oversized payloads (413) before they are fully + // buffered, so by here `body` is already within bounds. + let AddCheckpointRequest { + old_size, + consistency_proof, + checkpoint, + } = match parse_add_checkpoint_request(&body) { + Ok(r) => r, + Err(e) => { + log::warn!("add-checkpoint: malformed request: {e}"); + return Err(AppError::BadRequest(e.to_string())); + } + }; + + // Parse the checkpoint and look up the log by its origin. Using the + // validated `CheckpointText` origin (not a looser re-parse) keeps the + // lookup consistent with the size/hash used below. + let cp_text = match CheckpointText::from_bytes(checkpoint.text()) { + Ok(t) => t, + Err(e) => { + log::warn!("add-checkpoint: malformed checkpoint text: {e:?}"); + return Err(AppError::BadRequest(e.to_string())); + } + }; + let origin = cp_text.origin(); + let Some(verifiers) = log_verifiers(origin) else { + return Err(AppError::UnknownLogOrigin); + }; + + // Accept the checkpoint if at least one trusted log key signed it; + // unknown-key signatures are ignored (witness semantics). No valid + // trusted signature -> 403; a malformed signature line -> 400. + if let Err(e) = checkpoint.verify(&verifiers) { + match e { + NoteError::UnverifiedNote | NoteError::InvalidSignature { .. } => { + log::info!("add-checkpoint: rejecting note: {e:?}"); + return Err(AppError::NoValidSignatures); + } + _ => { + log::warn!("add-checkpoint: verify failed: {e:?}"); + return Err(AppError::BadRequest(e.to_string())); + } + } + } + + if old_size > cp_text.size() { + return Err(AppError::BadRequest(format!( + "old_size {old_size} > checkpoint size {}", + cp_text.size() + ))); + } + + // Atomic check-proof-and-update in the per-origin DO. Pass the whole + // signed note (via `to_bytes()`) so the DO retains the log's + // signature alongside size+hash, per spec. + let update = UpdatePendingRequest { + old_size, + new_size: cp_text.size(), + new_hash: *cp_text.hash(), + proof: consistency_proof, + signed_note_bytes: checkpoint.to_bytes(), + }; + if let Some(resp) = dispatch_update_pending(&env, origin, &update).await? { + return Ok(resp); + } + + // Empty body; the mirror cosigner MUST NOT sign here. + Ok(StatusCode::OK.into_response()) +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Maximum `add-checkpoint` request body, enforced by the route's +/// [`DefaultBodyLimit`] layer. A well-formed request is an `old ` +/// line, up to 63 base64 hash lines, and a checkpoint note of up to +/// `signed_note::MAX_NOTE_SIZE` (1 MiB); 1 MiB plus 16 KiB of headroom +/// covers that. +const MAX_ADD_CHECKPOINT_BODY_SIZE: usize = 1_024 * 1_024 + 16 * 1_024; + +/// POST the [`UpdatePendingRequest`] to the per-origin DO, translating +/// the DO's status code into either: +/// +/// * `Ok(None)`: success (200); the caller should return an empty +/// `Response`. +/// * `Ok(Some(resp))`: the DO responded with a non-200 status that +/// maps directly to the `add-checkpoint` HTTP response (409 with +/// `text/x.tlog.size` body, 422, or a forwarded 400). +/// * `Err(_)`: transport-level failure. +async fn dispatch_update_pending( + env: &Env, + origin: &str, + update: &UpdatePendingRequest, +) -> Result> { + let stub = state_stub(env, origin)?; + let mut resp = stub + .fetch_with_request(Request::new_with_init( + "http://do/update-pending", + &RequestInit { + method: Method::Post, + body: Some(serde_json::to_string(update)?.into()), + headers: { + let h = Headers::new(); + h.set("content-type", "application/json")?; + h + }, + ..Default::default() + }, + )?) + .await?; + match resp.status_code() { + // Drop `resp`; its destructor releases the body (no explicit read). + 200 => Ok(None), + 409 => { + let current: PendingCheckpoint = resp.json().await?; + Ok(Some(tlog_size_conflict(¤t))) + } + 422 => Ok(Some( + ( + StatusCode::UNPROCESSABLE_ENTITY, + "Unprocessable Entity: consistency proof failed", + ) + .into_response(), + )), + 400 => { + // Forward the DO's message verbatim; it already describes the + // specific violation (e.g. non-empty proof for a first pending). + let msg = resp.text().await.unwrap_or_else(|_| "Bad request".into()); + Ok(Some((StatusCode::BAD_REQUEST, msg).into_response())) + } + status => Ok(Some( + ( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Internal error: DO returned {status}"), + ) + .into_response(), + )), + } +} + +/// Build the 409 response body per the witness/mirror spec: +/// `text/x.tlog.size` content type, decimal latest size followed by a +/// newline. +fn tlog_size_conflict(current: &PendingCheckpoint) -> axum::response::Response { + let body = format!("{}\n", current.size); + ( + StatusCode::CONFLICT, + [(header::CONTENT_TYPE, CONTENT_TYPE_TLOG_SIZE)], + body, + ) + .into_response() +} diff --git a/crates/mirror_worker/src/lib.rs b/crates/mirror_worker/src/lib.rs index ff52e45..d4d5754 100644 --- a/crates/mirror_worker/src/lib.rs +++ b/crates/mirror_worker/src/lib.rs @@ -4,16 +4,34 @@ //! A transparency-log mirror implementing [c2sp.org/tlog-mirror][mirror] on //! Cloudflare Workers, specialized for MTC issuance logs. //! -//! The per-environment configuration is validated and embedded at build -//! time (see `build.rs` and the [`config`] sub-crate). The HTTP entry -//! point lives in [`frontend_worker`]. +//! 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`. +//! +//! Per-origin persistent state lives in a `MirrorState` Durable Object, +//! one per log origin. Its single-threaded execution model provides the +//! atomic check-old-state, verify, persist-new-state sequence the spec +//! requires. //! //! [mirror]: https://c2sp.org/tlog-mirror +//! [add-cp]: https://c2sp.org/tlog-mirror#add-checkpoint use config::AppConfig; -use std::sync::LazyLock; +use ml_dsa::pkcs8::{DecodePrivateKey as _, EncodePublicKey as _}; +use ml_dsa::{MlDsa44, VerifyingKey as MlDsaVerifyingKey}; +use pkcs8::{PrivateKeyInfoRef, SecretDocument, der::oid::db::fips204::ID_ML_DSA_44}; +use signed_note::{KeyName, NoteVerifier, VerifierList}; +use std::collections::HashMap; +use std::sync::{Arc, LazyLock, OnceLock}; +use tlog_cosignature::SubtreeV1NoteVerifier; +#[allow(clippy::wildcard_imports)] +use worker::*; mod frontend_worker; +mod mirror_state_do; + +/// The binding name used in `wrangler.jsonc` for the `MirrorState` DO. +pub(crate) const MIRROR_STATE_BINDING: &str = "MIRROR_STATE"; /// The compile-time-embedded worker configuration. /// @@ -24,3 +42,330 @@ pub(crate) static CONFIG: LazyLock = LazyLock::new(|| { serde_json::from_str(include_str!(concat!(env!("OUT_DIR"), "/config.json"))) .expect("config.json must be valid at build time") }); + +/// Per-origin cache of the parsed trusted log keys, keyed by the +/// concrete checkpoint origin. An MTC log-number window expands to one +/// entry per accepted log number (origin `.0.`), so the +/// map keys are the full set of origins this mirror serves. +/// +/// All log numbers of a CA share the same key(s), and a parsed +/// ML-DSA-44 [`LogKey`] is ~25 KiB (it precomputes the expanded NTT +/// matrix), so the value is an [`Arc`]: every origin in a window points +/// at one shared allocation rather than cloning the key per log number. +/// +/// Values are [`LogKey`] pairs rather than a pre-built `VerifierList`, +/// because `Box` is not `Sync` and so cannot live +/// inside a `LazyLock`. Building the `VerifierList` per request is cheap +/// (`SubtreeV1NoteVerifier::new` is just a key-ID hash). +pub(crate) static LOG_KEYS: LazyLock>>> = LazyLock::new(|| { + let mut map: HashMap>> = HashMap::new(); + for (log_key_name, log) in &CONFIG.logs { + let keys = Arc::new(parse_log_keys(log_key_name, log)); + for origin in log.origins(log_key_name) { + map.insert(origin, Arc::clone(&keys)); + } + } + map +}); + +/// A parsed trusted log key: the note-signature `name` (the +/// `log_key_name`, constant across an MTC CA's log-number window) and the +/// ML-DSA-44 verifying key. +#[derive(Clone)] +pub(crate) struct LogKey { + pub name: KeyName, + pub verifying_key: MlDsaVerifyingKey, +} + +/// Build the parsed keys for a single configured log. +/// +/// `build.rs` runs [`config::AppConfig::validate`] and refuses to compile +/// a malformed config, so the `log_key_name` and each SPKI are known to +/// parse; the `expect`s below would only fire if `validate` drifted out +/// of sync with this function. +fn parse_log_keys(log_key_name: &str, log: &config::LogParams) -> Vec { + use ml_dsa::pkcs8::DecodePublicKey as _; + let name = KeyName::new(log_key_name.to_owned()) + .expect("log_key_name validated as a signed-note KeyName by AppConfig::validate"); + log.log_public_keys + .iter() + .map(|spki| { + let verifying_key = MlDsaVerifyingKey::::from_public_key_der(spki) + .expect("SPKI validated as ML-DSA-44 by AppConfig::validate"); + LogKey { + name: name.clone(), + verifying_key, + } + }) + .collect() +} + +/// Build a [`VerifierList`] for a given origin from the cached keys, or +/// `None` if no log is configured at that origin. +pub(crate) fn log_verifiers(origin: &str) -> Option { + let keys = LOG_KEYS.get(origin)?; + let verifiers: Vec> = keys + .iter() + .map(|k| { + Box::new(SubtreeV1NoteVerifier::new( + k.name.clone(), + k.verifying_key.clone(), + )) as Box + }) + .collect(); + Some(VerifierList::new(verifiers)) +} + +// --------------------------------------------------------------------------- +// Mirror cosigner key +// --------------------------------------------------------------------------- + +/// The mirror's signing material. +/// +/// 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`. +/// +/// [mtc]: https://c2sp.org/mtc-tlog +/// [cosig]: https://c2sp.org/tlog-cosignature +pub(crate) struct MirrorSigner { + public_key_der: Vec, +} + +impl MirrorSigner { + /// DER-encoded `SubjectPublicKeyInfo` for the mirror's ML-DSA-44 + /// verifying key. + pub(crate) fn public_key_der(&self) -> &[u8] { + &self.public_key_der + } + + /// Stable string identifying the cosignature algorithm, published in + /// `/metadata`. Always `"subtree/v1"` (the only algorithm an MTC + /// mirror cosigner may use). + #[allow(clippy::unused_self)] + pub(crate) fn algorithm(&self) -> &'static str { + "subtree/v1" + } +} + +/// Cached mirror signer, so the PKCS#8 parse happens at most once per +/// worker instance. +static MIRROR_SIGNER: OnceLock = OnceLock::new(); + +/// Load (or return the already-cached) mirror signer. +/// +/// The `MIRROR_SIGNING_KEY` PKCS#8 PEM secret MUST carry an ML-DSA-44 key +/// (`id-ml-dsa-44`); the mirror cosigns with `subtree/v1`. +/// +/// # Errors +/// +/// Returns an error if the `MIRROR_SIGNING_KEY` secret is missing, the PEM +/// is malformed, or the key is not ML-DSA-44. +pub(crate) fn load_mirror_signer(env: &Env) -> Result<&'static MirrorSigner> { + if let Some(s) = MIRROR_SIGNER.get() { + return Ok(s); + } + let pem = env.secret("MIRROR_SIGNING_KEY")?.to_string(); + let signer = build_mirror_signer(&pem)?; + Ok(MIRROR_SIGNER.get_or_init(|| signer)) +} + +/// Build a [`MirrorSigner`] from a PKCS#8 PEM string. +/// +/// Split out from [`load_mirror_signer`] so unit tests can exercise the +/// parse/validation without a `worker::Env`. The key MUST be ML-DSA-44; +/// any other algorithm is rejected (the mirror's cosigner must be an MTC +/// cosigner, see [`MirrorSigner`]). +fn build_mirror_signer(pem: &str) -> Result { + 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()) + .map_err(|e| Error::from(format!("PrivateKeyInfo parse: {e}")))?; + 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. + 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 + .verifying_key() + .to_public_key_der() + .map_err(|e| Error::from(format!("ML-DSA-44 SPKI encode: {e}")))? + .to_vec(); + Ok(MirrorSigner { public_key_der }) + } + oid => Err(Error::from(format!( + "unsupported MIRROR_SIGNING_KEY algorithm OID {oid}: expected id-ml-dsa-44 \ + ({ID_ML_DSA_44}). The mirror's cosigner must be an MTC cosigner (ML-DSA-44, \ + subtree/v1)." + ))), + } +} + +/// Return the DER-encoded `SubjectPublicKeyInfo` for the mirror's own +/// verifying key. Used by the `/metadata` endpoint. +/// +/// # Errors +/// +/// Returns an error if the signing key is not available. +pub(crate) fn load_mirror_public_key_der(env: &Env) -> Result<&'static [u8]> { + Ok(load_mirror_signer(env)?.public_key_der()) +} + +#[cfg(test)] +mod signer_tests { + //! Unit tests for the mirror signer loader: only ML-DSA-44 keys are + //! accepted; other algorithms and malformed PEMs are rejected. + //! Ed25519 keys are generated only to exercise the rejection path, so + //! `ed25519-dalek` is a dev-dependency. + + use super::build_mirror_signer; + use ed25519_dalek::pkcs8::EncodePrivateKey as _; + + /// Generate a deterministic Ed25519 PEM from a seed byte. Used only to + /// check that a non-ML-DSA-44 key is rejected. + fn ed25519_pem(seed: u8) -> String { + let sk = ed25519_dalek::SigningKey::from_bytes(&[seed; 32]); + sk.to_pkcs8_pem(pkcs8::LineEnding::LF) + .expect("encode PEM") + .to_string() + } + + /// Generate a deterministic ML-DSA-44 PEM from a seed byte, using the + /// seed-only PKCS#8 encoding an operator gets from `openssl genpkey + /// -algorithm ML-DSA-44`, so this exercises the real load path. + fn ml_dsa_44_pem(seed: u8) -> String { + use ml_dsa::SigningKey; + use pkcs8::EncodePrivateKey as _; + let sk = SigningKey::::from_seed(&ml_dsa::B32::from([seed; 32])); + sk.to_pkcs8_pem(pkcs8::LineEnding::LF) + .expect("encode ML-DSA-44 PEM") + .to_string() + } + + #[test] + fn ml_dsa_44_pem_loads_with_subtree_v1_algorithm() { + let signer = build_mirror_signer(&ml_dsa_44_pem(2)).expect("build ML-DSA-44 signer"); + assert_eq!(signer.algorithm(), "subtree/v1"); + assert!( + !signer.public_key_der().is_empty(), + "ML-DSA-44 SPKI must be non-empty", + ); + } + + /// The mirror cosigner MUST be ML-DSA-44 (an MTC cosigner). An Ed25519 + /// key (a valid tlog cosigner key in general, but not an MTC one) is + /// refused, with an error naming the expected algorithm. + #[test] + fn ed25519_key_is_rejected() { + let Err(err) = build_mirror_signer(&ed25519_pem(3)) else { + panic!("Ed25519 MIRROR_SIGNING_KEY must be rejected") + }; + let msg = err.to_string(); + assert!( + msg.contains("unsupported") && msg.contains("id-ml-dsa-44"), + "unexpected error: {msg}", + ); + } + + #[test] + fn malformed_pem_is_rejected() { + let Err(err) = build_mirror_signer("not-a-pem") else { + panic!("malformed PEM must not parse") + }; + let msg = err.to_string(); + assert!( + msg.contains("PEM parse") || msg.contains("PrivateKeyInfo"), + "unexpected error: {msg}", + ); + } +} + +#[cfg(test)] +mod dev_config_tests { + //! Tests that pin invariants between `config.dev.json` / `.dev.vars` + //! and the integration-test fixtures that mirror them; a failure means + //! they have drifted and must be rotated together. + //! + //! The dev config models an MTC CA: the `logs` key is the CA cosigner + //! ID (`oid/1.3.6.1.4.1.32473.2`) whose ML-DSA-44 keypair signs + //! `subtree/v1` checkpoints. The mirror's own cosigner key in + //! `.dev.vars` is independent and is pinned only to "parses cleanly". + + use base64::prelude::*; + use ml_dsa::pkcs8::{DecodePrivateKey as _, EncodePublicKey as _}; + use ml_dsa::{Keypair as _, MlDsa44, SigningKey}; + + /// Raw `config.dev.json`, read directly rather than via `CONFIG`, + /// which is built from the `$DEPLOY_ENV` copy `build.rs` stages and + /// may not be `dev` during `cargo test`. + const DEV_CONFIG: &str = include_str!("../config.dev.json"); + + /// Raw `.dev.vars` contents. + const DEV_VARS: &str = include_str!("../.dev.vars"); + + /// Dev log PEM (ML-DSA-44, seed-only PKCS#8), duplicated from + /// `crates/integration_tests/tests/tlog_mirror.rs` so this test can + /// fail closed without that crate in scope. Rotate both copies and the + /// SPKI in `config.dev.json` together. + const DEV_LOG_SIGNING_KEY_PEM: &str = "-----BEGIN PRIVATE KEY-----\n\ + MDQCAQAwCwYJYIZIAWUDBAMRBCKAIBERERERERERERERERERERERERERERERERER\n\ + ERERERER\n\ + -----END PRIVATE KEY-----\n"; + + /// Extract a `.dev.vars` value by key. Lines are `KEY="value"` with + /// embedded newlines as literal `\n`, which we un-escape so the parsed + /// PEM round-trips. + fn dev_var(key: &str) -> String { + let line = DEV_VARS + .lines() + .find(|l| l.starts_with(&format!("{key}="))) + .unwrap_or_else(|| panic!(".dev.vars must define {key}")); + let rhs = line + .strip_prefix(&format!("{key}=")) + .unwrap() + .trim_matches('"'); + rhs.replace("\\n", "\n") + } + + #[test] + fn dev_config_spki_matches_embedded_pem() { + // Pull the log's first public key straight from the JSON, so the + // test is robust to unrelated config-shape changes. + let parsed: serde_json::Value = serde_json::from_str(DEV_CONFIG).unwrap(); + let b64 = parsed["logs"]["oid/1.3.6.1.4.1.32473.2"]["log_public_keys"][0] + .as_str() + .expect( + "config.dev.json must have logs[\"oid/1.3.6.1.4.1.32473.2\"].log_public_keys[0]", + ); + let config_spki = BASE64_STANDARD.decode(b64).expect("SPKI is base64"); + + // Derive the SPKI from the PEM and compare. + let sk = SigningKey::::from_pkcs8_pem(DEV_LOG_SIGNING_KEY_PEM) + .expect("parse dev log PEM"); + let derived_spki = sk.verifying_key().to_public_key_der().unwrap().to_vec(); + + assert_eq!( + config_spki, derived_spki, + "config.dev.json SPKI and DEV_LOG_SIGNING_KEY_PEM have drifted; \ + integration tests would 403", + ); + } + + /// `MIRROR_SIGNING_KEY` in `.dev.vars` parses cleanly through the + /// same `build_mirror_signer` code path that production uses. Pins + /// that an operator-typo in `.dev.vars` is caught at unit-test + /// time, not at the first request to a running `wrangler dev`. + #[test] + fn dev_vars_mirror_signing_key_parses() { + let pem = dev_var("MIRROR_SIGNING_KEY"); + let signer = super::build_mirror_signer(&pem) + .expect("MIRROR_SIGNING_KEY in .dev.vars must parse via build_mirror_signer"); + assert_eq!( + signer.algorithm(), + "subtree/v1", + "dev MIRROR_SIGNING_KEY must load as a subtree/v1 signer", + ); + } +} diff --git a/crates/mirror_worker/src/mirror_state_do.rs b/crates/mirror_worker/src/mirror_state_do.rs new file mode 100644 index 0000000..dc01602 --- /dev/null +++ b/crates/mirror_worker/src/mirror_state_do.rs @@ -0,0 +1,643 @@ +// Copyright (c) 2025-2026 Cloudflare, Inc. All rights reserved. +// SPDX-License-Identifier: BSD-3-Clause + +//! [`MirrorState`] Durable Object: per-origin atomic state for the +//! [c2sp.org/tlog-mirror][spec] protocol. +//! +//! The DO holds three pieces of per-origin state, always ordered +//! `committed.size <= next_entry.size <= pending.size`: +//! +//! - `pending`: the latest signed checkpoint accepted via +//! [`add-checkpoint`][add-cp], the source of truth for the consistency +//! proof check. +//! - `committed`: the latest *mirror checkpoint*, the state entries have +//! been fully ingested and cosigned for. Advanced monotonically. +//! - `next_entry`: the *persisted-entry frontier*, how far entry bundles +//! have been durably written. Advanced monotonically, including by +//! partial uploads that don't yet reach a signed pending size. +//! +//! It exposes internal RPCs (`/update-pending`, `/get-state`, +//! `/advance-next-entry`, `/commit`) consumed by the frontend handler in +//! the same worker; see each request type and handler for semantics. +//! Atomicity of the read-verify-compare-write sequences comes from the +//! DO's input/output gates. +//! +//! [spec]: https://c2sp.org/tlog-mirror +//! [add-cp]: https://c2sp.org/tlog-mirror#add-checkpoint + +use serde::{Deserialize, Serialize}; +use serde_with::{base64::Base64 as Base64As, serde_as}; +use tlog_core::{Hash, verify_consistency_proof}; +#[allow(clippy::wildcard_imports)] +use worker::*; + +use crate::MIRROR_STATE_BINDING; + +const PENDING_KEY: &str = "pending"; +const COMMITTED_KEY: &str = "committed"; +const NEXT_ENTRY_KEY: &str = "next_entry"; + +/// The persisted *pending checkpoint* for a single log origin. +/// +/// Stores the full signed-note bytes (not just size+hash) so the mirror +/// can serve them back to `add-entries` clients and retain the log's +/// signature per spec. +#[serde_as] +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct PendingCheckpoint { + /// Tree size, or zero if no pending checkpoint has been accepted for + /// this origin. + pub size: u64, + /// Root hash. All-zero if `size` is 0. + #[serde(with = "hash_hex")] + pub hash: Hash, + /// Full signed-note bytes, empty if `size` is 0. Base64 in the + /// on-disk JSON so the state stays valid UTF-8. + #[serde_as(as = "Base64As")] + pub signed_note_bytes: Vec, +} + +/// The persisted *committed checkpoint* (the *mirror checkpoint*) for a +/// single log origin: the state entries have been fully ingested and +/// cosigned for. Always at-or-behind [`PendingCheckpoint`], advanced +/// monotonically by `/commit`. +/// +/// Stores the full signed-note bytes so the mirror can serve a cosigned +/// checkpoint at `//checkpoint` without +/// looking up historic pending state. +#[serde_as] +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct CommittedCheckpoint { + /// Tree size, or zero if no entries have been committed for this + /// origin. + pub size: u64, + /// 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. + #[serde_as(as = "Base64As")] + pub signed_note_bytes: Vec, +} + +/// The persisted *next entry* frontier for a single log origin: how far +/// entry bundles have been durably written. +/// +/// Sits between the committed and pending checkpoints. A partial +/// `add-entries` upload advances this without advancing the mirror +/// checkpoint, which requires reaching a signed pending size. `hash` is +/// the tree root at `size`, retained so a resuming `add-entries` can +/// authenticate the frontier. +#[serde_as] +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct NextEntry { + /// Number of entries durably persisted as bundles for this origin. + pub size: u64, + /// Tree root at `size`. All-zero (and unused) when `size` is 0. + #[serde(with = "hash_hex")] + pub hash: Hash, +} + +/// Snapshot of the pending, committed, and next-entry state, returned by +/// `/get-state`. Used by the `add-entries` handler to early-reject +/// 409/404/422 cases and to resume appending from the persisted frontier. +#[derive(Serialize, Deserialize, Debug, Clone, Default)] +pub struct MirrorStateSnapshot { + pub pending: PendingCheckpoint, + pub committed: CommittedCheckpoint, + /// The persisted-entry frontier. `#[serde(default)]` so a snapshot + /// serialized by an older worker (before this field existed) still + /// deserializes to the zero frontier during a rolling deploy. + #[serde(default)] + pub next_entry: NextEntry, +} + +/// Body of the internal `/commit` RPC, advancing `committed`. See the +/// handler for the compare-and-swap semantics. +#[serde_as] +#[derive(Serialize, Deserialize, Debug)] +pub struct CommitRequest { + /// Proposed new committed tree size. + pub size: u64, + /// Proposed new committed root hash. + #[serde(with = "hash_hex")] + pub hash: Hash, + /// Full signed-note bytes for `(size, hash)`, served with the + /// mirror's cosignature. + #[serde_as(as = "Base64As")] + pub signed_note_bytes: Vec, +} + +/// Body of the internal `/advance-next-entry` RPC, advancing the +/// persisted-entry frontier. See the handler for the semantics. +#[serde_as] +#[derive(Serialize, Deserialize, Debug)] +pub struct AdvanceNextEntryRequest { + /// Proposed new persisted-entry frontier size. + pub size: u64, + /// Tree root at `size`. + #[serde(with = "hash_hex")] + pub hash: Hash, +} + +/// Body of the internal `/update-pending` RPC. +#[serde_as] +#[derive(Serialize, Deserialize, Debug)] +pub struct UpdatePendingRequest { + /// The client-claimed old size; must equal the persisted pending + /// size or the update is rejected (409 Conflict). + pub old_size: u64, + /// Proposed new tree size. + pub new_size: u64, + /// Proposed new root hash. + #[serde(with = "hash_hex")] + pub new_hash: Hash, + /// Consistency proof from `(old_size, stored_hash)` to + /// `(new_size, new_hash)`, per RFC 6962 section 2.1.2. MUST be empty if + /// `old_size == 0` or `old_size == new_size`, otherwise MUST verify. + #[serde(with = "hash_vec_hex")] + pub proof: Vec, + /// Full signed-note bytes of the new pending checkpoint. Persisted + /// alongside the size/hash so the mirror can serve them back to + /// `add-entries` clients. + #[serde_as(as = "Base64As")] + pub signed_note_bytes: Vec, +} + +/// A Durable Object holding the latest pending and committed checkpoint +/// state for a single log origin. +#[durable_object(fetch)] +struct MirrorState { + state: State, +} + +impl DurableObject for MirrorState { + fn new(state: State, _env: Env) -> Self { + Self { state } + } + + async fn fetch(&self, req: Request) -> Result { + self.fetch_inner(req).await + } +} + +impl MirrorState { + async fn fetch_inner(&self, mut req: Request) -> Result { + let path = req.path(); + match (req.method(), path.as_str()) { + (Method::Post, "/get-state") => { + let snapshot = self.read_snapshot().await?; + 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) + } + (Method::Post, "/advance-next-entry") => { + let body: AdvanceNextEntryRequest = req.json().await?; + self.advance_next_entry(body).await + } + (Method::Post, "/update-pending") => { + // The DO input/output gates make the read-verify-compare- + // write below atomic: concurrent requests for this origin + // cannot interleave, and the response is held until the + // write is durable, so a following add-entries cannot race + // it. + let body: UpdatePendingRequest = req.json().await?; + let current: PendingCheckpoint = self + .state + .storage() + .get(PENDING_KEY) + .await? + .unwrap_or_default(); + if current.size != body.old_size { + // Return the latest pending so the caller can build a + // 409 body. + return Response::from_json(¤t).map(|r| r.with_status(409)); + } + // Same size: hashes must match and the proof must be empty. + if body.old_size == body.new_size { + if current.hash.0 != body.new_hash.0 { + return Response::from_json(¤t).map(|r| r.with_status(409)); + } + if !body.proof.is_empty() { + return Response::error( + "consistency proof must be empty when old_size == checkpoint size", + 400, + ); + } + } else if body.old_size == 0 { + // First pending for this origin: proof must be empty. + if !body.proof.is_empty() { + return Response::error( + "consistency proof must be empty when old_size is 0 (first pending checkpoint for this origin)", + 400, + ); + } + } else { + // 0 < old_size < new_size: proof required. + // `verify_consistency_proof` takes the larger tree + // first, then the smaller. + if verify_consistency_proof( + &body.proof, + body.new_size, + body.new_hash, + body.old_size, + current.hash, + ) + .is_err() + { + return Response::error("consistency proof failed", 422); + } + } + let new_state = PendingCheckpoint { + size: body.new_size, + hash: body.new_hash, + signed_note_bytes: body.signed_note_bytes, + }; + self.state.storage().put(PENDING_KEY, &new_state).await?; + Response::from_json(&new_state) + } + _ => Response::error("not found", 404), + } + } +} + +impl MirrorState { + /// Handle `/advance-next-entry`: monotonically advance the + /// persisted-entry frontier. Compare-and-swap, like `/commit`: + /// + /// * `size > pending.size`: cannot persist beyond pending; 400. + /// * `size <= next_entry.size`: a concurrent `add-entries` already + /// reached here; return the current frontier without rewinding. + /// * otherwise: advance to `(size, hash)`. + async fn advance_next_entry(&self, body: AdvanceNextEntryRequest) -> Result { + let snapshot = self.read_snapshot().await?; + if body.size > snapshot.pending.size { + return Response::error( + format!( + "advance beyond pending: requested size {} > pending size {}", + body.size, snapshot.pending.size + ), + 400, + ); + } + if body.size <= snapshot.next_entry.size { + return Response::from_json(&snapshot.next_entry); + } + let new_next = NextEntry { + size: body.size, + hash: body.hash, + }; + self.state.storage().put(NEXT_ENTRY_KEY, &new_next).await?; + Response::from_json(&new_next) + } + + /// Read `pending`, `committed`, and `next_entry` from DO storage. + /// Missing keys default to the zero state ("no state yet"). + async fn read_snapshot(&self) -> Result { + let storage = self.state.storage(); + let pending: PendingCheckpoint = storage.get(PENDING_KEY).await?.unwrap_or_default(); + let committed: CommittedCheckpoint = storage.get(COMMITTED_KEY).await?.unwrap_or_default(); + let next_entry: NextEntry = storage.get(NEXT_ENTRY_KEY).await?.unwrap_or_default(); + Ok(MirrorStateSnapshot { + pending, + committed, + next_entry, + }) + } +} + +/// Lookup helper used by the frontend: get a stub for the DO serving a +/// particular log origin. +pub(crate) fn state_stub(env: &Env, origin: &str) -> Result { + let namespace = env.durable_object(MIRROR_STATE_BINDING)?; + namespace.id_from_name(origin)?.get_stub() +} + +// --------------------------------------------------------------------------- +// Serde helpers: emit/parse `Hash` as hex, so the DO's JSON state is +// human-readable in wrangler's dev console. +// --------------------------------------------------------------------------- +mod hash_hex { + use serde::{Deserialize, Deserializer, Serialize, Serializer}; + use tlog_core::{HASH_SIZE, Hash}; + + pub fn serialize(h: &Hash, ser: S) -> std::result::Result + where + S: Serializer, + { + hex::encode(h.0).serialize(ser) + } + + pub fn deserialize<'de, D>(de: D) -> std::result::Result + where + D: Deserializer<'de>, + { + let s = String::deserialize(de)?; + from_hex(&s).map_err(serde::de::Error::custom) + } + + /// Decode a single hex-encoded [`struct@Hash`]. Shared with the `Vec` + /// helper in the sibling `hash_vec_hex` module. + pub(super) fn from_hex(s: &str) -> std::result::Result { + let bytes: [u8; HASH_SIZE] = hex::decode(s) + .map_err(|e| e.to_string())? + .try_into() + .map_err(|v: Vec| format!("hash must be {} bytes, got {}", HASH_SIZE, v.len()))?; + Ok(Hash(bytes)) + } +} + +/// Serde helper for `Vec`. Encodes as a JSON array of hex strings +/// so the DO RPC body stays human-readable alongside the other +/// [`hash_hex`]-encoded fields. +mod hash_vec_hex { + use super::hash_hex::from_hex; + use serde::{Deserialize, Deserializer, Serialize, Serializer}; + use tlog_core::Hash; + + pub fn serialize(v: &[Hash], ser: S) -> std::result::Result + where + S: Serializer, + { + v.iter() + .map(|h| hex::encode(h.0)) + .collect::>() + .serialize(ser) + } + + pub fn deserialize<'de, D>(de: D) -> std::result::Result, D::Error> + where + D: Deserializer<'de>, + { + let strs = Vec::::deserialize(de)?; + strs.iter() + .map(|s| from_hex(s).map_err(serde::de::Error::custom)) + .collect() + } +} + +#[cfg(test)] +mod tests { + use super::{ + AdvanceNextEntryRequest, CommitRequest, CommittedCheckpoint, MirrorStateSnapshot, + NextEntry, PendingCheckpoint, UpdatePendingRequest, + }; + use tlog_core::{HASH_SIZE, Hash}; + + /// Pin the on-disk JSON layout of `PendingCheckpoint`. Changing + /// this format would make already-deployed mirrors unable to read + /// their persisted state after a worker upgrade, so any change + /// here must be paired with a migration plan. + #[test] + fn pending_checkpoint_json_format_unchanged() { + let mut bytes = [0u8; HASH_SIZE]; + for (i, b) in bytes.iter_mut().enumerate() { + *b = u8::try_from(i).unwrap(); + } + let pc = PendingCheckpoint { + size: 42, + hash: Hash(bytes), + signed_note_bytes: b"signed-note-bytes".to_vec(), + }; + let json = serde_json::to_string(&pc).unwrap(); + // Pin the expected canonical encoding, matching base64 of the + // signed-note bytes. + assert_eq!( + json, + r#"{"size":42,"hash":"000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f","signed_note_bytes":"c2lnbmVkLW5vdGUtYnl0ZXM="}"# + ); + + // Round-trip: an existing state blob must still parse after a + // rebuild. + let decoded: PendingCheckpoint = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded.size, 42); + assert_eq!(decoded.hash.0, bytes); + assert_eq!(decoded.signed_note_bytes, b"signed-note-bytes"); + } + + /// Pin the wire shape of the internal DO RPC body. The frontend + /// and the DO are in the same worker, but a format change still + /// needs both sides updated in lockstep. + #[test] + fn update_pending_request_json_format_unchanged() { + let req = UpdatePendingRequest { + old_size: 10, + new_size: 20, + new_hash: Hash([0xaa; HASH_SIZE]), + proof: vec![Hash([0xbb; HASH_SIZE]), Hash([0xcc; HASH_SIZE])], + signed_note_bytes: b"sn".to_vec(), + }; + let json = serde_json::to_string(&req).unwrap(); + assert_eq!( + json, + r#"{"old_size":10,"new_size":20,"new_hash":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","proof":["bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"],"signed_note_bytes":"c24="}"# + ); + let decoded: UpdatePendingRequest = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded.old_size, 10); + assert_eq!(decoded.new_size, 20); + assert_eq!(decoded.new_hash.0, [0xaa; HASH_SIZE]); + assert_eq!(decoded.proof.len(), 2); + assert_eq!(decoded.signed_note_bytes, b"sn"); + } + + /// The proof array is empty for first-pending and same-size cases; + /// make sure it round-trips as `[]` not omitted. + #[test] + fn update_pending_request_empty_proof_roundtrip() { + let req = UpdatePendingRequest { + old_size: 0, + new_size: 1, + new_hash: Hash([0u8; HASH_SIZE]), + proof: vec![], + signed_note_bytes: vec![], + }; + let json = serde_json::to_string(&req).unwrap(); + assert!( + json.contains(r#""proof":[]"#), + "proof must be serialized as an empty array, got: {json}" + ); + let decoded: UpdatePendingRequest = serde_json::from_str(&json).unwrap(); + assert!(decoded.proof.is_empty()); + } + + /// The default `PendingCheckpoint` represents "never accepted a + /// pending for this origin"; the frontend relies on the zero-sized + /// default when a DO has no stored state. + #[test] + fn pending_checkpoint_default_is_zero() { + let pc = PendingCheckpoint::default(); + assert_eq!(pc.size, 0); + assert_eq!(pc.hash.0, [0u8; HASH_SIZE]); + assert!(pc.signed_note_bytes.is_empty()); + } + + /// Pin the on-disk JSON layout of `CommittedCheckpoint`. Same + /// migration considerations as `PendingCheckpoint`: deployed + /// mirrors must keep parsing this after a worker upgrade. + #[test] + fn committed_checkpoint_json_format_unchanged() { + let mut bytes = [0u8; HASH_SIZE]; + for (i, b) in bytes.iter_mut().enumerate() { + *b = u8::try_from(i).unwrap(); + } + let cc = CommittedCheckpoint { + size: 42, + hash: Hash(bytes), + signed_note_bytes: b"signed-note-bytes".to_vec(), + }; + let json = serde_json::to_string(&cc).unwrap(); + assert_eq!( + json, + r#"{"size":42,"hash":"000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f","signed_note_bytes":"c2lnbmVkLW5vdGUtYnl0ZXM="}"# + ); + let decoded: CommittedCheckpoint = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded.size, 42); + assert_eq!(decoded.hash.0, bytes); + assert_eq!(decoded.signed_note_bytes, b"signed-note-bytes"); + } + + /// The default `CommittedCheckpoint` represents "never committed + /// any entries for this origin". + #[test] + fn committed_checkpoint_default_is_zero() { + let cc = CommittedCheckpoint::default(); + assert_eq!(cc.size, 0); + assert_eq!(cc.hash.0, [0u8; HASH_SIZE]); + assert!(cc.signed_note_bytes.is_empty()); + } + + /// Pin the wire shape of the `/get-state` response. + #[test] + fn mirror_state_snapshot_json_format() { + let snap = MirrorStateSnapshot { + pending: PendingCheckpoint { + size: 5, + hash: Hash([0xaa; HASH_SIZE]), + signed_note_bytes: b"p".to_vec(), + }, + committed: CommittedCheckpoint { + size: 3, + hash: Hash([0xbb; HASH_SIZE]), + signed_note_bytes: b"c".to_vec(), + }, + next_entry: NextEntry { + size: 4, + hash: Hash([0xcc; HASH_SIZE]), + }, + }; + let json = serde_json::to_string(&snap).unwrap(); + assert!( + json.contains(r#""pending":{"#) + && json.contains(r#""committed":{"#) + && json.contains(r#""next_entry":{"#), + "snapshot must include pending, committed, and next_entry: {json}" + ); + let decoded: MirrorStateSnapshot = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded.pending.size, 5); + assert_eq!(decoded.committed.size, 3); + assert_eq!(decoded.next_entry.size, 4); + } + + /// A snapshot serialized before `next_entry` existed must still + /// deserialize (to the zero frontier), so a rolling deploy that mixes + /// worker versions doesn't fail the `/get-state` round-trip. + #[test] + fn mirror_state_snapshot_defaults_next_entry() { + let legacy = r#"{"pending":{"size":5,"hash":"aa","signed_note_bytes":""},"committed":{"size":3,"hash":"bb","signed_note_bytes":""}}"# + .replace("\"aa\"", &format!("\"{}\"", "aa".repeat(HASH_SIZE))) + .replace("\"bb\"", &format!("\"{}\"", "bb".repeat(HASH_SIZE))); + let decoded: MirrorStateSnapshot = serde_json::from_str(&legacy).unwrap(); + assert_eq!(decoded.next_entry.size, 0); + assert_eq!(decoded.next_entry.hash.0, [0u8; HASH_SIZE]); + } + + /// Pin the on-disk JSON layout of `NextEntry`. Same migration + /// considerations as the other persisted checkpoint types. + #[test] + fn next_entry_json_format_unchanged() { + let ne = NextEntry { + size: 9, + hash: Hash([0xde; HASH_SIZE]), + }; + let json = serde_json::to_string(&ne).unwrap(); + assert_eq!( + json, + r#"{"size":9,"hash":"dededededededededededededededededededededededededededededededede"}"# + ); + let decoded: NextEntry = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded.size, 9); + assert_eq!(decoded.hash.0, [0xde; HASH_SIZE]); + } + + /// Pin the wire shape of the `/advance-next-entry` request body. + #[test] + fn advance_next_entry_request_json_format_unchanged() { + let req = AdvanceNextEntryRequest { + size: 11, + hash: Hash([0xef; HASH_SIZE]), + }; + let json = serde_json::to_string(&req).unwrap(); + assert_eq!( + json, + r#"{"size":11,"hash":"efefefefefefefefefefefefefefefefefefefefefefefefefefefefefefefef"}"# + ); + let decoded: AdvanceNextEntryRequest = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded.size, 11); + assert_eq!(decoded.hash.0, [0xef; HASH_SIZE]); + } + + /// Pin the wire shape of the `/commit` request body. + #[test] + fn commit_request_json_format_unchanged() { + let req = CommitRequest { + size: 7, + hash: Hash([0xcc; HASH_SIZE]), + signed_note_bytes: b"cm".to_vec(), + }; + let json = serde_json::to_string(&req).unwrap(); + assert_eq!( + json, + r#"{"size":7,"hash":"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc","signed_note_bytes":"Y20="}"# + ); + let decoded: CommitRequest = serde_json::from_str(&json).unwrap(); + assert_eq!(decoded.size, 7); + assert_eq!(decoded.hash.0, [0xcc; HASH_SIZE]); + assert_eq!(decoded.signed_note_bytes, b"cm"); + } +} diff --git a/crates/mirror_worker/wrangler.jsonc b/crates/mirror_worker/wrangler.jsonc index 3994f8e..e3b2411 100644 --- a/crates/mirror_worker/wrangler.jsonc +++ b/crates/mirror_worker/wrangler.jsonc @@ -17,7 +17,21 @@ // nightly >=2026-05-09 are not yet supported). "command": "cargo install -q worker-build@0.8.5 && DEPLOY_ENV=dev RUSTFLAGS='-Cllvm-args=-wasm-use-legacy-eh' worker-build --release" }, - "workers_dev": true + "workers_dev": true, + "durable_objects": { + "bindings": [ + { + "name": "MIRROR_STATE", + "class_name": "MirrorState" + } + ] + }, + "migrations": [ + { + "tag": "v1", + "new_sqlite_classes": ["MirrorState"] + } + ] } } } diff --git a/crates/witness_worker/src/witness_state_do.rs b/crates/witness_worker/src/witness_state_do.rs index a7ab681..01c0c83 100644 --- a/crates/witness_worker/src/witness_state_do.rs +++ b/crates/witness_worker/src/witness_state_do.rs @@ -143,8 +143,7 @@ impl WitnessState { .state .storage() .get(STATE_KEY) - .await - .unwrap_or(None) + .await? .unwrap_or_default(); if current.size != body.old_size { // Spec: respond with the latest size so the caller can