Harden gossip peer conformance checks

README.md

@@ -46,11 +46,12 @@ within a 1-minute window.
   [`ConformanceChallenge`](crates/binda-core/src/gossip.rs) — two
   synthetic tokens and a win condition), answerable only by actually
   running BINDA's own deterministic collision-resolution logic, and a
-  peer's rumors are adopted only if it answers that correctly. Getting it
-  wrong, or being malformed at all, means the same thing either way: for
-  that exchange, we assume we're not talking to a real BINDA node and
-  ignore everything it sent — with no memory of the failure carried into
-  the next exchange.
+  peer's rumors are adopted only if it answers that correctly and obeys the
+  request/response contract (no duplicates and no unrequested domains).
+  Getting any of that wrong, or being malformed at all, means the same
+  thing either way: for that exchange, we assume we're not talking to a
+  real BINDA node and ignore everything it sent — with no memory of the
+  failure carried into the next exchange.
 - **TOML zone files** — the same information a BIND9 zone file carries
   (SOA, records), expressed as TOML.
 

crates/binda-core/src/gossip.rs

@@ -9,7 +9,10 @@
 //!
 //! The test is not merely "did the message parse and stay within size
 //! bounds" (that's [`is_well_formed`], and it's necessary but not
-//! sufficient). Every [`GossipMessage::Request`] carries a
+//! sufficient). Gossip also enforces the request/response contract: a
+//! response may contain each requested domain at most once, and may not
+//! smuggle in an unrequested domain. Finally, every
+//! [`GossipMessage::Request`] carries a
 //! [`ConformanceChallenge`]: two synthetic registration tokens and a win
 //! condition. Real BINDA behaviour is a fully specified, deterministic
 //! function of that input (the same [`crate::collision::resolve`] every
@@ -20,6 +23,8 @@
 //! signal as a malformed message: this exchange's data is dropped,
 //! unconditionally, whether or not the rumors themselves look plausible.
 
+use std::collections::HashSet;
+
 use serde::{Deserialize, Serialize};
 
 use crate::collision::{resolve, WinCondition};
@@ -108,12 +113,36 @@ pub struct DigestEntry {
 /// *for that message*.
 pub fn is_well_formed(message: &GossipMessage) -> bool {
     match message {
-        GossipMessage::Digest { rumors } => rumors.len() <= MAX_DIGEST_ENTRIES,
-        GossipMessage::Request { domains, .. } => domains.len() <= MAX_REQUEST_ENTRIES,
-        GossipMessage::Rumors { rumors, .. } => rumors.len() <= MAX_DIGEST_ENTRIES,
+        GossipMessage::Digest { rumors } => {
+            rumors.len() <= MAX_DIGEST_ENTRIES && unique_domains(rumors.iter().map(|r| &r.domain))
+        }
+        GossipMessage::Request { domains, .. } => {
+            domains.len() <= MAX_REQUEST_ENTRIES && unique_domains(domains.iter())
+        }
+        GossipMessage::Rumors { rumors, .. } => {
+            rumors.len() <= MAX_DIGEST_ENTRIES && unique_domains(rumors.iter().map(|r| &r.domain))
+        }
     }
 }
 
+/// Check the semantic part of a gossip response: a peer may answer only for
+/// domains that were requested, and at most once for each requested domain.
+/// The caller still has to authenticate the exchange with the outstanding
+/// [`ConformanceChallenge`] before applying the returned facts.
+pub fn is_valid_rumor_response(requested: &[DomainName], rumors: &[RegistrationRumor]) -> bool {
+    let requested: HashSet<&DomainName> = requested.iter().collect();
+    rumors
+        .iter()
+        .map(|rumor| &rumor.domain)
+        .all(|domain| requested.contains(domain))
+        && unique_domains(rumors.iter().map(|rumor| &rumor.domain))
+}
+
+fn unique_domains<'a>(mut domains: impl Iterator<Item = &'a DomainName>) -> bool {
+    let mut seen = HashSet::new();
+    domains.all(|domain| seen.insert(domain))
+}
+
 /// Upper bound on how many entries a single digest or rumor batch may
 /// carry, so a malformed/hostile peer can't force unbounded allocation.
 pub const MAX_DIGEST_ENTRIES: usize = 4096;
@@ -198,6 +227,46 @@ mod tests {
         assert!(is_well_formed(&msg));
     }
 
+    #[test]
+    fn duplicate_domains_are_not_well_formed() {
+        let domain = DomainName::new("example.binda").unwrap();
+        let digest = GossipMessage::Digest {
+            rumors: vec![
+                DigestEntry {
+                    domain: domain.clone(),
+                    issued_at_millis: 1,
+                },
+                DigestEntry {
+                    domain,
+                    issued_at_millis: 2,
+                },
+            ],
+        };
+        assert!(!is_well_formed(&digest));
+    }
+
+    #[test]
+    fn rumor_response_must_be_request_scoped_and_unique() {
+        let requested = vec![DomainName::new("requested.binda").unwrap()];
+        let token = RegistrationToken::issue(1);
+        let valid = vec![RegistrationRumor {
+            domain: requested[0].clone(),
+            token,
+            client_key: "client".into(),
+        }];
+        assert!(is_valid_rumor_response(&requested, &valid));
+
+        let unrequested = vec![RegistrationRumor {
+            domain: DomainName::new("unrequested.binda").unwrap(),
+            token,
+            client_key: "client".into(),
+        }];
+        assert!(!is_valid_rumor_response(&requested, &unrequested));
+
+        let duplicate = vec![valid[0].clone(), valid[0].clone()];
+        assert!(!is_valid_rumor_response(&requested, &duplicate));
+    }
+
     #[test]
     fn conformance_challenge_has_exactly_one_correct_answer() {
         let challenge = ConformanceChallenge::random(1_000);

crates/binda/src/node.rs

@@ -9,9 +9,11 @@ use std::time::{Duration, Instant};
 use binda_core::api::handle_client_request;
 use binda_core::client_api::ClientRequest;
 use binda_core::dns;
+use binda_core::domain::DomainName;
 use binda_core::fcrdns::{self, RdnsVerifier};
 use binda_core::gossip::{
-    is_well_formed, ConformanceChallenge, DigestEntry, GossipMessage, RegistrationRumor,
+    is_valid_rumor_response, is_well_formed, ConformanceChallenge, DigestEntry, GossipMessage,
+    RegistrationRumor,
 };
 use binda_core::liveness::TimeSource;
 use binda_core::ntp::NtpTimeSource;
@@ -60,6 +62,8 @@ const PENDING_CHALLENGE_TIMEOUT: Duration = Duration::from_secs(10);
 /// this map without bound.
 const MAX_PENDING_CHALLENGES: usize = 10_000;
 
+type PendingChallenge = (ConformanceChallenge, Vec<DomainName>, Instant);
+
 /// Shared node state, cheap to clone (everything behind `Arc`).
 #[derive(Clone)]
 pub struct Node {
@@ -74,7 +78,7 @@ pub struct Node {
     /// [`GossipMessage::Request`]) and is waiting to see answered
     /// correctly in the matching [`GossipMessage::Rumors`] reply, keyed
     /// by the peer address the request was sent to.
-    pending_challenges: Arc<Mutex<HashMap<SocketAddr, (ConformanceChallenge, Instant)>>>,
+    pending_challenges: Arc<Mutex<HashMap<SocketAddr, PendingChallenge>>>,
     /// What checks a `Register` request's claimed `rdns` hostname against
     /// its actual source address. Real deployments must use the default
     /// ([`fcrdns::FcrdnsVerifier`]); swapping in
@@ -140,9 +144,12 @@ impl Node {
                         .await
                         .prune_older_than(RATE_LIMIT_PRUNE_AGE, now);
                 }
-                pending_challenges.lock().await.retain(|_, (_, issued_at)| {
-                    now.duration_since(*issued_at) < PENDING_CHALLENGE_TIMEOUT
-                });
+                pending_challenges
+                    .lock()
+                    .await
+                    .retain(|_, (_, _, issued_at)| {
+                        now.duration_since(*issued_at) < PENDING_CHALLENGE_TIMEOUT
+                    });
             }
         });
     }
@@ -247,13 +254,13 @@ impl Node {
                     if !pending.contains_key(&from) && pending.len() >= MAX_PENDING_CHALLENGES {
                         if let Some(oldest) = pending
                             .iter()
-                            .min_by_key(|(_, (_, issued_at))| *issued_at)
+                            .min_by_key(|(_, (_, _, issued_at))| *issued_at)
                             .map(|(addr, _)| *addr)
                         {
                             pending.remove(&oldest);
                         }
                     }
-                    pending.insert(from, (challenge, Instant::now()));
+                    pending.insert(from, (challenge, missing.clone(), Instant::now()));
                 }
 
                 let request = GossipMessage::Request {
@@ -299,7 +306,7 @@ impl Node {
                     let mut pending = self.pending_challenges.lock().await;
                     pending.remove(&from)
                 };
-                let Some((challenge, _)) = expected else {
+                let Some((challenge, requested, _)) = expected else {
                     // No outstanding challenge for this address: either
                     // we never asked, or it already timed out. Either
                     // way, there's nothing to verify this answer against,
@@ -314,6 +321,14 @@ impl Node {
                     return;
                 }
 
+                if !is_valid_rumor_response(&requested, &rumors) {
+                    // A correctly answered challenge proves only that the
+                    // sender can perform the deterministic BINDA operation.
+                    // It does not excuse violating the request/response
+                    // contract: never adopt an unrequested or duplicate fact.
+                    return;
+                }
+
                 let mut store = self.store.lock().await;
                 for rumor in rumors {
                     store.adopt_rumor(rumor.domain, rumor.client_key, rumor.token);
@@ -455,6 +470,7 @@ mod tests {
     use binda_core::client::ClientIdentity;
     use binda_core::domain::DomainName;
     use binda_core::liveness::SystemTimeSource;
+    use binda_core::token::RegistrationToken;
     use ed25519_dalek::SigningKey;
     use rand::rngs::OsRng;
     use std::time::Duration as StdDuration;
@@ -727,6 +743,51 @@ mod tests {
         assert_eq!(reg.token, token);
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn correctly_challenged_but_request_invalid_rumors_are_rejected() {
+        let victim_addr: SocketAddr = "127.0.0.1:29580".parse().unwrap();
+        let rogue_addr: SocketAddr = "127.0.0.1:29581".parse().unwrap();
+        let requested = DomainName::new("requested.binda").unwrap();
+        let unrequested = DomainName::new("unrequested.binda").unwrap();
+
+        let victim = test_node(vec![rogue_addr]);
+        spawn_gossip_task(&victim, victim_addr);
+        let rogue_socket = UdpSocket::bind(rogue_addr).await.unwrap();
+        tokio::time::sleep(StdDuration::from_millis(100)).await;
+
+        rogue_socket
+            .send_to(
+                &wire::encode(&GossipMessage::Digest {
+                    rumors: vec![DigestEntry {
+                        domain: requested.clone(),
+                        issued_at_millis: u64::MAX,
+                    }],
+                })
+                .unwrap(),
+                victim_addr,
+            )
+            .await
+            .unwrap();
+        let (_, challenge, from) = recv_request(&rogue_socket).await;
+        let invalid_response = GossipMessage::Rumors {
+            rumors: vec![RegistrationRumor {
+                domain: unrequested.clone(),
+                token: binda_core::token::RegistrationToken::issue(0),
+                client_key: "rogue".into(),
+            }],
+            challenge_answer: challenge.expected_answer(),
+        };
+        rogue_socket
+            .send_to(&wire::encode(&invalid_response).unwrap(), from)
+            .await
+            .unwrap();
+        tokio::time::sleep(StdDuration::from_millis(200)).await;
+
+        let store = victim.store.lock().await;
+        assert!(store.lookup(&requested).is_none());
+        assert!(store.lookup(&unrequested).is_none());
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn resolver_returns_registered_owner_and_records() {
         use binda_core::resolver::{ResolveAnswer, ResolveQuery};
@@ -864,6 +925,10 @@ mod tests {
         client_socket
             .set_read_timeout(Some(StdDuration::from_secs(3)))
             .unwrap();
+        // With no complete message ID there is no error response that can be
+        // addressed back to the sender; the listener must simply continue.
+        client_socket.send_to(&[0x99], dns_addr).unwrap();
+        std::thread::sleep(StdDuration::from_millis(50));
         // A 2-byte message can't possibly be a valid query, but it does
         // carry a recognizable ID for the error response to echo.
         client_socket.send_to(&[0x99, 0x88], dns_addr).unwrap();
@@ -1054,12 +1119,15 @@ mod tests {
             let mut pending = node.pending_challenges.lock().await;
             pending.insert(
                 fresh_peer,
-                (ConformanceChallenge::random(0), Instant::now()),
+                (ConformanceChallenge::random(0), Vec::new(), Instant::now()),
             );
             let long_ago = Instant::now()
                 .checked_sub(PENDING_CHALLENGE_TIMEOUT + Duration::from_secs(1))
                 .expect("test process has been up long enough for this");
-            pending.insert(stale_peer, (ConformanceChallenge::random(0), long_ago));
+            pending.insert(
+                stale_peer,
+                (ConformanceChallenge::random(0), Vec::new(), long_ago),
+            );
         }
 
         node.spawn_rate_limiter_maintenance();
@@ -1072,6 +1140,55 @@ mod tests {
         assert!(!pending.contains_key(&stale_peer));
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn pending_challenge_table_evicts_oldest_peer_at_capacity() {
+        let node = test_node(Vec::new());
+        let oldest_peer: SocketAddr = "127.0.0.1:1".parse().unwrap();
+        let newest_peer: SocketAddr = "127.0.0.1:2".parse().unwrap();
+        let incoming_peer: SocketAddr = "127.0.0.1:20000".parse().unwrap();
+        {
+            let mut pending = node.pending_challenges.lock().await;
+            for port in 1..=MAX_PENDING_CHALLENGES as u16 {
+                pending.insert(
+                    SocketAddr::from(([127, 0, 0, 1], port)),
+                    (
+                        ConformanceChallenge::random(port as u64),
+                        Vec::new(),
+                        Instant::now(),
+                    ),
+                );
+            }
+            // Make the ordering deterministic without waiting for the
+            // timeout-based maintenance task.
+            let oldest = pending.get_mut(&oldest_peer).unwrap();
+            oldest.2 = Instant::now()
+                .checked_sub(PENDING_CHALLENGE_TIMEOUT + Duration::from_secs(1))
+                .unwrap();
+            let newest = pending.get_mut(&newest_peer).unwrap();
+            newest.2 = Instant::now();
+        }
+
+        let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        let domain = DomainName::new("capacity-eviction.binda").unwrap();
+        node.handle_gossip_message(
+            &socket,
+            incoming_peer,
+            GossipMessage::Digest {
+                rumors: vec![DigestEntry {
+                    domain,
+                    issued_at_millis: u64::MAX,
+                }],
+            },
+        )
+        .await;
+
+        let pending = node.pending_challenges.lock().await;
+        assert_eq!(pending.len(), MAX_PENDING_CHALLENGES);
+        assert!(!pending.contains_key(&oldest_peer));
+        assert!(pending.contains_key(&newest_peer));
+        assert!(pending.contains_key(&incoming_peer));
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn gossip_push_loop_idles_safely_with_no_known_peers() {
         let gossip_addr: SocketAddr = "127.0.0.1:29610".parse().unwrap();
@@ -1103,6 +1220,37 @@ mod tests {
             .unwrap();
         tokio::time::sleep(StdDuration::from_millis(100)).await;
 
+        // A decodable message with duplicate domains is also invalid at the
+        // protocol layer. The listener must drop it and continue serving the
+        // next valid exchange.
+        let duplicate_domain = DomainName::new("duplicate.binda").unwrap();
+        let duplicate_digest = GossipMessage::Digest {
+            rumors: vec![
+                DigestEntry {
+                    domain: duplicate_domain.clone(),
+                    issued_at_millis: 1,
+                },
+                DigestEntry {
+                    domain: duplicate_domain,
+                    issued_at_millis: 2,
+                },
+            ],
+        };
+        socket
+            .send_to(&wire::encode(&duplicate_digest).unwrap(), gossip_addr)
+            .await
+            .unwrap();
+
+        // A rumor without an outstanding request must likewise be ignored.
+        let unsolicited = GossipMessage::Rumors {
+            rumors: Vec::new(),
+            challenge_answer: RegistrationToken::issue(0),
+        };
+        socket
+            .send_to(&wire::encode(&unsolicited).unwrap(), gossip_addr)
+            .await
+            .unwrap();
+
         // The loop must have survived the garbage: a well-formed message
         // sent right after should still get a real reply.
         let digest = GossipMessage::Digest {