From 8b687d16cd802d9c8761eca8a4ce340126526bfe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Garillot?= Date: Thu, 20 Aug 2026 17:34:12 -0400 Subject: [PATCH 01/10] refactor(validator): add role-specific DKG board facades --- .../commands/dkg/{board.rs => board/mod.rs} | 172 +++++++++++++++++- bin/validator/src/commands/dkg/runner.rs | 95 ++++++---- bin/validator/src/commands/dkg/tests.rs | 2 +- 3 files changed, 230 insertions(+), 39 deletions(-) rename bin/validator/src/commands/dkg/{board.rs => board/mod.rs} (89%) diff --git a/bin/validator/src/commands/dkg/board.rs b/bin/validator/src/commands/dkg/board/mod.rs similarity index 89% rename from bin/validator/src/commands/dkg/board.rs rename to bin/validator/src/commands/dkg/board/mod.rs index bf3462ac0..0722579ca 100644 --- a/bin/validator/src/commands/dkg/board.rs +++ b/bin/validator/src/commands/dkg/board/mod.rs @@ -65,10 +65,16 @@ const MAX_VALUES_PER_SLOT: usize = 2; #[derive(Clone, Debug, Deserialize, Serialize)] pub(super) struct BoardTicket { document: DocTicket, - pub(super) participant: u32, + participant: u32, upload_secret: [u8; 32], } +impl BoardTicket { + pub(super) fn participant(&self) -> u32 { + self.participant + } +} + #[derive(Deserialize, Serialize)] enum BoardTicketWireFormat { Variant0(BoardTicket), @@ -129,6 +135,44 @@ pub(super) enum ArtifactSlot { TranscriptAcceptance(u32), } +/// An artifact published by the ceremony coordinator. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum CommonArtifact { + Manifest, + DecryptionConfig, + ContextConfig, +} + +impl CommonArtifact { + fn slot(self) -> ArtifactSlot { + match self { + Self::Manifest => ArtifactSlot::Manifest, + Self::DecryptionConfig => ArtifactSlot::DecryptionConfig, + Self::ContextConfig => ArtifactSlot::ContextConfig, + } + } +} + +/// An artifact published by one ceremony participant. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum ParticipantArtifact { + Registration, + DecryptionDealing, + ContextDealing, + TranscriptAcceptance, +} + +impl ParticipantArtifact { + fn slot(self, participant: u32) -> ArtifactSlot { + match self { + Self::Registration => ArtifactSlot::Registration(participant), + Self::DecryptionDealing => ArtifactSlot::DecryptionDealing(participant), + Self::ContextDealing => ArtifactSlot::ContextDealing(participant), + Self::TranscriptAcceptance => ArtifactSlot::TranscriptAcceptance(participant), + } + } +} + impl ArtifactSlot { fn prefix(&self) -> String { match self { @@ -201,8 +245,24 @@ struct UploadProtocol { writer: BoardWriter, } +/// The read-only view shared by both board roles. +pub(super) struct BoardReader { + node: BoardNode, +} + +/// The board role that coordinates and publishes common ceremony artifacts. +pub(super) struct CoordinatorBoard { + reader: BoardReader, +} + +/// The board role held by one ceremony participant. +pub(super) struct ParticipantBoard { + participant: u32, + reader: BoardReader, +} + /// A persistent Iroh node joined to one ceremony document. -pub(super) struct BoardNode { +struct BoardNode { blobs: FsStore, document: Doc, downloader: Downloader, @@ -235,6 +295,114 @@ struct BoardEvents { task: tokio::task::JoinHandle<()>, } +impl BoardReader { + /// Waits until one unique artifact has synchronized locally. + pub(super) async fn wait_unique( + &self, + slot: &ArtifactSlot, + timeout: Duration, + ) -> anyhow::Result> { + self.node.wait_unique(slot, timeout).await + } +} + +impl CoordinatorBoard { + /// Creates or resumes a ceremony board and returns one scoped ticket per participant. + pub(super) async fn create( + data_directory: &Path, + participant_count: usize, + ) -> anyhow::Result<(Self, Vec)> { + let (node, tickets) = BoardNode::create(data_directory, participant_count).await?; + Ok((Self { reader: BoardReader { node } }, tickets)) + } + + #[cfg(test)] + pub(super) async fn create_with_network( + data_directory: &Path, + participant_count: usize, + use_network_services: bool, + ) -> anyhow::Result<(Self, Vec)> { + let (node, tickets) = + BoardNode::create_with_network(data_directory, participant_count, use_network_services) + .await?; + Ok((Self { reader: BoardReader { node } }, tickets)) + } + + pub(super) fn reader(&self) -> &BoardReader { + &self.reader + } + + /// Publishes one common artifact without replacing another value in the same slot. + pub(super) async fn publish( + &self, + artifact: CommonArtifact, + value: &[u8], + ) -> anyhow::Result<()> { + self.reader.node.publish(&artifact.slot(), value).await?; + Ok(()) + } + + /// Stops the board and flushes its persistent stores. + pub(super) async fn shutdown(self) -> anyhow::Result<()> { + self.reader.node.shutdown().await + } +} + +impl ParticipantBoard { + /// Joins or resumes a ceremony board through a read and upload ticket. + pub(super) async fn join( + data_directory: &Path, + ticket: BoardTicket, + participant_count: usize, + ) -> anyhow::Result { + let participant = ticket.participant(); + let node = BoardNode::join(data_directory, ticket, participant_count).await?; + Ok(Self { + participant, + reader: BoardReader { node }, + }) + } + + pub(super) async fn join_with_network( + data_directory: &Path, + ticket: BoardTicket, + participant_count: usize, + use_network_services: bool, + ) -> anyhow::Result { + let participant = ticket.participant(); + let node = BoardNode::join_with_network( + data_directory, + ticket, + participant_count, + use_network_services, + ) + .await?; + Ok(Self { + participant, + reader: BoardReader { node }, + }) + } + + pub(super) fn reader(&self) -> &BoardReader { + &self.reader + } + + /// Publishes one artifact for the participant named by this board's ticket. + pub(super) async fn publish( + &self, + artifact: ParticipantArtifact, + value: &[u8], + ) -> anyhow::Result<()> { + self.reader.node.publish(&artifact.slot(self.participant), value).await?; + Ok(()) + } + + /// Stops the board and flushes its persistent stores. + pub(super) async fn shutdown(self) -> anyhow::Result<()> { + self.reader.node.shutdown().await + } +} + impl Drop for BoardNode { fn drop(&mut self) { self.event_task.abort(); diff --git a/bin/validator/src/commands/dkg/runner.rs b/bin/validator/src/commands/dkg/runner.rs index 209256e8f..b884396fb 100644 --- a/bin/validator/src/commands/dkg/runner.rs +++ b/bin/validator/src/commands/dkg/runner.rs @@ -7,7 +7,15 @@ use golden_core::{EvrfProofBackend, ParticipantIndex}; use miden_protocol::crypto::dsa::ecdsa_k256_keccak::PublicKey; use miden_validator::ValidatorSigner; -use super::board::{ArtifactSlot, BoardNode, BoardTicket}; +use super::board::{ + ArtifactSlot, + BoardReader, + BoardTicket, + CommonArtifact, + CoordinatorBoard, + ParticipantArtifact, + ParticipantBoard, +}; use super::{ CONTEXT_CONFIG_FILE, CONTEXT_DEALING_FILE, @@ -137,11 +145,12 @@ pub(super) async fn run_validator(options: DkgRunOptions) -> anyhow::Result<()> pub(super) async fn serve_board(options: DkgBoardServeOptions) -> anyhow::Result<()> { let genesis = read_trusted_genesis(&options.genesis)?; let participant_count = genesis.inner().header().validator_keys().as_keys().len(); - let (board, tickets) = BoardNode::create(&options.data_directory, participant_count).await?; + let (board, tickets) = + CoordinatorBoard::create(&options.data_directory, participant_count).await?; publish_directory(&options.ticket_directory, |temporary| { for ticket in &tickets { write_new_file( - &temporary.join(format!("participant-{}.ticket", ticket.participant)), + &temporary.join(format!("participant-{}.ticket", ticket.participant())), ticket.to_string().as_bytes(), true, )?; @@ -173,7 +182,7 @@ pub(super) async fn serve_board(options: DkgBoardServeOptions) -> anyhow::Result /// Waits for signed registrations, prepares the ceremony, and publishes its common files. pub(super) async fn coordinate_common_files( - board: &BoardNode, + board: &CoordinatorBoard, data_directory: &Path, genesis_path: &Path, threshold: usize, @@ -191,7 +200,7 @@ pub(super) async fn coordinate_common_files( let ceremony_directory = data_directory.join(CEREMONY_DIRECTORY); if !ceremony_directory.exists() { - let registrations = wait_for_registrations(board, validator_keys, timeout).await?; + let registrations = wait_for_registrations(board.reader(), validator_keys, timeout).await?; let registration_directory = data_directory.join(REGISTRATIONS_DIRECTORY); materialize_or_compare(®istration_directory, ®istrations)?; let paths = registrations @@ -207,17 +216,17 @@ pub(super) async fn coordinate_common_files( "board threshold changed after creation" ); ensure!(ceremony.manifest.epoch == epoch, "board epoch changed after creation"); - publish_named_file(board, &ArtifactSlot::Manifest, &ceremony_directory.join(MANIFEST_FILE)) + publish_common_file(board, CommonArtifact::Manifest, &ceremony_directory.join(MANIFEST_FILE)) .await?; - publish_named_file( + publish_common_file( board, - &ArtifactSlot::DecryptionConfig, + CommonArtifact::DecryptionConfig, &ceremony_directory.join(DECRYPTION_CONFIG_FILE), ) .await?; - publish_named_file( + publish_common_file( board, - &ArtifactSlot::ContextConfig, + CommonArtifact::ContextConfig, &ceremony_directory.join(CONTEXT_CONFIG_FILE), ) .await?; @@ -225,7 +234,7 @@ pub(super) async fn coordinate_common_files( } async fn wait_for_registrations( - board: &BoardNode, + board: &BoardReader, validator_keys: &[PublicKey], timeout: Duration, ) -> anyhow::Result)>> { @@ -281,9 +290,10 @@ where let participant = prepare_local_identity(genesis_path, epoch, signer, work_directory).await?; let board_directory = work_directory.join(BOARD_DIRECTORY); let board = if use_network_services { - BoardNode::join(&board_directory, ticket, participant_count).await? + ParticipantBoard::join(&board_directory, ticket, participant_count).await? } else { - BoardNode::join_with_network(&board_directory, ticket, participant_count, false).await? + ParticipantBoard::join_with_network(&board_directory, ticket, participant_count, false) + .await? }; let result = run_validator_on_board::( &board, @@ -307,7 +317,7 @@ where reason = "the inputs and linear body mirror the ceremony policy and phase order" )] async fn run_validator_on_board( - board: &BoardNode, + board: &ParticipantBoard, genesis_path: &Path, signer: &ValidatorSigner, participant: ParticipantIndex, @@ -321,17 +331,20 @@ where B: EvrfProofBackend, { let identity_directory = work_directory.join(IDENTITY_DIRECTORY); - publish_named_file( + publish_participant_file( board, - &ArtifactSlot::Registration(participant.get()), + ParticipantArtifact::Registration, &identity_directory.join(REGISTRATION_FILE), ) .await?; let genesis = read_trusted_genesis(genesis_path)?; - let registrations = - wait_for_registrations(board, genesis.inner().header().validator_keys().as_keys(), timeout) - .await?; + let registrations = wait_for_registrations( + board.reader(), + genesis.inner().header().validator_keys().as_keys(), + timeout, + ) + .await?; let registration_directory = work_directory.join(REGISTRATIONS_DIRECTORY); materialize_or_compare(®istration_directory, ®istrations)?; let registration_paths = registrations @@ -345,15 +358,15 @@ where let common = vec![ ( MANIFEST_FILE.to_owned(), - board.wait_unique(&ArtifactSlot::Manifest, timeout).await?, + board.reader().wait_unique(&ArtifactSlot::Manifest, timeout).await?, ), ( DECRYPTION_CONFIG_FILE.to_owned(), - board.wait_unique(&ArtifactSlot::DecryptionConfig, timeout).await?, + board.reader().wait_unique(&ArtifactSlot::DecryptionConfig, timeout).await?, ), ( CONTEXT_CONFIG_FILE.to_owned(), - board.wait_unique(&ArtifactSlot::ContextConfig, timeout).await?, + board.reader().wait_unique(&ArtifactSlot::ContextConfig, timeout).await?, ), ]; materialize_or_compare(&ceremony_directory, &common)?; @@ -382,22 +395,22 @@ where &mut OsRng, )?; } - publish_named_file( + publish_participant_file( board, - &ArtifactSlot::DecryptionDealing(participant.get()), + ParticipantArtifact::DecryptionDealing, &dealings_directory.join(DECRYPTION_DEALING_FILE), ) .await?; - publish_named_file( + publish_participant_file( board, - &ArtifactSlot::ContextDealing(participant.get()), + ParticipantArtifact::ContextDealing, &dealings_directory.join(CONTEXT_DEALING_FILE), ) .await?; let participant_count = ceremony.manifest.participants.len(); let public_dealings_directory = work_directory.join(PUBLIC_DEALINGS_DIRECTORY); - let public_dealings = wait_for_dealings(board, participant_count, timeout).await?; + let public_dealings = wait_for_dealings(board.reader(), participant_count, timeout).await?; materialize_or_compare(&public_dealings_directory, &public_dealings)?; let decryption_dealings = participant_files( &public_dealings_directory, @@ -424,14 +437,14 @@ where ) .await?; } - publish_named_file( + publish_participant_file( board, - &ArtifactSlot::TranscriptAcceptance(participant.get()), + ParticipantArtifact::TranscriptAcceptance, &acceptance_directory.join(TRANSCRIPT_ACCEPTANCE_FILE), ) .await?; - let acceptances = wait_for_acceptances(board, participant_count, timeout).await?; + let acceptances = wait_for_acceptances(board.reader(), participant_count, timeout).await?; let public_acceptances_directory = work_directory.join(PUBLIC_ACCEPTANCES_DIRECTORY); materialize_or_compare(&public_acceptances_directory, &acceptances)?; let transcript_path = acceptance_directory.join(TRANSCRIPT_FILE); @@ -533,7 +546,7 @@ fn validate_local_identity( } async fn wait_for_dealings( - board: &BoardNode, + board: &BoardReader, participant_count: usize, timeout: Duration, ) -> anyhow::Result)>> { @@ -557,7 +570,7 @@ async fn wait_for_dealings( } async fn wait_for_acceptances( - board: &BoardNode, + board: &BoardReader, participant_count: usize, timeout: Duration, ) -> anyhow::Result)>> { @@ -574,14 +587,24 @@ async fn wait_for_acceptances( Ok(acceptances) } -async fn publish_named_file( - board: &BoardNode, - slot: &ArtifactSlot, +async fn publish_common_file( + board: &CoordinatorBoard, + artifact: CommonArtifact, + path: &Path, +) -> anyhow::Result<()> { + let bytes = fs_err::read(path) + .with_context(|| format!("failed to read ceremony artifact {}", path.display()))?; + board.publish(artifact, &bytes).await +} + +async fn publish_participant_file( + board: &ParticipantBoard, + artifact: ParticipantArtifact, path: &Path, ) -> anyhow::Result<()> { let bytes = fs_err::read(path) .with_context(|| format!("failed to read ceremony artifact {}", path.display()))?; - board.publish(slot, &bytes).await?; + board.publish(artifact, &bytes).await?; Ok(()) } diff --git a/bin/validator/src/commands/dkg/tests.rs b/bin/validator/src/commands/dkg/tests.rs index 761e97153..93c2b1b70 100644 --- a/bin/validator/src/commands/dkg/tests.rs +++ b/bin/validator/src/commands/dkg/tests.rs @@ -912,7 +912,7 @@ async fn iroh_ceremony_resumes_after_validator_restart() -> TestResult { let genesis = write_genesis(root.path())?; let board_directory = root.path().join("board"); let (board, tickets) = - board::BoardNode::create_with_network(&board_directory, 3, false).await?; + board::CoordinatorBoard::create_with_network(&board_directory, 3, false).await?; let timeout = Duration::from_mins(2); let restart_checkpoint_timeout = Duration::from_secs(10); let epoch = "66".repeat(32); From 95bf44d14bb0d8a52b9329147e04f4e6aba81361 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Garillot?= Date: Thu, 20 Aug 2026 17:38:25 -0400 Subject: [PATCH 02/10] fix(validator): reject mismatched DKG board tickets --- bin/validator/src/commands/dkg/board/mod.rs | 6 ++++ bin/validator/src/commands/dkg/runner.rs | 6 ++++ bin/validator/src/commands/dkg/tests.rs | 35 +++++++++++++++++++++ 3 files changed, 47 insertions(+) diff --git a/bin/validator/src/commands/dkg/board/mod.rs b/bin/validator/src/commands/dkg/board/mod.rs index 0722579ca..d87e325af 100644 --- a/bin/validator/src/commands/dkg/board/mod.rs +++ b/bin/validator/src/commands/dkg/board/mod.rs @@ -296,6 +296,12 @@ struct BoardEvents { } impl BoardReader { + /// Reads the unique content value published for one artifact slot. + #[cfg(test)] + pub(super) async fn read_unique(&self, slot: &ArtifactSlot) -> anyhow::Result>> { + self.node.read_unique(slot).await + } + /// Waits until one unique artifact has synchronized locally. pub(super) async fn wait_unique( &self, diff --git a/bin/validator/src/commands/dkg/runner.rs b/bin/validator/src/commands/dkg/runner.rs index b884396fb..5d8a33177 100644 --- a/bin/validator/src/commands/dkg/runner.rs +++ b/bin/validator/src/commands/dkg/runner.rs @@ -288,6 +288,12 @@ where ); decode_fixed_hex::<32>(epoch, "storage-key epoch")?; let participant = prepare_local_identity(genesis_path, epoch, signer, work_directory).await?; + ensure!( + ticket.participant() == participant.get(), + "storage key DKG board ticket belongs to participant {}, but the signing key belongs to participant {}", + ticket.participant(), + participant.get(), + ); let board_directory = work_directory.join(BOARD_DIRECTORY); let board = if use_network_services { ParticipantBoard::join(&board_directory, ticket, participant_count).await? diff --git a/bin/validator/src/commands/dkg/tests.rs b/bin/validator/src/commands/dkg/tests.rs index 93c2b1b70..5811b6034 100644 --- a/bin/validator/src/commands/dkg/tests.rs +++ b/bin/validator/src/commands/dkg/tests.rs @@ -901,6 +901,41 @@ async fn deal_rejects_unknown_identity_and_existing_output() -> TestResult { ); Ok(()) } + +#[tokio::test] +async fn runner_rejects_another_participants_ticket_before_publishing() -> TestResult { + let root = tempfile::tempdir()?; + let genesis = write_genesis(root.path())?; + let board_directory = root.path().join("board"); + let (board, tickets) = + board::CoordinatorBoard::create_with_network(&board_directory, 3, false).await?; + let signer = ValidatorSigner::new_local(genesis.signing_keys[0].clone()); + let error = runner::run_validator_with_network::( + tickets[1].clone(), + &genesis.path, + &signer, + 2, + &"66".repeat(32), + &root.path().join("work"), + &root.path().join("bundle"), + false, + Duration::from_secs(1), + ) + .await + .unwrap_err(); + + assert!(error.to_string().contains("ticket belongs to participant 2")); + assert!( + board + .reader() + .read_unique(&board::ArtifactSlot::Registration(2)) + .await? + .is_none() + ); + board.shutdown().await?; + Ok(()) +} + #[tokio::test] /// Proves a validator can resume from its saved identity after its ceremony process stops. #[expect( From 65c6c2bcd59952917ba48a9d9fd5723ff516e360 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Garillot?= Date: Thu, 20 Aug 2026 17:44:46 -0400 Subject: [PATCH 03/10] refactor(validator): isolate DKG board persistence --- .../commands/dkg/board/iroh/persistence.rs | 109 ++++++++++++++++ bin/validator/src/commands/dkg/board/mod.rs | 122 +++--------------- 2 files changed, 125 insertions(+), 106 deletions(-) create mode 100644 bin/validator/src/commands/dkg/board/iroh/persistence.rs diff --git a/bin/validator/src/commands/dkg/board/iroh/persistence.rs b/bin/validator/src/commands/dkg/board/iroh/persistence.rs new file mode 100644 index 000000000..9b215de83 --- /dev/null +++ b/bin/validator/src/commands/dkg/board/iroh/persistence.rs @@ -0,0 +1,109 @@ +use std::io::Write; +use std::path::Path; +use std::str::FromStr; + +use anyhow::{Context, ensure}; +use iroh::SecretKey; +use iroh_docs::api::Doc; + +use super::super::{decode_fixed_hex, publish_directory, sync_directory, write_new_file}; + +pub(super) const ENDPOINT_SECRET_FILE: &str = "endpoint-secret.hex"; +pub(super) const BOARD_METADATA_DIRECTORY: &str = "board-meta"; +pub(super) const DOCUMENT_ID_FILE: &str = "document-id.hex"; +pub(super) const BOARD_FORMAT_FILE: &str = "board-format"; +const BOARD_FORMAT: &[u8] = b"participant-upload-v4\n"; +pub(super) const UPLOAD_SECRETS_DIRECTORY: &str = "upload-secrets"; + +pub(super) fn load_or_create_endpoint_secret(data_directory: &Path) -> anyhow::Result { + let path = data_directory.join(ENDPOINT_SECRET_FILE); + if path.exists() { + let encoded = fs_err::read_to_string(&path) + .with_context(|| format!("failed to read Iroh endpoint secret {}", path.display()))?; + return SecretKey::from_str(encoded.trim()).context("invalid Iroh endpoint secret"); + } + + let secret = SecretKey::generate(); + let mut temporary = tempfile::Builder::new() + .prefix(".endpoint-secret-") + .tempfile_in(data_directory) + .context("failed to create temporary Iroh endpoint secret")?; + temporary + .write_all(hex::encode(secret.to_bytes()).as_bytes()) + .context("failed to write temporary Iroh endpoint secret")?; + temporary + .as_file() + .sync_all() + .context("failed to sync temporary Iroh endpoint secret")?; + temporary + .persist_noclobber(&path) + .map_err(|error| error.error) + .with_context(|| format!("failed to publish Iroh endpoint secret {}", path.display()))?; + sync_directory(data_directory)?; + Ok(secret) +} + +pub(super) fn publish_board_metadata( + path: &Path, + document: &Doc, + upload_secrets: &[[u8; 32]], +) -> anyhow::Result<()> { + publish_directory(path, |temporary| { + write_new_file( + &temporary.join(DOCUMENT_ID_FILE), + hex::encode(document.id().to_bytes()).as_bytes(), + true, + )?; + write_new_file(&temporary.join(BOARD_FORMAT_FILE), BOARD_FORMAT, true)?; + let upload_secrets_directory = temporary.join(UPLOAD_SECRETS_DIRECTORY); + fs_err::create_dir(&upload_secrets_directory).with_context(|| { + format!( + "failed to create DKG board upload secrets {}", + upload_secrets_directory.display() + ) + })?; + for (position, secret) in upload_secrets.iter().enumerate() { + write_new_file( + &upload_secrets_directory.join(format!("participant-{}.hex", position + 1)), + hex::encode(secret).as_bytes(), + true, + )?; + } + Ok(()) + }) +} + +pub(super) fn load_upload_secrets( + metadata_directory: &Path, + participant_count: usize, +) -> anyhow::Result> { + let path = metadata_directory.join(UPLOAD_SECRETS_DIRECTORY); + let entry_count = fs_err::read_dir(&path) + .with_context(|| format!("failed to read DKG board upload secrets {}", path.display()))? + .collect::, _>>()? + .len(); + ensure!( + entry_count == participant_count, + "DKG board upload secret count does not match the participant count" + ); + (1..=participant_count) + .map(|participant| { + let secret_path = path.join(format!("participant-{participant}.hex")); + let bytes = fs_err::read_to_string(&secret_path).with_context(|| { + format!("failed to read DKG board upload secret {}", secret_path.display()) + })?; + decode_fixed_hex::<32>(bytes.trim(), "DKG board upload secret") + }) + .collect() +} + +pub(super) fn require_current_board_format(metadata_directory: &Path) -> anyhow::Result<()> { + let path = metadata_directory.join(BOARD_FORMAT_FILE); + let format = fs_err::read(&path) + .with_context(|| format!("failed to read DKG board format {}", path.display()))?; + ensure!( + format == BOARD_FORMAT, + "unsupported DKG board format; start a new ceremony in a new data directory" + ); + Ok(()) +} diff --git a/bin/validator/src/commands/dkg/board/mod.rs b/bin/validator/src/commands/dkg/board/mod.rs index d87e325af..3345efc82 100644 --- a/bin/validator/src/commands/dkg/board/mod.rs +++ b/bin/validator/src/commands/dkg/board/mod.rs @@ -8,7 +8,6 @@ use std::collections::BTreeMap; use std::fmt; -use std::io::Write; use std::path::Path; use std::str::FromStr; use std::sync::Arc; @@ -32,20 +31,24 @@ use iroh_gossip::net::Gossip; use iroh_tickets::{ParseError, Ticket}; use serde::{Deserialize, Serialize}; -use super::{ - decode_fixed_hex, - durably_create_directory_all, - publish_directory, - sync_directory, - write_new_file, +#[path = "iroh/persistence.rs"] +mod persistence; + +#[cfg(test)] +use persistence::ENDPOINT_SECRET_FILE; +use persistence::{ + BOARD_FORMAT_FILE, + BOARD_METADATA_DIRECTORY, + DOCUMENT_ID_FILE, + UPLOAD_SECRETS_DIRECTORY, + load_or_create_endpoint_secret, + load_upload_secrets, + publish_board_metadata, + require_current_board_format, }; -const ENDPOINT_SECRET_FILE: &str = "endpoint-secret.hex"; -const BOARD_METADATA_DIRECTORY: &str = "board-meta"; -const DOCUMENT_ID_FILE: &str = "document-id.hex"; -const BOARD_FORMAT_FILE: &str = "board-format"; -const BOARD_FORMAT: &[u8] = b"participant-upload-v4\n"; -const UPLOAD_SECRETS_DIRECTORY: &str = "upload-secrets"; +use super::{decode_fixed_hex, durably_create_directory_all}; + const UPLOAD_ALPN: &[u8] = b"/miden/storage-key-dkg-board-upload/3"; const UPLOAD_HEADER_BYTES: usize = 32 + 1 + 4 + 8; const UPLOAD_RESPONSE_BYTES: usize = 1 + 32; @@ -1181,98 +1184,5 @@ async fn inspect_document_metadata( Ok(()) } -fn load_or_create_endpoint_secret(data_directory: &Path) -> anyhow::Result { - let path = data_directory.join(ENDPOINT_SECRET_FILE); - if path.exists() { - let encoded = fs_err::read_to_string(&path) - .with_context(|| format!("failed to read Iroh endpoint secret {}", path.display()))?; - return SecretKey::from_str(encoded.trim()).context("invalid Iroh endpoint secret"); - } - - let secret = SecretKey::generate(); - let mut temporary = tempfile::Builder::new() - .prefix(".endpoint-secret-") - .tempfile_in(data_directory) - .context("failed to create temporary Iroh endpoint secret")?; - temporary - .write_all(hex::encode(secret.to_bytes()).as_bytes()) - .context("failed to write temporary Iroh endpoint secret")?; - temporary - .as_file() - .sync_all() - .context("failed to sync temporary Iroh endpoint secret")?; - temporary - .persist_noclobber(&path) - .map_err(|error| error.error) - .with_context(|| format!("failed to publish Iroh endpoint secret {}", path.display()))?; - sync_directory(data_directory)?; - Ok(secret) -} - -fn publish_board_metadata( - path: &Path, - document: &Doc, - upload_secrets: &[[u8; 32]], -) -> anyhow::Result<()> { - publish_directory(path, |temporary| { - write_new_file( - &temporary.join(DOCUMENT_ID_FILE), - hex::encode(document.id().to_bytes()).as_bytes(), - true, - )?; - write_new_file(&temporary.join(BOARD_FORMAT_FILE), BOARD_FORMAT, true)?; - let upload_secrets_directory = temporary.join(UPLOAD_SECRETS_DIRECTORY); - fs_err::create_dir(&upload_secrets_directory).with_context(|| { - format!( - "failed to create DKG board upload secrets {}", - upload_secrets_directory.display() - ) - })?; - for (position, secret) in upload_secrets.iter().enumerate() { - write_new_file( - &upload_secrets_directory.join(format!("participant-{}.hex", position + 1)), - hex::encode(secret).as_bytes(), - true, - )?; - } - Ok(()) - }) -} - -fn load_upload_secrets( - metadata_directory: &Path, - participant_count: usize, -) -> anyhow::Result> { - let path = metadata_directory.join(UPLOAD_SECRETS_DIRECTORY); - let entry_count = fs_err::read_dir(&path) - .with_context(|| format!("failed to read DKG board upload secrets {}", path.display()))? - .collect::, _>>()? - .len(); - ensure!( - entry_count == participant_count, - "DKG board upload secret count does not match the participant count" - ); - (1..=participant_count) - .map(|participant| { - let secret_path = path.join(format!("participant-{participant}.hex")); - let bytes = fs_err::read_to_string(&secret_path).with_context(|| { - format!("failed to read DKG board upload secret {}", secret_path.display()) - })?; - decode_fixed_hex::<32>(bytes.trim(), "DKG board upload secret") - }) - .collect() -} - -fn require_current_board_format(metadata_directory: &Path) -> anyhow::Result<()> { - let path = metadata_directory.join(BOARD_FORMAT_FILE); - let format = fs_err::read(&path) - .with_context(|| format!("failed to read DKG board format {}", path.display()))?; - ensure!( - format == BOARD_FORMAT, - "unsupported DKG board format; start a new ceremony in a new data directory" - ); - Ok(()) -} - #[cfg(test)] mod tests; From f72c68fd245382b95772949c1513131ecbcee143 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Garillot?= Date: Thu, 20 Aug 2026 17:50:53 -0400 Subject: [PATCH 04/10] refactor(validator): isolate DKG board upload protocol --- .../src/commands/dkg/board/iroh/upload.rs | 221 ++++++++++++++++++ bin/validator/src/commands/dkg/board/mod.rs | 220 +---------------- 2 files changed, 229 insertions(+), 212 deletions(-) create mode 100644 bin/validator/src/commands/dkg/board/iroh/upload.rs diff --git a/bin/validator/src/commands/dkg/board/iroh/upload.rs b/bin/validator/src/commands/dkg/board/iroh/upload.rs new file mode 100644 index 000000000..67556a802 --- /dev/null +++ b/bin/validator/src/commands/dkg/board/iroh/upload.rs @@ -0,0 +1,221 @@ +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context, ensure}; +use iroh::endpoint::Connection; +use iroh::protocol::{AcceptError, ProtocolHandler}; +use iroh::{Endpoint, EndpointAddr}; +use iroh_blobs::Hash; + +use super::{ArtifactSlot, BoardWriter, MAX_ARTIFACT_BYTES, validate_artifact_length}; + +pub(super) const UPLOAD_ALPN: &[u8] = b"/miden/storage-key-dkg-board-upload/3"; +const UPLOAD_HEADER_BYTES: usize = 32 + 1 + 4 + 8; +const UPLOAD_RESPONSE_BYTES: usize = 1 + 32; +const MAX_CONCURRENT_UPLOADS: usize = 3; +const MAX_UPLOAD_ERROR_BYTES: usize = 1024; +const UPLOAD_TIMEOUT: Duration = Duration::from_secs(30); + +#[derive(Clone, Debug)] +pub(super) struct UploadProtocol { + permits: Arc, + upload_secrets: Arc>, + writer: BoardWriter, +} + +impl UploadProtocol { + pub(super) fn new(upload_secrets: Vec<[u8; 32]>, writer: BoardWriter) -> Self { + Self { + permits: Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_UPLOADS)), + upload_secrets: Arc::new(upload_secrets), + writer, + } + } + + async fn receive(&self, recv: &mut iroh::endpoint::RecvStream) -> anyhow::Result { + let mut header = [0u8; UPLOAD_HEADER_BYTES]; + recv.read_exact(&mut header) + .await + .context("failed to read DKG board upload header")?; + let kind = header[32]; + let participant = u32::from_be_bytes(header[33..37].try_into().expect("fixed slice")); + let secret_position = usize::try_from(participant) + .context("participant index does not fit usize")? + .checked_sub(1) + .context("DKG board participant index must be nonzero")?; + let expected_secret = self + .upload_secrets + .get(secret_position) + .context("DKG board upload targets an unknown participant")?; + ensure!( + secrets_match(&header[..32], expected_secret), + "DKG board ticket does not authorize this participant" + ); + let length = u64::from_be_bytes(header[37..45].try_into().expect("fixed slice")); + ensure!( + length > 0 && length <= MAX_ARTIFACT_BYTES, + "DKG board artifact exceeds {MAX_ARTIFACT_BYTES} bytes" + ); + let slot = ArtifactSlot::from_upload_fields(kind, participant)?; + self.writer.validate_slot(&slot)?; + let length = usize::try_from(length).context("DKG board artifact length is too large")?; + let mut value = vec![0u8; length]; + recv.read_exact(&mut value) + .await + .context("failed to read DKG board upload body")?; + recv.read_to_end(0).await.context("DKG board upload has trailing bytes")?; + self.writer.store(&slot, &value).await + } + + async fn serve_connection(&self, connection: &Connection) -> Result<(), AcceptError> { + let _permit = self.permits.acquire().await.map_err(AcceptError::from_err)?; + let (mut send, mut recv) = connection.accept_bi().await?; + let response = match self.receive(&mut recv).await { + Ok(hash) => { + let mut response = Vec::with_capacity(UPLOAD_RESPONSE_BYTES); + response.push(0); + response.extend_from_slice(hash.as_bytes()); + response + }, + Err(error) => upload_error_response(&error), + }; + send.write_all(&response).await.map_err(AcceptError::from_err)?; + send.finish()?; + connection.closed().await; + Ok(()) + } +} + +impl ProtocolHandler for UploadProtocol { + async fn accept(&self, connection: Connection) -> Result<(), AcceptError> { + if let Ok(result) = + tokio::time::timeout(UPLOAD_TIMEOUT, self.serve_connection(&connection)).await + { + result + } else { + connection.close(1u32.into(), b"DKG board upload timed out"); + Ok(()) + } + } +} + +impl ArtifactSlot { + fn upload_fields(&self) -> anyhow::Result<(u8, u32)> { + let fields = match self { + Self::Registration(participant) => (1, *participant), + Self::DecryptionDealing(participant) => (2, *participant), + Self::ContextDealing(participant) => (3, *participant), + Self::TranscriptAcceptance(participant) => (4, *participant), + Self::Manifest | Self::DecryptionConfig | Self::ContextConfig => { + anyhow::bail!("only the DKG board may publish common ceremony artifacts") + }, + }; + Ok(fields) + } + + fn from_upload_fields(kind: u8, participant: u32) -> anyhow::Result { + ensure!(participant > 0, "DKG board participant index must be nonzero"); + match kind { + 1 => Ok(Self::Registration(participant)), + 2 => Ok(Self::DecryptionDealing(participant)), + 3 => Ok(Self::ContextDealing(participant)), + 4 => Ok(Self::TranscriptAcceptance(participant)), + _ => anyhow::bail!("DKG board upload contains an unknown artifact kind"), + } + } +} + +pub(super) async fn upload_artifact( + endpoint: &Endpoint, + target: &EndpointAddr, + authorized_participant: u32, + upload_secret: &[u8; 32], + slot: &ArtifactSlot, + value: &[u8], +) -> anyhow::Result { + validate_artifact_length(value.len())?; + let (kind, participant) = slot.upload_fields()?; + ensure!( + participant == authorized_participant, + "DKG board ticket does not authorize participant {participant}" + ); + upload_artifact_request( + endpoint, + target, + upload_secret, + kind, + participant, + u64::try_from(value.len()).context("artifact length does not fit u64")?, + value, + ) + .await +} + +pub(super) async fn upload_artifact_request( + endpoint: &Endpoint, + target: &EndpointAddr, + upload_secret: &[u8; 32], + kind: u8, + participant: u32, + declared_length: u64, + value: &[u8], +) -> anyhow::Result { + let connection = endpoint + .connect(target.clone(), UPLOAD_ALPN) + .await + .context("failed to connect to the DKG board upload service")?; + let (mut send, mut recv) = + connection.open_bi().await.context("failed to open a DKG board upload stream")?; + let mut header = [0u8; UPLOAD_HEADER_BYTES]; + header[..32].copy_from_slice(upload_secret); + header[32] = kind; + header[33..37].copy_from_slice(&participant.to_be_bytes()); + header[37..45].copy_from_slice(&declared_length.to_be_bytes()); + send.write_all(&header) + .await + .context("failed to write DKG board upload header")?; + send.write_all(value).await.context("failed to write DKG board upload body")?; + send.finish().context("failed to finish DKG board upload")?; + let response = tokio::time::timeout( + UPLOAD_TIMEOUT, + recv.read_to_end(UPLOAD_RESPONSE_BYTES + MAX_UPLOAD_ERROR_BYTES), + ) + .await + .context("timed out waiting for the DKG board upload response")? + .context("failed to read DKG board upload response")?; + connection.close(0u32.into(), b"upload complete"); + ensure!(!response.is_empty(), "DKG board returned an empty upload response"); + if response[0] != 0 { + let message = std::str::from_utf8(&response[1..]) + .context("DKG board returned a non-UTF-8 upload error")?; + anyhow::bail!("DKG board rejected the artifact: {message}"); + } + ensure!( + response.len() == UPLOAD_RESPONSE_BYTES, + "DKG board returned an invalid upload response" + ); + Ok(Hash::from_bytes(response[1..].try_into().expect("validated response length"))) +} + +fn upload_error_response(error: &anyhow::Error) -> Vec { + let mut message = format!("{error:#}"); + if message.len() > MAX_UPLOAD_ERROR_BYTES { + let mut end = MAX_UPLOAD_ERROR_BYTES; + while !message.is_char_boundary(end) { + end -= 1; + } + message.truncate(end); + } + let mut response = Vec::with_capacity(1 + message.len()); + response.push(1); + response.extend_from_slice(message.as_bytes()); + response +} + +fn secrets_match(candidate: &[u8], expected: &[u8; 32]) -> bool { + candidate + .iter() + .zip(expected) + .fold(0u8, |difference, (left, right)| difference | (left ^ right)) + == 0 +} diff --git a/bin/validator/src/commands/dkg/board/mod.rs b/bin/validator/src/commands/dkg/board/mod.rs index 3345efc82..0d409303d 100644 --- a/bin/validator/src/commands/dkg/board/mod.rs +++ b/bin/validator/src/commands/dkg/board/mod.rs @@ -15,8 +15,8 @@ use std::time::Duration; use anyhow::{Context, ensure}; use futures::StreamExt; -use iroh::endpoint::{Connection, presets}; -use iroh::protocol::{AcceptError, ProtocolHandler, Router}; +use iroh::endpoint::presets; +use iroh::protocol::Router; use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey}; use iroh_blobs::api::downloader::{DownloadProgressItem, Downloader}; use iroh_blobs::store::fs::FsStore; @@ -33,6 +33,8 @@ use serde::{Deserialize, Serialize}; #[path = "iroh/persistence.rs"] mod persistence; +#[path = "iroh/upload.rs"] +mod upload; #[cfg(test)] use persistence::ENDPOINT_SECRET_FILE; @@ -46,16 +48,13 @@ use persistence::{ publish_board_metadata, require_current_board_format, }; +#[cfg(test)] +use upload::upload_artifact_request; +use upload::{UPLOAD_ALPN, UploadProtocol, upload_artifact}; use super::{decode_fixed_hex, durably_create_directory_all}; -const UPLOAD_ALPN: &[u8] = b"/miden/storage-key-dkg-board-upload/3"; -const UPLOAD_HEADER_BYTES: usize = 32 + 1 + 4 + 8; -const UPLOAD_RESPONSE_BYTES: usize = 1 + 32; const MAX_ARTIFACT_BYTES: u64 = 64 * 1024 * 1024; -const MAX_CONCURRENT_UPLOADS: usize = 3; -const MAX_UPLOAD_ERROR_BYTES: usize = 1024; -const UPLOAD_TIMEOUT: Duration = Duration::from_secs(30); const PEER_READY_TIMEOUT: Duration = Duration::from_secs(30); const COMMON_ARTIFACT_COUNT: usize = 3; const ARTIFACTS_PER_PARTICIPANT: usize = 4; @@ -196,30 +195,6 @@ impl ArtifactSlot { fn key(&self, hash: Hash) -> String { format!("{}{}", self.prefix(), hash.to_hex()) } - - fn upload_fields(&self) -> anyhow::Result<(u8, u32)> { - let fields = match self { - Self::Registration(participant) => (1, *participant), - Self::DecryptionDealing(participant) => (2, *participant), - Self::ContextDealing(participant) => (3, *participant), - Self::TranscriptAcceptance(participant) => (4, *participant), - Self::Manifest | Self::DecryptionConfig | Self::ContextConfig => { - anyhow::bail!("only the DKG board may publish common ceremony artifacts") - }, - }; - Ok(fields) - } - - fn from_upload_fields(kind: u8, participant: u32) -> anyhow::Result { - ensure!(participant > 0, "DKG board participant index must be nonzero"); - match kind { - 1 => Ok(Self::Registration(participant)), - 2 => Ok(Self::DecryptionDealing(participant)), - 3 => Ok(Self::ContextDealing(participant)), - 4 => Ok(Self::TranscriptAcceptance(participant)), - _ => anyhow::bail!("DKG board upload contains an unknown artifact kind"), - } - } } #[derive(Clone, Debug)] @@ -241,13 +216,6 @@ enum Publisher { }, } -#[derive(Clone, Debug)] -struct UploadProtocol { - permits: Arc, - upload_secrets: Arc>, - writer: BoardWriter, -} - /// The read-only view shared by both board roles. pub(super) struct BoardReader { node: BoardNode, @@ -792,171 +760,6 @@ impl BoardWriter { } } -impl UploadProtocol { - async fn receive(&self, recv: &mut iroh::endpoint::RecvStream) -> anyhow::Result { - let mut header = [0u8; UPLOAD_HEADER_BYTES]; - recv.read_exact(&mut header) - .await - .context("failed to read DKG board upload header")?; - let kind = header[32]; - let participant = u32::from_be_bytes(header[33..37].try_into().expect("fixed slice")); - let secret_position = usize::try_from(participant) - .context("participant index does not fit usize")? - .checked_sub(1) - .context("DKG board participant index must be nonzero")?; - let expected_secret = self - .upload_secrets - .get(secret_position) - .context("DKG board upload targets an unknown participant")?; - ensure!( - secrets_match(&header[..32], expected_secret), - "DKG board ticket does not authorize this participant" - ); - let length = u64::from_be_bytes(header[37..45].try_into().expect("fixed slice")); - ensure!( - length > 0 && length <= MAX_ARTIFACT_BYTES, - "DKG board artifact exceeds {MAX_ARTIFACT_BYTES} bytes" - ); - let slot = ArtifactSlot::from_upload_fields(kind, participant)?; - self.writer.validate_slot(&slot)?; - let length = usize::try_from(length).context("DKG board artifact length is too large")?; - let mut value = vec![0u8; length]; - recv.read_exact(&mut value) - .await - .context("failed to read DKG board upload body")?; - recv.read_to_end(0).await.context("DKG board upload has trailing bytes")?; - self.writer.store(&slot, &value).await - } -} - -impl ProtocolHandler for UploadProtocol { - async fn accept(&self, connection: Connection) -> Result<(), AcceptError> { - if let Ok(result) = - tokio::time::timeout(UPLOAD_TIMEOUT, self.serve_connection(&connection)).await - { - result - } else { - connection.close(1u32.into(), b"DKG board upload timed out"); - Ok(()) - } - } -} - -impl UploadProtocol { - async fn serve_connection(&self, connection: &Connection) -> Result<(), AcceptError> { - let _permit = self.permits.acquire().await.map_err(AcceptError::from_err)?; - let (mut send, mut recv) = connection.accept_bi().await?; - let response = match self.receive(&mut recv).await { - Ok(hash) => { - let mut response = Vec::with_capacity(UPLOAD_RESPONSE_BYTES); - response.push(0); - response.extend_from_slice(hash.as_bytes()); - response - }, - Err(error) => upload_error_response(&error), - }; - send.write_all(&response).await.map_err(AcceptError::from_err)?; - send.finish()?; - connection.closed().await; - Ok(()) - } -} - -async fn upload_artifact( - endpoint: &Endpoint, - target: &EndpointAddr, - authorized_participant: u32, - upload_secret: &[u8; 32], - slot: &ArtifactSlot, - value: &[u8], -) -> anyhow::Result { - validate_artifact_length(value.len())?; - let (kind, participant) = slot.upload_fields()?; - ensure!( - participant == authorized_participant, - "DKG board ticket does not authorize participant {participant}" - ); - upload_artifact_request( - endpoint, - target, - upload_secret, - kind, - participant, - u64::try_from(value.len()).context("artifact length does not fit u64")?, - value, - ) - .await -} - -async fn upload_artifact_request( - endpoint: &Endpoint, - target: &EndpointAddr, - upload_secret: &[u8; 32], - kind: u8, - participant: u32, - declared_length: u64, - value: &[u8], -) -> anyhow::Result { - let connection = endpoint - .connect(target.clone(), UPLOAD_ALPN) - .await - .context("failed to connect to the DKG board upload service")?; - let (mut send, mut recv) = - connection.open_bi().await.context("failed to open a DKG board upload stream")?; - let mut header = [0u8; UPLOAD_HEADER_BYTES]; - header[..32].copy_from_slice(upload_secret); - header[32] = kind; - header[33..37].copy_from_slice(&participant.to_be_bytes()); - header[37..45].copy_from_slice(&declared_length.to_be_bytes()); - send.write_all(&header) - .await - .context("failed to write DKG board upload header")?; - send.write_all(value).await.context("failed to write DKG board upload body")?; - send.finish().context("failed to finish DKG board upload")?; - let response = tokio::time::timeout( - UPLOAD_TIMEOUT, - recv.read_to_end(UPLOAD_RESPONSE_BYTES + MAX_UPLOAD_ERROR_BYTES), - ) - .await - .context("timed out waiting for the DKG board upload response")? - .context("failed to read DKG board upload response")?; - connection.close(0u32.into(), b"upload complete"); - ensure!(!response.is_empty(), "DKG board returned an empty upload response"); - if response[0] != 0 { - let message = std::str::from_utf8(&response[1..]) - .context("DKG board returned a non-UTF-8 upload error")?; - anyhow::bail!("DKG board rejected the artifact: {message}"); - } - ensure!( - response.len() == UPLOAD_RESPONSE_BYTES, - "DKG board returned an invalid upload response" - ); - Ok(Hash::from_bytes(response[1..].try_into().expect("validated response length"))) -} - -fn upload_error_response(error: &anyhow::Error) -> Vec { - let mut message = format!("{error:#}"); - if message.len() > MAX_UPLOAD_ERROR_BYTES { - let mut end = MAX_UPLOAD_ERROR_BYTES; - while !message.is_char_boundary(end) { - end -= 1; - } - message.truncate(end); - } - let mut response = Vec::with_capacity(1 + message.len()); - response.push(1); - response.extend_from_slice(message.as_bytes()); - response -} - -fn secrets_match(candidate: &[u8], expected: &[u8; 32]) -> bool { - candidate - .iter() - .zip(expected) - .fold(0u8, |difference, (left, right)| difference | (left ^ right)) - == 0 -} - impl BoardRuntime { async fn start(data_directory: &Path, use_network_services: bool) -> anyhow::Result { durably_create_directory_all(data_directory).with_context(|| { @@ -1043,14 +846,7 @@ impl BoardRuntime { upload_secrets.len() == participant_count, "DKG board requires one upload secret per participant" ); - router = router.accept( - UPLOAD_ALPN, - UploadProtocol { - permits: Arc::new(tokio::sync::Semaphore::new(MAX_CONCURRENT_UPLOADS)), - upload_secrets: Arc::new(upload_secrets), - writer, - }, - ); + router = router.accept(UPLOAD_ALPN, UploadProtocol::new(upload_secrets, writer)); } let router = router.spawn(); let events = BoardEvents::start(&document).await?; From 1d4ed898d6f2049c69180477035e53b06029dcf1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Garillot?= Date: Thu, 20 Aug 2026 17:59:10 -0400 Subject: [PATCH 05/10] test(validator): choose a mismatched DKG ticket deterministically --- bin/validator/src/commands/dkg/tests.rs | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/bin/validator/src/commands/dkg/tests.rs b/bin/validator/src/commands/dkg/tests.rs index 5811b6034..7d42c1703 100644 --- a/bin/validator/src/commands/dkg/tests.rs +++ b/bin/validator/src/commands/dkg/tests.rs @@ -910,13 +910,23 @@ async fn runner_rejects_another_participants_ticket_before_publishing() -> TestR let (board, tickets) = board::CoordinatorBoard::create_with_network(&board_directory, 3, false).await?; let signer = ValidatorSigner::new_local(genesis.signing_keys[0].clone()); + let epoch = "66".repeat(32); + let work_directory = root.path().join("work"); + let participant = + runner::prepare_local_identity(&genesis.path, &epoch, &signer, &work_directory).await?; + let ticket = tickets + .iter() + .find(|ticket| ticket.participant() != participant.get()) + .expect("a three-participant ceremony has another participant") + .clone(); + let ticket_participant = ticket.participant(); let error = runner::run_validator_with_network::( - tickets[1].clone(), + ticket, &genesis.path, &signer, 2, - &"66".repeat(32), - &root.path().join("work"), + &epoch, + &work_directory, &root.path().join("bundle"), false, Duration::from_secs(1), @@ -924,11 +934,16 @@ async fn runner_rejects_another_participants_ticket_before_publishing() -> TestR .await .unwrap_err(); - assert!(error.to_string().contains("ticket belongs to participant 2")); + assert!( + error + .to_string() + .contains(&format!("ticket belongs to participant {ticket_participant}")), + "unexpected error: {error:#}" + ); assert!( board .reader() - .read_unique(&board::ArtifactSlot::Registration(2)) + .read_unique(&board::ArtifactSlot::Registration(ticket_participant)) .await? .is_none() ); From 642efe057b4588d11702366a6d684e2599d18d9a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Garillot?= Date: Thu, 20 Aug 2026 18:02:04 -0400 Subject: [PATCH 06/10] refactor(validator): centralize DKG board rules --- bin/validator/src/commands/dkg/board/core.rs | 160 ++++++++++++++++++ bin/validator/src/commands/dkg/board/mod.rs | 135 +++------------ bin/validator/src/commands/dkg/board/tests.rs | 17 ++ 3 files changed, 200 insertions(+), 112 deletions(-) create mode 100644 bin/validator/src/commands/dkg/board/core.rs diff --git a/bin/validator/src/commands/dkg/board/core.rs b/bin/validator/src/commands/dkg/board/core.rs new file mode 100644 index 000000000..6bcd5c787 --- /dev/null +++ b/bin/validator/src/commands/dkg/board/core.rs @@ -0,0 +1,160 @@ +use anyhow::{Context, ensure}; + +use super::{CommonArtifact, ParticipantArtifact}; + +pub(super) const MAX_ARTIFACT_BYTES: u64 = 64 * 1024 * 1024; +const COMMON_ARTIFACT_COUNT: usize = 3; +const ARTIFACTS_PER_PARTICIPANT: usize = 4; +const MAX_VALUES_PER_SLOT: usize = 2; + +/// One immutable location in a DKG ceremony board. +#[derive(Clone, Debug, Eq, PartialEq)] +pub(in crate::commands::dkg) enum ArtifactSlot { + Registration(u32), + Manifest, + DecryptionConfig, + ContextConfig, + DecryptionDealing(u32), + ContextDealing(u32), + TranscriptAcceptance(u32), +} + +impl ArtifactSlot { + pub(super) fn prefix(&self) -> String { + match self { + Self::Registration(participant) => format!("registration/{participant}/"), + Self::Manifest => "common/manifest/".to_owned(), + Self::DecryptionConfig => "common/decryption-config/".to_owned(), + Self::ContextConfig => "common/context-config/".to_owned(), + Self::DecryptionDealing(participant) => { + format!("dealing/{participant}/decryption/") + }, + Self::ContextDealing(participant) => format!("dealing/{participant}/context/"), + Self::TranscriptAcceptance(participant) => { + format!("acceptance/{participant}/signature/") + }, + } + } +} + +impl CommonArtifact { + pub(super) fn slot(self) -> ArtifactSlot { + match self { + Self::Manifest => ArtifactSlot::Manifest, + Self::DecryptionConfig => ArtifactSlot::DecryptionConfig, + Self::ContextConfig => ArtifactSlot::ContextConfig, + } + } +} + +impl ParticipantArtifact { + pub(super) fn slot(self, participant: u32) -> ArtifactSlot { + match self { + Self::Registration => ArtifactSlot::Registration(participant), + Self::DecryptionDealing => ArtifactSlot::DecryptionDealing(participant), + Self::ContextDealing => ArtifactSlot::ContextDealing(participant), + Self::TranscriptAcceptance => ArtifactSlot::TranscriptAcceptance(participant), + } + } +} + +#[derive(Clone, Debug)] +pub(super) struct BoardCore { + allowed_prefixes: Vec, + max_document_entries: usize, +} + +impl BoardCore { + pub(super) fn new(participant_count: usize) -> anyhow::Result { + ensure!(participant_count > 0, "DKG board requires at least one participant"); + let artifact_slot_count = participant_count + .checked_mul(ARTIFACTS_PER_PARTICIPANT) + .and_then(|count| count.checked_add(COMMON_ARTIFACT_COUNT)) + .context("DKG board participant count is too large")?; + let max_document_entries = artifact_slot_count + .checked_mul(MAX_VALUES_PER_SLOT) + .context("DKG board participant count is too large")?; + let mut allowed_prefixes = vec![ + ArtifactSlot::Manifest.prefix(), + ArtifactSlot::DecryptionConfig.prefix(), + ArtifactSlot::ContextConfig.prefix(), + ]; + for position in 0..participant_count { + let participant = u32::try_from(position + 1).context("too many DKG participants")?; + allowed_prefixes.extend([ + ArtifactSlot::Registration(participant).prefix(), + ArtifactSlot::DecryptionDealing(participant).prefix(), + ArtifactSlot::ContextDealing(participant).prefix(), + ArtifactSlot::TranscriptAcceptance(participant).prefix(), + ]); + } + Ok(Self { allowed_prefixes, max_document_entries }) + } + + pub(super) fn allowed_prefixes(&self) -> &[String] { + &self.allowed_prefixes + } + + pub(super) fn max_document_entries(&self) -> usize { + self.max_document_entries + } + + pub(super) fn validate_slot(&self, slot: &ArtifactSlot) -> anyhow::Result<()> { + ensure!( + self.allowed_prefixes.contains(&slot.prefix()), + "DKG board upload targets an unknown participant or artifact slot" + ); + Ok(()) + } +} + +pub(super) fn validate_artifact_length(length: usize) -> anyhow::Result<()> { + ensure!(length > 0, "DKG board artifact must not be empty"); + ensure!( + u64::try_from(length).context("artifact length does not fit u64")? <= MAX_ARTIFACT_BYTES, + "DKG board artifact exceeds {MAX_ARTIFACT_BYTES} bytes", + ); + Ok(()) +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum PublishAction { + AlreadyPresent, + Insert, +} + +pub(super) struct SlotValues { + values: Vec, +} + +impl SlotValues { + pub(super) fn from_values(values: impl IntoIterator) -> Self { + let mut unique = Vec::new(); + for value in values { + if !unique.contains(&value) { + unique.push(value); + } + } + Self { values: unique } + } + + pub(super) fn publish(&self, value: &T) -> anyhow::Result { + if self.values.contains(value) { + return Ok(PublishAction::AlreadyPresent); + } + ensure!( + self.values.len() < MAX_VALUES_PER_SLOT, + "DKG board artifact slot already contains conflicting values" + ); + Ok(PublishAction::Insert) + } + + pub(super) fn into_unique(self, slot: &ArtifactSlot) -> anyhow::Result> { + ensure!( + self.values.len() <= 1, + "DKG board contains conflicting artifacts for {}", + slot.prefix() + ); + Ok(self.values.into_iter().next()) + } +} diff --git a/bin/validator/src/commands/dkg/board/mod.rs b/bin/validator/src/commands/dkg/board/mod.rs index 0d409303d..e7bd302fb 100644 --- a/bin/validator/src/commands/dkg/board/mod.rs +++ b/bin/validator/src/commands/dkg/board/mod.rs @@ -31,11 +31,15 @@ use iroh_gossip::net::Gossip; use iroh_tickets::{ParseError, Ticket}; use serde::{Deserialize, Serialize}; +mod core; #[path = "iroh/persistence.rs"] mod persistence; #[path = "iroh/upload.rs"] mod upload; +pub(super) use core::ArtifactSlot; +use core::{BoardCore, MAX_ARTIFACT_BYTES, PublishAction, SlotValues, validate_artifact_length}; + #[cfg(test)] use persistence::ENDPOINT_SECRET_FILE; use persistence::{ @@ -54,11 +58,7 @@ use upload::{UPLOAD_ALPN, UploadProtocol, upload_artifact}; use super::{decode_fixed_hex, durably_create_directory_all}; -const MAX_ARTIFACT_BYTES: u64 = 64 * 1024 * 1024; const PEER_READY_TIMEOUT: Duration = Duration::from_secs(30); -const COMMON_ARTIFACT_COUNT: usize = 3; -const ARTIFACTS_PER_PARTICIPANT: usize = 4; -const MAX_VALUES_PER_SLOT: usize = 2; /// The board address and read capability, paired with one participant's upload permission. /// @@ -125,18 +125,6 @@ impl FromStr for BoardTicket { } } -/// One immutable location in a DKG ceremony document. -#[derive(Clone, Debug, Eq, PartialEq)] -pub(super) enum ArtifactSlot { - Registration(u32), - Manifest, - DecryptionConfig, - ContextConfig, - DecryptionDealing(u32), - ContextDealing(u32), - TranscriptAcceptance(u32), -} - /// An artifact published by the ceremony coordinator. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(super) enum CommonArtifact { @@ -145,16 +133,6 @@ pub(super) enum CommonArtifact { ContextConfig, } -impl CommonArtifact { - fn slot(self) -> ArtifactSlot { - match self { - Self::Manifest => ArtifactSlot::Manifest, - Self::DecryptionConfig => ArtifactSlot::DecryptionConfig, - Self::ContextConfig => ArtifactSlot::ContextConfig, - } - } -} - /// An artifact published by one ceremony participant. #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(super) enum ParticipantArtifact { @@ -164,34 +142,7 @@ pub(super) enum ParticipantArtifact { TranscriptAcceptance, } -impl ParticipantArtifact { - fn slot(self, participant: u32) -> ArtifactSlot { - match self { - Self::Registration => ArtifactSlot::Registration(participant), - Self::DecryptionDealing => ArtifactSlot::DecryptionDealing(participant), - Self::ContextDealing => ArtifactSlot::ContextDealing(participant), - Self::TranscriptAcceptance => ArtifactSlot::TranscriptAcceptance(participant), - } - } -} - impl ArtifactSlot { - fn prefix(&self) -> String { - match self { - Self::Registration(participant) => format!("registration/{participant}/"), - Self::Manifest => "common/manifest/".to_owned(), - Self::DecryptionConfig => "common/decryption-config/".to_owned(), - Self::ContextConfig => "common/context-config/".to_owned(), - Self::DecryptionDealing(participant) => { - format!("dealing/{participant}/decryption/") - }, - Self::ContextDealing(participant) => format!("dealing/{participant}/context/"), - Self::TranscriptAcceptance(participant) => { - format!("acceptance/{participant}/signature/") - }, - } - } - fn key(&self, hash: Hash) -> String { format!("{}{}", self.prefix(), hash.to_hex()) } @@ -200,8 +151,8 @@ impl ArtifactSlot { #[derive(Clone, Debug)] struct BoardWriter { author: iroh_docs::AuthorId, + core: Arc, document: Doc, - allowed_prefixes: Arc>, lock: Arc>, } @@ -235,12 +186,11 @@ pub(super) struct ParticipantBoard { /// A persistent Iroh node joined to one ceremony document. struct BoardNode { blobs: FsStore, + core: Arc, document: Doc, downloader: Downloader, event_error: tokio::sync::watch::Receiver>, event_task: tokio::task::JoinHandle<()>, - allowed_prefixes: Arc>, - max_document_entries: usize, peer_ready: tokio::sync::watch::Receiver, publisher: Publisher, remote_providers: std::sync::Arc>>>, @@ -643,14 +593,17 @@ impl BoardNode { ); values.entry(hash).or_insert_with(|| bytes.to_vec()); } - ensure!(values.len() <= 1, "DKG board contains conflicting artifacts for {prefix}"); - Ok(values.into_values().next()) + SlotValues::from_values(values.into_values()).into_unique(slot) } async fn validate_document_metadata(&self) -> anyhow::Result<()> { self.ensure_admitted()?; - inspect_document_metadata(&self.document, &self.allowed_prefixes, self.max_document_entries) - .await + inspect_document_metadata( + &self.document, + self.core.allowed_prefixes(), + self.core.max_document_entries(), + ) + .await } fn ensure_admitted(&self) -> anyhow::Result<()> { @@ -707,22 +660,9 @@ impl BoardNode { } } -fn validate_artifact_length(length: usize) -> anyhow::Result<()> { - ensure!(length > 0, "DKG board artifact must not be empty"); - ensure!( - u64::try_from(length).context("artifact length does not fit u64")? <= MAX_ARTIFACT_BYTES, - "DKG board artifact exceeds {MAX_ARTIFACT_BYTES} bytes", - ); - Ok(()) -} - impl BoardWriter { fn validate_slot(&self, slot: &ArtifactSlot) -> anyhow::Result<()> { - ensure!( - self.allowed_prefixes.contains(&slot.prefix()), - "DKG board upload targets an unknown participant or artifact slot" - ); - Ok(()) + self.core.validate_slot(slot) } async fn store(&self, slot: &ArtifactSlot, value: &[u8]) -> anyhow::Result { @@ -740,14 +680,11 @@ impl BoardWriter { let mut hashes = Vec::new(); while let Some(entry) = entries.next().await { let entry = entry.context("failed to read DKG board artifact slot")?; - if entry.content_hash() == expected_hash { - return Ok(expected_hash); - } hashes.push(entry.content_hash()); - ensure!( - hashes.len() < MAX_VALUES_PER_SLOT, - "DKG board artifact slot already contains conflicting values" - ); + } + match SlotValues::from_values(hashes).publish(&expected_hash)? { + PublishAction::AlreadyPresent => return Ok(expected_hash), + PublishAction::Insert => {}, } let stored_hash = self @@ -808,24 +745,17 @@ impl BoardRuntime { served_upload_secrets: Option>, remote_upload: Option<(EndpointAddr, u32, [u8; 32])>, ) -> anyhow::Result { - ensure!(participant_count > 0, "DKG board requires at least one participant"); ensure!( served_upload_secrets.is_some() ^ remote_upload.is_some(), "DKG board must either serve or submit uploads" ); - let artifact_slot_count = participant_count - .checked_mul(ARTIFACTS_PER_PARTICIPANT) - .and_then(|count| count.checked_add(COMMON_ARTIFACT_COUNT)) - .context("DKG board participant count is too large")?; - let max_document_entries = artifact_slot_count - .checked_mul(MAX_VALUES_PER_SLOT) - .context("DKG board participant count is too large")?; - let allowed_prefixes = Arc::new(allowed_slot_prefixes(participant_count)?); - inspect_document_metadata(&document, &allowed_prefixes, max_document_entries).await?; + let core = Arc::new(BoardCore::new(participant_count)?); + inspect_document_metadata(&document, core.allowed_prefixes(), core.max_document_entries()) + .await?; let writer = BoardWriter { author: self.author, + core: core.clone(), document: document.clone(), - allowed_prefixes: allowed_prefixes.clone(), lock: Arc::new(tokio::sync::Mutex::new(())), }; let publisher = match remote_upload { @@ -852,12 +782,11 @@ impl BoardRuntime { let events = BoardEvents::start(&document).await?; Ok(BoardNode { blobs: self.blobs, + core, document, downloader: self.downloader, event_error: events.error, event_task: events.task, - allowed_prefixes, - max_document_entries, peer_ready: events.peer_ready, publisher, remote_providers: events.remote_providers, @@ -929,24 +858,6 @@ impl BoardEvents { } } -fn allowed_slot_prefixes(participant_count: usize) -> anyhow::Result> { - let mut prefixes = vec![ - ArtifactSlot::Manifest.prefix(), - ArtifactSlot::DecryptionConfig.prefix(), - ArtifactSlot::ContextConfig.prefix(), - ]; - for position in 0..participant_count { - let participant = u32::try_from(position + 1).context("too many DKG participants")?; - prefixes.extend([ - ArtifactSlot::Registration(participant).prefix(), - ArtifactSlot::DecryptionDealing(participant).prefix(), - ArtifactSlot::ContextDealing(participant).prefix(), - ArtifactSlot::TranscriptAcceptance(participant).prefix(), - ]); - } - Ok(prefixes) -} - async fn inspect_document_metadata( document: &Doc, allowed_prefixes: &[String], diff --git a/bin/validator/src/commands/dkg/board/tests.rs b/bin/validator/src/commands/dkg/board/tests.rs index 51e559ec5..5e52af460 100644 --- a/bin/validator/src/commands/dkg/board/tests.rs +++ b/bin/validator/src/commands/dkg/board/tests.rs @@ -268,6 +268,23 @@ fn oversized_artifacts_are_rejected_before_allocation() { assert!(validate_artifact_length(oversized).is_err()); } +#[test] +fn shared_core_enforces_slots_idempotency_and_conflicts() -> anyhow::Result<()> { + let core = BoardCore::new(2)?; + core.validate_slot(&ArtifactSlot::Registration(2))?; + assert!(core.validate_slot(&ArtifactSlot::Registration(3)).is_err()); + + let first = b"first".to_vec(); + let second = b"second".to_vec(); + let one_value = SlotValues::from_values([first.clone()]); + assert_eq!(one_value.publish(&first)?, PublishAction::AlreadyPresent); + assert_eq!(one_value.publish(&second)?, PublishAction::Insert); + + let conflicting = SlotValues::from_values([first, second]); + assert!(conflicting.into_unique(&ArtifactSlot::Manifest).is_err()); + Ok(()) +} + #[tokio::test] async fn oversized_upload_is_rejected_before_body_allocation() -> anyhow::Result<()> { let root = tempfile::tempdir()?; From f0d5872b4da33f2b1273def11e9950eef9d37cd8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Garillot?= Date: Thu, 20 Aug 2026 18:12:15 -0400 Subject: [PATCH 07/10] refactor(validator): contain Iroh board adapter --- .../src/commands/dkg/board/iroh/mod.rs | 766 +++++++++++++++++ .../commands/dkg/board/iroh/persistence.rs | 2 +- .../commands/dkg/board/{ => iroh}/tests.rs | 0 .../src/commands/dkg/board/iroh/upload.rs | 3 +- bin/validator/src/commands/dkg/board/mod.rs | 781 +----------------- 5 files changed, 813 insertions(+), 739 deletions(-) create mode 100644 bin/validator/src/commands/dkg/board/iroh/mod.rs rename bin/validator/src/commands/dkg/board/{ => iroh}/tests.rs (100%) diff --git a/bin/validator/src/commands/dkg/board/iroh/mod.rs b/bin/validator/src/commands/dkg/board/iroh/mod.rs new file mode 100644 index 000000000..efb6767d4 --- /dev/null +++ b/bin/validator/src/commands/dkg/board/iroh/mod.rs @@ -0,0 +1,766 @@ +//! The Iroh adapter for the storage key DKG bulletin board. +//! +//! The board process is the only writer to the Iroh document. Validators receive a read-only +//! document ticket plus a participant-scoped secret for the board's bounded upload protocol. Each +//! [`ArtifactSlot`] is valid only while it holds at most one content-addressed value. A second +//! distinct value poisons that slot and stops the ceremony. This module only moves and stores +//! artifacts. The ceremony phases that use those artifacts are ordered in `runner`. + +use std::collections::BTreeMap; +use std::fmt; +use std::path::Path; +use std::str::FromStr; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{Context, ensure}; +use futures::StreamExt; +use iroh::endpoint::presets; +use iroh::protocol::Router; +use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey}; +use iroh_blobs::api::downloader::{DownloadProgressItem, Downloader}; +use iroh_blobs::store::fs::FsStore; +use iroh_blobs::{BlobsProtocol, Hash}; +use iroh_docs::DocTicket; +use iroh_docs::api::Doc; +use iroh_docs::api::protocol::{AddrInfoOptions, ShareMode}; +use iroh_docs::engine::LiveEvent; +use iroh_docs::protocol::Docs; +use iroh_docs::store::{DownloadPolicy, Query}; +use iroh_gossip::net::Gossip; +use iroh_tickets::{ParseError, Ticket}; +use serde::{Deserialize, Serialize}; + +mod persistence; +mod upload; + +#[cfg(test)] +use persistence::ENDPOINT_SECRET_FILE; +use persistence::{ + BOARD_FORMAT_FILE, + BOARD_METADATA_DIRECTORY, + DOCUMENT_ID_FILE, + UPLOAD_SECRETS_DIRECTORY, + load_or_create_endpoint_secret, + load_upload_secrets, + publish_board_metadata, + require_current_board_format, +}; +#[cfg(test)] +use upload::upload_artifact_request; +use upload::{UPLOAD_ALPN, UploadProtocol, upload_artifact}; + +use super::super::{decode_fixed_hex, durably_create_directory_all}; +use super::core::{ + ArtifactSlot, + BoardCore, + MAX_ARTIFACT_BYTES, + PublishAction, + SlotValues, + validate_artifact_length, +}; + +const PEER_READY_TIMEOUT: Duration = Duration::from_secs(30); + +/// The board address and read capability, paired with one participant's upload permission. +/// +/// This credential contains no DKG private material. Its holder can read public ceremony artifacts +/// and upload only to the named participant's slots. +#[derive(Clone, Debug, Deserialize, Serialize)] +pub(super) struct BoardTicket { + document: DocTicket, + participant: u32, + upload_secret: [u8; 32], +} + +impl BoardTicket { + pub(super) fn participant(&self) -> u32 { + self.participant + } +} + +#[derive(Deserialize, Serialize)] +enum BoardTicketWireFormat { + Variant0(BoardTicket), +} + +impl Ticket for BoardTicket { + const KIND: &'static str = "miden-storage-key-dkg-board"; + + fn encode_bytes(&self) -> Vec { + postcard::to_stdvec(&BoardTicketWireFormat::Variant0(self.clone())) + .expect("postcard serialization failed") + } + + fn decode_bytes(bytes: &[u8]) -> Result { + let BoardTicketWireFormat::Variant0(ticket) = postcard::from_bytes(bytes)?; + if ticket.participant == 0 { + return Err(ParseError::verification_failed( + "DKG board participant index must be nonzero", + )); + } + if !matches!(ticket.document.capability, iroh_docs::Capability::Read(_)) { + return Err(ParseError::verification_failed( + "DKG board document ticket must be read-only", + )); + } + if ticket.document.nodes.is_empty() { + return Err(ParseError::verification_failed( + "DKG board document addressing info cannot be empty", + )); + } + Ok(ticket) + } +} + +impl fmt::Display for BoardTicket { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(&Ticket::encode_string(self)) + } +} + +impl FromStr for BoardTicket { + type Err = ParseError; + + fn from_str(value: &str) -> Result { + Ticket::decode_string(value) + } +} + +pub(super) fn validate_ticket(value: &str) -> anyhow::Result { + Ok(BoardTicket::from_str(value)?.participant()) +} + +pub(super) async fn create( + data_directory: &Path, + participant_count: usize, + use_network_services: bool, +) -> anyhow::Result<(BoardNode, Vec<(String, u32)>)> { + let (node, tickets) = + BoardNode::create_with_network(data_directory, participant_count, use_network_services) + .await?; + let tickets = tickets + .into_iter() + .map(|ticket| { + let participant = ticket.participant(); + (ticket.to_string(), participant) + }) + .collect(); + Ok((node, tickets)) +} + +pub(super) async fn join( + data_directory: &Path, + encoded_ticket: &str, + participant_count: usize, + use_network_services: bool, +) -> anyhow::Result { + let ticket = BoardTicket::from_str(encoded_ticket)?; + BoardNode::join_with_network(data_directory, ticket, participant_count, use_network_services) + .await +} + +impl ArtifactSlot { + fn key(&self, hash: Hash) -> String { + format!("{}{}", self.prefix(), hash.to_hex()) + } +} + +#[derive(Clone, Debug)] +struct BoardWriter { + author: iroh_docs::AuthorId, + core: Arc, + document: Doc, + lock: Arc>, +} + +#[derive(Debug)] +enum Publisher { + Local(BoardWriter), + Remote { + endpoint: Endpoint, + participant: u32, + target: EndpointAddr, + upload_secret: [u8; 32], + }, +} + +/// A persistent Iroh node joined to one ceremony document. +pub(super) struct BoardNode { + blobs: FsStore, + core: Arc, + document: Doc, + downloader: Downloader, + event_error: tokio::sync::watch::Receiver>, + event_task: tokio::task::JoinHandle<()>, + peer_ready: tokio::sync::watch::Receiver, + publisher: Publisher, + remote_providers: std::sync::Arc>>>, + router: Router, + sync_generation: tokio::sync::watch::Receiver, + sync_targets: Vec, +} + +struct BoardRuntime { + author: iroh_docs::AuthorId, + blobs: FsStore, + docs: Docs, + downloader: Downloader, + endpoint: Endpoint, + gossip: Gossip, +} + +struct BoardEvents { + error: tokio::sync::watch::Receiver>, + peer_ready: tokio::sync::watch::Receiver, + remote_providers: Arc>>>, + sync_generation: tokio::sync::watch::Receiver, + task: tokio::task::JoinHandle<()>, +} + +impl Drop for BoardNode { + fn drop(&mut self) { + self.event_task.abort(); + } +} + +impl BoardNode { + pub(super) async fn create_with_network( + data_directory: &Path, + participant_count: usize, + use_network_services: bool, + ) -> anyhow::Result<(Self, Vec)> { + let runtime = BoardRuntime::start(data_directory, use_network_services).await?; + let metadata_directory = data_directory.join(BOARD_METADATA_DIRECTORY); + let (document, upload_secrets) = if metadata_directory.exists() { + require_current_board_format(&metadata_directory)?; + let document_id_path = metadata_directory.join(DOCUMENT_ID_FILE); + let id = fs_err::read_to_string(&document_id_path).with_context(|| { + format!("failed to read Iroh document ID {}", document_id_path.display()) + })?; + let id = decode_fixed_hex::<32>(id.trim(), "Iroh document ID")?; + let document = runtime + .docs + .open(iroh_docs::NamespaceId::from(&id)) + .await + .context("failed to open Iroh document")? + .context("persisted Iroh document is missing")?; + let upload_secrets = load_upload_secrets(&metadata_directory, participant_count)?; + (document, upload_secrets) + } else { + ensure!( + !data_directory.join(DOCUMENT_ID_FILE).exists() + && !data_directory.join(BOARD_FORMAT_FILE).exists() + && !data_directory.join(UPLOAD_SECRETS_DIRECTORY).exists(), + "unsupported DKG board format; start a new ceremony in a new data directory" + ); + let document = runtime.docs.create().await.context("failed to create Iroh document")?; + let upload_secrets = (0..participant_count) + .map(|_| SecretKey::generate().to_bytes()) + .collect::>(); + publish_board_metadata(&metadata_directory, &document, &upload_secrets)?; + (document, upload_secrets) + }; + document + .set_download_policy(DownloadPolicy::NothingExcept(Vec::new())) + .await + .context("failed to restrict DKG board downloads")?; + let mut document_ticket = document + .share( + ShareMode::Read, + if use_network_services { + AddrInfoOptions::RelayAndAddresses + } else { + AddrInfoOptions::Id + }, + ) + .await + .context("failed to create Iroh document ticket")?; + if !use_network_services { + let mut socket = runtime + .endpoint + .bound_sockets() + .into_iter() + .find(std::net::SocketAddr::is_ipv4) + .context("Iroh test endpoint has no IPv4 socket")?; + socket.set_ip(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)); + document_ticket.nodes = vec![iroh::EndpointAddr::from_parts( + runtime.endpoint.id(), + [iroh::TransportAddr::Ip(socket)], + )]; + } + let tickets = upload_secrets + .iter() + .enumerate() + .map(|(position, upload_secret)| { + Ok(BoardTicket { + document: document_ticket.clone(), + participant: u32::try_from(position + 1) + .context("too many DKG participants")?, + upload_secret: *upload_secret, + }) + }) + .collect::>>()?; + let board = runtime + .attach(document, participant_count, Vec::new(), Some(upload_secrets), None) + .await?; + board + .document + .start_sync(Vec::new()) + .await + .context("failed to start DKG board synchronization")?; + Ok((board, tickets)) + } + + pub(super) async fn join_with_network( + data_directory: &Path, + ticket: BoardTicket, + participant_count: usize, + use_network_services: bool, + ) -> anyhow::Result { + let runtime = BoardRuntime::start(data_directory, use_network_services).await?; + let BoardTicket { document, participant, upload_secret } = ticket; + ensure!( + usize::try_from(participant).context("participant index does not fit usize")? + <= participant_count, + "DKG board ticket names an unknown participant" + ); + let DocTicket { capability, nodes } = document; + let target = nodes.first().cloned().context("DKG board ticket has no endpoint")?; + let document = runtime + .docs + .import_namespace(capability) + .await + .context("failed to join Iroh ceremony document")?; + document + .set_download_policy(DownloadPolicy::NothingExcept(Vec::new())) + .await + .context("failed to restrict DKG board downloads")?; + let mut board = runtime + .attach( + document, + participant_count, + nodes.clone(), + None, + Some((target, participant, upload_secret)), + ) + .await?; + board + .document + .start_sync(nodes) + .await + .context("failed to start DKG board synchronization")?; + board.wait_for_peer().await?; + Ok(board) + } + + /// Publishes one artifact without replacing another value in the same slot. + pub(super) async fn publish(&self, slot: &ArtifactSlot, value: &[u8]) -> anyhow::Result { + self.ensure_admitted()?; + validate_artifact_length(value.len())?; + let expected_hash = Hash::new(value); + let sync_generation = *self.sync_generation.borrow(); + let stored_hash = match &self.publisher { + Publisher::Local(writer) => writer.store(slot, value).await?, + Publisher::Remote { + endpoint, + participant, + target, + upload_secret, + } => { + upload_artifact(endpoint, target, *participant, upload_secret, slot, value).await? + }, + }; + ensure!(stored_hash == expected_hash, "Iroh stored artifact under an unexpected hash"); + self.document + .start_sync(self.sync_targets.clone()) + .await + .context("failed to synchronize DKG board artifact")?; + if !self.sync_targets.is_empty() || *self.peer_ready.borrow() { + let mut completed = self.sync_generation.clone(); + tokio::time::timeout( + PEER_READY_TIMEOUT, + completed.wait_for(|generation| *generation > sync_generation), + ) + .await + .context("timed out synchronizing DKG board artifact")? + .context("DKG board synchronization monitor stopped")?; + } + Ok(stored_hash) + } + + /// Reads the unique content value published for one artifact slot. + pub(super) async fn read_unique(&self, slot: &ArtifactSlot) -> anyhow::Result>> { + self.validate_document_metadata().await?; + let prefix = slot.prefix(); + let entries = self + .document + .get_many(Query::key_prefix(prefix.as_bytes())) + .await + .context("failed to query DKG board artifacts")?; + futures::pin_mut!(entries); + let mut values = BTreeMap::new(); + while let Some(entry) = entries.next().await { + let entry = entry.context("failed to read DKG board entry")?; + ensure!( + entry.content_len() > 0 && entry.content_len() <= MAX_ARTIFACT_BYTES, + "DKG board artifact exceeds {MAX_ARTIFACT_BYTES} bytes", + ); + let expected_key = slot.key(entry.content_hash()); + ensure!( + entry.key() == expected_key.as_bytes(), + "DKG board key does not match its content hash" + ); + let hash = entry.content_hash(); + if self.blobs.blobs().get_bytes(hash).await.is_err() { + let mut providers = + self.remote_providers.read().await.get(&hash).cloned().unwrap_or_default(); + let sync_peers = self + .document + .get_sync_peers() + .await + .context("failed to list DKG board peers")? + .unwrap_or_default() + .into_iter() + .map(|id| EndpointId::from_bytes(&id).context("invalid DKG board peer ID")) + .collect::>>()?; + for peer in sync_peers { + if !providers.contains(&peer) { + providers.push(peer); + } + } + if providers.is_empty() { + return Ok(None); + } + let Ok(mut progress) = self.downloader.download(hash, providers).stream().await + else { + return Ok(None); + }; + while let Some(item) = progress.next().await { + match item { + DownloadProgressItem::Progress(downloaded) => ensure!( + downloaded <= MAX_ARTIFACT_BYTES, + "DKG board artifact exceeds {MAX_ARTIFACT_BYTES} bytes", + ), + DownloadProgressItem::Error(_) | DownloadProgressItem::DownloadError => { + return Ok(None); + }, + DownloadProgressItem::TryProvider { .. } + | DownloadProgressItem::ProviderFailed { .. } + | DownloadProgressItem::PartComplete { .. } => {}, + } + } + } + let bytes = self + .blobs + .blobs() + .get_bytes(hash) + .await + .context("downloaded DKG board artifact is missing")?; + ensure!( + u64::try_from(bytes.len()).context("artifact length does not fit u64")? + == entry.content_len(), + "DKG board artifact length does not match its entry" + ); + values.entry(hash).or_insert_with(|| bytes.to_vec()); + } + SlotValues::from_values(values.into_values()).into_unique(slot) + } + + async fn validate_document_metadata(&self) -> anyhow::Result<()> { + self.ensure_admitted()?; + inspect_document_metadata( + &self.document, + self.core.allowed_prefixes(), + self.core.max_document_entries(), + ) + .await + } + + fn ensure_admitted(&self) -> anyhow::Result<()> { + if let Some(error) = self.event_error.borrow().as_ref() { + anyhow::bail!("DKG board synchronization stopped: {error}"); + } + Ok(()) + } + + async fn wait_for_peer(&mut self) -> anyhow::Result<()> { + if *self.peer_ready.borrow() { + return Ok(()); + } + tokio::time::timeout(PEER_READY_TIMEOUT, self.peer_ready.wait_for(|ready| *ready)) + .await + .context("timed out waiting for the DKG board peer")? + .context("DKG board peer monitor stopped")?; + Ok(()) + } + + /// Waits until one unique artifact has synchronized locally. + pub(super) async fn wait_unique( + &self, + slot: &ArtifactSlot, + timeout: Duration, + ) -> anyhow::Result> { + let mut events = self + .document + .subscribe() + .await + .context("failed to subscribe to DKG board updates")?; + tokio::time::timeout(timeout, async { + loop { + if let Some(value) = self.read_unique(slot).await? { + return Ok(value); + } + tokio::select! { + event = events.next() => { + event.transpose()?.context("DKG board update stream ended")?; + }, + () = tokio::time::sleep(Duration::from_millis(250)) => {}, + } + } + }) + .await + .with_context(|| format!("timed out waiting for DKG board slot {}", slot.prefix()))? + } + + /// Stops the board node and flushes its persistent stores. + pub(super) async fn shutdown(self) -> anyhow::Result<()> { + self.event_task.abort(); + self.router.shutdown().await.context("failed to stop Iroh board node")?; + Ok(()) + } +} + +impl BoardWriter { + fn validate_slot(&self, slot: &ArtifactSlot) -> anyhow::Result<()> { + self.core.validate_slot(slot) + } + + async fn store(&self, slot: &ArtifactSlot, value: &[u8]) -> anyhow::Result { + validate_artifact_length(value.len())?; + self.validate_slot(slot)?; + let prefix = slot.prefix(); + let expected_hash = Hash::new(value); + let _guard = self.lock.lock().await; + let entries = self + .document + .get_many(Query::key_prefix(prefix.as_bytes())) + .await + .context("failed to inspect DKG board artifact slot")?; + futures::pin_mut!(entries); + let mut hashes = Vec::new(); + while let Some(entry) = entries.next().await { + let entry = entry.context("failed to read DKG board artifact slot")?; + hashes.push(entry.content_hash()); + } + match SlotValues::from_values(hashes).publish(&expected_hash)? { + PublishAction::AlreadyPresent => return Ok(expected_hash), + PublishAction::Insert => {}, + } + + let stored_hash = self + .document + .set_bytes(self.author, slot.key(expected_hash), value.to_vec()) + .await + .context("failed to publish DKG board artifact")?; + ensure!(stored_hash == expected_hash, "Iroh stored artifact under an unexpected hash"); + Ok(stored_hash) + } +} + +impl BoardRuntime { + async fn start(data_directory: &Path, use_network_services: bool) -> anyhow::Result { + durably_create_directory_all(data_directory).with_context(|| { + format!("failed to create Iroh data directory {}", data_directory.display()) + })?; + let secret = load_or_create_endpoint_secret(data_directory)?; + let builder = if use_network_services { + Endpoint::builder(presets::N0) + } else { + Endpoint::builder(presets::Minimal) + }; + let endpoint = builder + .secret_key(secret) + .bind() + .await + .context("failed to bind Iroh endpoint")?; + let blobs_directory = data_directory.join("blobs"); + let docs_directory = data_directory.join("docs"); + fs_err::create_dir_all(&blobs_directory).context("failed to create Iroh blob directory")?; + fs_err::create_dir_all(&docs_directory) + .context("failed to create Iroh document directory")?; + let blobs = + FsStore::load(blobs_directory).await.context("failed to load Iroh blob store")?; + let downloader = blobs.downloader(&endpoint); + let gossip = Gossip::builder().spawn(endpoint.clone()); + let docs = Docs::persistent(docs_directory) + .spawn(endpoint.clone(), blobs.as_ref().clone(), gossip.clone()) + .await + .context("failed to load Iroh document store")?; + let author = docs.author_default().await.context("failed to load Iroh author")?; + Ok(Self { + author, + blobs, + docs, + downloader, + endpoint, + gossip, + }) + } + + async fn attach( + self, + document: Doc, + participant_count: usize, + sync_targets: Vec, + served_upload_secrets: Option>, + remote_upload: Option<(EndpointAddr, u32, [u8; 32])>, + ) -> anyhow::Result { + ensure!( + served_upload_secrets.is_some() ^ remote_upload.is_some(), + "DKG board must either serve or submit uploads" + ); + let core = Arc::new(BoardCore::new(participant_count)?); + inspect_document_metadata(&document, core.allowed_prefixes(), core.max_document_entries()) + .await?; + let writer = BoardWriter { + author: self.author, + core: core.clone(), + document: document.clone(), + lock: Arc::new(tokio::sync::Mutex::new(())), + }; + let publisher = match remote_upload { + Some((target, participant, upload_secret)) => Publisher::Remote { + endpoint: self.endpoint.clone(), + participant, + target, + upload_secret, + }, + None => Publisher::Local(writer.clone()), + }; + let mut router = Router::builder(self.endpoint) + .accept(iroh_blobs::ALPN, BlobsProtocol::new(self.blobs.as_ref(), None)) + .accept(iroh_gossip::ALPN, self.gossip) + .accept(iroh_docs::ALPN, self.docs.clone()); + if let Some(upload_secrets) = served_upload_secrets { + ensure!( + upload_secrets.len() == participant_count, + "DKG board requires one upload secret per participant" + ); + router = router.accept(UPLOAD_ALPN, UploadProtocol::new(upload_secrets, writer)); + } + let router = router.spawn(); + let events = BoardEvents::start(&document).await?; + Ok(BoardNode { + blobs: self.blobs, + core, + document, + downloader: self.downloader, + event_error: events.error, + event_task: events.task, + peer_ready: events.peer_ready, + publisher, + remote_providers: events.remote_providers, + router, + sync_generation: events.sync_generation, + sync_targets, + }) + } +} + +impl BoardEvents { + async fn start(document: &Doc) -> anyhow::Result { + let mut events = + document.subscribe().await.context("failed to start DKG board event monitor")?; + let (event_tx, error) = tokio::sync::watch::channel(None); + let (peer_ready_tx, peer_ready) = tokio::sync::watch::channel(false); + let (sync_generation_tx, sync_generation) = tokio::sync::watch::channel(0u64); + let remote_providers = + Arc::new(tokio::sync::RwLock::>>::default()); + let monitored_providers = remote_providers.clone(); + let task = tokio::spawn(async move { + let mut neighbor_ready = false; + let mut sync_ready = false; + while let Some(event) = events.next().await { + let event = match event { + Ok(event) => event, + Err(error) => { + event_tx.send_replace(Some(error.to_string())); + break; + }, + }; + match &event { + LiveEvent::NeighborUp(_) => { + neighbor_ready = true; + if sync_ready { + peer_ready_tx.send_replace(true); + } + }, + LiveEvent::NeighborDown(_) => { + neighbor_ready = false; + sync_ready = false; + peer_ready_tx.send_replace(false); + }, + LiveEvent::SyncFinished(sync) if sync.result.is_ok() => { + sync_ready = true; + sync_generation_tx.send_modify(|generation| *generation += 1); + if neighbor_ready { + peer_ready_tx.send_replace(true); + } + }, + _ => {}, + } + if let LiveEvent::InsertRemote { from, entry, .. } = &event { + let mut providers = monitored_providers.write().await; + let providers = providers.entry(entry.content_hash()).or_default(); + if !providers.contains(from) { + providers.push(*from); + } + } + } + }); + Ok(Self { + error, + peer_ready, + remote_providers, + sync_generation, + task, + }) + } +} + +async fn inspect_document_metadata( + document: &Doc, + allowed_prefixes: &[String], + max_document_entries: usize, +) -> anyhow::Result<()> { + let entries = document.get_many(Query::all()).await.context("failed to inspect DKG board")?; + futures::pin_mut!(entries); + let mut slots = BTreeMap::new(); + let mut count = 0usize; + while let Some(entry) = entries.next().await { + let entry = entry.context("failed to read DKG board entry")?; + count += 1; + ensure!(count <= max_document_entries, "DKG board contains too many entries"); + ensure!( + entry.content_len() > 0 && entry.content_len() <= MAX_ARTIFACT_BYTES, + "DKG board artifact exceeds {MAX_ARTIFACT_BYTES} bytes", + ); + let key = std::str::from_utf8(entry.key()).context("DKG board key is not UTF-8")?; + let (prefix, hash) = allowed_prefixes + .iter() + .find_map(|prefix| key.strip_prefix(prefix).map(|hash| (prefix, hash))) + .context("DKG board contains an unrecognized artifact slot")?; + ensure!( + hash.len() == 64 && hash.bytes().all(|byte| byte.is_ascii_hexdigit()), + "DKG board key has an invalid content hash" + ); + if let Some(previous) = slots.insert(prefix.clone(), hash.to_owned()) { + ensure!(previous == hash, "DKG board contains conflicting artifacts for {prefix}"); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests; diff --git a/bin/validator/src/commands/dkg/board/iroh/persistence.rs b/bin/validator/src/commands/dkg/board/iroh/persistence.rs index 9b215de83..3e01f99aa 100644 --- a/bin/validator/src/commands/dkg/board/iroh/persistence.rs +++ b/bin/validator/src/commands/dkg/board/iroh/persistence.rs @@ -6,7 +6,7 @@ use anyhow::{Context, ensure}; use iroh::SecretKey; use iroh_docs::api::Doc; -use super::super::{decode_fixed_hex, publish_directory, sync_directory, write_new_file}; +use super::super::super::{decode_fixed_hex, publish_directory, sync_directory, write_new_file}; pub(super) const ENDPOINT_SECRET_FILE: &str = "endpoint-secret.hex"; pub(super) const BOARD_METADATA_DIRECTORY: &str = "board-meta"; diff --git a/bin/validator/src/commands/dkg/board/tests.rs b/bin/validator/src/commands/dkg/board/iroh/tests.rs similarity index 100% rename from bin/validator/src/commands/dkg/board/tests.rs rename to bin/validator/src/commands/dkg/board/iroh/tests.rs diff --git a/bin/validator/src/commands/dkg/board/iroh/upload.rs b/bin/validator/src/commands/dkg/board/iroh/upload.rs index 67556a802..21a4ce8af 100644 --- a/bin/validator/src/commands/dkg/board/iroh/upload.rs +++ b/bin/validator/src/commands/dkg/board/iroh/upload.rs @@ -7,7 +7,8 @@ use iroh::protocol::{AcceptError, ProtocolHandler}; use iroh::{Endpoint, EndpointAddr}; use iroh_blobs::Hash; -use super::{ArtifactSlot, BoardWriter, MAX_ARTIFACT_BYTES, validate_artifact_length}; +use super::super::core::{ArtifactSlot, MAX_ARTIFACT_BYTES, validate_artifact_length}; +use super::BoardWriter; pub(super) const UPLOAD_ALPN: &[u8] = b"/miden/storage-key-dkg-board-upload/3"; const UPLOAD_HEADER_BYTES: usize = 32 + 1 + 4 + 8; diff --git a/bin/validator/src/commands/dkg/board/mod.rs b/bin/validator/src/commands/dkg/board/mod.rs index e7bd302fb..169950132 100644 --- a/bin/validator/src/commands/dkg/board/mod.rs +++ b/bin/validator/src/commands/dkg/board/mod.rs @@ -1,127 +1,60 @@ //! A bounded, append-only exchange for storage key DKG artifacts. //! -//! The board process is the only writer to the Iroh document. Validators receive a read-only -//! document ticket plus a participant-scoped secret for the board's bounded upload protocol. Each -//! [`ArtifactSlot`] is valid only while it holds at most one content-addressed value. A second -//! distinct value poisons that slot and stops the ceremony. This module only moves and stores -//! artifacts. The ceremony phases that use those artifacts are ordered in `runner`. +//! The ceremony runner sees role-specific boards and typed artifact slots. Transport setup, +//! credentials, synchronization, and persistence stay behind the private adapters. -use std::collections::BTreeMap; use std::fmt; use std::path::Path; use std::str::FromStr; -use std::sync::Arc; use std::time::Duration; -use anyhow::{Context, ensure}; -use futures::StreamExt; -use iroh::endpoint::presets; -use iroh::protocol::Router; -use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey}; -use iroh_blobs::api::downloader::{DownloadProgressItem, Downloader}; -use iroh_blobs::store::fs::FsStore; -use iroh_blobs::{BlobsProtocol, Hash}; -use iroh_docs::DocTicket; -use iroh_docs::api::Doc; -use iroh_docs::api::protocol::{AddrInfoOptions, ShareMode}; -use iroh_docs::engine::LiveEvent; -use iroh_docs::protocol::Docs; -use iroh_docs::store::{DownloadPolicy, Query}; -use iroh_gossip::net::Gossip; -use iroh_tickets::{ParseError, Ticket}; -use serde::{Deserialize, Serialize}; - mod core; -#[path = "iroh/persistence.rs"] -mod persistence; -#[path = "iroh/upload.rs"] -mod upload; +mod iroh; pub(super) use core::ArtifactSlot; -use core::{BoardCore, MAX_ARTIFACT_BYTES, PublishAction, SlotValues, validate_artifact_length}; - -#[cfg(test)] -use persistence::ENDPOINT_SECRET_FILE; -use persistence::{ - BOARD_FORMAT_FILE, - BOARD_METADATA_DIRECTORY, - DOCUMENT_ID_FILE, - UPLOAD_SECRETS_DIRECTORY, - load_or_create_endpoint_secret, - load_upload_secrets, - publish_board_metadata, - require_current_board_format, -}; -#[cfg(test)] -use upload::upload_artifact_request; -use upload::{UPLOAD_ALPN, UploadProtocol, upload_artifact}; - -use super::{decode_fixed_hex, durably_create_directory_all}; -const PEER_READY_TIMEOUT: Duration = Duration::from_secs(30); - -/// The board address and read capability, paired with one participant's upload permission. -/// -/// This credential contains no DKG private material. Its holder can read public ceremony artifacts -/// and upload only to the named participant's slots. -#[derive(Clone, Debug, Deserialize, Serialize)] +/// An opaque board address and one participant's publish permission. +#[derive(Clone)] pub(super) struct BoardTicket { - document: DocTicket, + encoded: String, participant: u32, - upload_secret: [u8; 32], } impl BoardTicket { + fn new(encoded: String, participant: u32) -> Self { + Self { encoded, participant } + } + pub(super) fn participant(&self) -> u32 { self.participant } -} - -#[derive(Deserialize, Serialize)] -enum BoardTicketWireFormat { - Variant0(BoardTicket), -} -impl Ticket for BoardTicket { - const KIND: &'static str = "miden-storage-key-dkg-board"; - - fn encode_bytes(&self) -> Vec { - postcard::to_stdvec(&BoardTicketWireFormat::Variant0(self.clone())) - .expect("postcard serialization failed") + fn into_encoded(self) -> String { + self.encoded } +} - fn decode_bytes(bytes: &[u8]) -> Result { - let BoardTicketWireFormat::Variant0(ticket) = postcard::from_bytes(bytes)?; - if ticket.participant == 0 { - return Err(ParseError::verification_failed( - "DKG board participant index must be nonzero", - )); - } - if !matches!(ticket.document.capability, iroh_docs::Capability::Read(_)) { - return Err(ParseError::verification_failed( - "DKG board document ticket must be read-only", - )); - } - if ticket.document.nodes.is_empty() { - return Err(ParseError::verification_failed( - "DKG board document addressing info cannot be empty", - )); - } - Ok(ticket) +impl fmt::Debug for BoardTicket { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter + .debug_struct("BoardTicket") + .field("participant", &self.participant) + .finish_non_exhaustive() } } impl fmt::Display for BoardTicket { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { - formatter.write_str(&Ticket::encode_string(self)) + formatter.write_str(&self.encoded) } } impl FromStr for BoardTicket { - type Err = ParseError; + type Err = anyhow::Error; fn from_str(value: &str) -> Result { - Ticket::decode_string(value) + let participant = iroh::validate_ticket(value)?; + Ok(Self::new(value.to_owned(), participant)) } } @@ -142,78 +75,9 @@ pub(super) enum ParticipantArtifact { TranscriptAcceptance, } -impl ArtifactSlot { - fn key(&self, hash: Hash) -> String { - format!("{}{}", self.prefix(), hash.to_hex()) - } -} - -#[derive(Clone, Debug)] -struct BoardWriter { - author: iroh_docs::AuthorId, - core: Arc, - document: Doc, - lock: Arc>, -} - -#[derive(Debug)] -enum Publisher { - Local(BoardWriter), - Remote { - endpoint: Endpoint, - participant: u32, - target: EndpointAddr, - upload_secret: [u8; 32], - }, -} - /// The read-only view shared by both board roles. pub(super) struct BoardReader { - node: BoardNode, -} - -/// The board role that coordinates and publishes common ceremony artifacts. -pub(super) struct CoordinatorBoard { - reader: BoardReader, -} - -/// The board role held by one ceremony participant. -pub(super) struct ParticipantBoard { - participant: u32, - reader: BoardReader, -} - -/// A persistent Iroh node joined to one ceremony document. -struct BoardNode { - blobs: FsStore, - core: Arc, - document: Doc, - downloader: Downloader, - event_error: tokio::sync::watch::Receiver>, - event_task: tokio::task::JoinHandle<()>, - peer_ready: tokio::sync::watch::Receiver, - publisher: Publisher, - remote_providers: std::sync::Arc>>>, - router: Router, - sync_generation: tokio::sync::watch::Receiver, - sync_targets: Vec, -} - -struct BoardRuntime { - author: iroh_docs::AuthorId, - blobs: FsStore, - docs: Docs, - downloader: Downloader, - endpoint: Endpoint, - gossip: Gossip, -} - -struct BoardEvents { - error: tokio::sync::watch::Receiver>, - peer_ready: tokio::sync::watch::Receiver, - remote_providers: Arc>>>, - sync_generation: tokio::sync::watch::Receiver, - task: tokio::task::JoinHandle<()>, + node: iroh::BoardNode, } impl BoardReader { @@ -233,25 +97,31 @@ impl BoardReader { } } +/// The board role that coordinates and publishes common ceremony artifacts. +pub(super) struct CoordinatorBoard { + reader: BoardReader, +} + impl CoordinatorBoard { /// Creates or resumes a ceremony board and returns one scoped ticket per participant. pub(super) async fn create( data_directory: &Path, participant_count: usize, ) -> anyhow::Result<(Self, Vec)> { - let (node, tickets) = BoardNode::create(data_directory, participant_count).await?; - Ok((Self { reader: BoardReader { node } }, tickets)) + Self::create_with_network(data_directory, participant_count, true).await } - #[cfg(test)] pub(super) async fn create_with_network( data_directory: &Path, participant_count: usize, use_network_services: bool, ) -> anyhow::Result<(Self, Vec)> { let (node, tickets) = - BoardNode::create_with_network(data_directory, participant_count, use_network_services) - .await?; + iroh::create(data_directory, participant_count, use_network_services).await?; + let tickets = tickets + .into_iter() + .map(|(encoded, participant)| BoardTicket::new(encoded, participant)) + .collect(); Ok((Self { reader: BoardReader { node } }, tickets)) } @@ -275,6 +145,12 @@ impl CoordinatorBoard { } } +/// The board role held by one ceremony participant. +pub(super) struct ParticipantBoard { + participant: u32, + reader: BoardReader, +} + impl ParticipantBoard { /// Joins or resumes a ceremony board through a read and upload ticket. pub(super) async fn join( @@ -282,12 +158,7 @@ impl ParticipantBoard { ticket: BoardTicket, participant_count: usize, ) -> anyhow::Result { - let participant = ticket.participant(); - let node = BoardNode::join(data_directory, ticket, participant_count).await?; - Ok(Self { - participant, - reader: BoardReader { node }, - }) + Self::join_with_network(data_directory, ticket, participant_count, true).await } pub(super) async fn join_with_network( @@ -297,9 +168,9 @@ impl ParticipantBoard { use_network_services: bool, ) -> anyhow::Result { let participant = ticket.participant(); - let node = BoardNode::join_with_network( + let node = iroh::join( data_directory, - ticket, + &ticket.into_encoded(), participant_count, use_network_services, ) @@ -329,567 +200,3 @@ impl ParticipantBoard { self.reader.node.shutdown().await } } - -impl Drop for BoardNode { - fn drop(&mut self) { - self.event_task.abort(); - } -} - -impl BoardNode { - /// Creates a new ceremony document and returns one scoped ticket per participant. - pub(super) async fn create( - data_directory: &Path, - participant_count: usize, - ) -> anyhow::Result<(Self, Vec)> { - Self::create_with_network(data_directory, participant_count, true).await - } - - pub(super) async fn create_with_network( - data_directory: &Path, - participant_count: usize, - use_network_services: bool, - ) -> anyhow::Result<(Self, Vec)> { - let runtime = BoardRuntime::start(data_directory, use_network_services).await?; - let metadata_directory = data_directory.join(BOARD_METADATA_DIRECTORY); - let (document, upload_secrets) = if metadata_directory.exists() { - require_current_board_format(&metadata_directory)?; - let document_id_path = metadata_directory.join(DOCUMENT_ID_FILE); - let id = fs_err::read_to_string(&document_id_path).with_context(|| { - format!("failed to read Iroh document ID {}", document_id_path.display()) - })?; - let id = decode_fixed_hex::<32>(id.trim(), "Iroh document ID")?; - let document = runtime - .docs - .open(iroh_docs::NamespaceId::from(&id)) - .await - .context("failed to open Iroh document")? - .context("persisted Iroh document is missing")?; - let upload_secrets = load_upload_secrets(&metadata_directory, participant_count)?; - (document, upload_secrets) - } else { - ensure!( - !data_directory.join(DOCUMENT_ID_FILE).exists() - && !data_directory.join(BOARD_FORMAT_FILE).exists() - && !data_directory.join(UPLOAD_SECRETS_DIRECTORY).exists(), - "unsupported DKG board format; start a new ceremony in a new data directory" - ); - let document = runtime.docs.create().await.context("failed to create Iroh document")?; - let upload_secrets = (0..participant_count) - .map(|_| SecretKey::generate().to_bytes()) - .collect::>(); - publish_board_metadata(&metadata_directory, &document, &upload_secrets)?; - (document, upload_secrets) - }; - document - .set_download_policy(DownloadPolicy::NothingExcept(Vec::new())) - .await - .context("failed to restrict DKG board downloads")?; - let mut document_ticket = document - .share( - ShareMode::Read, - if use_network_services { - AddrInfoOptions::RelayAndAddresses - } else { - AddrInfoOptions::Id - }, - ) - .await - .context("failed to create Iroh document ticket")?; - if !use_network_services { - let mut socket = runtime - .endpoint - .bound_sockets() - .into_iter() - .find(std::net::SocketAddr::is_ipv4) - .context("Iroh test endpoint has no IPv4 socket")?; - socket.set_ip(std::net::IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)); - document_ticket.nodes = vec![iroh::EndpointAddr::from_parts( - runtime.endpoint.id(), - [iroh::TransportAddr::Ip(socket)], - )]; - } - let tickets = upload_secrets - .iter() - .enumerate() - .map(|(position, upload_secret)| { - Ok(BoardTicket { - document: document_ticket.clone(), - participant: u32::try_from(position + 1) - .context("too many DKG participants")?, - upload_secret: *upload_secret, - }) - }) - .collect::>>()?; - let board = runtime - .attach(document, participant_count, Vec::new(), Some(upload_secrets), None) - .await?; - board - .document - .start_sync(Vec::new()) - .await - .context("failed to start DKG board synchronization")?; - Ok((board, tickets)) - } - - /// Joins an existing ceremony document through its read and upload ticket. - pub(super) async fn join( - data_directory: &Path, - ticket: BoardTicket, - participant_count: usize, - ) -> anyhow::Result { - Self::join_with_network(data_directory, ticket, participant_count, true).await - } - - pub(super) async fn join_with_network( - data_directory: &Path, - ticket: BoardTicket, - participant_count: usize, - use_network_services: bool, - ) -> anyhow::Result { - let runtime = BoardRuntime::start(data_directory, use_network_services).await?; - let BoardTicket { document, participant, upload_secret } = ticket; - ensure!( - usize::try_from(participant).context("participant index does not fit usize")? - <= participant_count, - "DKG board ticket names an unknown participant" - ); - let DocTicket { capability, nodes } = document; - let target = nodes.first().cloned().context("DKG board ticket has no endpoint")?; - let document = runtime - .docs - .import_namespace(capability) - .await - .context("failed to join Iroh ceremony document")?; - document - .set_download_policy(DownloadPolicy::NothingExcept(Vec::new())) - .await - .context("failed to restrict DKG board downloads")?; - let mut board = runtime - .attach( - document, - participant_count, - nodes.clone(), - None, - Some((target, participant, upload_secret)), - ) - .await?; - board - .document - .start_sync(nodes) - .await - .context("failed to start DKG board synchronization")?; - board.wait_for_peer().await?; - Ok(board) - } - - /// Publishes one artifact without replacing another value in the same slot. - pub(super) async fn publish(&self, slot: &ArtifactSlot, value: &[u8]) -> anyhow::Result { - self.ensure_admitted()?; - validate_artifact_length(value.len())?; - let expected_hash = Hash::new(value); - let sync_generation = *self.sync_generation.borrow(); - let stored_hash = match &self.publisher { - Publisher::Local(writer) => writer.store(slot, value).await?, - Publisher::Remote { - endpoint, - participant, - target, - upload_secret, - } => { - upload_artifact(endpoint, target, *participant, upload_secret, slot, value).await? - }, - }; - ensure!(stored_hash == expected_hash, "Iroh stored artifact under an unexpected hash"); - self.document - .start_sync(self.sync_targets.clone()) - .await - .context("failed to synchronize DKG board artifact")?; - if !self.sync_targets.is_empty() || *self.peer_ready.borrow() { - let mut completed = self.sync_generation.clone(); - tokio::time::timeout( - PEER_READY_TIMEOUT, - completed.wait_for(|generation| *generation > sync_generation), - ) - .await - .context("timed out synchronizing DKG board artifact")? - .context("DKG board synchronization monitor stopped")?; - } - Ok(stored_hash) - } - - /// Reads the unique content value published for one artifact slot. - pub(super) async fn read_unique(&self, slot: &ArtifactSlot) -> anyhow::Result>> { - self.validate_document_metadata().await?; - let prefix = slot.prefix(); - let entries = self - .document - .get_many(Query::key_prefix(prefix.as_bytes())) - .await - .context("failed to query DKG board artifacts")?; - futures::pin_mut!(entries); - let mut values = BTreeMap::new(); - while let Some(entry) = entries.next().await { - let entry = entry.context("failed to read DKG board entry")?; - ensure!( - entry.content_len() > 0 && entry.content_len() <= MAX_ARTIFACT_BYTES, - "DKG board artifact exceeds {MAX_ARTIFACT_BYTES} bytes", - ); - let expected_key = slot.key(entry.content_hash()); - ensure!( - entry.key() == expected_key.as_bytes(), - "DKG board key does not match its content hash" - ); - let hash = entry.content_hash(); - if self.blobs.blobs().get_bytes(hash).await.is_err() { - let mut providers = - self.remote_providers.read().await.get(&hash).cloned().unwrap_or_default(); - let sync_peers = self - .document - .get_sync_peers() - .await - .context("failed to list DKG board peers")? - .unwrap_or_default() - .into_iter() - .map(|id| EndpointId::from_bytes(&id).context("invalid DKG board peer ID")) - .collect::>>()?; - for peer in sync_peers { - if !providers.contains(&peer) { - providers.push(peer); - } - } - if providers.is_empty() { - return Ok(None); - } - let Ok(mut progress) = self.downloader.download(hash, providers).stream().await - else { - return Ok(None); - }; - while let Some(item) = progress.next().await { - match item { - DownloadProgressItem::Progress(downloaded) => ensure!( - downloaded <= MAX_ARTIFACT_BYTES, - "DKG board artifact exceeds {MAX_ARTIFACT_BYTES} bytes", - ), - DownloadProgressItem::Error(_) | DownloadProgressItem::DownloadError => { - return Ok(None); - }, - DownloadProgressItem::TryProvider { .. } - | DownloadProgressItem::ProviderFailed { .. } - | DownloadProgressItem::PartComplete { .. } => {}, - } - } - } - let bytes = self - .blobs - .blobs() - .get_bytes(hash) - .await - .context("downloaded DKG board artifact is missing")?; - ensure!( - u64::try_from(bytes.len()).context("artifact length does not fit u64")? - == entry.content_len(), - "DKG board artifact length does not match its entry" - ); - values.entry(hash).or_insert_with(|| bytes.to_vec()); - } - SlotValues::from_values(values.into_values()).into_unique(slot) - } - - async fn validate_document_metadata(&self) -> anyhow::Result<()> { - self.ensure_admitted()?; - inspect_document_metadata( - &self.document, - self.core.allowed_prefixes(), - self.core.max_document_entries(), - ) - .await - } - - fn ensure_admitted(&self) -> anyhow::Result<()> { - if let Some(error) = self.event_error.borrow().as_ref() { - anyhow::bail!("DKG board synchronization stopped: {error}"); - } - Ok(()) - } - - async fn wait_for_peer(&mut self) -> anyhow::Result<()> { - if *self.peer_ready.borrow() { - return Ok(()); - } - tokio::time::timeout(PEER_READY_TIMEOUT, self.peer_ready.wait_for(|ready| *ready)) - .await - .context("timed out waiting for the DKG board peer")? - .context("DKG board peer monitor stopped")?; - Ok(()) - } - - /// Waits until one unique artifact has synchronized locally. - pub(super) async fn wait_unique( - &self, - slot: &ArtifactSlot, - timeout: Duration, - ) -> anyhow::Result> { - let mut events = self - .document - .subscribe() - .await - .context("failed to subscribe to DKG board updates")?; - tokio::time::timeout(timeout, async { - loop { - if let Some(value) = self.read_unique(slot).await? { - return Ok(value); - } - tokio::select! { - event = events.next() => { - event.transpose()?.context("DKG board update stream ended")?; - }, - () = tokio::time::sleep(Duration::from_millis(250)) => {}, - } - } - }) - .await - .with_context(|| format!("timed out waiting for DKG board slot {}", slot.prefix()))? - } - - /// Stops the board node and flushes its persistent stores. - pub(super) async fn shutdown(self) -> anyhow::Result<()> { - self.event_task.abort(); - self.router.shutdown().await.context("failed to stop Iroh board node")?; - Ok(()) - } -} - -impl BoardWriter { - fn validate_slot(&self, slot: &ArtifactSlot) -> anyhow::Result<()> { - self.core.validate_slot(slot) - } - - async fn store(&self, slot: &ArtifactSlot, value: &[u8]) -> anyhow::Result { - validate_artifact_length(value.len())?; - self.validate_slot(slot)?; - let prefix = slot.prefix(); - let expected_hash = Hash::new(value); - let _guard = self.lock.lock().await; - let entries = self - .document - .get_many(Query::key_prefix(prefix.as_bytes())) - .await - .context("failed to inspect DKG board artifact slot")?; - futures::pin_mut!(entries); - let mut hashes = Vec::new(); - while let Some(entry) = entries.next().await { - let entry = entry.context("failed to read DKG board artifact slot")?; - hashes.push(entry.content_hash()); - } - match SlotValues::from_values(hashes).publish(&expected_hash)? { - PublishAction::AlreadyPresent => return Ok(expected_hash), - PublishAction::Insert => {}, - } - - let stored_hash = self - .document - .set_bytes(self.author, slot.key(expected_hash), value.to_vec()) - .await - .context("failed to publish DKG board artifact")?; - ensure!(stored_hash == expected_hash, "Iroh stored artifact under an unexpected hash"); - Ok(stored_hash) - } -} - -impl BoardRuntime { - async fn start(data_directory: &Path, use_network_services: bool) -> anyhow::Result { - durably_create_directory_all(data_directory).with_context(|| { - format!("failed to create Iroh data directory {}", data_directory.display()) - })?; - let secret = load_or_create_endpoint_secret(data_directory)?; - let builder = if use_network_services { - Endpoint::builder(presets::N0) - } else { - Endpoint::builder(presets::Minimal) - }; - let endpoint = builder - .secret_key(secret) - .bind() - .await - .context("failed to bind Iroh endpoint")?; - let blobs_directory = data_directory.join("blobs"); - let docs_directory = data_directory.join("docs"); - fs_err::create_dir_all(&blobs_directory).context("failed to create Iroh blob directory")?; - fs_err::create_dir_all(&docs_directory) - .context("failed to create Iroh document directory")?; - let blobs = - FsStore::load(blobs_directory).await.context("failed to load Iroh blob store")?; - let downloader = blobs.downloader(&endpoint); - let gossip = Gossip::builder().spawn(endpoint.clone()); - let docs = Docs::persistent(docs_directory) - .spawn(endpoint.clone(), blobs.as_ref().clone(), gossip.clone()) - .await - .context("failed to load Iroh document store")?; - let author = docs.author_default().await.context("failed to load Iroh author")?; - Ok(Self { - author, - blobs, - docs, - downloader, - endpoint, - gossip, - }) - } - - async fn attach( - self, - document: Doc, - participant_count: usize, - sync_targets: Vec, - served_upload_secrets: Option>, - remote_upload: Option<(EndpointAddr, u32, [u8; 32])>, - ) -> anyhow::Result { - ensure!( - served_upload_secrets.is_some() ^ remote_upload.is_some(), - "DKG board must either serve or submit uploads" - ); - let core = Arc::new(BoardCore::new(participant_count)?); - inspect_document_metadata(&document, core.allowed_prefixes(), core.max_document_entries()) - .await?; - let writer = BoardWriter { - author: self.author, - core: core.clone(), - document: document.clone(), - lock: Arc::new(tokio::sync::Mutex::new(())), - }; - let publisher = match remote_upload { - Some((target, participant, upload_secret)) => Publisher::Remote { - endpoint: self.endpoint.clone(), - participant, - target, - upload_secret, - }, - None => Publisher::Local(writer.clone()), - }; - let mut router = Router::builder(self.endpoint) - .accept(iroh_blobs::ALPN, BlobsProtocol::new(self.blobs.as_ref(), None)) - .accept(iroh_gossip::ALPN, self.gossip) - .accept(iroh_docs::ALPN, self.docs.clone()); - if let Some(upload_secrets) = served_upload_secrets { - ensure!( - upload_secrets.len() == participant_count, - "DKG board requires one upload secret per participant" - ); - router = router.accept(UPLOAD_ALPN, UploadProtocol::new(upload_secrets, writer)); - } - let router = router.spawn(); - let events = BoardEvents::start(&document).await?; - Ok(BoardNode { - blobs: self.blobs, - core, - document, - downloader: self.downloader, - event_error: events.error, - event_task: events.task, - peer_ready: events.peer_ready, - publisher, - remote_providers: events.remote_providers, - router, - sync_generation: events.sync_generation, - sync_targets, - }) - } -} - -impl BoardEvents { - async fn start(document: &Doc) -> anyhow::Result { - let mut events = - document.subscribe().await.context("failed to start DKG board event monitor")?; - let (event_tx, error) = tokio::sync::watch::channel(None); - let (peer_ready_tx, peer_ready) = tokio::sync::watch::channel(false); - let (sync_generation_tx, sync_generation) = tokio::sync::watch::channel(0u64); - let remote_providers = - Arc::new(tokio::sync::RwLock::>>::default()); - let monitored_providers = remote_providers.clone(); - let task = tokio::spawn(async move { - let mut neighbor_ready = false; - let mut sync_ready = false; - while let Some(event) = events.next().await { - let event = match event { - Ok(event) => event, - Err(error) => { - event_tx.send_replace(Some(error.to_string())); - break; - }, - }; - match &event { - LiveEvent::NeighborUp(_) => { - neighbor_ready = true; - if sync_ready { - peer_ready_tx.send_replace(true); - } - }, - LiveEvent::NeighborDown(_) => { - neighbor_ready = false; - sync_ready = false; - peer_ready_tx.send_replace(false); - }, - LiveEvent::SyncFinished(sync) if sync.result.is_ok() => { - sync_ready = true; - sync_generation_tx.send_modify(|generation| *generation += 1); - if neighbor_ready { - peer_ready_tx.send_replace(true); - } - }, - _ => {}, - } - if let LiveEvent::InsertRemote { from, entry, .. } = &event { - let mut providers = monitored_providers.write().await; - let providers = providers.entry(entry.content_hash()).or_default(); - if !providers.contains(from) { - providers.push(*from); - } - } - } - }); - Ok(Self { - error, - peer_ready, - remote_providers, - sync_generation, - task, - }) - } -} - -async fn inspect_document_metadata( - document: &Doc, - allowed_prefixes: &[String], - max_document_entries: usize, -) -> anyhow::Result<()> { - let entries = document.get_many(Query::all()).await.context("failed to inspect DKG board")?; - futures::pin_mut!(entries); - let mut slots = BTreeMap::new(); - let mut count = 0usize; - while let Some(entry) = entries.next().await { - let entry = entry.context("failed to read DKG board entry")?; - count += 1; - ensure!(count <= max_document_entries, "DKG board contains too many entries"); - ensure!( - entry.content_len() > 0 && entry.content_len() <= MAX_ARTIFACT_BYTES, - "DKG board artifact exceeds {MAX_ARTIFACT_BYTES} bytes", - ); - let key = std::str::from_utf8(entry.key()).context("DKG board key is not UTF-8")?; - let (prefix, hash) = allowed_prefixes - .iter() - .find_map(|prefix| key.strip_prefix(prefix).map(|hash| (prefix, hash))) - .context("DKG board contains an unrecognized artifact slot")?; - ensure!( - hash.len() == 64 && hash.bytes().all(|byte| byte.is_ascii_hexdigit()), - "DKG board key has an invalid content hash" - ); - if let Some(previous) = slots.insert(prefix.clone(), hash.to_owned()) { - ensure!(previous == hash, "DKG board contains conflicting artifacts for {prefix}"); - } - } - Ok(()) -} - -#[cfg(test)] -mod tests; From 90c20e75475bd14c3cf4feb564b1eeb2eb0b6855 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Garillot?= Date: Thu, 20 Aug 2026 18:20:59 -0400 Subject: [PATCH 08/10] test(validator): add memory DKG board adapter --- .../src/commands/dkg/board/iroh/mod.rs | 1 + .../src/commands/dkg/board/memory.rs | 67 +++++++++++++++ bin/validator/src/commands/dkg/board/mod.rs | 86 +++++++++++++++++-- bin/validator/src/commands/dkg/board/tests.rs | 79 +++++++++++++++++ 4 files changed, 226 insertions(+), 7 deletions(-) create mode 100644 bin/validator/src/commands/dkg/board/memory.rs create mode 100644 bin/validator/src/commands/dkg/board/tests.rs diff --git a/bin/validator/src/commands/dkg/board/iroh/mod.rs b/bin/validator/src/commands/dkg/board/iroh/mod.rs index efb6767d4..079b857fb 100644 --- a/bin/validator/src/commands/dkg/board/iroh/mod.rs +++ b/bin/validator/src/commands/dkg/board/iroh/mod.rs @@ -391,6 +391,7 @@ impl BoardNode { /// Reads the unique content value published for one artifact slot. pub(super) async fn read_unique(&self, slot: &ArtifactSlot) -> anyhow::Result>> { + self.core.validate_slot(slot)?; self.validate_document_metadata().await?; let prefix = slot.prefix(); let entries = self diff --git a/bin/validator/src/commands/dkg/board/memory.rs b/bin/validator/src/commands/dkg/board/memory.rs new file mode 100644 index 000000000..4e1151bdd --- /dev/null +++ b/bin/validator/src/commands/dkg/board/memory.rs @@ -0,0 +1,67 @@ +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::Duration; + +use super::core::{ArtifactSlot, BoardCore, PublishAction, SlotValues, validate_artifact_length}; + +#[derive(Clone)] +pub(super) struct BoardNode { + inner: Arc, +} + +struct Inner { + core: BoardCore, + changed: tokio::sync::Notify, + values: tokio::sync::Mutex>>>, +} + +pub(super) fn create(participant_count: usize) -> anyhow::Result> { + let inner = Arc::new(Inner { + core: BoardCore::new(participant_count)?, + changed: tokio::sync::Notify::new(), + values: tokio::sync::Mutex::new(BTreeMap::new()), + }); + Ok((0..=participant_count).map(|_| BoardNode { inner: inner.clone() }).collect()) +} + +impl BoardNode { + pub(super) async fn publish(&self, slot: &ArtifactSlot, value: &[u8]) -> anyhow::Result<()> { + validate_artifact_length(value.len())?; + self.inner.core.validate_slot(slot)?; + let mut state = self.inner.values.lock().await; + let values = state.entry(slot.prefix()).or_default(); + match SlotValues::from_values(values.iter().cloned()).publish(&value.to_vec())? { + PublishAction::AlreadyPresent => return Ok(()), + PublishAction::Insert => values.push(value.to_vec()), + } + drop(state); + self.inner.changed.notify_waiters(); + Ok(()) + } + + pub(super) async fn read_unique(&self, slot: &ArtifactSlot) -> anyhow::Result>> { + self.inner.core.validate_slot(slot)?; + let state = self.inner.values.lock().await; + let values = state.get(&slot.prefix()).cloned().unwrap_or_default(); + SlotValues::from_values(values).into_unique(slot) + } + + pub(super) async fn wait_unique( + &self, + slot: &ArtifactSlot, + timeout: Duration, + ) -> anyhow::Result> { + tokio::time::timeout(timeout, async { + loop { + let changed = self.inner.changed.notified(); + if let Some(value) = self.read_unique(slot).await? { + return Ok(value); + } + changed.await; + } + }) + .await + .map_err(Into::into) + .and_then(|result| result) + } +} diff --git a/bin/validator/src/commands/dkg/board/mod.rs b/bin/validator/src/commands/dkg/board/mod.rs index 169950132..822863f3a 100644 --- a/bin/validator/src/commands/dkg/board/mod.rs +++ b/bin/validator/src/commands/dkg/board/mod.rs @@ -10,6 +10,10 @@ use std::time::Duration; mod core; mod iroh; +#[cfg(test)] +mod memory; +#[cfg(test)] +mod tests; pub(super) use core::ArtifactSlot; @@ -77,7 +81,50 @@ pub(super) enum ParticipantArtifact { /// The read-only view shared by both board roles. pub(super) struct BoardReader { - node: iroh::BoardNode, + node: Transport, +} + +enum Transport { + Iroh(Box), + #[cfg(test)] + Memory(memory::BoardNode), +} + +impl Transport { + async fn publish(&self, slot: &ArtifactSlot, value: &[u8]) -> anyhow::Result<()> { + match self { + Self::Iroh(node) => { + node.publish(slot, value).await?; + Ok(()) + }, + #[cfg(test)] + Self::Memory(node) => node.publish(slot, value).await, + } + } + + #[cfg(test)] + async fn read_unique(&self, slot: &ArtifactSlot) -> anyhow::Result>> { + match self { + Self::Iroh(node) => node.read_unique(slot).await, + Self::Memory(node) => node.read_unique(slot).await, + } + } + + async fn wait_unique(&self, slot: &ArtifactSlot, timeout: Duration) -> anyhow::Result> { + match self { + Self::Iroh(node) => node.wait_unique(slot, timeout).await, + #[cfg(test)] + Self::Memory(node) => node.wait_unique(slot, timeout).await, + } + } + + async fn shutdown(self) -> anyhow::Result<()> { + match self { + Self::Iroh(node) => (*node).shutdown().await, + #[cfg(test)] + Self::Memory(_) => Ok(()), + } + } } impl BoardReader { @@ -122,7 +169,34 @@ impl CoordinatorBoard { .into_iter() .map(|(encoded, participant)| BoardTicket::new(encoded, participant)) .collect(); - Ok((Self { reader: BoardReader { node } }, tickets)) + Ok(( + Self { + reader: BoardReader { node: Transport::Iroh(Box::new(node)) }, + }, + tickets, + )) + } + + #[cfg(test)] + pub(super) fn create_memory( + participant_count: usize, + ) -> anyhow::Result<(Self, Vec)> { + let mut nodes = memory::create(participant_count)?.into_iter(); + let coordinator = Self { + reader: BoardReader { + node: Transport::Memory(nodes.next().expect("memory board includes a coordinator")), + }, + }; + let participants = nodes + .enumerate() + .map(|(position, node)| { + Ok(ParticipantBoard { + participant: u32::try_from(position + 1)?, + reader: BoardReader { node: Transport::Memory(node) }, + }) + }) + .collect::>>()?; + Ok((coordinator, participants)) } pub(super) fn reader(&self) -> &BoardReader { @@ -135,8 +209,7 @@ impl CoordinatorBoard { artifact: CommonArtifact, value: &[u8], ) -> anyhow::Result<()> { - self.reader.node.publish(&artifact.slot(), value).await?; - Ok(()) + self.reader.node.publish(&artifact.slot(), value).await } /// Stops the board and flushes its persistent stores. @@ -177,7 +250,7 @@ impl ParticipantBoard { .await?; Ok(Self { participant, - reader: BoardReader { node }, + reader: BoardReader { node: Transport::Iroh(Box::new(node)) }, }) } @@ -191,8 +264,7 @@ impl ParticipantBoard { artifact: ParticipantArtifact, value: &[u8], ) -> anyhow::Result<()> { - self.reader.node.publish(&artifact.slot(self.participant), value).await?; - Ok(()) + self.reader.node.publish(&artifact.slot(self.participant), value).await } /// Stops the board and flushes its persistent stores. diff --git a/bin/validator/src/commands/dkg/board/tests.rs b/bin/validator/src/commands/dkg/board/tests.rs new file mode 100644 index 000000000..b572f99a7 --- /dev/null +++ b/bin/validator/src/commands/dkg/board/tests.rs @@ -0,0 +1,79 @@ +use std::time::Duration; + +use super::*; + +const WAIT_TIMEOUT: Duration = Duration::from_secs(10); + +async fn assert_board_contract( + coordinator: &CoordinatorBoard, + participants: &[ParticipantBoard], +) -> anyhow::Result<()> { + coordinator.publish(CommonArtifact::Manifest, b"manifest").await?; + coordinator.publish(CommonArtifact::Manifest, b"manifest").await?; + assert_eq!( + participants[0] + .reader() + .wait_unique(&ArtifactSlot::Manifest, WAIT_TIMEOUT) + .await?, + b"manifest" + ); + + participants[0] + .publish(ParticipantArtifact::Registration, b"registration") + .await?; + participants[0] + .publish(ParticipantArtifact::Registration, b"registration") + .await?; + assert_eq!( + coordinator + .reader() + .wait_unique(&ArtifactSlot::Registration(1), WAIT_TIMEOUT) + .await?, + b"registration" + ); + + assert!(coordinator.publish(CommonArtifact::ContextConfig, b"").await.is_err()); + assert!(coordinator.reader().read_unique(&ArtifactSlot::Registration(3)).await.is_err()); + + coordinator.publish(CommonArtifact::Manifest, b"other manifest").await?; + assert!(coordinator.reader().read_unique(&ArtifactSlot::Manifest).await.is_err()); + Ok(()) +} + +async fn shutdown_boards( + coordinator: CoordinatorBoard, + participants: Vec, +) -> anyhow::Result<()> { + for participant in participants { + participant.shutdown().await?; + } + coordinator.shutdown().await +} + +#[tokio::test] +async fn memory_adapter_obeys_board_contract() -> anyhow::Result<()> { + let (coordinator, participants) = CoordinatorBoard::create_memory(2)?; + assert_board_contract(&coordinator, &participants).await?; + shutdown_boards(coordinator, participants).await +} + +#[tokio::test] +async fn iroh_adapter_obeys_board_contract() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let (coordinator, tickets) = + CoordinatorBoard::create_with_network(&root.path().join("coordinator"), 2, false).await?; + let mut participants = Vec::new(); + for (position, ticket) in tickets.into_iter().enumerate() { + participants.push( + ParticipantBoard::join_with_network( + &root.path().join(format!("participant-{}", position + 1)), + ticket, + 2, + false, + ) + .await?, + ); + } + assert_board_contract(&coordinator, &participants).await?; + shutdown_boards(coordinator, participants).await +} From a42aa55565560bfdd8b95db129d3b91ffe5dd4a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Garillot?= Date: Thu, 20 Aug 2026 18:25:15 -0400 Subject: [PATCH 09/10] test(validator): cover blocked DKG board reads --- bin/validator/src/commands/dkg/board/tests.rs | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/bin/validator/src/commands/dkg/board/tests.rs b/bin/validator/src/commands/dkg/board/tests.rs index b572f99a7..1cc4e684a 100644 --- a/bin/validator/src/commands/dkg/board/tests.rs +++ b/bin/validator/src/commands/dkg/board/tests.rs @@ -8,15 +8,17 @@ async fn assert_board_contract( coordinator: &CoordinatorBoard, participants: &[ParticipantBoard], ) -> anyhow::Result<()> { + let wait_for_manifest = + participants[0].reader().wait_unique(&ArtifactSlot::Manifest, WAIT_TIMEOUT); + let publish_manifest = async { + tokio::task::yield_now().await; + coordinator.publish(CommonArtifact::Manifest, b"manifest").await + }; + let (manifest, publish_result) = tokio::join!(wait_for_manifest, publish_manifest); + publish_result?; + assert_eq!(manifest?, b"manifest"); + coordinator.publish(CommonArtifact::Manifest, b"manifest").await?; - coordinator.publish(CommonArtifact::Manifest, b"manifest").await?; - assert_eq!( - participants[0] - .reader() - .wait_unique(&ArtifactSlot::Manifest, WAIT_TIMEOUT) - .await?, - b"manifest" - ); participants[0] .publish(ParticipantArtifact::Registration, b"registration") From 2eaf083a55870795fa2a9f6e41671aab7ba572d2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Garillot?= Date: Thu, 20 Aug 2026 18:30:47 -0400 Subject: [PATCH 10/10] refactor(validator): inject DKG boards into runner --- .../src/commands/dkg/board/iroh/tests.rs | 32 ---- bin/validator/src/commands/dkg/runner.rs | 17 +- bin/validator/src/commands/dkg/tests.rs | 153 +++++++++++++++--- 3 files changed, 134 insertions(+), 68 deletions(-) diff --git a/bin/validator/src/commands/dkg/board/iroh/tests.rs b/bin/validator/src/commands/dkg/board/iroh/tests.rs index 5e52af460..238a51369 100644 --- a/bin/validator/src/commands/dkg/board/iroh/tests.rs +++ b/bin/validator/src/commands/dkg/board/iroh/tests.rs @@ -99,21 +99,6 @@ async fn artifact_syncs_between_board_nodes() -> anyhow::Result<()> { Ok(()) } -#[tokio::test] -async fn conflicting_artifacts_are_rejected() -> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let (host, _) = BoardNode::create_for_test(&root.path().join("host")).await?; - let slot = ArtifactSlot::Manifest; - - host.publish(&slot, b"first").await?; - host.publish(&slot, b"second").await?; - let error = host.read_unique(&slot).await.unwrap_err(); - assert!(error.to_string().contains("conflicting artifacts")); - - host.shutdown().await?; - Ok(()) -} - #[tokio::test] async fn board_reopens_the_same_document_after_restart() -> anyhow::Result<()> { let root = tempfile::tempdir()?; @@ -268,23 +253,6 @@ fn oversized_artifacts_are_rejected_before_allocation() { assert!(validate_artifact_length(oversized).is_err()); } -#[test] -fn shared_core_enforces_slots_idempotency_and_conflicts() -> anyhow::Result<()> { - let core = BoardCore::new(2)?; - core.validate_slot(&ArtifactSlot::Registration(2))?; - assert!(core.validate_slot(&ArtifactSlot::Registration(3)).is_err()); - - let first = b"first".to_vec(); - let second = b"second".to_vec(); - let one_value = SlotValues::from_values([first.clone()]); - assert_eq!(one_value.publish(&first)?, PublishAction::AlreadyPresent); - assert_eq!(one_value.publish(&second)?, PublishAction::Insert); - - let conflicting = SlotValues::from_values([first, second]); - assert!(conflicting.into_unique(&ArtifactSlot::Manifest).is_err()); - Ok(()) -} - #[tokio::test] async fn oversized_upload_is_rejected_before_body_allocation() -> anyhow::Result<()> { let root = tempfile::tempdir()?; diff --git a/bin/validator/src/commands/dkg/runner.rs b/bin/validator/src/commands/dkg/runner.rs index 5d8a33177..df690379f 100644 --- a/bin/validator/src/commands/dkg/runner.rs +++ b/bin/validator/src/commands/dkg/runner.rs @@ -128,7 +128,7 @@ pub(super) async fn run_validator(options: DkgRunOptions) -> anyhow::Result<()> ensure!(!board.is_empty(), "storage key DKG board ticket must not be empty"); let board = board.parse::().context("invalid storage key DKG board ticket")?; let signer = options.signing_key.into_signer().await?; - run_validator_with_network::( + run_validator_with_ticket::( board, &options.genesis, &signer, @@ -136,7 +136,6 @@ pub(super) async fn run_validator(options: DkgRunOptions) -> anyhow::Result<()> &options.epoch, &options.work_directory, &options.output_directory, - true, CEREMONY_WAIT_TIMEOUT, ) .await @@ -260,9 +259,9 @@ async fn wait_for_registrations( /// Runs the restartable validator state machine over one board. #[expect( clippy::too_many_arguments, - reason = "the inputs separate ceremony policy, durable paths, and test networking" + reason = "the inputs separate ceremony policy from durable paths" )] -pub(super) async fn run_validator_with_network( +pub(super) async fn run_validator_with_ticket( ticket: BoardTicket, genesis_path: &Path, signer: &ValidatorSigner, @@ -270,7 +269,6 @@ pub(super) async fn run_validator_with_network( epoch: &str, work_directory: &Path, output_directory: &Path, - use_network_services: bool, timeout: Duration, ) -> anyhow::Result<()> where @@ -295,12 +293,7 @@ where participant.get(), ); let board_directory = work_directory.join(BOARD_DIRECTORY); - let board = if use_network_services { - ParticipantBoard::join(&board_directory, ticket, participant_count).await? - } else { - ParticipantBoard::join_with_network(&board_directory, ticket, participant_count, false) - .await? - }; + let board = ParticipantBoard::join(&board_directory, ticket, participant_count).await?; let result = run_validator_on_board::( &board, genesis_path, @@ -322,7 +315,7 @@ where clippy::too_many_lines, reason = "the inputs and linear body mirror the ceremony policy and phase order" )] -async fn run_validator_on_board( +pub(super) async fn run_validator_on_board( board: &ParticipantBoard, genesis_path: &Path, signer: &ValidatorSigner, diff --git a/bin/validator/src/commands/dkg/tests.rs b/bin/validator/src/commands/dkg/tests.rs index 7d42c1703..0a3bc9362 100644 --- a/bin/validator/src/commands/dkg/tests.rs +++ b/bin/validator/src/commands/dkg/tests.rs @@ -920,7 +920,7 @@ async fn runner_rejects_another_participants_ticket_before_publishing() -> TestR .expect("a three-participant ceremony has another participant") .clone(); let ticket_participant = ticket.participant(); - let error = runner::run_validator_with_network::( + let error = runner::run_validator_with_ticket::( ticket, &genesis.path, &signer, @@ -928,7 +928,6 @@ async fn runner_rejects_another_participants_ticket_before_publishing() -> TestR &epoch, &work_directory, &root.path().join("bundle"), - false, Duration::from_secs(1), ) .await @@ -951,6 +950,104 @@ async fn runner_rejects_another_participants_ticket_before_publishing() -> TestR Ok(()) } +fn assert_completed_bundles(bundle_directories: &[PathBuf]) -> TestResult { + let shared_setup = fs_err::read(bundle_directories[0].join(SETUP_CONTEXT_FILE))?; + let shared_public_keys = fs_err::read(bundle_directories[0].join(PUBLIC_KEY_SET_FILE))?; + let secret_shares = bundle_directories + .iter() + .map(|bundle| fs_err::read(bundle.join(SECRET_SHARE_FILE))) + .collect::, _>>()?; + assert!(bundle_directories.iter().all(|bundle| { + fs_err::read(bundle.join(SETUP_CONTEXT_FILE)).unwrap() == shared_setup + && fs_err::read(bundle.join(PUBLIC_KEY_SET_FILE)).unwrap() == shared_public_keys + })); + assert_ne!(secret_shares[0], secret_shares[1]); + assert_ne!(secret_shares[1], secret_shares[2]); + assert_ne!(secret_shares[0], secret_shares[2]); + Ok(()) +} + +#[tokio::test] +async fn memory_board_runs_complete_ceremony() -> TestResult { + let root = tempfile::tempdir()?; + let genesis = write_genesis(root.path())?; + let board_directory = root.path().join("board"); + fs_err::create_dir(&board_directory)?; + let (board, participant_boards) = board::CoordinatorBoard::create_memory(3)?; + let timeout = Duration::from_mins(2); + let epoch = "66".repeat(32); + let signers = genesis + .signing_keys + .iter() + .cloned() + .map(ValidatorSigner::new_local) + .collect::>(); + let work_directories = (1..=3) + .map(|participant| root.path().join(format!("work-{participant}"))) + .collect::>(); + let bundle_directories = (1..=3) + .map(|participant| root.path().join(format!("bundle-{participant}"))) + .collect::>(); + let mut participants = Vec::new(); + for (signer, work_directory) in signers.iter().zip(&work_directories) { + fs_err::create_dir(work_directory)?; + let participant = + runner::prepare_local_identity(&genesis.path, &epoch, signer, work_directory).await?; + let position = usize::try_from(participant.get() - 1)?; + participants.push((participant, position)); + } + + let coordinate = runner::coordinate_common_files( + &board, + &board_directory, + &genesis.path, + 2, + &epoch, + timeout, + ); + let first = runner::run_validator_on_board::( + &participant_boards[participants[0].1], + &genesis.path, + &signers[0], + participants[0].0, + 2, + &epoch, + &work_directories[0], + &bundle_directories[0], + timeout, + ); + let second = runner::run_validator_on_board::( + &participant_boards[participants[1].1], + &genesis.path, + &signers[1], + participants[1].0, + 2, + &epoch, + &work_directories[1], + &bundle_directories[1], + timeout, + ); + let third = runner::run_validator_on_board::( + &participant_boards[participants[2].1], + &genesis.path, + &signers[2], + participants[2].0, + 2, + &epoch, + &work_directories[2], + &bundle_directories[2], + timeout, + ); + tokio::try_join!(coordinate, first, second, third)?; + assert_completed_bundles(&bundle_directories)?; + + for participant_board in participant_boards { + participant_board.shutdown().await?; + } + board.shutdown().await?; + Ok(()) +} + #[tokio::test] /// Proves a validator can resume from its saved identity after its ceremony process stops. #[expect( @@ -1015,6 +1112,23 @@ async fn iroh_ceremony_resumes_after_validator_restart() -> TestResult { let bundle_directories = (1..=3) .map(|participant| root.path().join(format!("bundle-{participant}"))) .collect::>(); + let mut participant_indices = Vec::new(); + let mut participant_boards = Vec::new(); + for ((signer, ticket), work_directory) in signers.iter().zip(&tickets).zip(&work_directories) { + fs_err::create_dir_all(work_directory)?; + participant_indices.push( + runner::prepare_local_identity(&genesis.path, &epoch, signer, work_directory).await?, + ); + participant_boards.push( + board::ParticipantBoard::join_with_network( + &work_directory.join("board"), + ticket.clone(), + 3, + false, + ) + .await?, + ); + } let coordinate = runner::coordinate_common_files( &board, &board_directory, @@ -1023,54 +1137,42 @@ async fn iroh_ceremony_resumes_after_validator_restart() -> TestResult { &epoch, timeout, ); - let first = runner::run_validator_with_network::( - tickets[0].clone(), + let first = runner::run_validator_on_board::( + &participant_boards[0], &genesis.path, &signers[0], + participant_indices[0], 2, &epoch, &work_directories[0], &bundle_directories[0], - false, timeout, ); - let second = runner::run_validator_with_network::( - tickets[1].clone(), + let second = runner::run_validator_on_board::( + &participant_boards[1], &genesis.path, &signers[1], + participant_indices[1], 2, &epoch, &work_directories[1], &bundle_directories[1], - false, timeout, ); - let third = runner::run_validator_with_network::( - tickets[2].clone(), + let third = runner::run_validator_on_board::( + &participant_boards[2], &genesis.path, &signers[2], + participant_indices[2], 2, &epoch, &work_directories[2], &bundle_directories[2], - false, timeout, ); tokio::try_join!(coordinate, first, second, third)?; - let shared_setup = fs_err::read(bundle_directories[0].join(SETUP_CONTEXT_FILE))?; - let shared_public_keys = fs_err::read(bundle_directories[0].join(PUBLIC_KEY_SET_FILE))?; - let secret_shares = bundle_directories - .iter() - .map(|bundle| fs_err::read(bundle.join(SECRET_SHARE_FILE))) - .collect::, _>>()?; - assert!(bundle_directories.iter().all(|bundle| { - fs_err::read(bundle.join(SETUP_CONTEXT_FILE)).unwrap() == shared_setup - && fs_err::read(bundle.join(PUBLIC_KEY_SET_FILE)).unwrap() == shared_public_keys - })); - assert_ne!(secret_shares[0], secret_shares[1]); - assert_ne!(secret_shares[1], secret_shares[2]); - assert_ne!(secret_shares[0], secret_shares[2]); + assert_completed_bundles(&bundle_directories)?; let common_files = [MANIFEST_FILE, DECRYPTION_CONFIG_FILE, CONTEXT_CONFIG_FILE]; assert!(common_files.iter().all(|name| { let expected = fs_err::read(board_directory.join("ceremony").join(name)).unwrap(); @@ -1079,6 +1181,9 @@ async fn iroh_ceremony_resumes_after_validator_restart() -> TestResult { .all(|work| fs_err::read(work.join("ceremony").join(name)).unwrap() == expected) })); + for participant_board in participant_boards { + participant_board.shutdown().await?; + } board.shutdown().await?; Ok(()) }