diff options
| author | tv <tv@krebsco.de> | 2026-10-24 03:19:28 +0200 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-10-24 03:19:28 +0200 |
| commit | a1460b0a2861d5bb34ba25d409b234468f269e4e (patch) | |
| tree | fb447e712aa587b4136c12f4b918cd9c6cf160f0 | |
| parent | 84a307a09eb3ba95920b028a49d0ec57bf316cdc (diff) | |
use nonce in handshake
To prevent stale hello replies to cause joins and thus stale sessions,
each run gets a random ID, which used as nonce in the handshake.
| -rw-r--r-- | src/main.rs | 1 | ||||
| -rw-r--r-- | src/state.rs | 3 | ||||
| -rw-r--r-- | src/swarm/peer.rs | 16 | ||||
| -rw-r--r-- | src/transport/mod.rs | 8 | ||||
| -rw-r--r-- | src/transport/mycelium.rs | 13 |
5 files changed, 39 insertions, 2 deletions
diff --git a/src/main.rs b/src/main.rs index d5043a8..6082e67 100644 --- a/src/main.rs +++ b/src/main.rs @@ -185,6 +185,7 @@ async fn main() -> Result<()> { timing: Timing::from_keepalive(Duration::from_secs(args.keepalive_secs)), remake: Arc::new(remake::RemakeState::default()), swarm_key, + run_id: uuid::Uuid::new_v4().into_bytes(), stale: Default::default(), catch_up_source: Default::default(), }; diff --git a/src/state.rs b/src/state.rs index bbda4e4..cfa5ddf 100644 --- a/src/state.rs +++ b/src/state.rs @@ -492,6 +492,9 @@ pub(crate) struct AppState { /// knows the swarm ID. pub(crate) swarm_key: Option<Arc<crate::swarm::auth::SwarmKey>>, + /// Random per run of this node; our HELLOs carry it and replies must echo it. + pub(crate) run_id: transport::RunId, + /// While rejoining a newer epoch, the previous document, served read-only (and /// marked stale) instead of answering 503. pub(crate) stale: Arc<std::sync::RwLock<Option<StaleView>>>, diff --git a/src/swarm/peer.rs b/src/swarm/peer.rs index c8c86ac..13a980c 100644 --- a/src/swarm/peer.rs +++ b/src/swarm/peer.rs @@ -271,6 +271,7 @@ async fn transmit_hello(state: &AppState, peer: &str) -> Result<()> { epoch: state.local_epoch(), auth, ready: state.is_ready().await, + run_id: state.run_id, }), ) .await; @@ -298,7 +299,8 @@ async fn ensure_handshake(state: &AppState, peer: &str) { } } -async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> { +/// Answers a HELLO that carried `run_id`. +async fn send_hello_reply(state: &AppState, peer: &str, run_id: transport::RunId) -> Result<()> { let swarm_id = state .swarm_id() .await @@ -335,6 +337,7 @@ async fn send_hello_reply(state: &AppState, peer: &str) -> Result<()> { epoch: state.local_epoch(), auth, ready, + run_id, }; debug!(to = %peer, ready, "Sending HELLO_REPLY"); @@ -661,6 +664,15 @@ pub(crate) async fn handle_message( return Ok(()); } + // A reply to a HELLO from an earlier run of ours (late, or queued in Mycelium while + // we were down) says nothing about now; the peer answers our current HELLO. + if let transport::ParsedMessage::HelloReply(reply) = &message { + if reply.run_id != state.run_id { + debug!(peer = %peer, "Ignoring a HELLO_REPLY to a HELLO from an earlier run"); + return Ok(()); + } + } + if !admit(state, &peer, &message).await { return Ok(()); } @@ -700,7 +712,7 @@ pub(crate) async fn handle_message( } establish_peer(state, &peer, swarm_id, true, hello.ready).await?; - if let Err(error) = send_hello_reply(state, &peer).await { + if let Err(error) = send_hello_reply(state, &peer, hello.run_id).await { // The peer retries its HELLO. warn!(peer = %peer, error = %format!("{error:#}"), "HELLO_REPLY failed"); } diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 83c4e8d..e8d654d 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -12,6 +12,9 @@ pub use crate::document::replica::Epoch; pub mod firewall; pub mod mycelium; +/// Random identifier of one run of a node (from start to exit). +pub type RunId = [u8; 16]; + /// Most heads listed in a keepalive; more are sent as a digest. A head takes 34 bytes on /// the wire, so 32 of them keep a keepalive within one Mycelium chunk (1300 bytes; see /// the size test in `mycelium`). There are rarely more heads than nodes that wrote @@ -120,6 +123,8 @@ pub struct HelloRequest { pub auth: Vec<u8>, /// The sender is ready (same meaning as `SyncEnvelope::ready`). pub ready: bool, + /// Random per run of the sender; the reply echoes it. + pub run_id: RunId, } #[derive(Debug, Clone)] @@ -131,6 +136,9 @@ pub struct HelloReply { pub auth: Vec<u8>, /// The sender is ready. pub ready: bool, + /// The `run_id` of the HELLO answered. A reply to a HELLO from an earlier run of the + /// recipient (delivered late, or queued while it was down) is ignored. + pub run_id: RunId, } #[derive(Debug, Clone)] diff --git a/src/transport/mycelium.rs b/src/transport/mycelium.rs index 4c03839..51bbfab 100644 --- a/src/transport/mycelium.rs +++ b/src/transport/mycelium.rs @@ -877,6 +877,7 @@ pub struct HelloRequestShadow { epoch: EpochShadow, auth: ByteBuf, ready: bool, + run_id: ByteBuf, } #[derive(Serialize, Deserialize)] @@ -890,6 +891,7 @@ pub struct HelloReplyShadow<'a> { #[serde(default)] auth: ByteBuf, ready: bool, + run_id: ByteBuf, } #[derive(Serialize, Deserialize)] @@ -940,6 +942,7 @@ pub fn hello_request_to_bytes(msg: &HelloRequest) -> Result<Vec<u8>, rmp_serde:: epoch: epoch_to_shadow(&msg.epoch), auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, + run_id: ByteBuf::from(msg.run_id.to_vec()), }) } pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportError> { @@ -949,9 +952,17 @@ pub fn bytes_to_hello_request(bytes: &[u8]) -> Result<HelloRequest, TransportErr epoch: epoch_from_shadow(shadow.epoch)?, auth: shadow.auth.into_vec(), ready: shadow.ready, + run_id: run_id_from_shadow(&shadow.run_id)?, }) } +fn run_id_from_shadow(bytes: &ByteBuf) -> Result<crate::transport::RunId, TransportError> { + bytes + .as_slice() + .try_into() + .map_err(|_| TransportError::InvalidPayload("run id must be 16 bytes".into())) +} + // --- HelloReply --- pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::encode::Error> { let shadow = HelloReplyShadow { @@ -960,6 +971,7 @@ pub fn hello_reply_to_bytes(msg: &HelloReply) -> Result<Vec<u8>, rmp_serde::enco epoch: epoch_to_shadow(&msg.epoch), auth: ByteBuf::from(msg.auth.clone()), ready: msg.ready, + run_id: ByteBuf::from(msg.run_id.to_vec()), }; rmp_serde::to_vec(&shadow) } @@ -971,6 +983,7 @@ pub fn bytes_to_hello_reply(bytes: &[u8]) -> Result<HelloReply, TransportError> epoch: epoch_from_shadow(shadow.epoch)?, auth: shadow.auth.into_vec(), ready: shadow.ready, + run_id: run_id_from_shadow(&shadow.run_id)?, }) } |
