//! The identity layer: between the node and its transports, 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 may have several //! transports; a peer is reached on one both have, so nodes on several transports //! carry the traffic between the others. //! //! 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, to another recipient or //! on another transport, 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 one of the transports 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 addresses //! it proved itself at, best first (`Paths::rank`, `Paths::send_first`: the next //! when a send fails); with only hints, those are asked WHO, and the send is refused //! until one answers (the node retries refused sends). A node proven on some //! transports is asked WHO at its hints on the others too, now and then //! (`OTHER_PATHS_ASK_INTERVAL`), so that each transport both have becomes a path. //! //! Addresses and the transports behind them are the paths layer's //! (`transport::paths`): the books keep addresses as paths keep them. 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::firewall::Firewall; use crate::transport::paths::{Gate, PathHandler, Paths, scheme_of}; use crate::transport::stats::{IdentityCounts, IdentityStats, TransportStats}; use crate::transport::traffic::Traffic; use crate::transport::{ Destination, 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); /// A node's addresses on transports it hasn't proven itself on are asked WHO at most /// this often (`Book::unproven_hints`): over Tor, a WHO is a dial that takes seconds, /// and an address that doesn't answer shouldn't be pressed. const OTHER_PATHS_ASK_INTERVAL: Duration = Duration::from_secs(60); /// 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; /// A send refused because the peer hasn't proven where it is yet (`send_encoded`): /// expected when a peer known only from hints is first contacted, and retried. #[derive(Debug, thiserror::Error)] #[error("{0} hasn't proven where it is yet; asked where it is said to be")] pub(crate) struct NotProven(String); /// Whether `error` is (or was caused by) a [`NotProven`]. pub(crate) fn is_not_proven(error: &anyhow::Error) -> bool { error.chain().any(|cause| cause.is::()) } /// What a node under the identity layer needs (`node::NodeConfig::identity`). #[derive(Clone)] pub(crate) struct IdentityConfig { pub(crate) key: Arc, /// The address scheme of the node's transport (`NodeConfig::transport`), e.g. /// `mycelium`. pub(crate) scheme: &'static str, /// Further transports, by scheme, after the first in preference. pub(crate) more: Vec<(&'static str, Arc)>, } /// 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 { 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, } struct Shared { paths: Arc, /// The debug firewall, by node ID; paths applies it (`Gate`). firewall: Option>, key: Arc, node_id: String, /// WHO and ME are counted here, under the address (`traffic`). traffic: Arc, // A std mutex: every access is short and never spans an await. book: Mutex, } /// Who is where, by address. #[derive(Default)] struct Book { /// The node that proved itself at an address, and when. at: HashMap, /// The addresses each node last proved itself at, one per scheme. proven: HashMap>, /// Where others say a node is. hints: HashMap>, /// When each address was last asked WHO. asked: HashMap, /// Seeds being resolved: who waits for a ME from where. waiting: HashMap>>, /// Since start (`stats`). counts: IdentityCounts, } impl Book { /// `node` proved itself at `address`. Returns whether that is news. fn bind(&mut self, address: &str, node: &str, now: Instant) -> bool { let news = self.at.get(address).is_none_or(|(known, _)| known != node); // One address per node and scheme, and one node per address: a node that // moved, or an address taken over by another node, leaves no stale entry. let scheme = scheme_of(address); let addresses = self.proven.entry(node.to_owned()).or_default(); if let Some(slot) = addresses.iter_mut().find(|at| scheme_of(at) == scheme) { if *slot != address { let previous = std::mem::replace(slot, address.to_owned()); self.at.remove(&previous); self.counts.address_changes += 1; } } else { addresses.push(address.to_owned()); } if let Some((previous, _)) = self.at.insert(address.to_owned(), (node.to_owned(), now)) && previous != node { self.forget_proof(&previous, address); } if self.at.len() > MAX_ENTRIES && let Some(oldest) = self .at .iter() .min_by_key(|(_, (_, at))| *at) .map(|(address, _)| address.clone()) && let Some((node, _)) = self.at.remove(&oldest) { self.forget_proof(&node, &oldest); } for waiter in self.waiting.remove(address).into_iter().flatten() { let _ = waiter.send(node.to_owned()); } news } fn forget_proof(&mut self, node: &str, address: &str) { if let Some(addresses) = self.proven.get_mut(node) { addresses.retain(|at| at != address); if addresses.is_empty() { self.proven.remove(node); } } } /// What the books hold now, and the counts. fn stats(&self) -> IdentityStats { let mut proven = std::collections::BTreeMap::new(); for addresses in self.proven.values() { for address in addresses { *proven.entry(scheme_of(address).to_owned()).or_insert(0) += 1; } } IdentityStats { proven, hinted: self .hints .keys() .filter(|node| !self.proven.contains_key(*node)) .count(), counts: self.counts, } } /// Where `node` is said to be on transports it hasn't proven itself on yet. fn unproven_hints(&self, node: &str) -> Vec { let proven = self.proven.get(node); self.hints .get(node) .into_iter() .flatten() .filter(|hint| { proven.is_none_or(|proven| { !proven .iter() .any(|address| scheme_of(address) == scheme_of(hint)) }) }) .cloned() .collect() } /// Whether to ask `address` WHO now (always, if `force`), noting that it was asked. fn ask(&mut self, address: &str, now: Instant, force: bool) -> bool { let interval = if force { Duration::ZERO } else { ASK_INTERVAL }; self.ask_every(address, now, interval) } /// Whether to ask `address` WHO now, if it wasn't within `interval`, noting that it /// was asked. fn ask_every(&mut self, address: &str, now: Instant, interval: Duration) -> bool { if self .asked .get(address) .is_some_and(|at| now.duration_since(*at) < interval) { return false; } if self.asked.len() >= MAX_ENTRIES { // The longest interval: other paths' asks are kept until they are due again. self.asked .retain(|_, at| now.duration_since(*at) < OTHER_PATHS_ASK_INTERVAL); } self.asked.insert(address.to_owned(), now); true } } impl IdentityTransport { /// Over `transport`, whose addresses have `config.scheme`, and `config.more`. /// `firewall`: the debug firewall, if any, which paths applies by address. pub(crate) fn new( transport: Arc, config: IdentityConfig, traffic: Arc, firewall: Option>, ) -> Self { let node_id = config.key.id(); let shared = Arc::new_cyclic(|shared: &std::sync::Weak| { let gate: std::sync::Weak = shared.clone(); Shared { paths: Arc::new(Paths::new( std::iter::once((config.scheme, transport)) .chain(config.more) .collect(), Some(Arc::clone(&traffic)), firewall.is_some().then_some(gate), )), firewall, key: config.key, node_id, traffic, book: Mutex::default(), } }); Self { shared } } } /// The firewall's rules are by node ID: an address is blocked if the node proven at it /// is. Addresses no node proved itself at aren't (nothing could name them). impl Gate for Shared { fn blocks_inbound(&self, address: &str) -> bool { self.blocks(address, Firewall::blocks_inbound) } fn blocks_outbound(&self, address: &str) -> bool { self.blocks(address, Firewall::blocks_outbound) } } impl Shared { fn blocks(&self, address: &str, rule: fn(&Firewall, &str) -> bool) -> bool { let Some(firewall) = &self.firewall else { return false; }; let node = self.lock().at.get(address).map(|(node, _)| node.clone()); node.is_some_and(|node| rule(firewall, &node)) } } impl Shared { fn lock(&self) -> MutexGuard<'_, Book> { self.book.lock_or_recover() } /// Our proof for `address` (as paths keep it), from our address on its transport. fn proof(&self, label: &[u8], address: &str) -> Option { let message = signed(label, &self.paths.own_address(address)?, address); Some(IdentityProof { key: self.key.public(), signature: self.key.sign(&message), }) } /// Notes the node `proof` proves to be at `address` (as paths keep it); None if it /// proves nothing. fn prove(&self, address: &str, label: &[u8], proof: &IdentityProof) -> Option { let message = signed(label, address, &self.paths.own_address(address)?); if !identity::verify(&proof.key, &message, &proof.signature) { self.lock().counts.proofs_rejected += 1; debug!(from = %address, "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 = %address, "Ignoring a proof of our own node key"); return None; } if self.lock().bind(address, &node, Instant::now()) { debug!(node = %node, address = %address, "Node proved its address"); } Some(node) } /// Asks `address` WHO, unless it was asked recently (`force`: anyway). fn ask(self: &Arc, address: &str, force: bool) { if !self.lock().ask(address, Instant::now(), force) { return; } self.send_who(address); } /// Asks `node`'s addresses on transports it hasn't proven itself on WHO, now and then /// (`OTHER_PATHS_ASK_INTERVAL`): its ME there proves another path to it, which the /// paths layer then measures, and fails over to. fn ask_other_paths(self: &Arc, node: &str) { let due: Vec = { let mut book = self.lock(); let now = Instant::now(); book.unproven_hints(node) .into_iter() .filter(|address| book.ask_every(address, now, OTHER_PATHS_ASK_INTERVAL)) .collect() }; for address in due { self.send_who(&address); } } fn send_who(self: &Arc, address: &str) { let shared = Arc::clone(self); let address = address.to_owned(); tokio::spawn(async move { shared.send_proof(MessageKind::Who, &address).await }); } /// Sends a WHO or ME (`kind`) to `address`. async fn send_proof(&self, kind: MessageKind, address: &str) { let Some(address) = self.paths.canonical(address) else { return; }; let address = address.as_str(); let message = if kind == MessageKind::Who { self.proof(WHO_LABEL, address).map(ParsedMessage::Who) } else { self.proof(ME_LABEL, address).map(ParsedMessage::Me) }; let Some(message) = message else { return; }; let encoded = match self.paths.encode(address, &message) { Ok(encoded) => encoded, Err(error) => { debug!(to = %address, error = %format!("{error:#}"), "Couldn't encode an identity proof"); return; } }; match self.paths.send(&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"); } } } /// A message from `address`. async fn receive( self: Arc, address: String, message: ParsedMessage, payload_len: usize, handler: &HandlerFn, ) -> Result<()> { match message { ParsedMessage::Who(proof) => { self.traffic .received(&address, MessageKind::Who, payload_len); if self.prove(&address, WHO_LABEL, &proof).is_some() { let shared = Arc::clone(&self); tokio::spawn(async move { shared.send_proof(MessageKind::Me, &address).await }); } Ok(()) } ParsedMessage::Me(proof) => { self.traffic .received(&address, MessageKind::Me, payload_len); self.prove(&address, ME_LABEL, &proof); Ok(()) } message => { let node = self.lock().at.get(&address).map(|(node, _)| node.clone()); match node { Some(node) => handler(node, message, payload_len).await, None => { self.lock().counts.dropped_unproven += 1; debug!( from = %address, kind = ?message.kind(), "Dropping a message from an address no node proved itself at; asking who is there" ); self.ask(&address, false); Ok(()) } } } } } /// Where to send to `peer`: an address it proved itself at, else one it is said to /// be at; the most preferred transport's first. fn destination(&self, peer: &str) -> Option { self.proven_at(peer) .into_iter() .next() .or_else(|| self.hinted_at(peer).into_iter().next()) } /// The addresses `peer` proved itself at, best first (`Paths::rank`). fn proven_at(&self, peer: &str) -> Vec { let proven = self.lock().proven.get(peer).cloned().unwrap_or_default(); self.paths.rank(&proven) } /// Where `peer` is said to be, best first. fn hinted_at(&self, peer: &str) -> Vec { let hints = self.lock().hints.get(peer).cloned().unwrap_or_default(); self.paths.rank(&hints) } } #[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.paths.delivery_limit() } fn local_addresses(&self) -> Vec { self.shared.paths.local_addresses() } /// Where `peer` proved itself, else where it is said to be, as others can use it. fn addresses_of(&self, peer: &str) -> Vec { let addresses = { let book = self.shared.lock(); book.proven .get(peer) .or_else(|| book.hints.get(peer)) .cloned() .unwrap_or_default() }; let mut locations = Vec::new(); for address in addresses { for location in self.shared.paths.locations(&address) { if !locations.contains(&location) { locations.push(location); } } } locations } fn learn_addresses(&self, peer: &str, addresses: &[String]) { if peer == self.shared.node_id || !identity::is_node_id(peer) { return; } let mut known: Vec = Vec::new(); for address in addresses { if let Some(address) = self.shared.paths.canonical(address) && !self.shared.paths.is_own(&address) && !known.contains(&address) { known.push(address); } } known.truncate(MAX_HINTS); if known.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(), known); } } fn can_reach(&self, peer: &str) -> bool { self.shared.destination(peer).is_some() } fn transports(&self) -> Vec { self.shared.paths.transports() } /// The address `encode` picks, the transport's connection there, and every path /// to `peer` as paths measures it, best first. fn destination(&self, peer: &str) -> Option { let proven = self.shared.proven_at(peer); let (address, is_proven) = match proven.first() { Some(address) => (address.clone(), true), None => (self.shared.hinted_at(peer).into_iter().next()?, false), }; let connection = self.shared.paths.connection(&address); Some(Destination { address: self.shared.paths.location(&address), proven: is_proven, connection, paths: self.shared.paths.info(&proven), }) } /// The paths layer answers them (`Paths::receive`). fn answers_pings(&self) -> bool { true } /// The best path's, where messages to `peer` go. fn rtt(&self, peer: &str) -> Option { let address = self.shared.proven_at(peer).into_iter().next()?; self.shared.paths.rtt(&address) } /// The identity layer's, and each transport's. fn stats(&self) -> TransportStats { let mut stats = self.shared.paths.stats(); stats.merge(TransportStats { identity: Some(self.shared.lock().stats()), ..TransportStats::default() }); stats } fn discovered_seeds(&self) -> Vec { self.shared.paths.discovered_seeds() } /// A node ID resolves to itself if this node knows where it is; anything else is /// an address (`:...`) or what one of the transports resolves to one (the /// first that does), asked WHO. async fn resolve_seed(&self, seed: &str) -> Result { if identity::is_node_id(seed) { if self.shared.destination(seed).is_some() { return Ok(seed.to_owned()); } bail!("a node ID alone doesn't say where that node is; give its address"); } let address = self.shared.paths.resolve(seed).await?; if self.shared.paths.is_own(&address) { bail!("{seed} is this node"); } let receiver = { let mut book = self.shared.lock(); if let Some((node, _)) = book.at.get(&address) { return Ok(node.clone()); } let (sender, receiver) = oneshot::channel(); let waiting = book.waiting.entry(address.clone()).or_default(); waiting.retain(|waiter| !waiter.is_closed()); waiting.push(sender); receiver }; self.shared.ask(&address, true); match tokio::time::timeout(RESOLVE_TIMEOUT, receiver).await { Ok(Ok(node)) => { info!(seed = %seed, node = %node, "Seed is node"); Ok(node) } _ => { self.shared.lock().counts.seeds_unanswered += 1; Err(anyhow!( "no node answered at {address} within {RESOLVE_TIMEOUT:?}" )) } } } /// For the address `destination` picks (a send there is refused until `peer` /// proved itself there, `send_encoded`). fn encode(&self, peer: &str, message: &ParsedMessage) -> Result { let Some(address) = self.shared.destination(peer) else { bail!("no address known for {peer} on this node's transports"); }; let mut encoded = self.shared.paths.encode(&address, message)?; encoded.peer = peer.to_owned(); Ok(encoded) } /// To the addresses `peer` proved itself at, best first, the next when a send /// fails (`Paths::send_first`); refused if it proved itself nowhere yet. async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { let proven = self.shared.proven_at(&message.peer); if !proven.is_empty() { self.shared.ask_other_paths(&message.peer); } if proven.is_empty() { let hints = { let mut book = self.shared.lock(); book.counts.refused_unproven += 1; book.hints.get(&message.peer).cloned().unwrap_or_default() }; for address in hints { self.shared.ask(&address, false); } return Err(NotProven(message.peer.clone()).into()); } self.shared .paths .send_first(message, &proven) .await .map(|_| ()) } /// Every transport's consumer (`Paths::start_consumer`); returns when the first /// one does. async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { let handler: Arc = Arc::from(handler); let shared = Arc::clone(&self.shared); let receive: Arc = Arc::new(move |address, message, payload_len| { let shared = Arc::clone(&shared); let handler = Arc::clone(&handler); Box::pin(async move { shared .receive(address, message, payload_len, &*handler) .await }) }); self.shared.paths.start_consumer(receive).await } /// When the first transport exits. async fn wait_for_exit(&self) -> std::io::Result { self.shared.paths.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_scheme_and_one_node_per_address() { let now = Instant::now(); let mut book = Book::default(); assert!(book.bind("s:x", "n1", now)); assert!(!book.bind("s:x", "n1", now)); // n1 moved to s:y: s:x is no longer n1's. assert!(book.bind("s:y", "n1", now)); assert!(!book.at.contains_key("s:x")); assert_eq!(book.proven["n1"], ["s:y"]); // On another transport too. assert!(book.bind("t:x", "n1", now)); assert_eq!(book.proven["n1"], ["s:y", "t:x"]); // n2 took over s:y: n1 is proven on t only. assert!(book.bind("s:y", "n2", now)); assert_eq!(book.proven["n1"], ["t:x"]); assert_eq!(book.at["s:y"].0, "n2"); // Only n1's move from s:x to s:y changed an address. assert_eq!(book.counts.address_changes, 1); let stats = book.stats(); assert_eq!(stats.proven["s"], 1); assert_eq!(stats.proven["t"], 1); } #[test] fn other_paths_are_the_hints_on_transports_not_proven() { let now = Instant::now(); let mut book = Book::default(); book.hints .insert("n1".into(), vec!["s:x".into(), "t:x".into(), "u:x".into()]); // Proven nowhere: every hint. assert_eq!(book.unproven_hints("n1"), ["s:x", "t:x", "u:x"]); // Proven on s (even at another address): the others. book.bind("s:y", "n1", now); assert_eq!(book.unproven_hints("n1"), ["t:x", "u:x"]); assert!(book.ask_every("t:x", now, OTHER_PATHS_ASK_INTERVAL)); assert!(!book.ask_every("t:x", now + ASK_INTERVAL, OTHER_PATHS_ASK_INTERVAL)); assert!(book.ask_every( "t:x", now + OTHER_PATHS_ASK_INTERVAL, OTHER_PATHS_ASK_INTERVAL )); assert!(book.unproven_hints("n2").is_empty()); } #[test] fn asking_is_rate_limited() { let now = Instant::now(); let mut book = Book::default(); assert!(book.ask("s:x", now, false)); assert!(!book.ask("s:x", now, false)); assert!(book.ask("s:x", now, true)); assert!(book.ask("s: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("s:x".into(), vec![sender]); book.bind("s:x", "n1", Instant::now()); assert_eq!(receiver.await.unwrap(), "n1"); } }