Refuse an epoch-mismatch msg1 that would destroy a live peering

The epoch in msg1 is sealed, so a msg1 announcing a different epoch than
the one we have stored for that peer is authentic. It is also replayable:
a captured one stays valid indefinitely, and accepting it removed the
peer entry, its index registrations and the FSP session state behind it,
from off the path and repeatedly.

Gate the teardown on two receiver-local conditions. Refuse while the
peering the msg1 would destroy has shown authenticated inbound traffic
recently, which is the evidence that the old session is not in fact dead;
last_seen moves only on a successful decrypt, so a peer that genuinely
restarted clears the gate by having stopped sending, while a peering
under replay is by construction still heartbeating. And refuse a second
accepted epoch change for the same peer identity inside the same
interval, which bounds the churn a peer can drive on its own.

The stamp is written only on acceptance: a refusal that slid the window
would let a sustained replay starve a genuinely restarting peer. The
refusal drops the msg1 silently rather than resending msg2, which is
bound to the original msg1's ephemeral. Both thresholds come from one
constant at 15 seconds, sized so a restarting peer's msg1 resends still
land inside its first handshake window and so the liveness gate cannot
outlive the link reaper.

One residual stays open: a msg1 captured before an accepted epoch change
can still be replayed once per interval per peer. Closing it needs a
seen-msg1 cache scoped to this arm, which is receiver-side too and is not
part of this change.

CHANGELOG.md

@@ -574,6 +574,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   build with overflow checks on, such as the test harness. Behaviour is
   unchanged for every counter an honest peer can emit.
 
+- An epoch-mismatch msg1 no longer tears down a peering that is still
+  carrying authenticated traffic, and a second epoch change for the same peer
+  identity inside 15 seconds is refused. The epoch travels inside the AEAD, so
+  such a msg1 is authentic, but it stays authentic after capture: replaying
+  one destroyed a working peering, and with it the FSP session state that
+  peering carried, from off the path. The peering's last authenticated inbound
+  frame is the evidence that it is still alive, and nothing an unauthenticated
+  sender emits can refresh it, so a peer that genuinely restarted clears the
+  gate by having stopped sending. The refusal is a silent drop: no msg2 is
+  returned, since the stored msg2 is bound to the original msg1's ephemeral
+  and answering a sender-chosen address is free amplification. The interval is
+  stamped only when an epoch change is accepted, so a sustained replay cannot
+  starve a genuinely restarting peer. Both thresholds come from one constant,
+  sized so a restarting peer's msg1 resends still land inside its own first
+  handshake window and below `link_dead_timeout_secs`, and nothing changes on
+  the wire.
+
 - A session setup message naming an already-established peer no longer replaces
   that peer's session. The handler did this whenever `node.rekey.enabled` was
   false: it ran a fresh responder handshake and overwrote the entry, discarding

src/node/handlers/handshake.rs

@@ -9,9 +9,33 @@ use crate::node::wire::{Msg1Header, Msg2Header, build_msg2};
 use crate::node::{Node, NodeError};
 use crate::peer::{ActivePeer, PeerConnection, PromotionResult, cross_connection_winner};
 use crate::transport::{Link, LinkDirection, LinkId, ReceivedPacket};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use tracing::{debug, info, warn};
 
+/// Minimum interval between accepted epoch changes for one peer identity,
+/// and the recency threshold at which the peering an epoch change would
+/// destroy still counts as live.
+///
+/// An epoch-mismatch msg1 is authentic but replayable: a captured one stays
+/// valid indefinitely, and accepting it tears down a working peering. Both
+/// conditions are receiver-local. The liveness half is the one that closes
+/// the replay, since a peering under attack is by construction still
+/// heartbeating; the interval half bounds the churn a peer can drive on its
+/// own.
+///
+/// Sized against the peer's own recovery rather than against a round number:
+/// a genuinely restarting peer's msg1 resends fire at roughly t+1, t+3, t+7
+/// and t+15 seconds and its attempt is reaped at `handshake_timeout_secs`
+/// (30), so 15 is the largest value at which a real restart still re-peers
+/// inside its first handshake window with no reconnect backoff. It also sits
+/// below `link_dead_timeout_secs` (30), so the liveness gate can never
+/// outlive the reaper that would have removed the peering anyway.
+///
+/// Raising it lengthens the outage an attacker's accepted replay causes,
+/// because the genuine peer's recovery msg1 hits the same arm. Lowering it
+/// weakens both halves and, below the resend ladder, buys nothing.
+const EPOCH_RESTART_MIN_INTERVAL_SECS: u64 = 15;
+
 /// Why an inbound msg1 got past the `accept_connections` gate, and against
 /// what identity the post-DH confirmation must check it.
 ///
@@ -444,19 +468,58 @@ impl Node {
         if possible_restart && let Some(existing_peer) = self.peers.get(&peer_node_addr) {
             let new_epoch = conn.remote_epoch();
             let existing_epoch = existing_peer.remote_epoch();
+            let now_ms = Self::now_ms();
+            // How long the peering this msg1 would destroy has gone without
+            // authenticated inbound traffic. `last_seen` moves only on a
+            // successful decrypt, so nothing an unauthenticated sender emits
+            // can refresh it.
+            let peering_idle_ms = existing_peer.idle_time(now_ms);
 
             match (existing_epoch, new_epoch) {
                 (Some(existing), Some(new)) if existing != new => {
                     // Epoch mismatch — peer restarted. Tear down stale session.
+                    //
+                    // Two receiver-local conditions have to hold first. The
+                    // epoch is sealed, so this msg1 is authentic, but a
+                    // captured one stays authentic forever and replaying it
+                    // destroys a working peering. Refuse while the peering is
+                    // still carrying authenticated traffic — a peer that has
+                    // genuinely restarted stopped feeding `last_seen` when it
+                    // died, so this self-clears — and refuse a second epoch
+                    // change inside the dampening interval, which bounds the
+                    // churn a peer can drive on its own.
+                    let dampened = self
+                        .restart_dampener
+                        .get(&peer_node_addr)
+                        .is_some_and(|t| t.elapsed().as_secs() < EPOCH_RESTART_MIN_INTERVAL_SECS);
+                    let peering_is_live = peering_idle_ms < EPOCH_RESTART_MIN_INTERVAL_SECS * 1000;
+                    if peering_is_live || dampened {
+                        debug!(
+                            peer = %self.peer_display_name(&peer_node_addr),
+                            idle_ms = peering_idle_ms,
+                            dampened,
+                            "Epoch mismatch dampened, dropping msg1"
+                        );
+                        // No msg2 is sent: the stored msg2 is bound to the
+                        // original msg1's ephemeral, and answering an address
+                        // the sender chose is free amplification.
+                        self.connections.remove(&link_id);
+                        self.links.remove(&link_id);
+                        self.stats_mut()
+                            .record_reject(RejectReason::Handshake(HandshakeReject::BadState));
+                        return;
+                    }
                     debug!(
                         peer = %self.peer_display_name(&peer_node_addr),
                         "Peer restart detected (epoch mismatch), removing stale session"
                     );
                     self.remove_active_peer(&peer_node_addr);
-                    let now_ms = std::time::SystemTime::now()
-                        .duration_since(std::time::UNIX_EPOCH)
-                        .map(|d| d.as_millis() as u64)
-                        .unwrap_or(0);
+                    // Stamped on acceptance only. A refusal that slid the
+                    // window would let a sustained replay starve a genuinely
+                    // restarting peer for as long as it kept sending.
+                    let cutoff = Duration::from_secs(EPOCH_RESTART_MIN_INTERVAL_SECS);
+                    self.restart_dampener.retain(|_, t| t.elapsed() < cutoff);
+                    self.restart_dampener.insert(peer_node_addr, Instant::now());
                     self.schedule_reconnect(peer_node_addr, now_ms);
                     // Fall through to process as new connection
                 }

src/node/mod.rs

@@ -488,6 +488,12 @@ pub struct Node {
     /// Pending outbound handshakes by our sender_idx.
     /// Tracks which LinkId corresponds to which session index.
     pending_outbound: HashMap<(TransportId, u32), LinkId>,
+    /// When each peer identity's last ACCEPTED epoch change tore down its
+    /// peering. Keyed on identity rather than address, and held here rather
+    /// than on `ActivePeer`, because the teardown being dampened destroys
+    /// the peer entry itself. Pruned on insert; see
+    /// `EPOCH_RESTART_MIN_INTERVAL_SECS`.
+    restart_dampener: HashMap<NodeAddr, std::time::Instant>,
 
     // === Rate Limiting ===
     /// Rate limiter for msg1 processing (DoS protection).
@@ -788,6 +794,7 @@ impl Node {
             index_allocator: IndexAllocator::new(),
             peers_by_index: HashMap::new(),
             pending_outbound: HashMap::new(),
+            restart_dampener: HashMap::new(),
             msg1_rate_limiter,
             setup_rate_limiter,
             icmp_rate_limiter: IcmpRateLimiter::new(),
@@ -947,6 +954,7 @@ impl Node {
             index_allocator: IndexAllocator::new(),
             peers_by_index: HashMap::new(),
             pending_outbound: HashMap::new(),
+            restart_dampener: HashMap::new(),
             msg1_rate_limiter,
             setup_rate_limiter,
             icmp_rate_limiter: IcmpRateLimiter::new(),

src/node/tests/handshake.rs

@@ -1811,3 +1811,212 @@ async fn a_stale_reverse_address_entry_does_not_hide_a_peer_reachable_by_address
          returns above the insert that would overwrite the stale entry"
     );
 }
+
+// ===== Epoch-restart dampening =====
+//
+// An epoch-mismatch msg1 is authentic, because the epoch travels inside the
+// AEAD, but it is replayable: a captured one stays valid forever and
+// accepting it destroys a working peering. Two receiver-local conditions
+// gate the teardown, and each of the first two cases below breaks one of
+// them.
+
+/// Install a peering for `initiator` that carries an epoch its genuine msg1
+/// does not, and that has gone `idle_secs` without authenticated inbound
+/// traffic. Returns the link the peering is bound to.
+fn install_peering_at_a_different_epoch(
+    node: &mut Node,
+    initiator: &Node,
+    transport_id: TransportId,
+    source_addr: &TransportAddr,
+    idle_secs: u64,
+) -> LinkId {
+    use crate::peer::ActivePeer;
+
+    let identity = PeerIdentity::from_pubkey_full(initiator.identity().pubkey_full());
+    let node_addr = *identity.node_addr();
+    let link_id = node.allocate_link_id();
+    let authenticated_at = Node::now_ms().saturating_sub(idle_secs * 1000);
+    let mut peer = ActivePeer::new(identity, link_id, authenticated_at);
+    peer.set_current_addr(transport_id, source_addr.clone());
+    // Anything but the epoch the initiator's msg1 carries, so the msg1 reads
+    // as a restart.
+    peer.set_remote_epoch(Some([0xAA; 8]));
+    node.peers.insert(node_addr, peer);
+    node.addr_to_link
+        .insert((transport_id, source_addr.clone()), link_id);
+    link_id
+}
+
+/// A peering long enough past its last authenticated inbound frame that the
+/// liveness gate does not hold the restart back.
+const IDLE_SECS: u64 = 60;
+
+#[tokio::test]
+async fn an_epoch_mismatch_msg1_against_a_live_peering_leaves_it_intact() {
+    let transport_id = TransportId::new(1);
+    let mut node = make_node();
+    let initiator = make_node();
+    let initiator_addr = node_addr_of(&initiator);
+    let source_addr = TransportAddr::from_string("127.0.0.1:41001");
+
+    // The peering is carrying authenticated traffic: it decrypted a frame a
+    // moment ago. Under replay that is always the case, because the genuine
+    // peer is heartbeating.
+    let peer_link =
+        install_peering_at_a_different_epoch(&mut node, &initiator, transport_id, &source_addr, 0);
+
+    let bad_state_before = node.stats().handshake.bad_state;
+    node.handle_msg1(ReceivedPacket::with_timestamp(
+        transport_id,
+        source_addr.clone(),
+        genuine_msg1(&initiator, &node),
+        Node::now_ms(),
+    ))
+    .await;
+
+    let peer = node
+        .get_peer(&initiator_addr)
+        .expect("a live peering must survive an epoch-mismatch msg1");
+    assert_eq!(
+        peer.link_id(),
+        peer_link,
+        "the peering must be the one that was already established, not a \
+         replacement promoted from the msg1"
+    );
+    assert_eq!(
+        peer.remote_epoch(),
+        Some([0xAA; 8]),
+        "the stored epoch must not have moved to the one the msg1 carried"
+    );
+    assert_eq!(
+        node.connection_count(),
+        0,
+        "the dropped msg1 must leave no connection behind"
+    );
+    assert_eq!(
+        node.stats().handshake.bad_state - bad_state_before,
+        1,
+        "the drop must be counted"
+    );
+}
+
+#[tokio::test]
+async fn a_second_epoch_change_inside_the_dampening_interval_leaves_the_peering_intact() {
+    let transport_id = TransportId::new(1);
+    let mut node = make_node();
+    let initiator = make_node();
+    let initiator_addr = node_addr_of(&initiator);
+    let source_addr = TransportAddr::from_string("127.0.0.1:41002");
+
+    // First epoch change: the peering is genuinely idle, so it is accepted
+    // and stamps the dampener.
+    let first_link = install_peering_at_a_different_epoch(
+        &mut node,
+        &initiator,
+        transport_id,
+        &source_addr,
+        IDLE_SECS,
+    );
+    node.handle_msg1(ReceivedPacket::with_timestamp(
+        transport_id,
+        source_addr.clone(),
+        genuine_msg1(&initiator, &node),
+        Node::now_ms(),
+    ))
+    .await;
+    let promoted = node
+        .get_peer(&initiator_addr)
+        .expect("the first epoch change must be accepted");
+    assert_ne!(
+        promoted.link_id(),
+        first_link,
+        "the first epoch change must have replaced the peering"
+    );
+
+    // The peer moves epoch again straight away. Nothing about the second
+    // msg1 is distinguishable from the first, which is why the interval,
+    // not the message, has to be what refuses it.
+    let second_link = install_peering_at_a_different_epoch(
+        &mut node,
+        &initiator,
+        transport_id,
+        &source_addr,
+        IDLE_SECS,
+    );
+
+    let bad_state_before = node.stats().handshake.bad_state;
+    node.handle_msg1(ReceivedPacket::with_timestamp(
+        transport_id,
+        source_addr.clone(),
+        genuine_msg1(&initiator, &node),
+        Node::now_ms(),
+    ))
+    .await;
+
+    let peer = node
+        .get_peer(&initiator_addr)
+        .expect("a second epoch change inside the interval must not tear the peering down");
+    assert_eq!(
+        peer.link_id(),
+        second_link,
+        "the peering must be the one that was already established"
+    );
+    assert_eq!(
+        peer.remote_epoch(),
+        Some([0xAA; 8]),
+        "the stored epoch must not have moved to the one the msg1 carried"
+    );
+    assert_eq!(
+        node.connection_count(),
+        0,
+        "the dropped msg1 must leave no connection behind"
+    );
+    assert_eq!(
+        node.stats().handshake.bad_state - bad_state_before,
+        1,
+        "the drop must be counted"
+    );
+}
+
+/// Healthy path, and NOT discriminating: this passes with or without the
+/// gates. It is here so that tightening either one, or a bug that stamps the
+/// dampener on a refusal, reds the suite instead of silently refusing every
+/// genuine restart.
+#[tokio::test]
+async fn a_first_epoch_change_against_a_silent_peering_still_restarts_it() {
+    let transport_id = TransportId::new(1);
+    let mut node = make_node();
+    let initiator = make_node();
+    let initiator_addr = node_addr_of(&initiator);
+    let source_addr = TransportAddr::from_string("127.0.0.1:41003");
+
+    let stale_link = install_peering_at_a_different_epoch(
+        &mut node,
+        &initiator,
+        transport_id,
+        &source_addr,
+        IDLE_SECS,
+    );
+
+    node.handle_msg1(ReceivedPacket::with_timestamp(
+        transport_id,
+        source_addr.clone(),
+        genuine_msg1(&initiator, &node),
+        Node::now_ms(),
+    ))
+    .await;
+
+    let peer = node
+        .get_peer(&initiator_addr)
+        .expect("a restart with no prior epoch change must be promoted");
+    assert_ne!(
+        peer.link_id(),
+        stale_link,
+        "the stale peering must have been torn down and replaced"
+    );
+    assert_eq!(
+        peer.remote_epoch(),
+        Some(initiator.startup_epoch()),
+        "the replacement must carry the epoch the msg1 announced"
+    );
+}