strangers messages small

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()
     }