fix(server): place rooms on adopted members, so a node cannot mint an id into a replica set (C25)

crates/core/src/lib.rs

@@ -63,7 +63,8 @@ pub use marks::{MarkState, ResolvedMark};
 pub use op::{destrand_split, split_groups, Op, OpId, OpKind, Tx, TxId, MAX_TX_MEMBERS};
 pub use protocol::{
     decode_header, decode_message, encode_header, encode_message, select_codec, BranchInfo,
-    Channel, DiffKind, ErrorCode, MemberState, Message, ProtocolError, CODEC_V1, SUPPORTED_CODECS,
+    Channel, DiffKind, ErrorCode, MemberAdvert, MemberState, Message, ProtocolError, CODEC_V1,
+    SUPPORTED_CODECS,
 };
 pub use ranged::{RangeAnchor, RangedElement, RangedInit, RangedPayload};
 pub use register::Register;

crates/core/src/protocol.rs

@@ -131,6 +131,13 @@ pub enum MemberState {
     Dead,
 }
 
+/// One cluster member as a [`Message::Gossip`] frame advertises it: its node id, the
+/// advertise address a peer dials it at, its incarnation, its [`MemberState`], and
+/// whether the **sender of that frame** has itself completed an identity-checked peer
+/// link to it. The last is a first-hand claim about the sender, so a receiver
+/// attributes it to the member its link is bound to and to nobody the payload names.
+pub type MemberAdvert = (Vec<u8>, Vec<u8>, u64, MemberState, bool);
+
 /// A closed set of failure reasons the server reports to a client.
 #[derive(Clone, Copy, PartialEq, Eq, Debug)]
 pub enum ErrorCode {
@@ -415,18 +422,26 @@ pub enum Message {
     },
     /// A node's advertisement of the cluster members it knows, for gossip
     /// membership discovery and SWIM-style failure detection: `members` is a set of
-    /// `(node_id, advertise_addr, incarnation, state)` tuples — the node id a peer
-    /// places with, the address it dials to reach that member, a monotonic per-node
-    /// refutation counter, and the member's [`MemberState`]. A receiver merges each
-    /// tuple into its own liveness view (anti-entropy: a higher incarnation wins,
-    /// and at equal incarnation the more-suspicious state wins), so a node that
-    /// boots knowing only a seed peer learns the whole cluster — and a node's
-    /// failure propagates to every node — within a few gossip rounds. Node-to-node
-    /// — never a client frame; a client that sends one commits a protocol
-    /// violation.
-    Gossip {
-        members: Vec<(Vec<u8>, Vec<u8>, u64, MemberState)>,
-    },
+    /// `(node_id, advertise_addr, incarnation, state, verified)` tuples — the node id
+    /// a peer places with, the address it dials to reach that member, a monotonic
+    /// per-node refutation counter, the member's [`MemberState`], and whether the
+    /// *sender of this frame* has itself completed an identity-checked peer link to
+    /// that member. A receiver merges each tuple into its own liveness view
+    /// (anti-entropy: a higher incarnation wins, and at equal incarnation the
+    /// more-suspicious state wins), so a node that boots knowing only a seed peer
+    /// learns the whole cluster — and a node's failure propagates to every node —
+    /// within a few gossip rounds.
+    ///
+    /// `verified` is a **first-hand claim about the sender itself**, not a relayed
+    /// fact: it says "I have verified this member", so a receiver attributes it to
+    /// the member its link is bound to and to nobody else. That is what makes it
+    /// safe to act on — a member can assert only its own verifications, never
+    /// another's — and it is the evidence a room's replica set is placed over, so
+    /// that a node cannot mint an id into the placement ring by introducing itself.
+    ///
+    /// Node-to-node — never a client frame; a client that sends one commits a
+    /// protocol violation.
+    Gossip { members: Vec<MemberAdvert> },
     /// Requests the branches of `room` — the app-facing enumeration a client runs
     /// to discover a room's forks and published targets before subscribing one.
     /// Room-keyed rather than channel-keyed: branch management is a room-level
@@ -848,11 +863,12 @@ pub fn encode_message(m: &Message) -> Vec<u8> {
                 &mut out,
                 u32::try_from(members.len()).expect("member count exceeds u32"),
             );
-            for (node, addr, incarnation, state) in members {
+            for (node, addr, incarnation, state, verified) in members {
                 put_bytes(&mut out, node);
                 put_bytes(&mut out, addr);
                 put_u64(&mut out, *incarnation);
                 put_u8(&mut out, member_state_tag(*state));
+                put_u8(&mut out, u8::from(*verified));
             }
         }
         Message::BranchList { room } => {
@@ -1249,7 +1265,19 @@ pub fn decode_message(bytes: &[u8]) -> Result<Message, ProtocolError> {
                 let addr = cur.bytes()?;
                 let incarnation = cur.u64()?;
                 let state = member_state(cur.u8()?)?;
-                members.push((node, addr, incarnation, state));
+                // One byte string per message: the encoder writes 0 or 1, so anything
+                // else is not a frame this cluster produced.
+                let verified = match cur.u8()? {
+                    0 => false,
+                    1 => true,
+                    tag => {
+                        return Err(ProtocolError::BadTag {
+                            what: "gossip member verification",
+                            tag,
+                        })
+                    }
+                };
+                members.push((node, addr, incarnation, state, verified));
             }
             Message::Gossip { members }
         }

crates/core/tests/protocol_gossip.rs

@@ -2,14 +2,16 @@
 //! each with its liveness, for anti-entropy failure detection.
 //!
 //! `Gossip { members }` carries a length-prefixed list of `(node_id,
-//! advertise_addr, incarnation, state)` tuples: the id a peer places with, the
-//! address it dials, a monotonic per-node refutation counter, and the member's
-//! SWIM state (`Alive` / `Suspect` / `Dead`). A receiver merges each tuple into
-//! its own liveness view — a higher incarnation wins, and at equal incarnation a
-//! more-suspicious state wins — so both the member set and its liveness converge
-//! cluster-wide. Node-to-node, never a client frame. Decoding is total — a
-//! truncation, a trailing byte, or a bad state tag is a `ProtocolError`, never a
-//! panic.
+//! advertise_addr, incarnation, state, verified)` tuples: the id a peer places
+//! with, the address it dials, a monotonic per-node refutation counter, the
+//! member's SWIM state (`Alive` / `Suspect` / `Dead`), and whether the sender has
+//! itself completed an identity-checked peer link to that member. A receiver
+//! merges each tuple into its own liveness view — a higher incarnation wins, and at
+//! equal incarnation a more-suspicious state wins — so both the member set and its
+//! liveness converge cluster-wide; the verification flag is recorded against the
+//! sender, and enough of them place the member in the ring. Node-to-node, never a
+//! client frame. Decoding is total — a truncation, a trailing byte, or a bad state
+//! tag is a `ProtocolError`, never a panic.
 
 use crdtsync_core::protocol::{
     decode_message, encode_message, MemberState, Message, ProtocolError,
@@ -26,8 +28,18 @@ fn member(
     addr: &[u8],
     incarnation: u64,
     state: MemberState,
-) -> (Vec<u8>, Vec<u8>, u64, MemberState) {
-    (node.to_vec(), addr.to_vec(), incarnation, state)
+) -> (Vec<u8>, Vec<u8>, u64, MemberState, bool) {
+    (node.to_vec(), addr.to_vec(), incarnation, state, false)
+}
+
+/// The same advertisement, carrying the sender's claim to have verified the member.
+fn verified_member(
+    node: &[u8],
+    addr: &[u8],
+    incarnation: u64,
+    state: MemberState,
+) -> (Vec<u8>, Vec<u8>, u64, MemberState, bool) {
+    (node.to_vec(), addr.to_vec(), incarnation, state, true)
 }
 
 #[test]
@@ -80,6 +92,7 @@ fn gossip_round_trips_binary_and_empty_fields() {
                 vec![0xFF, 0x00, 0x80, 0x7F],
                 999,
                 MemberState::Dead,
+                false,
             ),
         ],
     });
@@ -131,12 +144,14 @@ fn a_gossip_count_past_the_payload_is_an_error() {
 #[test]
 fn a_bad_state_tag_is_an_error_not_a_panic() {
     // The state is a single tag byte; an unknown value is a decode error, never a
-    // silent misread or a panic. Build a one-member frame and clobber its trailing
-    // state byte with an out-of-range tag.
+    // silent misread or a panic. Build a one-member frame and clobber its state byte
+    // — the last but one, the verification flag trailing it — with an out-of-range
+    // tag.
     let mut bytes = encode_message(&Message::Gossip {
         members: vec![member(b"node-a", b"10.0.0.1:9000", 0, MemberState::Alive)],
     });
-    *bytes.last_mut().unwrap() = 200;
+    let state_byte = bytes.len() - 2;
+    bytes[state_byte] = 200;
     assert!(matches!(
         decode_message(&bytes),
         Err(ProtocolError::BadTag { .. })
@@ -145,12 +160,12 @@ fn a_bad_state_tag_is_an_error_not_a_panic() {
 
 #[test]
 fn each_state_tag_is_distinct() {
-    // The three states encode to three distinct trailing tag bytes.
+    // The three states encode to three distinct tag bytes.
     let tag_of = |state| {
         let bytes = encode_message(&Message::Gossip {
             members: vec![member(b"n", b"a", 0, state)],
         });
-        *bytes.last().unwrap()
+        bytes[bytes.len() - 2]
     };
     let alive = tag_of(MemberState::Alive);
     let suspect = tag_of(MemberState::Suspect);
@@ -160,6 +175,48 @@ fn each_state_tag_is_distinct() {
     assert_ne!(alive, dead);
 }
 
+#[test]
+fn gossip_round_trips_a_verification_claim() {
+    // The sender's claim to have verified a member rides per tuple and survives the
+    // round trip independently of the liveness beside it — a frame mixing verified
+    // and unverified members comes back as it went.
+    round_trips(Message::Gossip {
+        members: vec![
+            verified_member(b"node-a", b"10.0.0.1:9000", 4, MemberState::Alive),
+            member(b"node-b", b"10.0.0.2:9000", 4, MemberState::Suspect),
+        ],
+    });
+}
+
+#[test]
+fn a_verification_flag_is_a_distinct_trailing_byte() {
+    // Verified and unverified encode differently, in the byte after the state tag —
+    // so a decoder cannot read the claim off the state, nor drop it silently.
+    let flag_of = |m| {
+        let bytes = encode_message(&Message::Gossip { members: vec![m] });
+        *bytes.last().unwrap()
+    };
+    assert_ne!(
+        flag_of(member(b"n", b"a", 0, MemberState::Alive)),
+        flag_of(verified_member(b"n", b"a", 0, MemberState::Alive)),
+    );
+}
+
+#[test]
+fn a_truncated_verification_flag_is_an_error_not_a_panic() {
+    // The flag is the last byte of a tuple, so a frame cut just before it is a
+    // truncation inside a field — an error, never a silently unverified member.
+    let bytes = encode_message(&Message::Gossip {
+        members: vec![verified_member(
+            b"node-a",
+            b"10.0.0.1:9000",
+            0,
+            MemberState::Alive,
+        )],
+    });
+    assert!(decode_message(&bytes[..bytes.len() - 1]).is_err());
+}
+
 #[test]
 fn the_gossip_tag_is_distinct() {
     // Tag 26 is the gossip frame; the frame's first byte is that tag.