diff options
| -rw-r--r-- | src/state.rs | 5 | ||||
| -rw-r--r-- | src/swarm/auth.rs | 31 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 114 | ||||
| -rw-r--r-- | src/swarm/sync.rs | 1 | ||||
| -rw-r--r-- | src/transport/mod.rs | 3 | ||||
| -rw-r--r-- | src/transport/mycelium.rs | 5 |
6 files changed, 74 insertions, 85 deletions
diff --git a/src/state.rs b/src/state.rs index 542cadb..64fa8ea 100644 --- a/src/state.rs +++ b/src/state.rs @@ -96,10 +96,6 @@ pub(crate) struct PeerState { pub(crate) standing: Standing, pub(crate) hello_state: HelloState, - /// The peer sent HELLO; we still owe it a HELLO_REPLY (e.g. we had no swarm ID yet, - /// or sending failed). - pub(crate) reply_owed: bool, - /// Everything about the current Automerge sync with this peer; a reset replaces it. pub(crate) sync: SyncSession, @@ -318,7 +314,6 @@ impl PeerState { leads, standing: Standing::default(), hello_state: HelloState::default(), - reply_owed: false, sync: SyncSession::new(leads, SessionCause::NewPeer), liveness: Liveness::Active, added_at: Instant::now(), diff --git a/src/swarm/auth.rs b/src/swarm/auth.rs index ea95406..31230a5 100644 --- a/src/swarm/auth.rs +++ b/src/swarm/auth.rs @@ -8,8 +8,8 @@ //! The tag is `HMAC-SHA256(key, label ‖ swarm_id ‖ from_pk ‖ to_pk)`. It needs no //! challenge: it is bound to both keys, and only `from_pk` can send as `from_pk`, so a //! captured tag is useless to anyone else. Different labels for HELLO and HELLO_REPLY -//! keep one from being passed off as the other. A HELLO has no swarm ID yet (a new node -//! doesn't know it), so its tag binds an empty one. +//! keep one from being passed off as the other. A HELLO binds the swarm ID its sender +//! claims, or an empty one from a new node that doesn't know it yet. //! //! Limits of a shared key: every member holds the same secret, so a single member can't //! be revoked (forgetting it lasts until it says HELLO again), and changing the key @@ -82,9 +82,9 @@ impl SwarmKey { Some(std::mem::take(&mut rejections.suppressed)) } - /// Tag for a HELLO from `from` to `to`. - pub(crate) fn hello_tag(&self, from: &str, to: &str) -> Vec<u8> { - self.tag(HELLO_LABEL, "", from, to).to_vec() + /// Tag for a HELLO from `from` to `to`, claiming `swarm_id` ("" if unknown). + pub(crate) fn hello_tag(&self, swarm_id: &str, from: &str, to: &str) -> Vec<u8> { + self.tag(HELLO_LABEL, swarm_id, from, to).to_vec() } /// Tag for a HELLO_REPLY from `from` to `to`, naming the swarm. @@ -92,8 +92,8 @@ impl SwarmKey { self.tag(HELLO_REPLY_LABEL, swarm_id, from, to).to_vec() } - pub(crate) fn verify_hello(&self, tag: &[u8], from: &str, to: &str) -> bool { - constant_time_eq(tag, &self.tag(HELLO_LABEL, "", from, to)) + pub(crate) fn verify_hello(&self, tag: &[u8], swarm_id: &str, from: &str, to: &str) -> bool { + constant_time_eq(tag, &self.tag(HELLO_LABEL, swarm_id, from, to)) } pub(crate) fn verify_hello_reply( @@ -176,18 +176,19 @@ mod tests { let key = SwarmKey::new(vec![1; 32]); let other = SwarmKey::new(vec![2; 32]); - let hello = key.hello_tag("A", "B"); - assert!(key.verify_hello(&hello, "A", "B")); - assert!(!other.verify_hello(&hello, "A", "B")); - assert!(!key.verify_hello(&hello, "B", "A")); - assert!(!key.verify_hello(&hello, "A", "C")); - assert!(!key.verify_hello(&[], "A", "B")); + let hello = key.hello_tag("swarm", "A", "B"); + assert!(key.verify_hello(&hello, "swarm", "A", "B")); + assert!(!key.verify_hello(&hello, "", "A", "B")); + assert!(!other.verify_hello(&hello, "swarm", "A", "B")); + assert!(!key.verify_hello(&hello, "swarm", "B", "A")); + assert!(!key.verify_hello(&hello, "swarm", "A", "C")); + assert!(!key.verify_hello(&[], "swarm", "A", "B")); let reply = key.hello_reply_tag("swarm", "A", "B"); assert!(key.verify_hello_reply(&reply, "swarm", "A", "B")); assert!(!key.verify_hello_reply(&reply, "other-swarm", "A", "B")); // A HELLO tag is no HELLO_REPLY tag, and the other way round. - assert!(!key.verify_hello_reply(&hello, "", "A", "B")); - assert!(!key.verify_hello(&key.hello_reply_tag("", "A", "B"), "A", "B")); + assert!(!key.verify_hello_reply(&hello, "swarm", "A", "B")); + assert!(!key.verify_hello(&key.hello_reply_tag("swarm", "A", "B"), "swarm", "A", "B")); } } diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index 4c07d7d..bf552e9 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -253,17 +253,20 @@ async fn retry_hello_if_due(state: &AppState, peer: &str) -> Result<()> { /// Success and failure both end in AwaitingReply: Mycelium accepting the message /// says nothing about whether the peer received it. async fn transmit_hello(state: &AppState, peer: &str, delay: Duration) -> Result<()> { + let swarm_id = state.swarm_id().await; + let auth = state + .swarm_key + .as_ref() + .map(|key| key.hello_tag(swarm_id.as_deref().unwrap_or(""), state.local_id(), peer)) + .unwrap_or_default(); let result = state .transport .send_message( peer, &ParsedMessage::Hello(transport::HelloRequest { + swarm_id, epoch: state.local_epoch(), - auth: state - .swarm_key - .as_ref() - .map(|key| key.hello_tag(state.local_id(), peer)) - .unwrap_or_default(), + auth, }), ) .await; @@ -294,35 +297,6 @@ async fn ensure_handshake(state: &AppState, peer: &str) { } } -/// Sends a HELLO_REPLY if one is owed and our swarm ID is known. -/// Otherwise it stays owed and the peer worker tries again. -async fn send_owed_hello_reply(state: &AppState, peer: &str) { - if state.swarm_id().await.is_none() { - return; - } - - let Some(peer_state_mutex) = state.get_peer_state(peer).await else { - return; - }; - - let owed = std::mem::take(&mut peer_state_mutex.lock().await.reply_owed); - if !owed { - return; - } - - if let Err(error) = send_hello_reply(state, peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed; will retry"); - peer_state_mutex.lock().await.reply_owed = true; - } -} - -async fn flush_owed_hello_replies(state: &AppState) { - let peers: Vec<String> = state.peer_states.read().await.keys().cloned().collect(); - for peer in &peers { - send_owed_hello_reply(state, peer).await; - } -} - async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> { let swarm_id = state .swarm_id() @@ -460,8 +434,6 @@ pub(crate) async fn run_peer_worker(state: AppState) { if let Err(error) = retry_hello_if_due(&state, peer).await { warn!(peer = %peer, error = %format!("{error:#}"), "HELLO retry failed"); } - - send_owed_hello_reply(&state, peer).await; } if last_forget_check.is_none_or(|t| t.elapsed() >= FORGET_CHECK_INTERVAL) { @@ -473,7 +445,16 @@ pub(crate) async fn run_peer_worker(state: AppState) { // --- peer membership --------------------------------------------------------- -async fn establish_peer(state: &AppState, peer: &str, swarm_id: String) -> Result<()> { +/// Makes `peer` a member of `swarm_id`, after its valid HELLO (`cause: PeerHello`) or +/// HELLO_REPLY (`HandshakeComplete`). Either completes the handshake: one round trip +/// proves both sides to each other (the HELLO the sender's key, the reply the swarm ID +/// and the responder's key), so neither side needs a HELLO of its own. +async fn establish_peer( + state: &AppState, + peer: &str, + swarm_id: String, + cause: SessionCause, +) -> Result<()> { let expected_swarm = state.swarm_id().await; if let Some(expected) = &expected_swarm { @@ -510,12 +491,12 @@ async fn establish_peer(state: &AppState, peer: &str, swarm_id: String) -> Resul }; } - // First reply completes the handshake: stop retries and sync from a clean slate. - // Duplicate replies (caused by our own retries) don't reset again. - if peer_state.hello_state != HelloState::Established { + // Stop our retries and sync from a clean slate. A HELLO always resets: the peer + // starts over with us. Duplicate replies (caused by our own retries) don't. + if cause == SessionCause::PeerHello || peer_state.hello_state != HelloState::Established { peer_state.hello_state = HelloState::Established; - peer_state.reset_sync(SessionCause::HandshakeComplete); - info!(peer = %peer, "Handshake complete; sync state reset"); + peer_state.reset_sync(cause); + info!(peer = %peer, ?cause, "Handshake complete; sync state reset"); } info!(peer = %peer, swarm_id = %swarm_id, "Peer membership established"); @@ -604,7 +585,12 @@ async fn admit(state: &AppState, peer: &str, message: &ParsedMessage) -> bool { let local = state.local_id(); let proof = match message { - ParsedMessage::Hello(hello) => Some(key.verify_hello(&hello.auth, peer, local)), + ParsedMessage::Hello(hello) => Some(key.verify_hello( + &hello.auth, + hello.swarm_id.as_deref().unwrap_or(""), + peer, + local, + )), ParsedMessage::HelloReply(reply) => { Some(key.verify_hello_reply(&reply.auth, &reply.swarm_id, peer, local)) } @@ -696,37 +682,41 @@ pub(crate) async fn handle_message( //debug!(from = %peer, message = ?message, "Received message"); match message { - transport::ParsedMessage::Hello(_) => { + transport::ParsedMessage::Hello(hello) => { // A HELLO means the peer is (re)starting its handshake with us, e.g. after a - // restart. Whatever our sync state believed about it is stale. - { - let peer_state_mutex = state.get_or_create_peer_state(&peer).await; - let mut peer_state = peer_state_mutex.lock().await; - peer_state.reset_sync(SessionCause::PeerHello); - peer_state.reply_owed = true; + // restart. It becomes a member right away; our reply makes us one for it. + let Some(swarm_id) = state.swarm_id().await else { + // We are joining ourselves; the peer retries until we know our swarm. + debug!(peer = %peer, "HELLO before we know our swarm; not answering yet"); + return Ok(()); + }; + if let Some(theirs) = &hello.swarm_id { + if theirs != &swarm_id { + bail!("Incompatible swarm at {peer}: ours is {swarm_id}, it claims {theirs}"); + } } + establish_peer(state, &peer, swarm_id, SessionCause::PeerHello).await?; - // Sends now if we know our swarm ID; otherwise stays owed. - send_owed_hello_reply(state, &peer).await; - - // Our own HELLO goes out regardless of whether the reply could be sent. - if let Err(error) = send_hello_once(state, &peer).await { - warn!(peer = %peer, error = %format!("{error:#}"), "HELLO failed; will retry"); + if let Err(error) = send_hello_reply(state, &peer).await { + // The peer retries its HELLO. + warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed"); } - // Restart sync with the fresh state (no-op until the peer is a validated member). let _ = state .sync_events - .send(SyncEvent::PeerNeedsSync(peer.clone())); + .send(SyncEvent::PeerValidated(peer.clone())); } transport::ParsedMessage::HelloReply(reply) => { - establish_peer(state, &peer, reply.swarm_id.clone()).await?; + establish_peer( + state, + &peer, + reply.swarm_id.clone(), + SessionCause::HandshakeComplete, + ) + .await?; discover_peers(state, &peer, &reply.peers).await; - // Our swarm ID may have just become known: pay any replies we owe. - flush_owed_hello_replies(state).await; - state .sync_events .send(SyncEvent::PeerValidated(peer.clone())) diff --git a/src/swarm/sync.rs b/src/swarm/sync.rs index 04cb31c..8a85b8c 100644 --- a/src/swarm/sync.rs +++ b/src/swarm/sync.rs @@ -97,7 +97,6 @@ async fn deliver_sync_to_peer(state: &AppState, peer: &str, retry_delivery: bool if peer_state.sync.opening_overdue(now) { debug!( cause = ?peer_state.sync.cause, - reply_owed = peer_state.reply_owed, likely = peer_state.sync.cause.missing_opening_hint(), "The peer didn't open the sync within the grace period; opening ourselves" ); diff --git a/src/transport/mod.rs b/src/transport/mod.rs index d7e7135..d96b059 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -112,6 +112,9 @@ impl ParsedMessage { #[derive(Debug, Clone)] pub struct HelloRequest { + /// The sender's swarm, if it knows it (a new node doesn't yet). A receiver in + /// another swarm doesn't answer. + pub swarm_id: Option<String>, pub epoch: Epoch, /// Proof of the swarm key (`swarm::auth`); empty without one. pub auth: Vec<u8>, diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index 2683f82..2406312 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -827,9 +827,8 @@ fn bytes_to_keepalive(bytes: &[u8]) -> Result<Keepalive, TransportError> { #[derive(Serialize, Deserialize)] pub struct HelloRequestShadow { - #[serde(default)] + swarm_id: Option<String>, epoch: EpochShadow, - #[serde(default)] auth: ByteBuf, } @@ -889,6 +888,7 @@ mod automerge_msg_codec { // --- HelloRequest --- pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde::encode::Error> { rmp_serde::to_vec(&HelloRequestShadow { + swarm_id: msg.swarm_id.clone(), epoch: epoch_to_shadow(&msg.epoch), auth: ByteBuf::from(msg.auth.clone()), }) @@ -896,6 +896,7 @@ pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde:: pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> { let shadow: HelloRequestShadow = rmp_serde::from_slice(bytes)?; Ok(HelloRequest { + swarm_id: shadow.swarm_id, epoch: epoch_from_shadow(shadow.epoch)?, auth: shadow.auth.into_vec(), }) |