From d0760c918ab75c837548ca27f19a46d387f77004 Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 6 Nov 2026 04:31:04 +0100 Subject: strangers messages small --- src/transport/identity.rs | 36 +++++++++++++++++++++++- src/transport/libp2p/mod.rs | 25 +++++++++++++---- src/transport/mod.rs | 64 +++++++++++++++++++++++++++++++++++++++++++ src/transport/mycelium/mod.rs | 22 +++++++++++++++ src/transport/tor.rs | 19 +++++++++++-- src/transport/traffic.rs | 4 +++ 6 files changed, 162 insertions(+), 8 deletions(-) (limited to 'src/transport') 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, // A std mutex: every access is short and never spans an await. book: Mutex, + /// 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>, } /// 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| { let gate: std::sync::Weak = 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>, incoming: tokio::sync::Mutex>, stopped: watch::Receiver, + /// Whose messages may be large (`transport::Admission`). + admission: Arc>, } 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> { +/// 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> { 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 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; @@ -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, } /// 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 = 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, probes: mpsc::Sender, + /// Whose messages may be large (`transport::Admission`). + admission: std::sync::OnceLock, } fn lock(mutex: &Mutex) -> 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() } -- cgit v1.3.1