diff options
| -rw-r--r-- | src/api/stats/prometheus.rs | 1 | ||||
| -rw-r--r-- | src/identity.rs | 43 | ||||
| -rw-r--r-- | src/main.rs | 6 | ||||
| -rw-r--r-- | src/node.rs | 24 | ||||
| -rw-r--r-- | src/simtest/identity.rs | 128 | ||||
| -rw-r--r-- | src/simtest/mod.rs | 26 | ||||
| -rw-r--r-- | src/swarm/peer/inbound.rs | 8 | ||||
| -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 |
12 files changed, 832 insertions, 41 deletions
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index f9eb3bf..8d31eb0 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -172,6 +172,7 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou (direction, "ping", by.ping), (direction, "pong", by.pong), (direction, "blob", by.blob), + (direction, "identity", by.identity), ] }; let mut series = Vec::new(); diff --git a/src/identity.rs b/src/identity.rs index 8f7fdda..93d8739 100644 --- a/src/identity.rs +++ b/src/identity.rs @@ -59,7 +59,8 @@ impl NodeKey { Ok(Self::from_seed(seed)) } - fn from_seed(seed: [u8; KEY_BYTES]) -> Self { + /// A key from its 32 secret bytes: for simulated nodes, which have no key files. + pub(crate) fn from_seed(seed: [u8; KEY_BYTES]) -> Self { Self { signing: SigningKey::from_bytes(&seed), } @@ -87,11 +88,32 @@ impl NodeKey { /// This node's ID. pub(crate) fn id(&self) -> String { - format!( - "{NODE_ID_PREFIX}{}", - Hex(self.signing.verifying_key().as_bytes()) - ) + node_id_of(&self.public()) } + + /// The public key, as others check signatures with it. + pub(crate) fn public(&self) -> [u8; 32] { + self.signing.verifying_key().to_bytes() + } + + pub(crate) fn sign(&self, message: &[u8]) -> [u8; 64] { + use ed25519_dalek::Signer; + self.signing.sign(message).to_bytes() + } +} + +/// The node ID of the node with public key `key`. +pub(crate) fn node_id_of(key: &[u8; 32]) -> String { + format!("{NODE_ID_PREFIX}{}", Hex(key)) +} + +/// Whether `signature` is `key`'s over `message`. +pub(crate) fn verify(key: &[u8; 32], message: &[u8], signature: &[u8; 64]) -> bool { + let Ok(key) = ed25519_dalek::VerifyingKey::from_bytes(key) else { + return false; + }; + key.verify_strict(message, &ed25519_dalek::Signature::from_bytes(signature)) + .is_ok() } /// Whether `id` has the form of a node ID. @@ -144,6 +166,17 @@ mod tests { } #[test] + fn signatures() { + let key = NodeKey::from_seed([2; KEY_BYTES]); + let signature = key.sign(b"message"); + assert!(verify(&key.public(), b"message", &signature)); + assert!(!verify(&key.public(), b"massage", &signature)); + let other = NodeKey::from_seed([3; KEY_BYTES]); + assert!(!verify(&other.public(), b"message", &signature)); + assert_eq!(node_id_of(&key.public()), key.id()); + } + + #[test] fn node_ids() { let id = NodeKey::from_seed([1; KEY_BYTES]).id(); assert!(is_node_id(&id)); diff --git a/src/main.rs b/src/main.rs index a891110..b67ac1d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -161,7 +161,7 @@ async fn main() -> Result<()> { debug!(epoch = %initial_epoch, "Local epoch"); let mycelium_transport = Arc::new(mycelium::MyceliumTransport::connect(mycelium_setup).await?); - debug!(local_id = mycelium_transport.local_id(), "Transport ready"); + debug!(address = mycelium_transport.local_id(), "Transport ready"); let delivery_limit = mycelium_transport.delivery_limit(); let document_file = args.document_file.clone().map(|path| { @@ -174,6 +174,10 @@ async fn main() -> Result<()> { }); let node = node::Node::new(node::NodeConfig { transport: mycelium_transport, + identity: Some(transport::identity::IdentityConfig { + key: Arc::new(node_key), + scheme: "mycelium", + }), initial_epoch, initial_doc, earlier_actors, diff --git a/src/node.rs b/src/node.rs index 57ee302..8e4ace7 100644 --- a/src/node.rs +++ b/src/node.rs @@ -24,12 +24,18 @@ use crate::swarm::sync::SyncEvent; use crate::swarm::{self, peer, sync}; use crate::transport::Transport; use crate::transport::firewall::{Firewall, FirewallTransport}; +use crate::transport::identity::{IdentityConfig, IdentityTransport}; use crate::transport::traffic::{CountingTransport, Traffic}; /// What a node is built from. pub(crate) struct NodeConfig { - /// The node's transport, without the counting and firewall layers (added here). + /// The node's transport, without the identity, counting and firewall layers (added + /// here). pub(crate) transport: Arc<dyn Transport>, + /// The node key and the transport's address scheme: the transport goes under the + /// identity layer (`transport::identity`), and the node is its node ID. Without it + /// (simulations only), the node is its transport identity. + pub(crate) identity: Option<IdentityConfig>, pub(crate) initial_epoch: Epoch, pub(crate) initial_doc: AutoCommit, /// Actors this node wrote with in earlier runs (`--document-file`). @@ -66,6 +72,14 @@ impl Node { let firewall = Arc::new(Firewall::default()); let traffic = Arc::new(Traffic::default()); let initial_actor = config.initial_doc.get_actor().clone(); + let transport = match config.identity { + Some(identity) => Arc::new(IdentityTransport::new( + config.transport, + identity, + Arc::clone(&traffic), + )) as Arc<dyn Transport>, + None => config.transport, + }; let replica = Replica::new(config.initial_epoch, config.initial_doc); let stamp = replica.stamps(); @@ -78,12 +92,10 @@ impl Node { })), peer_states: Arc::new(RwLock::new(HashMap::new())), // Counted outside the firewall: what it drops outbound counts as sent (it - // plays a lossy network), what it drops inbound never arrived. + // plays a lossy network), what it drops inbound never arrived. Both above + // the identity layer, so by node ID. transport: Arc::new(CountingTransport::new( - Arc::new(FirewallTransport::new( - config.transport, - Arc::clone(&firewall), - )), + Arc::new(FirewallTransport::new(transport, Arc::clone(&firewall))), Arc::clone(&traffic), )), firewall, diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs new file mode 100644 index 0000000..a33783f --- /dev/null +++ b/src/simtest/identity.rs @@ -0,0 +1,128 @@ +//! Nodes under the identity layer (`transport::identity`): named by node IDs, found +//! through addresses (seeds, HELLO_REPLYs, the member list), and believed only once a +//! node proved itself at an address. + +use std::time::Duration; + +use serde_json::json; + +use super::{Sim, read, sim_key, write}; +use crate::identity::{self, NodeKey}; +use crate::state::AppState; +use crate::swarm::bootstrap::Start; +use crate::swarm::members; +use crate::transport::identity::{WHO_LABEL, signed}; +use crate::transport::{HeadsSummary, IdentityProof, Keepalive, ParsedMessage}; + +const SOON: Duration = Duration::from_secs(20); + +async fn peers_are_node_ids(node: &AppState) { + for (peer, _) in node.peers_snapshot().await { + assert!(identity::is_node_id(&peer), "peer {peer}"); + } +} + +/// Seeds by name (resolved by the net) and by address; the third node finds the second +/// only through what the first lists in its HELLO_REPLY. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn nodes_are_their_node_ids() { + let sim = Sim::with_identity(51); + let a = sim.start("a", &[], Start::Genesis); + let b = sim.start("b", &["a"], Start::Join); + let c = sim.start("c", &["sim:a"], Start::Join); + sim.wait_ready(&[&a, &b, &c], SOON).await; + assert_eq!(a.local_id(), sim_key("a").id()); + + write(&b, "x", json!(1)).await; + sim.wait_converged(&[&a, &b, &c], SOON).await; + assert_eq!(read(&c, "x").await, Some(json!(1))); + for node in [&a, &b, &c] { + peers_are_node_ids(node).await; + } + let (b_id, c_id) = (b.local_id().to_owned(), c.local_id().to_owned()); + let a = &a; + sim.wait_until( + "b and c are registered with their addresses", + SOON, + move || { + let (b_id, c_id) = (b_id.clone(), c_id.clone()); + async move { + let replica = a.doc.read().await; + members::addresses(&replica.doc, &b_id) == ["sim:b"] + && members::addresses(&replica.doc, &c_id) == ["sim:c"] + } + }, + ) + .await; +} + +/// With two neighbours each, most pairs of six nodes never meet through seeds: they +/// find each other's addresses in the member list. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn members_find_each_other_through_the_member_list() { + let sim = Sim::with_identity(52); + let two = |config: &mut crate::node::NodeConfig| config.neighbours_k = 2; + let a = sim.start_with("a", &[], Start::Genesis, two); + let others: Vec<AppState> = ["b", "c", "d", "e", "f"] + .into_iter() + .map(|id| sim.start_with(id, &["a"], Start::Join, two)) + .collect(); + let mut all: Vec<&AppState> = vec![&a]; + all.extend(others.iter()); + sim.wait_ready(&all, SOON).await; + + write(&others[4], "from/f", json!("f")).await; + write(&others[0], "from/b", json!("b")).await; + sim.wait_converged(&all, SOON).await; + for node in &all { + assert_eq!(read(node, "from/f").await, Some(json!("f"))); + assert_eq!(read(node, "from/b").await, Some(json!("b"))); + peers_are_node_ids(node).await; + } +} + +fn ping(node: &AppState) -> ParsedMessage { + ParsedMessage::Ping(Keepalive { + heads: HeadsSummary::of(Vec::new()), + epoch: node.local_epoch(), + epoch_seq: 0, + ready: false, + syncs_with_you: false, + }) +} + +fn who(key: &NodeKey, from: &str, to: &str) -> ParsedMessage { + ParsedMessage::Who(IdentityProof { + key: key.public(), + signature: key.sign(&signed(WHO_LABEL, from, to)), + }) +} + +/// A proof counts only for the address it was made at: messages from an address with +/// a proof made elsewhere are dropped, and the node never hears of them. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_proof_holds_only_at_its_address() { + let sim = Sim::with_identity(53); + let a = sim.start("a", &[], Start::Genesis); + sim.wait_ready(&[&a], SOON).await; + let ghost = NodeKey::from_seed([9; 32]); + + // Made for another sender address, or for another recipient. + for forged in [ + who(&ghost, "sim:elsewhere", "sim:a"), + who(&ghost, "sim:ghost", "sim:b"), + ] { + sim.net.inject("ghost", "a", &forged).await.unwrap(); + sim.net.inject("ghost", "a", &ping(&a)).await.unwrap(); + assert!(a.get_peer_state(&ghost.id()).await.is_none()); + } + + sim.net + .inject("ghost", "a", &who(&ghost, "sim:ghost", "sim:a")) + .await + .unwrap(); + sim.net.inject("ghost", "a", &ping(&a)).await.unwrap(); + assert!(a.get_peer_state(&ghost.id()).await.is_some()); + // Known by its node ID, not by its address. + assert!(a.get_peer_state("ghost").await.is_none()); +} diff --git a/src/simtest/mod.rs b/src/simtest/mod.rs index 531d1c6..4d7374a 100644 --- a/src/simtest/mod.rs +++ b/src/simtest/mod.rs @@ -10,6 +10,7 @@ //! (all nodes interleaved; messages name peers by ID). mod deltas; +mod identity; mod remakes; mod soak; mod timing; @@ -26,10 +27,12 @@ use tokio::time::Instant; use crate::document::replica::Epoch; use crate::document::store; use crate::files::Files; +use crate::identity::NodeKey; use crate::locks::RwLockExt; use crate::node::{self, Node, NodeConfig}; use crate::state::{AppState, PeerState, Timing}; use crate::swarm::bootstrap::{self, Start}; +use crate::transport::identity::IdentityConfig; use crate::transport::sim::SimNet; /// Keepalive of simulated nodes: dormancy after 1 s, probes of dormant peers every @@ -45,6 +48,9 @@ pub(crate) const DELIVERY_LIMIT: Duration = Duration::from_secs(3); /// A simulated swarm: the net and the nodes started on it. pub(crate) struct Sim { pub(crate) net: SimNet, + /// Nodes under the identity layer (`transport::identity`), named by node IDs; else + /// by their names on the net. + identity: bool, /// Every node started, for the description of a failed wait. nodes: std::sync::Mutex<Vec<(String, AppState)>>, } @@ -54,10 +60,20 @@ impl Sim { init_logging(); Self { net: SimNet::new(seed), + identity: false, nodes: Default::default(), } } + /// [`new`](Self::new), with every node under the identity layer: its key derived + /// from its name, its node ID as `local_id`. + pub(crate) fn with_identity(seed: u64) -> Self { + Self { + identity: true, + ..Self::new(seed) + } + } + /// Starts node `id` with the test defaults: builds it, starts its workers, and /// bootstraps it in the background (through `seeds`, other nodes' IDs). Returns at /// once; [`wait_ready`] waits for it. @@ -75,6 +91,10 @@ impl Sim { ) -> AppState { let mut config = NodeConfig { transport: self.net.transport(id), + identity: self.identity.then(|| IdentityConfig { + key: Arc::new(sim_key(id)), + scheme: "sim", + }), initial_epoch: Epoch::GENESIS, initial_doc: AutoCommit::new(), earlier_actors: Vec::new(), @@ -239,6 +259,12 @@ impl Sim { } } +/// The node key of simulated node `id`. +pub(crate) fn sim_key(id: &str) -> NodeKey { + use sha2::Digest; + NodeKey::from_seed(sha2::Sha256::digest(id.as_bytes()).into()) +} + fn path(path: &str) -> Vec<String> { path.split('/').map(str::to_owned).collect() } diff --git a/src/swarm/peer/inbound.rs b/src/swarm/peer/inbound.rs index a1f6646..b04aad9 100644 --- a/src/swarm/peer/inbound.rs +++ b/src/swarm/peer/inbound.rs @@ -30,6 +30,10 @@ pub(crate) async fn handle_message( debug!("Ignoring message from self"); return Ok(()); } + // WHO and ME end at the identity layer (`transport::identity`). + let Some((epoch, epoch_seq)) = message.epoch_stamp() else { + return Ok(()); + }; // A reply to a HELLO from an earlier run of ours (late, or queued by the transport // while we were down) says nothing about now; the peer answers our current HELLO. @@ -70,8 +74,7 @@ pub(crate) async fn handle_message( transport::ParsedMessage::HelloReply(reply) => Some(reply.sender_run_id), _ => None, }; - let same_epoch = - observe_peer_epoch(state, &peer, message.epoch(), message.epoch_seq(), run).await; + let same_epoch = observe_peer_epoch(state, &peer, epoch, epoch_seq, run).await; //debug!(from = %peer, message = ?message, "Received message"); @@ -155,6 +158,7 @@ pub(crate) async fn handle_message( state.lock_rtt().pong_received(&peer, Instant::now()); on_keepalive(state, &peer, &remote, reactivated).await; } + transport::ParsedMessage::Who(_) | transport::ParsedMessage::Me(_) => {} } Ok(()) 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])); } } |