summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authortv <tv@krebsco.de>2026-10-24 03:19:28 +0200
committertv <tv@krebsco.de>2026-10-24 03:19:28 +0200
commita1460b0a2861d5bb34ba25d409b234468f269e4e (patch)
treefb447e712aa587b4136c12f4b918cd9c6cf160f0
parent84a307a09eb3ba95920b028a49d0ec57bf316cdc (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.rs1
-rw-r--r--src/state.rs3
-rw-r--r--src/swarm/peer.rs16
-rw-r--r--src/transport/mod.rs8
-rw-r--r--src/transport/mycelium.rs13
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)?,
})
}