diff options
Diffstat (limited to 'src/transport')
| -rw-r--r-- | src/transport/identity.rs | 510 | ||||
| -rw-r--r-- | src/transport/mod.rs | 62 | ||||
| -rw-r--r-- | src/transport/mycelium/mod.rs | 1 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 10 | ||||
| -rw-r--r-- | src/transport/wire.rs | 54 |
5 files changed, 610 insertions, 27 deletions
diff --git a/src/transport/identity.rs b/src/transport/identity.rs new file mode 100644 index 0000000..712e1f9 --- /dev/null +++ b/src/transport/identity.rs @@ -0,0 +1,510 @@ +//! The identity layer: between the node and a transport whose identities are addresses +//! (Mycelium's public keys), so that the rest of the node deals in node IDs +//! (`crate::identity`) only, whatever transport reaches a peer. +//! +//! A node proves who it is to an address with WHO (which asks back) or ME (the answer): +//! its public node key, and a signature over its own address and the recipient's +//! (`signed`). The transport authenticates the sender's address, so a proof is good for +//! that pair of addresses only: replayed from another address, or to another recipient, +//! it fails. Once a node proved itself at an address, messages from there are its. +//! +//! * A message from an address no node proved itself at is dropped, and the address is +//! asked WHO (at most every `ASK_INTERVAL`): after a restart, the first message from +//! a peer brings its proof back, and the peer's node retries what was lost. +//! * A seed (`resolve_seed`) is an address, or something the transport resolves to one; +//! it is asked WHO, and its ME says which node is there. +//! * Where other nodes are comes from the member list and HELLO_REPLYs +//! (`learn_addresses`): hints, not proofs. A message to a node goes to the address it +//! proved itself at; with only hints, those are asked WHO, and the send is refused +//! until one answers (the node retries refused sends). +//! +//! Addresses are `<scheme>:<transport identity>`, e.g. `mycelium:<public key>`. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use anyhow::{Result, anyhow, bail}; +use async_trait::async_trait; +use tokio::sync::oneshot; +use tokio::time::Instant; +use tracing::{debug, info}; + +use crate::identity::{self, NodeKey}; +use crate::locks::MutexExt; +use crate::transport::traffic::Traffic; +use crate::transport::{ + EncodedMessage, HandlerFn, IdentityProof, MessageHandler, MessageKind, ParsedMessage, Transport, +}; + +pub(crate) const WHO_LABEL: &[u8] = b"continuum/who/1"; +pub(crate) const ME_LABEL: &[u8] = b"continuum/me/1"; +/// An address is asked WHO at most this often, unless a seed is being resolved. +const ASK_INTERVAL: Duration = Duration::from_secs(5); +/// How long resolving a seed waits for its ME. +const RESOLVE_TIMEOUT: Duration = Duration::from_secs(10); +/// Addresses kept per node from hints: about one per transport. +const MAX_HINTS: usize = 8; +/// Addresses with a proven node, nodes with hints, and addresses asked, each at most: +/// anyone can send us proofs (keys are free), so the books are bounded. +const MAX_ENTRIES: usize = 4096; + +/// What a node under the identity layer needs (`node::NodeConfig::identity`). +#[derive(Clone)] +pub(crate) struct IdentityConfig { + pub(crate) key: Arc<NodeKey>, + /// The transport's address scheme, e.g. `mycelium`. + pub(crate) scheme: &'static str, +} + +/// The bytes a WHO (`label`: `WHO_LABEL`) or ME proof signs: the label, then the +/// sender's and the recipient's address, each with its length. +pub(crate) fn signed(label: &[u8], from: &str, to: &str) -> Vec<u8> { + let mut bytes = label.to_vec(); + for address in [from, to] { + bytes.extend_from_slice(&(address.len() as u32).to_be_bytes()); + bytes.extend_from_slice(address.as_bytes()); + } + bytes +} + +pub(crate) struct IdentityTransport { + shared: Arc<Shared>, +} + +struct Shared { + inner: Arc<dyn Transport>, + scheme: &'static str, + key: Arc<NodeKey>, + node_id: String, + /// WHO and ME are counted here, under the address (`traffic`). + traffic: Arc<Traffic>, + // A std mutex: every access is short and never spans an await. + book: Mutex<Book>, +} + +/// Who is where. Keys are transport identities (an address without its scheme). +#[derive(Default)] +struct Book { + /// The node that proved itself at a transport identity, and when. + at: HashMap<String, (String, Instant)>, + /// The transport identity each node last proved itself at. + proven: HashMap<String, String>, + /// Where others say a node is. + hints: HashMap<String, Vec<String>>, + /// When each transport identity was last asked WHO. + asked: HashMap<String, Instant>, + /// Seeds being resolved: who waits for a ME from where. + waiting: HashMap<String, Vec<oneshot::Sender<String>>>, +} + +impl Book { + /// `node` proved itself at `target`. Returns whether that is news. + fn bind(&mut self, target: &str, node: &str, now: Instant) -> bool { + let news = self.at.get(target).is_none_or(|(known, _)| known != node); + // One address per node and one node per address: a node that moved, or an + // address taken over by another node, leaves no stale entry behind. + if let Some(previous) = self.proven.insert(node.to_owned(), target.to_owned()) + && previous != target + { + self.at.remove(&previous); + } + if let Some((previous, _)) = self.at.insert(target.to_owned(), (node.to_owned(), now)) + && previous != node + && self.proven.get(&previous).is_some_and(|at| at == target) + { + self.proven.remove(&previous); + } + if self.at.len() > MAX_ENTRIES + && let Some(oldest) = self + .at + .iter() + .min_by_key(|(_, (_, at))| *at) + .map(|(target, _)| target.clone()) + && let Some((node, _)) = self.at.remove(&oldest) + { + self.proven.remove(&node); + } + for waiter in self.waiting.remove(target).into_iter().flatten() { + let _ = waiter.send(node.to_owned()); + } + news + } + + /// Whether to ask `target` WHO now (always, if `force`), noting that it was asked. + fn ask(&mut self, target: &str, now: Instant, force: bool) -> bool { + if !force + && self + .asked + .get(target) + .is_some_and(|at| now.duration_since(*at) < ASK_INTERVAL) + { + return false; + } + if self.asked.len() >= MAX_ENTRIES { + self.asked + .retain(|_, at| now.duration_since(*at) < ASK_INTERVAL); + } + self.asked.insert(target.to_owned(), now); + true + } +} + +impl IdentityTransport { + pub(crate) fn new( + inner: Arc<dyn Transport>, + config: IdentityConfig, + traffic: Arc<Traffic>, + ) -> Self { + let node_id = config.key.id(); + Self { + shared: Arc::new(Shared { + inner, + scheme: config.scheme, + key: config.key, + node_id, + traffic, + book: Mutex::default(), + }), + } + } +} + +impl Shared { + fn lock(&self) -> MutexGuard<'_, Book> { + self.book.lock_or_recover() + } + + fn address(&self, target: &str) -> String { + format!("{}:{target}", self.scheme) + } + + /// The transport identity in `address`, if it is one of this transport's. + fn target_of<'a>(&self, address: &'a str) -> Option<&'a str> { + address + .strip_prefix(self.scheme) + .and_then(|rest| rest.strip_prefix(':')) + .filter(|target| self.inner.is_peer_id(target)) + } + + fn proof(&self, label: &[u8], target: &str) -> IdentityProof { + let message = signed( + label, + &self.address(self.inner.local_id()), + &self.address(target), + ); + IdentityProof { + key: self.key.public(), + signature: self.key.sign(&message), + } + } + + /// Notes the node `proof` proves to be at `sender`; None if it proves nothing. + fn prove(&self, sender: &str, label: &[u8], proof: &IdentityProof) -> Option<String> { + let message = signed( + label, + &self.address(sender), + &self.address(self.inner.local_id()), + ); + if !identity::verify(&proof.key, &message, &proof.signature) { + debug!(from = %self.address(sender), "Ignoring an identity proof that doesn't hold"); + return None; + } + let node = identity::node_id_of(&proof.key); + if node == self.node_id { + debug!(from = %self.address(sender), "Ignoring a proof of our own node key"); + return None; + } + if self.lock().bind(sender, &node, Instant::now()) { + debug!(node = %node, address = %self.address(sender), "Node proved its address"); + } + Some(node) + } + + /// Asks `target` WHO, unless it was asked recently (`force`: anyway). + fn ask(self: &Arc<Self>, target: &str, force: bool) { + if !self.lock().ask(target, Instant::now(), force) { + return; + } + let shared = Arc::clone(self); + let target = target.to_owned(); + tokio::spawn(async move { shared.send_proof(MessageKind::Who, &target).await }); + } + + /// Sends a WHO or ME (`kind`) to `target`. + async fn send_proof(&self, kind: MessageKind, target: &str) { + let message = if kind == MessageKind::Who { + ParsedMessage::Who(self.proof(WHO_LABEL, target)) + } else { + ParsedMessage::Me(self.proof(ME_LABEL, target)) + }; + let encoded = match self.inner.encode(target, &message) { + Ok(encoded) => encoded, + Err(error) => { + debug!(to = %target, error = %format!("{error:#}"), "Couldn't encode an identity proof"); + return; + } + }; + let address = self.address(target); + match self.inner.send_encoded(&encoded).await { + Ok(()) => self + .traffic + .sent(&address, encoded.kind, encoded.payload_len), + Err(error) => { + self.traffic + .refused(&address, encoded.kind, encoded.payload_len); + debug!(to = %address, error = %format!("{error:#}"), "Couldn't send an identity proof"); + } + } + } + + async fn receive( + self: Arc<Self>, + sender: String, + message: ParsedMessage, + payload_len: usize, + handler: &HandlerFn, + ) -> Result<()> { + match message { + ParsedMessage::Who(proof) => { + self.traffic + .received(&self.address(&sender), MessageKind::Who, payload_len); + if self.prove(&sender, WHO_LABEL, &proof).is_some() { + let shared = Arc::clone(&self); + tokio::spawn(async move { shared.send_proof(MessageKind::Me, &sender).await }); + } + Ok(()) + } + ParsedMessage::Me(proof) => { + self.traffic + .received(&self.address(&sender), MessageKind::Me, payload_len); + self.prove(&sender, ME_LABEL, &proof); + Ok(()) + } + message => { + let node = self.lock().at.get(&sender).map(|(node, _)| node.clone()); + match node { + Some(node) => handler(node, message, payload_len).await, + None => { + debug!( + from = %self.address(&sender), + kind = ?message.kind(), + "Dropping a message from an address no node proved itself at; asking who is there" + ); + self.ask(&sender, false); + Ok(()) + } + } + } + } + } +} + +#[async_trait] +impl Transport for IdentityTransport { + fn local_id(&self) -> &str { + &self.shared.node_id + } + + fn is_peer_id(&self, id: &str) -> bool { + identity::is_node_id(id) + } + + fn delivery_limit(&self) -> Duration { + self.shared.inner.delivery_limit() + } + + fn local_addresses(&self) -> Vec<String> { + vec![self.shared.address(self.shared.inner.local_id())] + } + + fn addresses_of(&self, peer: &str) -> Vec<String> { + let book = self.shared.lock(); + match book.proven.get(peer) { + Some(target) => vec![self.shared.address(target)], + None => book + .hints + .get(peer) + .into_iter() + .flatten() + .map(|target| self.shared.address(target)) + .collect(), + } + } + + fn learn_addresses(&self, peer: &str, addresses: &[String]) { + if peer == self.shared.node_id || !identity::is_node_id(peer) { + return; + } + let local = self.shared.inner.local_id(); + let mut targets: Vec<String> = addresses + .iter() + .filter_map(|address| self.shared.target_of(address)) + .filter(|target| *target != local) + .map(str::to_owned) + .collect(); + targets.dedup(); + targets.truncate(MAX_HINTS); + if targets.is_empty() { + return; + } + let mut book = self.shared.lock(); + if book.hints.len() < MAX_ENTRIES || book.hints.contains_key(peer) { + book.hints.insert(peer.to_owned(), targets); + } + } + + /// A node ID resolves to itself if this node knows where it is; anything else is + /// an address (`<scheme>:...`) or what the transport resolves to one, asked WHO. + async fn resolve_seed(&self, seed: &str) -> Result<String> { + if identity::is_node_id(seed) { + let book = self.shared.lock(); + if book.proven.contains_key(seed) || book.hints.contains_key(seed) { + return Ok(seed.to_owned()); + } + bail!("a node ID alone doesn't say where that node is; give its address"); + } + let target = match self.shared.target_of(seed) { + Some(target) => target.to_owned(), + None => self.shared.inner.resolve_seed(seed).await?, + }; + if target == self.shared.inner.local_id() { + bail!("{seed} is this node"); + } + let receiver = { + let mut book = self.shared.lock(); + if let Some((node, _)) = book.at.get(&target) { + return Ok(node.clone()); + } + let (sender, receiver) = oneshot::channel(); + let waiting = book.waiting.entry(target.clone()).or_default(); + waiting.retain(|waiter| !waiter.is_closed()); + waiting.push(sender); + receiver + }; + self.shared.ask(&target, true); + match tokio::time::timeout(RESOLVE_TIMEOUT, receiver).await { + Ok(Ok(node)) => { + info!(seed = %seed, node = %node, "Seed is node"); + Ok(node) + } + _ => Err(anyhow!( + "no node answered at {} within {RESOLVE_TIMEOUT:?}", + self.shared.address(&target) + )), + } + } + + /// For the address `peer` proved itself at, or else one it is said to be at (a send + /// there is refused until it proves itself, `send_encoded`). + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + let target = { + let book = self.shared.lock(); + book.proven + .get(peer) + .or_else(|| book.hints.get(peer).and_then(|hints| hints.first())) + .cloned() + }; + let Some(target) = target else { + bail!("no address known for {peer}"); + }; + let mut encoded = self.shared.inner.encode(&target, message)?; + encoded.peer = peer.to_owned(); + Ok(encoded) + } + + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + let hints = { + let book = self.shared.lock(); + if book.proven.get(&message.peer) == Some(&message.address) { + None + } else { + Some(book.hints.get(&message.peer).cloned().unwrap_or_default()) + } + }; + if let Some(hints) = hints { + for target in hints { + self.shared.ask(&target, false); + } + bail!( + "{} hasn't proven where it is yet; asked where it is said to be", + message.peer + ); + } + let mut inner = message.clone(); + inner.peer = message.address.clone(); + self.shared.inner.send_encoded(&inner).await + } + + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + let handler: Arc<HandlerFn> = Arc::from(handler); + let shared = Arc::clone(&self.shared); + self.shared + .inner + .start_consumer(Box::new(move |sender, message, payload_len| { + let shared = Arc::clone(&shared); + let handler = Arc::clone(&handler); + Box::pin(async move { + shared + .receive(sender, message, payload_len, &*handler) + .await + }) + })) + .await + } + + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + self.shared.inner.wait_for_exit().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn signed_bytes_say_both_addresses() { + // A length before each address: "a:bc" to "d" isn't "a:b" to "cd". + assert_ne!( + signed(WHO_LABEL, "a:bc", "d"), + signed(WHO_LABEL, "a:b", "cd") + ); + assert_ne!(signed(WHO_LABEL, "a", "b"), signed(ME_LABEL, "a", "b")); + assert_ne!(signed(WHO_LABEL, "a", "b"), signed(WHO_LABEL, "b", "a")); + } + + #[test] + fn one_address_per_node_and_one_node_per_address() { + let now = Instant::now(); + let mut book = Book::default(); + assert!(book.bind("x", "n1", now)); + assert!(!book.bind("x", "n1", now)); + // n1 moved to y: x is no longer n1's. + assert!(book.bind("y", "n1", now)); + assert!(!book.at.contains_key("x")); + assert_eq!(book.proven["n1"], "y"); + // n2 took over y: n1 is nowhere proven. + assert!(book.bind("y", "n2", now)); + assert!(!book.proven.contains_key("n1")); + assert_eq!(book.at["y"].0, "n2"); + } + + #[test] + fn asking_is_rate_limited() { + let now = Instant::now(); + let mut book = Book::default(); + assert!(book.ask("x", now, false)); + assert!(!book.ask("x", now, false)); + assert!(book.ask("x", now, true)); + assert!(book.ask("x", now + ASK_INTERVAL, false)); + } + + #[tokio::test] + async fn a_waiting_seed_learns_the_node() { + let mut book = Book::default(); + let (sender, receiver) = oneshot::channel(); + book.waiting.insert("x".into(), vec![sender]); + book.bind("x", "n1", Instant::now()); + assert_eq!(receiver.await.unwrap(), "n1"); + } +} diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 29c48d6..52db406 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -12,6 +12,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; pub(crate) use crate::document::replica::Epoch; pub(crate) mod firewall; +pub(crate) mod identity; pub(crate) mod mycelium; #[cfg(test)] pub(crate) mod sim; @@ -137,6 +138,31 @@ pub(crate) enum ParsedMessage { Ping(Keepalive), Pong(Keepalive), Blob(BlobMessage), + /// Asks the recipient who it is, saying who the sender is (`transport::identity`). + Who(IdentityProof), + /// Answers a WHO. + Me(IdentityProof), +} + +/// A node's proof that it holds the key of its node ID (`identity`), sending from its +/// transport address to the recipient's: a signature over both addresses +/// (`transport::identity::signed`). WHO and ME end at the identity layer; the rest of +/// the node never sees them. +#[derive(Clone, Serialize, Deserialize)] +pub(crate) struct IdentityProof { + /// The sender's public node key. + #[serde(with = "crate::transport::wire::codec::bytes32")] + pub(crate) key: [u8; 32], + #[serde(with = "crate::transport::wire::codec::bytes64")] + pub(crate) signature: [u8; 64], +} + +impl std::fmt::Debug for IdentityProof { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("IdentityProof") + .field("key", &Hex(&self.key)) + .finish_non_exhaustive() + } } /// Content-addressed bytes that any node holding them can serve, in chunks @@ -237,6 +263,8 @@ pub(crate) enum MessageKind { Ping, Pong, Blob, + Who, + Me, } impl ParsedMessage { @@ -248,28 +276,23 @@ impl ParsedMessage { ParsedMessage::Ping(_) => MessageKind::Ping, ParsedMessage::Pong(_) => MessageKind::Pong, ParsedMessage::Blob(_) => MessageKind::Blob, + ParsedMessage::Who(_) => MessageKind::Who, + ParsedMessage::Me(_) => MessageKind::Me, } } - /// Every message carries the sender's epoch. - pub(crate) fn epoch(&self) -> Epoch { - match self { - ParsedMessage::Hello(hello) => hello.epoch, - ParsedMessage::HelloReply(reply) => reply.epoch, - ParsedMessage::Sync(envelope) => envelope.epoch, - ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch, - ParsedMessage::Blob(blob) => blob.epoch, - } - } - - /// `SyncEnvelope::epoch_seq`, as every message carries it. - pub(crate) fn epoch_seq(&self) -> u32 { + /// The sender's epoch and `SyncEnvelope::epoch_seq`, as every message carries them + /// but WHO and ME (which the node never sees). + pub(crate) fn epoch_stamp(&self) -> Option<(Epoch, u32)> { match self { - ParsedMessage::Hello(hello) => hello.epoch_seq, - ParsedMessage::HelloReply(reply) => reply.epoch_seq, - ParsedMessage::Sync(envelope) => envelope.epoch_seq, - ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => keepalive.epoch_seq, - ParsedMessage::Blob(blob) => blob.epoch_seq, + ParsedMessage::Hello(hello) => Some((hello.epoch, hello.epoch_seq)), + ParsedMessage::HelloReply(reply) => Some((reply.epoch, reply.epoch_seq)), + ParsedMessage::Sync(envelope) => Some((envelope.epoch, envelope.epoch_seq)), + ParsedMessage::Ping(keepalive) | ParsedMessage::Pong(keepalive) => { + Some((keepalive.epoch, keepalive.epoch_seq)) + } + ParsedMessage::Blob(blob) => Some((blob.epoch, blob.epoch_seq)), + ParsedMessage::Who(_) | ParsedMessage::Me(_) => None, } } } @@ -472,6 +495,9 @@ impl std::fmt::Debug for SyncPayload { #[derive(Clone, Debug)] pub(crate) struct EncodedMessage { pub(crate) peer: String, + /// Where the transport sends it: `peer` itself for a transport whose identities are + /// addresses, the peer's address under the identity layer (`transport::identity`). + pub(crate) address: String, pub(crate) body: bytes::Bytes, pub(crate) kind: MessageKind, /// Bytes the message takes on the wire (without the transport's own framing). diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index 17344c7..354aebc 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -446,6 +446,7 @@ pub(crate) fn encode_message(peer: &str, message: &ParsedMessage) -> Result<Enco let (outbound, payload_len) = format_message(peer, message)?; Ok(EncodedMessage { peer: peer.to_owned(), + address: peer.to_owned(), body: serde_json::to_vec(&outbound)?.into(), kind: message.kind(), payload_len, diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index c27f0d7..59acaf2 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -40,6 +40,9 @@ pub(crate) struct ByKind { pub(crate) ping: Counter, pub(crate) pong: Counter, pub(crate) blob: Counter, + /// WHO and ME (`transport::identity`), counted by the identity layer under the + /// address they went to or came from. + pub(crate) identity: Counter, } impl ByKind { @@ -51,6 +54,7 @@ impl ByKind { MessageKind::Ping => &mut self.ping, MessageKind::Pong => &mut self.pong, MessageKind::Blob => &mut self.blob, + MessageKind::Who | MessageKind::Me => &mut self.identity, } } } @@ -92,15 +96,15 @@ impl Traffic { } } - fn sent(&self, peer: &str, kind: MessageKind, bytes: usize) { + pub(crate) fn sent(&self, peer: &str, kind: MessageKind, bytes: usize) { self.count(peer, bytes, |t| t.sent.get_mut(kind)); } - fn refused(&self, peer: &str, kind: MessageKind, bytes: usize) { + pub(crate) fn refused(&self, peer: &str, kind: MessageKind, bytes: usize) { self.count(peer, bytes, |t| t.refused.get_mut(kind)); } - fn received(&self, peer: &str, kind: MessageKind, bytes: usize) { + pub(crate) fn received(&self, peer: &str, kind: MessageKind, bytes: usize) { self.count(peer, bytes, |t| t.received.get_mut(kind)); } diff --git a/src/transport/wire.rs b/src/transport/wire.rs index ceeaf87..5577f42 100644 --- a/src/transport/wire.rs +++ b/src/transport/wire.rs @@ -14,7 +14,7 @@ use super::{MessageKind, ParsedMessage}; /// The wire format's version, the first byte of every payload. Bump it with every change /// to a message type or a codec (`golden_bytes` fails then). -pub(crate) const VERSION: u8 = 4; +pub(crate) const VERSION: u8 = 5; /// What the second byte of a payload says. fn tag(kind: MessageKind) -> u8 { @@ -25,6 +25,8 @@ fn tag(kind: MessageKind) -> u8 { MessageKind::Ping => 3, MessageKind::Pong => 4, MessageKind::Blob => 5, + MessageKind::Who => 6, + MessageKind::Me => 7, } } @@ -57,6 +59,9 @@ pub(crate) fn encode(message: &ParsedMessage) -> Result<Vec<u8>, WireError> { rmp_serde::encode::write(&mut payload, keepalive)? } ParsedMessage::Blob(blob) => rmp_serde::encode::write(&mut payload, blob)?, + ParsedMessage::Who(proof) | ParsedMessage::Me(proof) => { + rmp_serde::encode::write(&mut payload, proof)? + } } Ok(payload) } @@ -81,6 +86,10 @@ pub(crate) fn decode(payload: &[u8]) -> Result<ParsedMessage, WireError> { ParsedMessage::Pong(rmp_serde::from_slice(body)?) } else if is(MessageKind::Blob) { ParsedMessage::Blob(rmp_serde::from_slice(body)?) + } else if is(MessageKind::Who) { + ParsedMessage::Who(rmp_serde::from_slice(body)?) + } else if is(MessageKind::Me) { + ParsedMessage::Me(rmp_serde::from_slice(body)?) } else { return Err(WireError::UnknownKind(kind)); }) @@ -125,6 +134,19 @@ pub(crate) mod codec { } } + /// A 64-byte signature. + pub(crate) mod bytes64 { + use super::*; + + pub(crate) fn serialize<S: Serializer>(value: &[u8; 64], s: S) -> Result<S::Ok, S::Error> { + Bytes::new(value).serialize(s) + } + + pub(crate) fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 64], D::Error> { + fixed(&ByteBuf::deserialize(d)?, "a signature") + } + } + /// 32-byte hashes. pub(crate) mod hashes32 { use super::*; @@ -299,8 +321,8 @@ mod tests { use super::*; use crate::document::replica::Epoch; use crate::transport::{ - BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, Keepalive, - ListedPeer, MAX_LISTED_HEADS, + BlobBody, BlobId, BlobMessage, HeadsSummary, HelloReply, HelloRequest, IdentityProof, + Keepalive, ListedPeer, MAX_LISTED_HEADS, }; use automerge::ChangeHash; @@ -447,7 +469,7 @@ mod tests { }); assert_eq!( hex(&encode(&hello).unwrap()), - format!("040097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) + format!("050097a5737761726d{epoch_hex}01c403010203c3{}07", run("04")) ); let reply = ParsedMessage::HelloReply(HelloReply { @@ -467,7 +489,7 @@ mod tests { assert_eq!( hex(&encode(&reply).unwrap()), format!( - "040199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", + "050199a5737761726d9192a17091a161{epoch_hex}01c400c2{}05{}", run("04"), run("06") ) @@ -482,7 +504,27 @@ mod tests { }); assert_eq!( hex(&encode(&ping).unwrap()), - format!("0403959290c400{epoch_hex}01c3c2") + format!("0503959290c400{epoch_hex}01c3c2") ); + + let who = ParsedMessage::Who(IdentityProof { + key: [8; 32], + signature: [9; 64], + }); + assert_eq!( + hex(&encode(&who).unwrap()), + format!("050692c420{}c440{}", "08".repeat(32), "09".repeat(64)) + ); + let ParsedMessage::Me(proof) = decode( + &encode(&ParsedMessage::Me(IdentityProof { + key: [8; 32], + signature: [9; 64], + })) + .unwrap(), + ) + .unwrap() else { + panic!("not a ME"); + }; + assert_eq!((proof.key, proof.signature), ([8; 32], [9; 64])); } } |