tor claims bounded

1 files changed, 45 insertions, 14 deletions

diff --git a/src/transport/tor.rs b/src/transport/tor.rs
index bc95626..a08a5d0 100644
--- a/src/transport/tor.rs
+++ b/src/transport/tor.rs
@@ -19,7 +19,9 @@
 //! handshake the key there), so a claim holds if dialing the claimed onion service found
 //! the same key, now or before; if it wasn't dialed yet, the listener dials it to check,
 //! and closes that connection again. A claim whose check failed is refused without a
-//! dial for a while (`FAILED_CHECK_FOR`). Then messages go as on libp2p: their length (4
+//! dial for a while (`FAILED_CHECK_FOR`), and so is one of an onion service where
+//! another node key was found within that time; at most `MAX_CHECKS` are dialed at once.
+//! Then messages go as on libp2p: their length (4
 //! bytes, big endian), then the message as `wire` encodes it. A connection that breaks
 //! is dialed again on the next send. Both ends may dial at once; each then sends on the
 //! newest connection and reads from both until the older one closes.
@@ -88,6 +90,9 @@ const READS: usize = 32;
 const FAILED_CHECK_FOR: Duration = Duration::from_secs(600);
 /// Onion services remembered (with the node key found there, `Shared::proven`), at most.
 const MAX_PEERS: usize = 4096;
+/// Claims checked by dialing at once (`Shared::check`); one more is refused, and its
+/// dialer dials again. Anyone may claim any onion service, and each check costs a dial.
+const MAX_CHECKS: usize = 4;
 
 /// How long a message of `bytes` may take.
 fn stream_timeout(bytes: usize) -> Duration {
@@ -168,8 +173,11 @@ struct Shared {
     client: Arc<TorClient<PreferredRuntime>>,
     /// Our onion service, `<onion>.onion:<port>`: our transport identity.
     local_id: String,
-    /// The node key found at each onion service dialed: proven there (`check`).
-    proven: Mutex<HashMap<String, [u8; 32]>>,
+    /// The node key found at each onion service dialed, and when: proven there
+    /// (`check`).
+    proven: Mutex<HashMap<String, ([u8; 32], Instant)>>,
+    /// Claims being checked by dialing (`MAX_CHECKS`).
+    checks: tokio::sync::Semaphore,
     /// Claims whose check failed lately (`FAILED_CHECK_FOR`): an onion service and the
     /// node key claiming it, and when.
     failed_checks: Mutex<HashMap<(String, [u8; 32]), Instant>>,
@@ -287,6 +295,7 @@ impl TorTransport {
             client,
             local_id: onion.clone(),
             proven: Mutex::default(),
+            checks: tokio::sync::Semaphore::new(MAX_CHECKS),
             failed_checks: Mutex::default(),
             connections: Mutex::default(),
             dialing: Mutex::default(),
@@ -372,31 +381,31 @@ impl Shared {
             .with_context(|| format!("Handshake with {location}"))?;
         let mut proven = lock(&self.proven);
         if proven.len() < MAX_PEERS || proven.contains_key(location) {
-            proven.insert(location.to_owned(), key);
+            proven.insert(location.to_owned(), (key, Instant::now()));
         }
         Ok((key, stream))
     }
 
     /// Whether the node with `key` is at the onion service at `location` (a dialer
     /// claims it is: module docs). Dialed to find out, unless that was done already, or
-    /// failed within `FAILED_CHECK_FOR`.
+    /// failed within `FAILED_CHECK_FOR`, or another node key was found there within that
+    /// time (a node that made itself a new key is believed after it), or `MAX_CHECKS`
+    /// are being dialed: claims of a member's onion service under fresh keys don't make
+    /// us dial it once each, nor claims of fresh ones dial without bound.
     async fn check(&self, location: &str, key: [u8; 32]) -> Result<bool> {
-        if lock(&self.proven).get(location) == Some(&key) {
-            return Ok(true);
-        }
         let claim = (location.to_owned(), key);
-        if let Some(failed) = self.failed_lately(&claim) {
-            bail!("its check failed {}s ago", failed.as_secs());
+        if self.decided(&claim)? {
+            return Ok(true);
         }
+        let Ok(_checking) = self.checks.try_acquire() else {
+            bail!("{MAX_CHECKS} claims are being checked already");
+        };
         let dialing = self.dialing(location);
         let _one_at_a_time = dialing.lock().await;
         // A dial or check meanwhile may have found out.
-        if lock(&self.proven).get(location) == Some(&key) {
+        if self.decided(&claim)? {
             return Ok(true);
         }
-        if let Some(failed) = self.failed_lately(&claim) {
-            bail!("its check failed {}s ago", failed.as_secs());
-        }
         let checked = self
             .dial(location, "")
             .await
@@ -413,6 +422,28 @@ impl Shared {
         checked
     }
 
+    /// Ok(true) if `claim` holds as far as known (its key was found at its onion
+    /// service); an error if it failed lately, or another key was found there lately
+    /// (`FAILED_CHECK_FOR`); Ok(false) if only a dial can tell.
+    fn decided(&self, claim: &(String, [u8; 32])) -> Result<bool> {
+        let (location, key) = claim;
+        if let Some((found, at)) = lock(&self.proven).get(location) {
+            if found == key {
+                return Ok(true);
+            }
+            if at.elapsed() < FAILED_CHECK_FOR {
+                bail!(
+                    "another node key was found there {}s ago",
+                    at.elapsed().as_secs()
+                );
+            }
+        }
+        if let Some(failed) = self.failed_lately(claim) {
+            bail!("its check failed {}s ago", failed.as_secs());
+        }
+        Ok(false)
+    }
+
     /// How long ago the check of `claim` failed, if within `FAILED_CHECK_FOR`.
     fn failed_lately(&self, claim: &(String, [u8; 32])) -> Option<Duration> {
         let mut failed = lock(&self.failed_checks);