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.rs b/bin/validator/src/commands/dkg/board/iroh/mod.rs similarity index 61% rename from bin/validator/src/commands/dkg/board.rs rename to bin/validator/src/commands/dkg/board/iroh/mod.rs index bf3462ac0..079b857fb 100644 --- a/bin/validator/src/commands/dkg/board.rs +++ b/bin/validator/src/commands/dkg/board/iroh/mod.rs @@ -1,4 +1,4 @@ -//! A bounded, append-only exchange for storage key DKG artifacts. +//! 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 @@ -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; @@ -16,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; @@ -32,31 +31,36 @@ 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, +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 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"; -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; -const MAX_VALUES_PER_SLOT: usize = 2; /// The board address and read capability, paired with one participant's upload permission. /// @@ -65,10 +69,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), @@ -117,69 +127,50 @@ 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), +pub(super) fn validate_ticket(value: &str) -> anyhow::Result { + Ok(BoardTicket::from_str(value)?.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/") - }, - } - } +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()) } - - 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)] struct BoardWriter { author: iroh_docs::AuthorId, + core: Arc, document: Doc, - allowed_prefixes: Arc>, lock: Arc>, } @@ -194,22 +185,14 @@ enum Publisher { }, } -#[derive(Clone, Debug)] -struct UploadProtocol { - permits: Arc, - upload_secrets: Arc>, - writer: BoardWriter, -} - /// 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<()>, - allowed_prefixes: Arc>, - max_document_entries: usize, peer_ready: tokio::sync::watch::Receiver, publisher: Publisher, remote_providers: std::sync::Arc>>>, @@ -242,14 +225,6 @@ impl Drop for BoardNode { } 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, @@ -337,15 +312,6 @@ impl BoardNode { 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, @@ -425,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 @@ -498,14 +465,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<()> { @@ -562,22 +532,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 { @@ -595,14 +552,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 @@ -615,171 +569,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(|| { @@ -828,24 +617,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 { @@ -866,25 +648,17 @@ 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?; 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, @@ -956,24 +730,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], @@ -1007,98 +763,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; 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..3e01f99aa --- /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::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/iroh/tests.rs b/bin/validator/src/commands/dkg/board/iroh/tests.rs new file mode 100644 index 000000000..238a51369 --- /dev/null +++ b/bin/validator/src/commands/dkg/board/iroh/tests.rs @@ -0,0 +1,339 @@ +use super::*; + +impl BoardNode { + async fn create_for_test(data_directory: &Path) -> anyhow::Result<(Self, Vec)> { + Self::create_with_network(data_directory, 3, false).await + } + + async fn join_for_test(data_directory: &Path, ticket: BoardTicket) -> anyhow::Result { + Self::join_with_network(data_directory, ticket, 3, false).await + } + + fn local_writer_for_test(&self) -> &BoardWriter { + match &self.publisher { + Publisher::Local(writer) => writer, + Publisher::Remote { .. } => panic!("expected local DKG board writer"), + } + } + + async fn upload_raw_for_test( + &self, + kind: u8, + participant: u32, + declared_length: u64, + value: &[u8], + ) -> anyhow::Result { + match &self.publisher { + Publisher::Remote { endpoint, target, upload_secret, .. } => { + upload_artifact_request( + endpoint, + target, + upload_secret, + kind, + participant, + declared_length, + value, + ) + .await + }, + Publisher::Local(_) => anyhow::bail!("expected remote DKG board publisher"), + } + } + + async fn publish_hash_for_test( + &self, + slot: &ArtifactSlot, + hash: Hash, + size: u64, + ) -> anyhow::Result<()> { + let writer = self.local_writer_for_test(); + self.document + .set_hash(writer.author, slot.key(hash), hash, size) + .await + .context("failed to publish raw test hash") + } +} + +fn ticket_for(tickets: &[BoardTicket], participant: u32) -> BoardTicket { + tickets + .iter() + .find(|ticket| ticket.participant == participant) + .expect("participant ticket must exist") + .clone() +} + +#[test] +fn endpoint_secret_is_persisted_privately() -> anyhow::Result<()> { + let data_directory = tempfile::tempdir()?; + let first = load_or_create_endpoint_secret(data_directory.path())?; + let second = load_or_create_endpoint_secret(data_directory.path())?; + assert_eq!(first.to_bytes(), second.to_bytes()); + + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + + let mode = fs_err::metadata(data_directory.path().join(ENDPOINT_SECRET_FILE))? + .permissions() + .mode(); + assert_eq!(mode & 0o777, 0o600); + } + Ok(()) +} + +#[tokio::test] +async fn artifact_syncs_between_board_nodes() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let (host, tickets) = BoardNode::create_for_test(&root.path().join("host")).await?; + let ticket = ticket_for(&tickets, 1); + assert!(matches!(ticket.document.capability, iroh_docs::Capability::Read(_))); + let client = BoardNode::join_for_test(&root.path().join("client"), ticket).await?; + let slot = ArtifactSlot::Registration(1); + let value = b"signed registration"; + + client.publish(&slot, value).await?; + assert_eq!(host.wait_unique(&slot, Duration::from_secs(10)).await?, value,); + + client.shutdown().await?; + host.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn board_reopens_the_same_document_after_restart() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let data_directory = root.path().join("host"); + let (host, first_tickets) = BoardNode::create_for_test(&data_directory).await?; + host.publish(&ArtifactSlot::Manifest, b"manifest").await?; + host.shutdown().await?; + + let (host, second_tickets) = BoardNode::create_for_test(&data_directory).await?; + assert_eq!(first_tickets.len(), second_tickets.len()); + for (first, second) in first_tickets.iter().zip(&second_tickets) { + assert_eq!(first.participant, second.participant); + assert_eq!(first.document.capability.id(), second.document.capability.id()); + assert_eq!(first.upload_secret, second.upload_secret); + } + assert_eq!(host.read_unique(&ArtifactSlot::Manifest).await?, Some(b"manifest".to_vec())); + + host.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn board_metadata_is_published_as_one_directory() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let data_directory = root.path().join("host"); + let (host, _) = BoardNode::create_for_test(&data_directory).await?; + + let metadata_directory = data_directory.join(BOARD_METADATA_DIRECTORY); + assert!(metadata_directory.join(DOCUMENT_ID_FILE).is_file()); + assert!(metadata_directory.join(BOARD_FORMAT_FILE).is_file()); + assert!(metadata_directory.join(UPLOAD_SECRETS_DIRECTORY).is_dir()); + assert!(!data_directory.join(DOCUMENT_ID_FILE).exists()); + assert!(!data_directory.join(BOARD_FORMAT_FILE).exists()); + assert!(!data_directory.join(UPLOAD_SECRETS_DIRECTORY).exists()); + + host.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn incomplete_board_metadata_is_rejected() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let data_directory = root.path().join("host"); + let (host, _) = BoardNode::create_for_test(&data_directory).await?; + host.shutdown().await?; + fs_err::remove_file(data_directory.join(BOARD_METADATA_DIRECTORY).join(BOARD_FORMAT_FILE))?; + + let error = BoardNode::create_for_test(&data_directory) + .await + .err() + .context("incomplete board metadata unexpectedly reopened")?; + assert!(error.to_string().contains("failed to read DKG board format")); + Ok(()) +} + +#[tokio::test] +async fn legacy_board_metadata_is_rejected() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let data_directory = root.path().join("host"); + let upload_secrets_directory = data_directory.join(UPLOAD_SECRETS_DIRECTORY); + fs_err::create_dir_all(&upload_secrets_directory)?; + fs_err::write(data_directory.join(DOCUMENT_ID_FILE), hex::encode([0; 32]))?; + fs_err::write(data_directory.join(BOARD_FORMAT_FILE), b"participant-upload-v3\n")?; + for participant in 1..=3 { + fs_err::write( + upload_secrets_directory.join(format!("participant-{participant}.hex")), + hex::encode([0; 32]), + )?; + } + + let error = BoardNode::create_for_test(&data_directory) + .await + .err() + .context("legacy board metadata unexpectedly reopened")?; + assert!(error.to_string().contains("unsupported DKG board format")); + assert!(!data_directory.join(BOARD_METADATA_DIRECTORY).exists()); + Ok(()) +} + +#[tokio::test] +async fn previous_board_format_is_not_reopened() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let data_directory = root.path().join("host"); + let (host, _) = BoardNode::create_for_test(&data_directory).await?; + host.shutdown().await?; + fs_err::write( + data_directory.join(BOARD_METADATA_DIRECTORY).join(BOARD_FORMAT_FILE), + b"participant-upload-v3\n", + )?; + + let error = BoardNode::create_for_test(&data_directory) + .await + .err() + .context("old board format unexpectedly reopened")?; + assert!(error.to_string().contains("unsupported DKG board format")); + Ok(()) +} + +#[test] +fn legacy_board_ticket_is_rejected() { + BoardTicket::from_str("miden-storage-key-dkg-board-v3:1:00:invalid") + .expect_err("old board ticket unexpectedly parsed"); +} + +#[tokio::test] +async fn board_ticket_round_trips_and_validates_fields() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let (host, mut tickets) = BoardNode::create_for_test(root.path()).await?; + let ticket = tickets.remove(0); + let encoded = ticket.to_string(); + let decoded = BoardTicket::from_str(&encoded)?; + assert_eq!(decoded.to_string(), encoded); + + let mut invalid = ticket.clone(); + invalid.participant = 0; + let error = BoardTicket::from_str(&invalid.to_string()) + .expect_err("zero participant ticket unexpectedly parsed"); + assert!(error.to_string().contains("must be nonzero")); + + invalid = ticket; + invalid.document.nodes.clear(); + let error = BoardTicket::from_str(&invalid.to_string()) + .expect_err("ticket without addressing info unexpectedly parsed"); + assert!(error.to_string().contains("addressing info cannot be empty")); + + host.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn unknown_participants_and_artifact_kinds_are_rejected_before_body_allocation() +-> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let (host, tickets) = BoardNode::create_for_test(&root.path().join("host")).await?; + let ticket = ticket_for(&tickets, 1); + let client = BoardNode::join_for_test(&root.path().join("client"), ticket).await?; + + let error = client.upload_raw_for_test(1, 99, MAX_ARTIFACT_BYTES, &[]).await.unwrap_err(); + assert!(error.to_string().contains("unknown participant")); + let error = client.upload_raw_for_test(255, 1, 16, b"private artifact").await.unwrap_err(); + assert!(error.to_string().contains("unknown artifact kind")); + assert!(host.read_unique(&ArtifactSlot::Registration(1)).await?.is_none()); + + client.shutdown().await?; + host.shutdown().await?; + Ok(()) +} + +#[test] +fn oversized_artifacts_are_rejected_before_allocation() { + let oversized = usize::try_from(MAX_ARTIFACT_BYTES).unwrap() + 1; + assert!(validate_artifact_length(oversized).is_err()); +} + +#[tokio::test] +async fn oversized_upload_is_rejected_before_body_allocation() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let (host, tickets) = BoardNode::create_for_test(&root.path().join("host")).await?; + let ticket = ticket_for(&tickets, 1); + let client = BoardNode::join_for_test(&root.path().join("client"), ticket).await?; + let error = client.upload_raw_for_test(1, 1, MAX_ARTIFACT_BYTES + 1, &[]).await.unwrap_err(); + assert!(error.to_string().contains("exceeds")); + assert!(host.read_unique(&ArtifactSlot::Registration(1)).await?.is_none()); + + client.shutdown().await?; + host.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn invalid_upload_secret_is_rejected_before_storage() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let (host, tickets) = BoardNode::create_for_test(&root.path().join("host")).await?; + let mut ticket = ticket_for(&tickets, 1); + ticket.upload_secret[0] ^= 1; + let client = BoardNode::join_for_test(&root.path().join("client"), ticket).await?; + + let error = client + .publish(&ArtifactSlot::Registration(1), b"signed registration") + .await + .unwrap_err(); + assert!(error.to_string().contains("does not authorize this participant")); + assert!(host.read_unique(&ArtifactSlot::Registration(1)).await?.is_none()); + + client.shutdown().await?; + host.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn participant_ticket_cannot_publish_another_participants_slot() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let (host, tickets) = BoardNode::create_for_test(&root.path().join("host")).await?; + let first = + BoardNode::join_for_test(&root.path().join("first"), ticket_for(&tickets, 1)).await?; + + let error = first + .upload_raw_for_test( + 1, + 2, + u64::try_from(b"wrong registration".len())?, + b"wrong registration", + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("does not authorize this participant")); + assert!(host.read_unique(&ArtifactSlot::Registration(2)).await?.is_none()); + + let second = + BoardNode::join_for_test(&root.path().join("second"), ticket_for(&tickets, 2)).await?; + second.publish(&ArtifactSlot::Registration(2), b"signed registration").await?; + assert_eq!( + host.wait_unique(&ArtifactSlot::Registration(2), Duration::from_secs(10)) + .await?, + b"signed registration" + ); + + first.shutdown().await?; + second.shutdown().await?; + host.shutdown().await?; + Ok(()) +} + +#[tokio::test] +async fn invalid_download_metadata_is_rejected() -> anyhow::Result<()> { + let root = tempfile::tempdir()?; + let (host, _) = BoardNode::create_for_test(&root.path().join("host")).await?; + let slot = ArtifactSlot::Manifest; + let value = b"manifest"; + let hash = host.publish(&slot, value).await?; + + host.publish_hash_for_test(&slot, hash, 1).await?; + let error = host.read_unique(&slot).await.unwrap_err(); + assert!(error.to_string().contains("length does not match")); + + host.shutdown().await?; + Ok(()) +} 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..21a4ce8af --- /dev/null +++ b/bin/validator/src/commands/dkg/board/iroh/upload.rs @@ -0,0 +1,222 @@ +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::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; +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/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 new file mode 100644 index 000000000..822863f3a --- /dev/null +++ b/bin/validator/src/commands/dkg/board/mod.rs @@ -0,0 +1,274 @@ +//! A bounded, append-only exchange for storage key DKG artifacts. +//! +//! The ceremony runner sees role-specific boards and typed artifact slots. Transport setup, +//! credentials, synchronization, and persistence stay behind the private adapters. + +use std::fmt; +use std::path::Path; +use std::str::FromStr; +use std::time::Duration; + +mod core; +mod iroh; +#[cfg(test)] +mod memory; +#[cfg(test)] +mod tests; + +pub(super) use core::ArtifactSlot; + +/// An opaque board address and one participant's publish permission. +#[derive(Clone)] +pub(super) struct BoardTicket { + encoded: String, + participant: u32, +} + +impl BoardTicket { + fn new(encoded: String, participant: u32) -> Self { + Self { encoded, participant } + } + + pub(super) fn participant(&self) -> u32 { + self.participant + } + + fn into_encoded(self) -> String { + self.encoded + } +} + +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(&self.encoded) + } +} + +impl FromStr for BoardTicket { + type Err = anyhow::Error; + + fn from_str(value: &str) -> Result { + let participant = iroh::validate_ticket(value)?; + Ok(Self::new(value.to_owned(), participant)) + } +} + +/// An artifact published by the ceremony coordinator. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum CommonArtifact { + Manifest, + DecryptionConfig, + ContextConfig, +} + +/// An artifact published by one ceremony participant. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(super) enum ParticipantArtifact { + Registration, + DecryptionDealing, + ContextDealing, + TranscriptAcceptance, +} + +/// The read-only view shared by both board roles. +pub(super) struct BoardReader { + 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 { + /// 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, + slot: &ArtifactSlot, + timeout: Duration, + ) -> anyhow::Result> { + self.node.wait_unique(slot, timeout).await + } +} + +/// 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)> { + 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 (node, tickets) = + 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: 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 { + &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 + } + + /// Stops the board and flushes its persistent stores. + pub(super) async fn shutdown(self) -> anyhow::Result<()> { + self.reader.node.shutdown().await + } +} + +/// 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( + 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 participant = ticket.participant(); + let node = iroh::join( + data_directory, + &ticket.into_encoded(), + participant_count, + use_network_services, + ) + .await?; + Ok(Self { + participant, + reader: BoardReader { node: Transport::Iroh(Box::new(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 + } + + /// Stops the board and flushes its persistent stores. + pub(super) async fn shutdown(self) -> anyhow::Result<()> { + self.reader.node.shutdown().await + } +} diff --git a/bin/validator/src/commands/dkg/board/tests.rs b/bin/validator/src/commands/dkg/board/tests.rs index 51e559ec5..1cc4e684a 100644 --- a/bin/validator/src/commands/dkg/board/tests.rs +++ b/bin/validator/src/commands/dkg/board/tests.rs @@ -1,354 +1,81 @@ -use super::*; - -impl BoardNode { - async fn create_for_test(data_directory: &Path) -> anyhow::Result<(Self, Vec)> { - Self::create_with_network(data_directory, 3, false).await - } - - async fn join_for_test(data_directory: &Path, ticket: BoardTicket) -> anyhow::Result { - Self::join_with_network(data_directory, ticket, 3, false).await - } - - fn local_writer_for_test(&self) -> &BoardWriter { - match &self.publisher { - Publisher::Local(writer) => writer, - Publisher::Remote { .. } => panic!("expected local DKG board writer"), - } - } - - async fn upload_raw_for_test( - &self, - kind: u8, - participant: u32, - declared_length: u64, - value: &[u8], - ) -> anyhow::Result { - match &self.publisher { - Publisher::Remote { endpoint, target, upload_secret, .. } => { - upload_artifact_request( - endpoint, - target, - upload_secret, - kind, - participant, - declared_length, - value, - ) - .await - }, - Publisher::Local(_) => anyhow::bail!("expected remote DKG board publisher"), - } - } - - async fn publish_hash_for_test( - &self, - slot: &ArtifactSlot, - hash: Hash, - size: u64, - ) -> anyhow::Result<()> { - let writer = self.local_writer_for_test(); - self.document - .set_hash(writer.author, slot.key(hash), hash, size) - .await - .context("failed to publish raw test hash") - } -} - -fn ticket_for(tickets: &[BoardTicket], participant: u32) -> BoardTicket { - tickets - .iter() - .find(|ticket| ticket.participant == participant) - .expect("participant ticket must exist") - .clone() -} - -#[test] -fn endpoint_secret_is_persisted_privately() -> anyhow::Result<()> { - let data_directory = tempfile::tempdir()?; - let first = load_or_create_endpoint_secret(data_directory.path())?; - let second = load_or_create_endpoint_secret(data_directory.path())?; - assert_eq!(first.to_bytes(), second.to_bytes()); - - #[cfg(unix)] - { - use std::os::unix::fs::PermissionsExt; - - let mode = fs_err::metadata(data_directory.path().join(ENDPOINT_SECRET_FILE))? - .permissions() - .mode(); - assert_eq!(mode & 0o777, 0o600); - } - Ok(()) -} - -#[tokio::test] -async fn artifact_syncs_between_board_nodes() -> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let (host, tickets) = BoardNode::create_for_test(&root.path().join("host")).await?; - let ticket = ticket_for(&tickets, 1); - assert!(matches!(ticket.document.capability, iroh_docs::Capability::Read(_))); - let client = BoardNode::join_for_test(&root.path().join("client"), ticket).await?; - let slot = ArtifactSlot::Registration(1); - let value = b"signed registration"; - - client.publish(&slot, value).await?; - assert_eq!(host.wait_unique(&slot, Duration::from_secs(10)).await?, value,); - - client.shutdown().await?; - host.shutdown().await?; - Ok(()) -} +use std::time::Duration; -#[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()?; - let data_directory = root.path().join("host"); - let (host, first_tickets) = BoardNode::create_for_test(&data_directory).await?; - host.publish(&ArtifactSlot::Manifest, b"manifest").await?; - host.shutdown().await?; - - let (host, second_tickets) = BoardNode::create_for_test(&data_directory).await?; - assert_eq!(first_tickets.len(), second_tickets.len()); - for (first, second) in first_tickets.iter().zip(&second_tickets) { - assert_eq!(first.participant, second.participant); - assert_eq!(first.document.capability.id(), second.document.capability.id()); - assert_eq!(first.upload_secret, second.upload_secret); - } - assert_eq!(host.read_unique(&ArtifactSlot::Manifest).await?, Some(b"manifest".to_vec())); - - host.shutdown().await?; - Ok(()) -} - -#[tokio::test] -async fn board_metadata_is_published_as_one_directory() -> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let data_directory = root.path().join("host"); - let (host, _) = BoardNode::create_for_test(&data_directory).await?; - - let metadata_directory = data_directory.join(BOARD_METADATA_DIRECTORY); - assert!(metadata_directory.join(DOCUMENT_ID_FILE).is_file()); - assert!(metadata_directory.join(BOARD_FORMAT_FILE).is_file()); - assert!(metadata_directory.join(UPLOAD_SECRETS_DIRECTORY).is_dir()); - assert!(!data_directory.join(DOCUMENT_ID_FILE).exists()); - assert!(!data_directory.join(BOARD_FORMAT_FILE).exists()); - assert!(!data_directory.join(UPLOAD_SECRETS_DIRECTORY).exists()); +use super::*; - host.shutdown().await?; - Ok(()) -} +const WAIT_TIMEOUT: Duration = Duration::from_secs(10); + +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?; + + 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" + ); -#[tokio::test] -async fn incomplete_board_metadata_is_rejected() -> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let data_directory = root.path().join("host"); - let (host, _) = BoardNode::create_for_test(&data_directory).await?; - host.shutdown().await?; - fs_err::remove_file(data_directory.join(BOARD_METADATA_DIRECTORY).join(BOARD_FORMAT_FILE))?; + assert!(coordinator.publish(CommonArtifact::ContextConfig, b"").await.is_err()); + assert!(coordinator.reader().read_unique(&ArtifactSlot::Registration(3)).await.is_err()); - let error = BoardNode::create_for_test(&data_directory) - .await - .err() - .context("incomplete board metadata unexpectedly reopened")?; - assert!(error.to_string().contains("failed to read DKG board format")); + coordinator.publish(CommonArtifact::Manifest, b"other manifest").await?; + assert!(coordinator.reader().read_unique(&ArtifactSlot::Manifest).await.is_err()); Ok(()) } -#[tokio::test] -async fn legacy_board_metadata_is_rejected() -> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let data_directory = root.path().join("host"); - let upload_secrets_directory = data_directory.join(UPLOAD_SECRETS_DIRECTORY); - fs_err::create_dir_all(&upload_secrets_directory)?; - fs_err::write(data_directory.join(DOCUMENT_ID_FILE), hex::encode([0; 32]))?; - fs_err::write(data_directory.join(BOARD_FORMAT_FILE), b"participant-upload-v3\n")?; - for participant in 1..=3 { - fs_err::write( - upload_secrets_directory.join(format!("participant-{participant}.hex")), - hex::encode([0; 32]), - )?; +async fn shutdown_boards( + coordinator: CoordinatorBoard, + participants: Vec, +) -> anyhow::Result<()> { + for participant in participants { + participant.shutdown().await?; } - - let error = BoardNode::create_for_test(&data_directory) - .await - .err() - .context("legacy board metadata unexpectedly reopened")?; - assert!(error.to_string().contains("unsupported DKG board format")); - assert!(!data_directory.join(BOARD_METADATA_DIRECTORY).exists()); - Ok(()) + coordinator.shutdown().await } #[tokio::test] -async fn previous_board_format_is_not_reopened() -> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let data_directory = root.path().join("host"); - let (host, _) = BoardNode::create_for_test(&data_directory).await?; - host.shutdown().await?; - fs_err::write( - data_directory.join(BOARD_METADATA_DIRECTORY).join(BOARD_FORMAT_FILE), - b"participant-upload-v3\n", - )?; - - let error = BoardNode::create_for_test(&data_directory) - .await - .err() - .context("old board format unexpectedly reopened")?; - assert!(error.to_string().contains("unsupported DKG board format")); - Ok(()) -} - -#[test] -fn legacy_board_ticket_is_rejected() { - BoardTicket::from_str("miden-storage-key-dkg-board-v3:1:00:invalid") - .expect_err("old board ticket unexpectedly parsed"); +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 board_ticket_round_trips_and_validates_fields() -> anyhow::Result<()> { +async fn iroh_adapter_obeys_board_contract() -> anyhow::Result<()> { let root = tempfile::tempdir()?; - let (host, mut tickets) = BoardNode::create_for_test(root.path()).await?; - let ticket = tickets.remove(0); - let encoded = ticket.to_string(); - let decoded = BoardTicket::from_str(&encoded)?; - assert_eq!(decoded.to_string(), encoded); - - let mut invalid = ticket.clone(); - invalid.participant = 0; - let error = BoardTicket::from_str(&invalid.to_string()) - .expect_err("zero participant ticket unexpectedly parsed"); - assert!(error.to_string().contains("must be nonzero")); - - invalid = ticket; - invalid.document.nodes.clear(); - let error = BoardTicket::from_str(&invalid.to_string()) - .expect_err("ticket without addressing info unexpectedly parsed"); - assert!(error.to_string().contains("addressing info cannot be empty")); - - host.shutdown().await?; - Ok(()) -} - -#[tokio::test] -async fn unknown_participants_and_artifact_kinds_are_rejected_before_body_allocation() --> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let (host, tickets) = BoardNode::create_for_test(&root.path().join("host")).await?; - let ticket = ticket_for(&tickets, 1); - let client = BoardNode::join_for_test(&root.path().join("client"), ticket).await?; - - let error = client.upload_raw_for_test(1, 99, MAX_ARTIFACT_BYTES, &[]).await.unwrap_err(); - assert!(error.to_string().contains("unknown participant")); - let error = client.upload_raw_for_test(255, 1, 16, b"private artifact").await.unwrap_err(); - assert!(error.to_string().contains("unknown artifact kind")); - assert!(host.read_unique(&ArtifactSlot::Registration(1)).await?.is_none()); - - client.shutdown().await?; - host.shutdown().await?; - Ok(()) -} - -#[test] -fn oversized_artifacts_are_rejected_before_allocation() { - let oversized = usize::try_from(MAX_ARTIFACT_BYTES).unwrap() + 1; - assert!(validate_artifact_length(oversized).is_err()); -} - -#[tokio::test] -async fn oversized_upload_is_rejected_before_body_allocation() -> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let (host, tickets) = BoardNode::create_for_test(&root.path().join("host")).await?; - let ticket = ticket_for(&tickets, 1); - let client = BoardNode::join_for_test(&root.path().join("client"), ticket).await?; - let error = client.upload_raw_for_test(1, 1, MAX_ARTIFACT_BYTES + 1, &[]).await.unwrap_err(); - assert!(error.to_string().contains("exceeds")); - assert!(host.read_unique(&ArtifactSlot::Registration(1)).await?.is_none()); - - client.shutdown().await?; - host.shutdown().await?; - Ok(()) -} - -#[tokio::test] -async fn invalid_upload_secret_is_rejected_before_storage() -> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let (host, tickets) = BoardNode::create_for_test(&root.path().join("host")).await?; - let mut ticket = ticket_for(&tickets, 1); - ticket.upload_secret[0] ^= 1; - let client = BoardNode::join_for_test(&root.path().join("client"), ticket).await?; - - let error = client - .publish(&ArtifactSlot::Registration(1), b"signed registration") - .await - .unwrap_err(); - assert!(error.to_string().contains("does not authorize this participant")); - assert!(host.read_unique(&ArtifactSlot::Registration(1)).await?.is_none()); - - client.shutdown().await?; - host.shutdown().await?; - Ok(()) -} - -#[tokio::test] -async fn participant_ticket_cannot_publish_another_participants_slot() -> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let (host, tickets) = BoardNode::create_for_test(&root.path().join("host")).await?; - let first = - BoardNode::join_for_test(&root.path().join("first"), ticket_for(&tickets, 1)).await?; - - let error = first - .upload_raw_for_test( - 1, - 2, - u64::try_from(b"wrong registration".len())?, - b"wrong registration", - ) - .await - .unwrap_err(); - assert!(error.to_string().contains("does not authorize this participant")); - assert!(host.read_unique(&ArtifactSlot::Registration(2)).await?.is_none()); - - let second = - BoardNode::join_for_test(&root.path().join("second"), ticket_for(&tickets, 2)).await?; - second.publish(&ArtifactSlot::Registration(2), b"signed registration").await?; - assert_eq!( - host.wait_unique(&ArtifactSlot::Registration(2), Duration::from_secs(10)) + 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?, - b"signed registration" - ); - - first.shutdown().await?; - second.shutdown().await?; - host.shutdown().await?; - Ok(()) -} - -#[tokio::test] -async fn invalid_download_metadata_is_rejected() -> anyhow::Result<()> { - let root = tempfile::tempdir()?; - let (host, _) = BoardNode::create_for_test(&root.path().join("host")).await?; - let slot = ArtifactSlot::Manifest; - let value = b"manifest"; - let hash = host.publish(&slot, value).await?; - - host.publish_hash_for_test(&slot, hash, 1).await?; - let error = host.read_unique(&slot).await.unwrap_err(); - assert!(error.to_string().contains("length does not match")); - - host.shutdown().await?; - Ok(()) + ); + } + assert_board_contract(&coordinator, &participants).await?; + shutdown_boards(coordinator, participants).await } diff --git a/bin/validator/src/commands/dkg/runner.rs b/bin/validator/src/commands/dkg/runner.rs index 209256e8f..df690379f 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, @@ -120,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, @@ -128,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 @@ -137,11 +144,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 +181,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 +199,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 +215,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 +233,7 @@ pub(super) async fn coordinate_common_files( } async fn wait_for_registrations( - board: &BoardNode, + board: &BoardReader, validator_keys: &[PublicKey], timeout: Duration, ) -> anyhow::Result)>> { @@ -251,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, @@ -261,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 @@ -279,12 +286,14 @@ 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 { - BoardNode::join(&board_directory, ticket, participant_count).await? - } else { - BoardNode::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, @@ -306,8 +315,8 @@ where clippy::too_many_lines, reason = "the inputs and linear body mirror the ceremony policy and phase order" )] -async fn run_validator_on_board( - board: &BoardNode, +pub(super) async fn run_validator_on_board( + board: &ParticipantBoard, genesis_path: &Path, signer: &ValidatorSigner, participant: ParticipantIndex, @@ -321,17 +330,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 +357,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 +394,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 +436,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 +545,7 @@ fn validate_local_identity( } async fn wait_for_dealings( - board: &BoardNode, + board: &BoardReader, participant_count: usize, timeout: Duration, ) -> anyhow::Result)>> { @@ -557,7 +569,7 @@ async fn wait_for_dealings( } async fn wait_for_acceptances( - board: &BoardNode, + board: &BoardReader, participant_count: usize, timeout: Duration, ) -> anyhow::Result)>> { @@ -574,14 +586,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..0a3bc9362 100644 --- a/bin/validator/src/commands/dkg/tests.rs +++ b/bin/validator/src/commands/dkg/tests.rs @@ -901,6 +901,153 @@ 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 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_ticket::( + ticket, + &genesis.path, + &signer, + 2, + &epoch, + &work_directory, + &root.path().join("bundle"), + Duration::from_secs(1), + ) + .await + .unwrap_err(); + + assert!( + error + .to_string() + .contains(&format!("ticket belongs to participant {ticket_participant}")), + "unexpected error: {error:#}" + ); + assert!( + board + .reader() + .read_unique(&board::ArtifactSlot::Registration(ticket_participant)) + .await? + .is_none() + ); + board.shutdown().await?; + 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( @@ -912,7 +1059,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); @@ -965,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, @@ -973,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(); @@ -1029,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(()) }