strangers messages small

1 files changed, 17 insertions, 2 deletions

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
     }