diff options
| -rw-r--r-- | docs/trust.md | 7 | ||||
| -rw-r--r-- | src/swarm/peer/handshake.rs | 4 | ||||
| -rw-r--r-- | src/transport/identity.rs | 36 | ||||
| -rw-r--r-- | src/transport/libp2p/mod.rs | 25 | ||||
| -rw-r--r-- | src/transport/mod.rs | 64 | ||||
| -rw-r--r-- | src/transport/mycelium/mod.rs | 22 | ||||
| -rw-r--r-- | src/transport/tor.rs | 19 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 4 |
8 files changed, 172 insertions, 9 deletions
diff --git a/docs/trust.md b/docs/trust.md index 165fc9b..4435f99 100644 --- a/docs/trust.md +++ b/docs/trust.md @@ -99,7 +99,12 @@ peers' and warns when its own is off (`--clock-skew-warn`; [`measurements.md`](m Send messages. Before it proves the swarm key, a node ignores everything from it but the handshake, and doesn't sync with it. A stranger can still cost a node the work of -reading and dropping its messages. +reading and dropping its messages, but no more than 256 KiB of each: a node that hasn't +proved itself a member (its HELLO, or HELLO_REPLY, with the key) can only usefully send +a handshake, WHO or ME, or PING or PONG, and libp2p and Tor refuse a larger frame from +it before reading it, by the length it announces (`transport::Admission`). Mycelium's +daemon has taken a message whole before the node sees it; a stranger's larger one is +dropped undecoded. A member's messages may be large (a sync message up to 64 MiB). ## What protects against mistakes diff --git a/src/swarm/peer/handshake.rs b/src/swarm/peer/handshake.rs index 769b010..92216fd 100644 --- a/src/swarm/peer/handshake.rs +++ b/src/swarm/peer/handshake.rs @@ -405,6 +405,10 @@ pub(super) async fn establish_peer( ); } + // Proven a member of our swarm (its HELLO or HELLO_REPLY passed `admit`): its + // messages may be large now (`transport::Admission`). + state.transport.admit(peer); + let peer_state_mutex = state.get_or_create_peer_state(peer).await; let mut peer_state = peer_state_mutex.lock().await; diff --git a/src/transport/identity.rs b/src/transport/identity.rs index 0b36967..1131a11 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -27,7 +27,7 @@ //! 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::collections::{HashMap, HashSet}; use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; @@ -112,6 +112,10 @@ struct Shared { traffic: Arc<Traffic>, // A std mutex: every access is short and never spans an await. book: Mutex<Book>, + /// Nodes that proved themselves members of our swarm (`Transport::admit`): at the + /// addresses they proved themselves at, their messages may be large + /// (`transport::Admission`). + admitted: Mutex<HashSet<String>>, } /// Who is where, by address. @@ -258,6 +262,16 @@ impl IdentityTransport { let node_id = config.key.id(); let shared = Arc::new_cyclic(|shared: &std::sync::Weak<Shared>| { let gate: std::sync::Weak<dyn Gate> = shared.clone(); + let admitting = shared.clone(); + let admission = crate::transport::Admission::new(move |address| { + admitting + .upgrade() + .is_some_and(|shared| shared.admits(address)) + }); + transport.set_admission(admission.clone()); + for (_, more) in &config.more { + more.set_admission(admission.clone()); + } Shared { paths: Arc::new(Paths::new( std::iter::once((config.scheme, transport)) @@ -273,6 +287,7 @@ impl IdentityTransport { node_id, traffic, book: Mutex::default(), + admitted: Mutex::default(), } }); Self { shared } @@ -292,6 +307,17 @@ impl Gate for Shared { } impl Shared { + /// Whether the node proven at `address` is admitted (`Transport::admit`). + fn admits(&self, address: &str) -> bool { + let node = self.lock().at.get(address).map(|(node, _)| node.clone()); + node.is_some_and(|node| { + self.admitted + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .contains(&node) + }) + } + fn blocks(&self, address: &str, rule: fn(&Firewall, &str) -> bool) -> bool { let Some(firewall) = &self.firewall else { return false; @@ -625,6 +651,14 @@ impl Transport for IdentityTransport { .min_by_key(|offset| offset.within_ms) } + fn admit(&self, node: &str) { + self.shared + .admitted + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .insert(node.to_owned()); + } + /// The identity layer's, and each transport's. fn stats(&self) -> TransportStats { let mut stats = self.shared.paths.stats(); diff --git a/src/transport/libp2p/mod.rs b/src/transport/libp2p/mod.rs index 056b1b2..2e670bf 100644 --- a/src/transport/libp2p/mod.rs +++ b/src/transport/libp2p/mod.rs @@ -238,6 +238,8 @@ pub(crate) struct Libp2pTransport { book: Arc<Mutex<Book>>, incoming: tokio::sync::Mutex<Option<libp2p_stream::IncomingStreams>>, stopped: watch::Receiver<bool>, + /// Whose messages may be large (`transport::Admission`). + admission: Arc<std::sync::OnceLock<crate::transport::Admission>>, } impl Libp2pTransport { @@ -308,6 +310,7 @@ impl Libp2pTransport { book, incoming: tokio::sync::Mutex::new(incoming), stopped, + admission: Arc::default(), }) } @@ -339,15 +342,16 @@ impl Libp2pTransport { } } -/// One message from an inbound stream. -async fn read_message(mut stream: ::libp2p::Stream) -> Result<Vec<u8>> { +/// One message from an inbound stream, of at most `limit` bytes (`MAX_MESSAGE`, or less +/// from a peer not admitted: `transport::Admission`). +async fn read_message(mut stream: ::libp2p::Stream, limit: usize) -> Result<Vec<u8>> { let mut length = [0u8; 4]; tokio::time::timeout(STREAM_TIMEOUT, stream.read_exact(&mut length)) .await .map_err(|_| anyhow!("no length within {STREAM_TIMEOUT:?}"))??; let length = u32::from_be_bytes(length) as usize; - if length > MAX_MESSAGE { - bail!("a message of {length} bytes; at most {MAX_MESSAGE} are read"); + if length > limit { + bail!("a message of {length} bytes; at most {limit} are read from this peer"); } let mut body = vec![0u8; length]; tokio::time::timeout(stream_timeout(length), stream.read_exact(&mut body)) @@ -635,6 +639,10 @@ impl Transport for Libp2pTransport { Ok(()) } + fn set_admission(&self, admission: crate::transport::Admission) { + let _ = self.admission.set(admission); + } + /// Reads inbound streams (`READS` at once) and handles their messages one at a /// time, in the order they were read; PINGs and PONGs at once (`is_probe`). async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { @@ -645,6 +653,7 @@ impl Transport for Libp2pTransport { let (sender, mut received) = mpsc::channel::<(PeerId, ParsedMessage, usize)>(READS); let reads = Arc::new(Semaphore::new(READS)); let probes = Arc::clone(&handler); + let admission = Arc::clone(&self.admission); tokio::spawn(async move { while let Some((peer, stream)) = incoming.next().await { let Ok(permit) = Arc::clone(&reads).acquire_owned().await else { @@ -652,8 +661,14 @@ impl Transport for Libp2pTransport { }; let sender = sender.clone(); let probes = Arc::clone(&probes); + let limit = crate::transport::Admission::limit( + admission.get(), + SCHEME, + &peer.to_string(), + MAX_MESSAGE, + ); tokio::spawn(async move { - match read_message(stream).await { + match read_message(stream, limit).await { Ok(body) => match wire::decode(&body) { Ok(message) if is_probe(&message) => { hand_on_probe( diff --git a/src/transport/mod.rs b/src/transport/mod.rs index e81adfb..00e07e5 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -871,6 +871,41 @@ pub(crate) fn hand_on_probe( }); } +/// Most bytes a message from a peer that isn't established may take (`Admission`): +/// before the handshake a peer can only say HELLO (or answer one: a HELLO_REPLY lists at +/// most 64 peers with 8 addresses each, some tens of KiB), WHO or ME, or PING or PONG. +/// Larger frames from it are refused before anything is allocated for them. +pub(crate) const MAX_UNADMITTED: usize = 256 << 10; + +/// Whether a transport address (`scheme:identity`) is a node that proved itself a +/// member of our swarm (its handshake, `swarm::peer`): only its messages may be large; +/// a stranger's are refused above `MAX_UNADMITTED`, unread. Set by the identity layer +/// (`identity::IdentityTransport`), which knows which node is at an address; without +/// it (tests on a bare transport), everyone is. +#[derive(Clone)] +pub(crate) struct Admission(std::sync::Arc<dyn Fn(&str) -> bool + Send + Sync>); + +impl Admission { + pub(crate) fn new(admits: impl Fn(&str) -> bool + Send + Sync + 'static) -> Self { + Self(std::sync::Arc::new(admits)) + } + + /// The most bytes a message from `scheme:identity` may take. + pub(crate) fn limit( + admission: Option<&Self>, + scheme: &str, + identity: &str, + most: usize, + ) -> usize { + match admission { + Some(admission) if !(admission.0)(&format!("{scheme}:{identity}")) => { + most.min(MAX_UNADMITTED) + } + _ => most, + } + } +} + #[async_trait] pub(crate) trait Transport: Send + Sync { /// This node's identity on the transport (for Mycelium: its public key). @@ -1038,6 +1073,18 @@ pub(crate) trait Transport: Send + Sync { stats::TransportStats::default() } + /// Who may send large messages (`Admission`), for a transport that reads messages + /// itself (it frames them). By default ignored. + fn set_admission(&self, admission: Admission) { + let _ = admission; + } + + /// `node` proved itself a member of our swarm (its handshake): its messages may be + /// large from now on (`Admission`). By default nothing. + fn admit(&self, node: &str) { + let _ = node; + } + /// Resolves a seed (anything that leads to a node, e.g. an address) to that node's /// identity, as `local_id` gives it on that node. async fn resolve_seed(&self, seed: &str) -> Result<String>; @@ -1121,6 +1168,23 @@ mod tests { assert_eq!(seen.len(), heads.len()); } + /// A stranger's messages are limited, a member's aren't; without an identity layer + /// (no admission), nobody's. + #[test] + fn strangers_messages_are_small() { + let admission = Admission::new(|address| address == "s:member"); + let most = 64 << 20; + assert_eq!( + Admission::limit(Some(&admission), "s", "member", most), + most + ); + assert_eq!( + Admission::limit(Some(&admission), "s", "stranger", most), + MAX_UNADMITTED + ); + assert_eq!(Admission::limit(None, "s", "stranger", most), most); + } + /// A tag names its epoch, not one of another generation or another ID, and resolves /// to the epoch it names among those known. #[test] diff --git a/src/transport/mycelium/mod.rs b/src/transport/mycelium/mod.rs index 7bae0a7..369d683 100644 --- a/src/transport/mycelium/mod.rs +++ b/src/transport/mycelium/mod.rs @@ -89,6 +89,8 @@ pub(crate) struct MyceliumTransport { messages_endpoint: Url, pubkey_basepath: Url, local_id: String, + /// Whose messages may be large (`transport::Admission`). + admission: std::sync::OnceLock<crate::transport::Admission>, } /// This transport's options, checked when the configuration is loaded @@ -247,6 +249,7 @@ impl MyceliumTransport { messages_endpoint: rest_url.join("/api/v1/messages")?, pubkey_basepath: rest_url.join("/api/v1/pubkey/")?, local_id: admin.node_pubkey, + admission: std::sync::OnceLock::new(), }; transport.discard_backlog().await; Ok(transport) @@ -317,6 +320,10 @@ impl MyceliumTransport { #[async_trait] impl Transport for MyceliumTransport { + fn set_admission(&self, admission: crate::transport::Admission) { + let _ = self.admission.set(admission); + } + fn local_id(&self) -> &str { &self.local_id } @@ -413,6 +420,21 @@ impl Transport for MyceliumTransport { // PINGs and PONGs aren't handled after the message before them (`is_probe`). let handler: std::sync::Arc<HandlerFn> = std::sync::Arc::from(handler); while let Some(inbound_message) = rx.recv().await { + // Mycelium has taken it whole already; a stranger's large one isn't decoded. + let limit = crate::transport::Admission::limit( + self.admission.get(), + SCHEME, + &inbound_message.src_pk, + MAX_INBOUND_BYTES, + ); + if base64_decoded_len(&inbound_message.payload) > limit { + tracing::debug!( + from = %Addressed(SCHEME, &inbound_message.src_pk), + limit, + "Dropping a large message from a peer not admitted" + ); + continue; + } match parse_message(&inbound_message) { Ok((sender, parsed)) => { // The message itself is logged above, in the paths layer. diff --git a/src/transport/tor.rs b/src/transport/tor.rs index 7ecd624..bc95626 100644 --- a/src/transport/tor.rs +++ b/src/transport/tor.rs @@ -182,6 +182,8 @@ struct Shared { /// handed on at once (`is_probe`). inbound: mpsc::Sender<Inbound>, probes: mpsc::Sender<Inbound>, + /// Whose messages may be large (`transport::Admission`). + admission: std::sync::OnceLock<crate::transport::Admission>, } fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> { @@ -291,6 +293,7 @@ impl TorTransport { next_connection: AtomicU64::new(0), inbound, probes, + admission: std::sync::OnceLock::new(), }); let (stopping, stopped) = watch::channel(false); { @@ -508,8 +511,16 @@ impl Shared { return error.into(); } let length = u32::from_be_bytes(length) as usize; - if length > MAX_MESSAGE { - return anyhow!("a message of {length} bytes; at most {MAX_MESSAGE} are read"); + let limit = crate::transport::Admission::limit( + self.admission.get(), + SCHEME, + location, + MAX_MESSAGE, + ); + if length > limit { + return anyhow!( + "a message of {length} bytes; at most {limit} are read from this peer" + ); } let mut body = vec![0u8; length]; match tokio::time::timeout(stream_timeout(length), reader.read_exact(&mut body)).await { @@ -558,6 +569,10 @@ impl Shared { #[async_trait] impl Transport for TorTransport { + fn set_admission(&self, admission: crate::transport::Admission) { + let _ = self.shared.admission.set(admission); + } + fn local_id(&self) -> &str { &self.shared.local_id } diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index 11aa70d..b427a14 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -425,6 +425,10 @@ impl Transport for CountingTransport { self.inner.connection(target) } + fn admit(&self, node: &str) { + self.inner.admit(node); + } + fn stats(&self) -> crate::transport::stats::TransportStats { self.inner.stats() } |