laantungir/fips
1
0
Fork 0
mirror of https://github.com/jmcorgan/fips.git synced 2026-08-09 08:14:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

FSP wire format revision and session-layer MMP implementation

Browse Source 
FSP wire format revision (TASK-2026-0007):

Introduce the FIPS Session Protocol (FSP) wire format with a 4-byte
common prefix [ver_phase:1][flags:1][payload_len:2 LE] replacing the
old 1-byte msg_type dispatch. All session messages share this prefix
with phase-based dispatch (Established, Setup, Ack, Unencrypted).

- New session_wire.rs: FSP constants, header types, parse/build helpers
- SessionMessageType enum: DataPacket (0x10), SenderReport (0x11),
  ReceiverReport (0x12), PathMtuNotification (0x13)
- FspFlags (CP/K/U) and FspInnerFlags (SP) for flag management
- SessionSenderReport, SessionReceiverReport, PathMtuNotification
  message structs with encode/decode
- FSP send pipeline: 12-byte header as AAD, 6-byte inner header
  (timestamp + msg_type + inner_flags), encrypt_with_aad()
- FSP receive pipeline: parse header, extract cleartext coords (CP),
  AEAD decrypt with AAD, strip inner header, msg_type dispatch
- Forwarding: transit nodes parse cleartext coords without decryption
- Removed DataPacket struct and associated types
- SessionEntry: session_start_ms, mark_established(), session_timestamp()
- FIPS_OVERHEAD: 144 → 150 bytes (+6 for FSP inner header)
- Design docs updated for new wire format

Session-layer MMP implementation (TASK-2026-0008):

Implement complete session-layer MMP reusing the link-layer algorithm
modules (SenderState, ReceiverState, MmpMetrics, SpinBitState) with
independent configuration and higher report interval clamps.

- SessionMmpConfig: separate config section (node.session_mmp.*)
- MmpSessionState: session-specific wrapper with PathMtuState tracking
- Session-layer constants (500ms-10s report intervals, 1s cold start)
- Parameterized interval methods (new_with_cold_start,
  update_report_interval_with_bounds) on SenderState/ReceiverState
- Bidirectional From conversions between link/session report types
- SessionEntry: mmp and is_initiator fields, initialized on Established
- send_session_msg() for reports/notifications
- Per-message RX recording with spin bit state tracking
- Handlers for SenderReport, ReceiverReport, PathMtuNotification
- path_mtu threaded from SessionDatagram envelope through to handlers
- check_session_mmp_reports() tick handler with collect-then-send pattern
- Periodic and teardown operator logging for session metrics
- PathMtuState: destination observes incoming MTU on all session messages,
  source seeded from outbound transport MTU, decrease-immediate /
  increase-requires-3-consecutive rules

Link-layer MMP fix:

- Stop feeding spin bit RTT samples into SRTT estimator; inter-frame
  timing in the mesh is irregular, inflating spin-bit RTT by variable
  processing delays; timestamp-echo provides accurate RTT

29 files changed, 602 tests pass, 0 clippy warnings.
This commit is contained in:
Johnathan Corgan
2026-02-19 03:15:05 +00:00
parent d8cb4d407e
commit 04d9fd625d
29 changed files with 2675 additions and 733 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/handlers/discovery.rs
+1 -1
View File
@@ -151,7 +151,7 @@ impl Node {
// If an established session exists, reset the warmup counter.
// Discovery has completed and transit nodes along the response
// path now have fresh coords. Reset warmup so the next N
// DataPackets include COORDS_PRESENT to re-warm the forward path.
// data packets include COORDS_PRESENT to re-warm the forward path.
if let Some(entry) = self.sessions.get_mut(&target)
&& entry.is_established()
{
src/node/handlers/encrypted.rs
+6 -4
View File
@@ -100,10 +100,12 @@ impl Node {
ce_flag,
now,
);
// Spin bit: feed to spin state, get optional RTT sample
if let Some(rtt) = mmp.spin_bit.rx_observe(sp_flag, header.counter, now) {
mmp.metrics.srtt.update(rtt.as_micros() as i64);
}
// Spin bit: advance state machine for correct TX reflection.
// RTT samples from spin bit are not used for SRTT because
// inter-frame timing in the mesh is irregular, inflating
// spin-bit RTT by variable processing delays on both sides.
// Timestamp-echo in ReceiverReport provides accurate RTT.
let _spin_rtt = mmp.spin_bit.rx_observe(sp_flag, header.counter, now);
}
// Update address for roaming support
src/node/handlers/forwarding.rs
+46 -39
View File
@@ -6,9 +6,12 @@
//! locally, and generates error signals on routing failure.
use crate::node::Node;
use crate::node::session_wire::{
parse_encrypted_coords, FspCommonPrefix, FSP_COMMON_PREFIX_SIZE, FSP_HEADER_SIZE,
FSP_PHASE_ESTABLISHED, FSP_PHASE_MSG1, FSP_PHASE_MSG2,
};
use crate::protocol::{
CoordsRequired, DataPacket, PathBroken, SessionAck, SessionDatagram, SessionMessageType,
SessionSetup,
CoordsRequired, PathBroken, SessionAck, SessionDatagram, SessionSetup,
};
use crate::NodeAddr;
use tracing::debug;
@@ -42,7 +45,7 @@ impl Node {
// Local delivery: dispatch to session layer handlers
if datagram.dest_addr == *self.node_addr() {
self.handle_session_payload(&datagram.src_addr, &datagram.payload)
self.handle_session_payload(&datagram.src_addr, &datagram.payload, datagram.path_mtu)
.await;
return;
}
@@ -81,28 +84,29 @@ impl Node {
/// Attempt to warm the coordinate cache from session-layer payload headers.
///
/// Transit routers can read the session message type byte and, for
/// SessionSetup and SessionAck, extract plaintext coordinate fields.
/// DataPacket with COORDS_PRESENT has plaintext coords before the
/// encrypted payload. Other types are ignored.
/// Transit routers parse the 4-byte FSP common prefix to identify message
/// type, then extract plaintext coordinate fields from:
/// - SessionSetup (phase 0x1): src_coords + dest_coords
/// - SessionAck (phase 0x2): src_coords
/// - Encrypted with CP flag (phase 0x0): cleartext coords between header and ciphertext
///
/// Decode failures are logged and silently ignored — they don't block
/// forwarding.
fn try_warm_coord_cache(&mut self, datagram: &SessionDatagram) {
if datagram.payload.is_empty() {
return;
}
let prefix = match FspCommonPrefix::parse(&datagram.payload) {
Some(p) => p,
None => return,
};
let msg_type = datagram.payload[0];
let inner = &datagram.payload[1..];
let inner = &datagram.payload[FSP_COMMON_PREFIX_SIZE..];
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as u64)
.unwrap_or(0);
match SessionMessageType::from_byte(msg_type) {
Some(SessionMessageType::SessionSetup) => {
match prefix.phase {
FSP_PHASE_MSG1 => {
match SessionSetup::decode(inner) {
Ok(setup) => {
self.coord_cache_mut().insert(
@@ -126,7 +130,7 @@ impl Node {
}
}
}
Some(SessionMessageType::SessionAck) => {
FSP_PHASE_MSG2 => {
match SessionAck::decode(inner) {
Ok(ack) => {
self.coord_cache_mut().insert(
@@ -144,38 +148,41 @@ impl Node {
}
}
}
Some(SessionMessageType::DataPacket) => {
match DataPacket::decode(inner) {
Ok(data) => {
if data.flags.coords_present {
if let Some(src_coords) = data.src_coords {
self.coord_cache_mut().insert(
datagram.src_addr,
src_coords,
now_ms,
);
}
if let Some(dest_coords) = data.dest_coords {
self.coord_cache_mut().insert(
datagram.dest_addr,
dest_coords,
now_ms,
);
}
debug!(
src = %datagram.src_addr,
dest = %datagram.dest_addr,
"Cached coords from DataPacket"
FSP_PHASE_ESTABLISHED if prefix.has_coords() => {
// CP flag set: coords in cleartext between header and ciphertext.
// Parse coords from the cleartext section after the 12-byte header.
// inner starts after the 4-byte prefix, so we need 8 more bytes
// for the counter (header is 12 total = 4 prefix + 8 counter).
let coord_data = &datagram.payload[FSP_HEADER_SIZE..];
match parse_encrypted_coords(coord_data) {
Ok((src_coords, dest_coords, _bytes_consumed)) => {
if let Some(coords) = src_coords {
self.coord_cache_mut().insert(
datagram.src_addr,
coords,
now_ms,
);
}
if let Some(coords) = dest_coords {
self.coord_cache_mut().insert(
datagram.dest_addr,
coords,
now_ms,
);
}
debug!(
src = %datagram.src_addr,
dest = %datagram.dest_addr,
"Cached coords from encrypted message"
);
}
Err(e) => {
debug!(error = %e, "Failed to decode DataPacket for cache warming");
debug!(error = %e, "Failed to parse coords for cache warming");
}
}
}
_ => {
// CoordsRequired, PathBroken, unknown: no coords to cache
// Phase 0x0 without CP, error signals, unknown: no coords to cache
}
}
}
src/node/handlers/mmp.rs
+163 -8
View File
@@ -5,8 +5,12 @@
//! and teardown metric logs.
use crate::mmp::MmpMode;
use crate::mmp::MmpSessionState;
use crate::mmp::report::{ReceiverReport, SenderReport};
use crate::node::Node;
use crate::protocol::{
PathMtuNotification, SessionMessageType, SessionReceiverReport, SessionSenderReport,
};
use crate::NodeAddr;
use std::time::Instant;
use tracing::{debug, info, warn};
@@ -139,17 +143,19 @@ impl Node {
let mode = mmp.mode();
// Sender reports: Full mode only
if mode == MmpMode::Full && mmp.sender.should_send_report(now) {
if let Some(sr) = mmp.sender.build_report(now) {
sender_reports.push((*node_addr, sr.encode()));
}
if mode == MmpMode::Full
&& mmp.sender.should_send_report(now)
&& let Some(sr) = mmp.sender.build_report(now)
{
sender_reports.push((*node_addr, sr.encode()));
}
// Receiver reports: Full and Lightweight modes
if mode != MmpMode::Minimal && mmp.receiver.should_send_report(now) {
if let Some(rr) = mmp.receiver.build_report(now) {
receiver_reports.push((*node_addr, rr.encode()));
}
if mode != MmpMode::Minimal
&& mmp.receiver.should_send_report(now)
&& let Some(rr) = mmp.receiver.build_report(now)
{
receiver_reports.push((*node_addr, rr.encode()));
}
// Periodic operator logging
@@ -243,4 +249,153 @@ impl Node {
"MMP link teardown"
);
}
// === Session-layer MMP ===
/// Check all sessions for pending MMP reports and send them.
///
/// Called from the tick handler. Also emits periodic session MMP logs.
/// Uses the collect-then-send pattern to avoid borrowing conflicts.
pub(in crate::node) async fn check_session_mmp_reports(&mut self) {
let now = Instant::now();
// Collect reports to send: (dest_addr, msg_type, encoded_body)
let mut reports: Vec<(NodeAddr, u8, Vec<u8>)> = Vec::new();
for (dest_addr, entry) in self.sessions.iter_mut() {
let Some(mmp) = entry.mmp_mut() else {
continue;
};
let mode = mmp.mode();
// Sender reports: Full mode only
if mode == MmpMode::Full
&& mmp.sender.should_send_report(now)
&& let Some(sr) = mmp.sender.build_report(now)
{
let session_sr: SessionSenderReport = SessionSenderReport::from(&sr);
reports.push((
*dest_addr,
SessionMessageType::SenderReport.to_byte(),
session_sr.encode(),
));
}
// Receiver reports: Full and Lightweight modes
if mode != MmpMode::Minimal
&& mmp.receiver.should_send_report(now)
&& let Some(rr) = mmp.receiver.build_report(now)
{
let session_rr: SessionReceiverReport = SessionReceiverReport::from(&rr);
reports.push((
*dest_addr,
SessionMessageType::ReceiverReport.to_byte(),
session_rr.encode(),
));
}
// PathMtu notifications (all modes)
if mmp.path_mtu.should_send_notification(now)
&& let Some(mtu_value) = mmp.path_mtu.build_notification(now)
{
let notif = PathMtuNotification::new(mtu_value);
reports.push((
*dest_addr,
SessionMessageType::PathMtuNotification.to_byte(),
notif.encode(),
));
}
// Periodic operator logging
if mmp.should_log(now) {
Self::log_session_mmp_metrics(dest_addr, mmp);
mmp.mark_logged(now);
}
}
// Send collected reports via session-layer encryption
for (dest_addr, msg_type, body) in reports {
if let Err(e) = self.send_session_msg(&dest_addr, msg_type, &body).await {
debug!(
dest = %dest_addr,
msg_type,
error = %e,
"Failed to send session MMP report"
);
}
}
}
/// Emit periodic session MMP metrics at info and debug levels.
fn log_session_mmp_metrics(dest_addr: &NodeAddr, mmp: &MmpSessionState) {
let m = &mmp.metrics;
let rtt_str = match m.srtt_ms() {
Some(rtt) => format!("{:.1}ms", rtt),
None => "n/a".to_string(),
};
let loss_pct = m.loss_rate() * 100.0;
let tx_pkts = mmp.sender.cumulative_packets_sent();
let rx_pkts = mmp.receiver.cumulative_packets_recv();
let goodput_str = format_throughput(m.goodput_bps());
let send_mtu = mmp.path_mtu.current_mtu();
let observed_mtu = mmp.path_mtu.last_observed_mtu();
info!(
session = %dest_addr,
rtt = %rtt_str,
loss = format_args!("{:.1}%", loss_pct),
goodput = %goodput_str,
send_mtu,
observed_mtu,
tx_pkts,
rx_pkts,
"MMP session metrics"
);
debug!(
session = %dest_addr,
jitter_us = mmp.receiver.jitter_us(),
rtt_trend = format_args!("{}", if m.rtt_trend.initialized() {
format!("short={:.1} long={:.1}", m.rtt_trend.short(), m.rtt_trend.long())
} else {
"n/a".to_string()
}),
loss_trend = format_args!("{}", if m.loss_trend.initialized() {
format!("short={:.4} long={:.4}", m.loss_trend.short(), m.loss_trend.long())
} else {
"n/a".to_string()
}),
delivery_fwd = format_args!("{:.3}", m.delivery_ratio_forward),
delivery_rev = format_args!("{:.3}", m.delivery_ratio_reverse),
mode = %mmp.mode(),
"MMP session metrics (detail)"
);
}
/// Emit a teardown log summarizing lifetime session MMP metrics.
pub(in crate::node) fn log_session_mmp_teardown(dest_addr: &NodeAddr, mmp: &MmpSessionState) {
let m = &mmp.metrics;
let rtt_str = match m.srtt_ms() {
Some(rtt) => format!("{:.1}ms", rtt),
None => "n/a".to_string(),
};
info!(
session = %dest_addr,
rtt = %rtt_str,
loss = format_args!("{:.1}%", m.loss_rate() * 100.0),
etx = format_args!("{:.2}", m.etx),
send_mtu = mmp.path_mtu.current_mtu(),
observed_mtu = mmp.path_mtu.last_observed_mtu(),
tx_pkts = mmp.sender.cumulative_packets_sent(),
tx_bytes = mmp.sender.cumulative_bytes_sent(),
rx_pkts = mmp.receiver.cumulative_packets_recv(),
rx_bytes = mmp.receiver.cumulative_bytes_recv(),
jitter_us = mmp.receiver.jitter_us(),
"MMP session teardown"
);
}
}
src/node/handlers/rx_loop.rs
+1
View File
@@ -84,6 +84,7 @@ impl Node {
self.check_tree_state().await;
self.check_bloom_state().await;
self.check_mmp_reports().await;
self.check_session_mmp_reports().await;
self.purge_stale_lookups(now_ms);
}
}
src/node/handlers/session.rs
+511 -139
View File
@@ -1,16 +1,24 @@
//! End-to-end session message handlers.
//!
//! Handles locally-delivered session payloads from SessionDatagram envelopes.
//! Dispatches based on session message type to specific handlers for
//! SessionSetup (Noise IK msg1), SessionAck (msg2), DataPacket, and
//! error signals (CoordsRequired, PathBroken).
//! Dispatches based on FSP common prefix phase to specific handlers for
//! SessionSetup (Noise IK msg1), SessionAck (msg2), encrypted data,
//! and error signals (CoordsRequired, PathBroken).
use crate::node::session::{EndToEndState, SessionEntry};
use crate::node::session_wire::{
build_fsp_header, fsp_prepend_inner_header, fsp_strip_inner_header,
parse_encrypted_coords, FspCommonPrefix, FspEncryptedHeader, FSP_COMMON_PREFIX_SIZE,
FSP_FLAG_CP, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED, FSP_PHASE_MSG1, FSP_PHASE_MSG2,
};
use crate::protocol::encode_coords;
use crate::node::{Node, NodeError};
use crate::noise::{HandshakeState, HANDSHAKE_MSG1_SIZE, HANDSHAKE_MSG2_SIZE};
use crate::mmp::report::ReceiverReport;
use crate::mmp::{MAX_SESSION_REPORT_INTERVAL_MS, MIN_SESSION_REPORT_INTERVAL_MS};
use crate::protocol::{
CoordsRequired, DataPacket, PathBroken, SessionAck, SessionDatagram, SessionMessageType,
SessionSetup,
CoordsRequired, FspInnerFlags, PathBroken, PathMtuNotification, SessionAck, SessionDatagram,
SessionMessageType, SessionReceiverReport, SessionSenderReport, SessionSetup,
};
use crate::NodeAddr;
use secp256k1::PublicKey;
@@ -20,42 +28,239 @@ impl Node {
/// Handle a locally-delivered session datagram payload.
///
/// Called from `handle_session_datagram()` when `dest_addr == self.node_addr()`.
/// Dispatches to the appropriate handler based on the session message type byte.
/// Dispatches based on the 4-byte FSP common prefix:
///
/// - Phase 0x1 → SessionSetup (handshake msg1)
/// - Phase 0x2 → SessionAck (handshake msg2)
/// - Phase 0x0 + U flag → plaintext error signal (CoordsRequired/PathBroken)
/// - Phase 0x0 + !U → encrypted session message (data, reports, etc.)
pub(in crate::node) async fn handle_session_payload(
&mut self,
src_addr: &NodeAddr,
payload: &[u8],
path_mtu: u16,
) {
if payload.is_empty() {
debug!("Empty session payload");
return;
}
let prefix = match FspCommonPrefix::parse(payload) {
Some(p) => p,
None => {
debug!(len = payload.len(), "Session payload too short for FSP prefix");
return;
}
};
let msg_type = payload[0];
let inner = &payload[1..];
let inner = &payload[FSP_COMMON_PREFIX_SIZE..];
match SessionMessageType::from_byte(msg_type) {
Some(SessionMessageType::SessionSetup) => {
match prefix.phase {
FSP_PHASE_MSG1 => {
self.handle_session_setup(src_addr, inner).await;
}
Some(SessionMessageType::SessionAck) => {
FSP_PHASE_MSG2 => {
self.handle_session_ack(src_addr, inner).await;
}
Some(SessionMessageType::DataPacket) => {
self.handle_data_packet(src_addr, inner).await;
FSP_PHASE_ESTABLISHED if prefix.is_unencrypted() => {
// Plaintext error signals: read msg_type from first byte after prefix
if inner.is_empty() {
debug!("Empty plaintext error signal");
return;
}
let error_type = inner[0];
let error_body = &inner[1..];
match SessionMessageType::from_byte(error_type) {
Some(SessionMessageType::CoordsRequired) => {
self.handle_coords_required(error_body).await;
}
Some(SessionMessageType::PathBroken) => {
self.handle_path_broken(error_body).await;
}
_ => {
debug!(error_type, "Unknown plaintext error signal type");
}
}
}
Some(SessionMessageType::CoordsRequired) => {
self.handle_coords_required(inner).await;
FSP_PHASE_ESTABLISHED => {
self.handle_encrypted_session_msg(src_addr, payload, path_mtu).await;
}
Some(SessionMessageType::PathBroken) => {
self.handle_path_broken(inner).await;
}
None => {
debug!(msg_type, "Unknown session message type");
_ => {
debug!(phase = prefix.phase, "Unknown FSP phase");
}
}
}
/// Handle an encrypted session message (phase 0x0, U flag clear).
///
/// Full FSP receive pipeline:
/// 1. Parse FspEncryptedHeader (12 bytes) → counter, flags, header_bytes
/// 2. If CP flag: parse cleartext coords, cache them
/// 3. Session lookup with Responding→Established transition
/// 4. AEAD decrypt with AAD = header_bytes
/// 5. Strip FSP inner header → timestamp, msg_type, inner_flags
/// 6. Dispatch by msg_type
async fn handle_encrypted_session_msg(&mut self, src_addr: &NodeAddr, payload: &[u8], path_mtu: u16) {
// Parse the 12-byte encrypted header (includes the 4-byte prefix)
let header = match FspEncryptedHeader::parse(payload) {
Some(h) => h,
None => {
debug!(len = payload.len(), "Encrypted session message too short for FSP header");
return;
}
};
// Determine where ciphertext starts (after header, optionally after coords)
let mut ciphertext_offset = FSP_HEADER_SIZE;
// If CP flag set, parse cleartext coords between header and ciphertext
if header.has_coords() {
let coord_data = &payload[FSP_HEADER_SIZE..];
match parse_encrypted_coords(coord_data) {
Ok((src_coords, dest_coords, bytes_consumed)) => {
let now_ms = Self::now_ms();
if let Some(coords) = src_coords {
self.coord_cache.insert(*src_addr, coords, now_ms);
}
if let Some(coords) = dest_coords {
self.coord_cache.insert(*self.node_addr(), coords, now_ms);
}
ciphertext_offset += bytes_consumed;
}
Err(e) => {
debug!(error = %e, "Failed to parse coords from encrypted session message");
return;
}
}
}
let ciphertext = &payload[ciphertext_offset..];
// Look up session entry, handle Responding→Established transition
let mut entry = match self.sessions.remove(src_addr) {
Some(e) => e,
None => {
debug!(src = %src_addr, "Encrypted session message for unknown session");
return;
}
};
if entry.state().is_responding() {
let old_state = entry.take_state();
let handshake = match old_state {
Some(EndToEndState::Responding(hs)) => hs,
_ => {
debug!(src = %src_addr, "Unexpected state during Responding transition");
return;
}
};
let noise_session = match handshake.into_session() {
Ok(s) => s,
Err(e) => {
debug!(error = %e, "Failed to create session from responding handshake");
return;
}
};
entry.set_state(EndToEndState::Established(noise_session));
entry.set_coords_warmup_remaining(self.config.node.session.coords_warmup_packets);
entry.mark_established(Self::now_ms());
entry.init_mmp(&self.config.node.session_mmp);
debug!(src = %src_addr, "Session established (responder, on first encrypted message)");
}
// Decrypt with AAD = the 12-byte header
let session = match entry.state_mut() {
EndToEndState::Established(s) => s,
_ => {
debug!(src = %src_addr, "Encrypted message but session not established");
self.sessions.insert(*src_addr, entry);
return;
}
};
let plaintext = match session.decrypt_with_replay_check_and_aad(
ciphertext,
header.counter,
&header.header_bytes,
) {
Ok(pt) => pt,
Err(e) => {
debug!(
error = %e, src = %src_addr, counter = header.counter,
"Session AEAD decryption failed"
);
self.sessions.insert(*src_addr, entry);
return;
}
};
entry.touch(Self::now_ms());
self.sessions.insert(*src_addr, entry);
// Strip FSP inner header (6 bytes)
let (timestamp, msg_type, inner_flags_byte, rest) = match fsp_strip_inner_header(&plaintext) {
Some(parts) => parts,
None => {
debug!(src = %src_addr, "Decrypted payload too short for FSP inner header");
return;
}
};
// MMP per-message recording on RX path
if let Some(entry) = self.sessions.get_mut(src_addr)
&& let Some(mmp) = entry.mmp_mut()
{
let now = std::time::Instant::now();
mmp.receiver.record_recv(
header.counter, timestamp, plaintext.len(), false, now,
);
// Spin bit: advance state machine for correct TX reflection.
// RTT samples not fed into SRTT — timestamp-echo provides
// accurate RTT; spin bit includes variable inter-frame delays.
let inner_flags = FspInnerFlags::from_byte(inner_flags_byte);
let _spin_rtt = mmp.spin_bit.rx_observe(
inner_flags.spin_bit, header.counter, now,
);
}
// Feed path_mtu from datagram envelope to MMP path MTU tracking.
// Done for ALL session messages, not just DataPackets, so the
// destination learns the path MTU even when only reports flow.
if let Some(entry) = self.sessions.get_mut(src_addr)
&& let Some(mmp) = entry.mmp_mut()
{
mmp.path_mtu.observe_incoming_mtu(path_mtu);
}
// Dispatch by msg_type
match SessionMessageType::from_byte(msg_type) {
Some(SessionMessageType::DataPacket) => {
// msg_type 0x10: deliver rest (IPv6 payload) to TUN
if let Some(tun_tx) = &self.tun_tx {
if let Err(e) = tun_tx.send(rest.to_vec()) {
debug!(error = %e, "Failed to deliver decrypted packet to TUN");
}
} else {
debug!(
src = %src_addr,
"DataPacket decrypted (no TUN interface, plaintext dropped)"
);
}
}
Some(SessionMessageType::SenderReport) => {
self.handle_session_sender_report(src_addr, rest);
}
Some(SessionMessageType::ReceiverReport) => {
self.handle_session_receiver_report(src_addr, rest);
}
Some(SessionMessageType::PathMtuNotification) => {
self.handle_session_path_mtu_notification(src_addr, rest);
}
_ => {
debug!(src = %src_addr, msg_type, "Unknown session message type, dropping");
}
}
// Flush any pending outbound packets (e.g., simultaneous initiation
// where responder also had queued outbound packets)
self.flush_pending_packets(src_addr).await;
}
/// Handle an incoming SessionSetup (Noise IK msg1).
///
/// The remote node wants to establish an end-to-end session with us.
@@ -143,18 +348,18 @@ impl Node {
let our_coords = self.tree_state.my_coords().clone();
let ack = SessionAck::new(our_coords).with_handshake(msg2);
let my_addr = *self.node_addr();
let datagram = SessionDatagram::new(my_addr, *src_addr, ack.encode())
let mut datagram = SessionDatagram::new(my_addr, *src_addr, ack.encode())
.with_ttl(self.config.node.session.default_ttl);
// Route the ack back to the initiator
if let Err(e) = self.send_session_datagram(&datagram).await {
if let Err(e) = self.send_session_datagram(&mut datagram).await {
debug!(error = %e, dest = %src_addr, "Failed to send SessionAck");
return;
}
// Store session entry in Responding state
let now_ms = Self::now_ms();
let entry = SessionEntry::new(*src_addr, remote_pubkey, EndToEndState::Responding(handshake), now_ms);
let entry = SessionEntry::new(*src_addr, remote_pubkey, EndToEndState::Responding(handshake), now_ms, false);
self.sessions.insert(*src_addr, entry);
debug!(src = %src_addr, "SessionSetup processed, SessionAck sent");
@@ -210,13 +415,13 @@ impl Node {
}
};
let now_ms = Self::now_ms();
entry.set_state(EndToEndState::Established(session));
entry.set_coords_warmup_remaining(self.config.node.session.coords_warmup_packets);
entry.touch(Self::now_ms());
entry.mark_established(now_ms);
entry.init_mmp(&self.config.node.session_mmp);
entry.touch(now_ms);
self.sessions.insert(*src_addr, entry);
// Cache the responder's coordinates
let now_ms = Self::now_ms();
self.coord_cache.insert(*src_addr, ack.src_coords, now_ms);
// Flush any queued outbound packets for this destination
@@ -225,6 +430,136 @@ impl Node {
debug!(src = %src_addr, "Session established (initiator)");
}
// === Session-layer MMP report handlers ===
/// Handle an incoming session-layer SenderReport (msg_type 0x11).
///
/// Informational only — the peer is telling us about what they sent.
/// Logged but not used for metrics (same pattern as link-layer).
fn handle_session_sender_report(&mut self, src_addr: &NodeAddr, body: &[u8]) {
let sr = match SessionSenderReport::decode(body) {
Ok(sr) => sr,
Err(e) => {
debug!(src = %src_addr, error = %e, "Malformed SessionSenderReport");
return;
}
};
debug!(
src = %src_addr,
cum_pkts = sr.cumulative_packets_sent,
interval_bytes = sr.interval_bytes_sent,
"Received SessionSenderReport"
);
}
/// Handle an incoming session-layer ReceiverReport (msg_type 0x12).
///
/// The peer is telling us about what they received from us. We feed
/// this to our metrics to compute RTT, loss rate, and trend indicators.
fn handle_session_receiver_report(&mut self, src_addr: &NodeAddr, body: &[u8]) {
let session_rr = match SessionReceiverReport::decode(body) {
Ok(rr) => rr,
Err(e) => {
debug!(src = %src_addr, error = %e, "Malformed SessionReceiverReport");
return;
}
};
// Convert to link-layer ReceiverReport for MmpMetrics processing
let rr: ReceiverReport = ReceiverReport::from(&session_rr);
let now_ms = Self::now_ms();
let entry = match self.sessions.get_mut(src_addr) {
Some(e) => e,
None => {
debug!(src = %src_addr, "SessionReceiverReport for unknown session");
return;
}
};
let our_timestamp_ms = entry.session_timestamp(now_ms);
let Some(mmp) = entry.mmp_mut() else {
return;
};
let now = std::time::Instant::now();
mmp.metrics.process_receiver_report(&rr, our_timestamp_ms, now);
// Feed SRTT back to sender/receiver report interval tuning (session-layer bounds)
if let Some(srtt_ms) = mmp.metrics.srtt_ms() {
let srtt_us = (srtt_ms * 1000.0) as i64;
mmp.sender.update_report_interval_with_bounds(
srtt_us,
MIN_SESSION_REPORT_INTERVAL_MS,
MAX_SESSION_REPORT_INTERVAL_MS,
);
mmp.receiver.update_report_interval_with_bounds(
srtt_us,
MIN_SESSION_REPORT_INTERVAL_MS,
MAX_SESSION_REPORT_INTERVAL_MS,
);
// Also update PathMtu notification interval from SRTT
mmp.path_mtu.update_interval_from_srtt(srtt_ms);
}
// Update reverse delivery ratio from our own receiver state
let our_recv_packets = mmp.receiver.cumulative_packets_recv();
let peer_highest = mmp.receiver.highest_counter();
if peer_highest > 0 {
let reverse_ratio = (our_recv_packets as f64) / (peer_highest as f64);
mmp.metrics.set_delivery_ratio_reverse(reverse_ratio);
}
debug!(
src = %src_addr,
rtt_ms = ?mmp.metrics.srtt_ms(),
loss = format_args!("{:.1}%", mmp.metrics.loss_rate() * 100.0),
"Processed SessionReceiverReport"
);
}
/// Handle an incoming PathMtuNotification (msg_type 0x13).
///
/// The destination is telling us the path MTU has changed.
/// Apply source-side rules (decrease immediate, increase validated).
fn handle_session_path_mtu_notification(&mut self, src_addr: &NodeAddr, body: &[u8]) {
let notif = match PathMtuNotification::decode(body) {
Ok(n) => n,
Err(e) => {
debug!(src = %src_addr, error = %e, "Malformed PathMtuNotification");
return;
}
};
let entry = match self.sessions.get_mut(src_addr) {
Some(e) => e,
None => {
debug!(src = %src_addr, "PathMtuNotification for unknown session");
return;
}
};
let Some(mmp) = entry.mmp_mut() else {
return;
};
let old_mtu = mmp.path_mtu.current_mtu();
let now = std::time::Instant::now();
mmp.path_mtu.apply_notification(notif.path_mtu, now);
let new_mtu = mmp.path_mtu.current_mtu();
if new_mtu != old_mtu {
debug!(
src = %src_addr,
old_mtu,
new_mtu,
"Path MTU changed via notification"
);
}
}
/// Complete an initiator-side Noise IK handshake given msg2.
fn complete_initiator_handshake(
mut handshake: HandshakeState,
@@ -238,90 +573,6 @@ impl Node {
.map_err(|e| format!("into_session failed: {}", e))
}
/// Handle an incoming DataPacket.
///
/// Decrypts the payload using the established session key and delivers
/// to the TUN interface.
async fn handle_data_packet(&mut self, src_addr: &NodeAddr, inner: &[u8]) {
let packet = match DataPacket::decode(inner) {
Ok(p) => p,
Err(e) => {
debug!(error = %e, "Malformed DataPacket");
return;
}
};
// Remove entry to take ownership for potential state transition
let mut entry = match self.sessions.remove(src_addr) {
Some(e) => e,
None => {
debug!(src = %src_addr, "DataPacket for unknown session");
return;
}
};
// If in Responding state, transition to Established first
// (responder wrote msg2, handshake is complete from our side)
if entry.state().is_responding() {
let old_state = entry.take_state();
let handshake = match old_state {
Some(EndToEndState::Responding(hs)) => hs,
_ => {
debug!(src = %src_addr, "Unexpected state in DataPacket handler");
return;
}
};
let noise_session = match handshake.into_session() {
Ok(s) => s,
Err(e) => {
debug!(error = %e, "Failed to create session from responding handshake");
return;
}
};
entry.set_state(EndToEndState::Established(noise_session));
entry.set_coords_warmup_remaining(self.config.node.session.coords_warmup_packets);
debug!(src = %src_addr, "Session established (responder, on first data)");
}
// Decrypt
let session = match entry.state_mut() {
EndToEndState::Established(s) => s,
_ => {
debug!(src = %src_addr, "DataPacket but session not established");
self.sessions.insert(*src_addr, entry);
return;
}
};
let plaintext = match session.decrypt_with_replay_check(&packet.payload, packet.counter) {
Ok(pt) => pt,
Err(e) => {
debug!(error = %e, src = %src_addr, counter = packet.counter, "Session decryption failed");
self.sessions.insert(*src_addr, entry);
return;
}
};
entry.touch(Self::now_ms());
self.sessions.insert(*src_addr, entry);
// Deliver to TUN
if let Some(tun_tx) = &self.tun_tx {
if let Err(e) = tun_tx.send(plaintext) {
debug!(error = %e, "Failed to deliver decrypted packet to TUN");
}
} else {
debug!(
src = %src_addr,
"DataPacket decrypted (no TUN interface, plaintext dropped)"
);
}
// Flush any pending outbound packets (e.g., simultaneous initiation
// where responder also had queued outbound packets)
self.flush_pending_packets(src_addr).await;
}
/// Handle a CoordsRequired error signal from a transit router.
///
/// The router couldn't route our packet because it lacks cached
@@ -431,18 +682,18 @@ impl Node {
// Wrap in SessionDatagram
let my_addr = *self.node_addr();
let datagram = SessionDatagram::new(my_addr, dest_addr, setup.encode())
let mut datagram = SessionDatagram::new(my_addr, dest_addr, setup.encode())
.with_ttl(self.config.node.session.default_ttl);
// Route toward destination
self.send_session_datagram(&datagram).await?;
self.send_session_datagram(&mut datagram).await?;
// Register destination identity for TUN → session routing
self.register_identity(dest_addr, dest_pubkey);
// Store session entry
let now_ms = Self::now_ms();
let entry = SessionEntry::new(dest_addr, dest_pubkey, EndToEndState::Initiating(handshake), now_ms);
let entry = SessionEntry::new(dest_addr, dest_pubkey, EndToEndState::Initiating(handshake), now_ms, true);
self.sessions.insert(dest_addr, entry);
debug!(dest = %dest_addr, "Session initiation started");
@@ -451,18 +702,28 @@ impl Node {
/// Send application data over an established session.
///
/// Encrypts the payload with the session key, wraps in DataPacket
/// and SessionDatagram, routes toward destination.
/// Uses the FSP pipeline: builds a 12-byte cleartext header (used as AAD),
/// prepends the 6-byte inner header to the plaintext, encrypts with AAD,
/// optionally inserts cleartext coords, and wraps in a SessionDatagram.
pub(in crate::node) async fn send_session_data(
&mut self,
dest_addr: &NodeAddr,
plaintext: &[u8],
) -> Result<(), NodeError> {
let now_ms = Self::now_ms();
let entry = self.sessions.get_mut(dest_addr).ok_or_else(|| NodeError::SendFailed {
node_addr: *dest_addr,
reason: "no session".into(),
})?;
// Check warmup counter and get session timestamp
let include_coords = entry.coords_warmup_remaining() > 0;
if include_coords {
entry.set_coords_warmup_remaining(entry.coords_warmup_remaining() - 1);
}
let timestamp = entry.session_timestamp(now_ms);
let spin_bit = entry.mmp().is_some_and(|m| m.spin_bit.tx_bit());
let session = match entry.state_mut() {
EndToEndState::Established(s) => s,
_ => {
@@ -476,35 +737,129 @@ impl Node {
// Get counter before encrypting (encrypt will increment it)
let counter = session.current_send_counter();
// Encrypt with session key
let ciphertext = session.encrypt(plaintext).map_err(|e| NodeError::SendFailed {
node_addr: *dest_addr,
reason: format!("session encrypt failed: {}", e),
// FSP inner header: [timestamp:4 LE][msg_type:1][inner_flags:1] + plaintext
let msg_type = SessionMessageType::DataPacket.to_byte(); // 0x10
let inner_flags = FspInnerFlags { spin_bit }.to_byte();
let inner_plaintext = fsp_prepend_inner_header(timestamp, msg_type, inner_flags, plaintext);
// Build FSP flags
let flags = if include_coords { FSP_FLAG_CP } else { 0 };
// Build 12-byte FSP header (used as AAD for AEAD)
let payload_len = inner_plaintext.len() as u16;
let header = build_fsp_header(counter, flags, payload_len);
// Encrypt with AAD binding to the FSP header
let ciphertext = session.encrypt_with_aad(&inner_plaintext, &header).map_err(|e| {
NodeError::SendFailed {
node_addr: *dest_addr,
reason: format!("session encrypt failed: {}", e),
}
})?;
// Check warmup counter and decrement (while entry is still borrowed)
let include_coords = entry.coords_warmup_remaining() > 0;
if include_coords {
entry.set_coords_warmup_remaining(entry.coords_warmup_remaining() - 1);
}
// Build DataPacket with explicit counter for replay protection
let mut data_packet = DataPacket::new(counter, ciphertext);
// Assemble: header(12) + [coords] + ciphertext
let mut fsp_payload = Vec::with_capacity(FSP_HEADER_SIZE + ciphertext.len() + 200);
fsp_payload.extend_from_slice(&header);
if include_coords {
let my_coords = self.tree_state.my_coords().clone();
let dest_coords = self.get_dest_coords(dest_addr);
data_packet = data_packet.with_coords(my_coords, dest_coords);
encode_coords(&my_coords, &mut fsp_payload);
encode_coords(&dest_coords, &mut fsp_payload);
}
fsp_payload.extend_from_slice(&ciphertext);
let my_addr = *self.node_addr();
let datagram = SessionDatagram::new(my_addr, *dest_addr, data_packet.encode())
let mut datagram = SessionDatagram::new(my_addr, *dest_addr, fsp_payload)
.with_ttl(self.config.node.session.default_ttl);
self.send_session_datagram(&datagram).await?;
self.send_session_datagram(&mut datagram).await?;
// Re-borrow after send (which borrowed &mut self)
if let Some(entry) = self.sessions.get_mut(dest_addr) {
entry.touch(Self::now_ms());
if let Some(mmp) = entry.mmp_mut() {
mmp.sender.record_sent(counter, timestamp, ciphertext.len());
}
entry.touch(now_ms);
}
Ok(())
}
/// Send a non-data session message (reports, notifications) over an established session.
///
/// Similar to `send_session_data()` but:
/// - Takes an explicit `msg_type` byte (0x11, 0x12, 0x13, etc.)
/// - Never includes COORDS_PRESENT (reports are lightweight)
/// - Reads spin bit from MMP state for the inner header
/// - Records the send in MMP sender state
pub(in crate::node) async fn send_session_msg(
&mut self,
dest_addr: &NodeAddr,
msg_type: u8,
payload: &[u8],
) -> Result<(), NodeError> {
let now_ms = Self::now_ms();
// Read spin bit and session timestamp from entry
let entry = self.sessions.get(dest_addr).ok_or_else(|| NodeError::SendFailed {
node_addr: *dest_addr,
reason: "no session".into(),
})?;
let timestamp = entry.session_timestamp(now_ms);
let spin_bit = entry.mmp().is_some_and(|m| m.spin_bit.tx_bit());
// Build inner flags with spin bit
let inner_flags = FspInnerFlags { spin_bit }.to_byte();
// Get mutable access for encryption
let entry = self.sessions.get_mut(dest_addr).ok_or_else(|| NodeError::SendFailed {
node_addr: *dest_addr,
reason: "no session".into(),
})?;
let session = match entry.state_mut() {
EndToEndState::Established(s) => s,
_ => {
return Err(NodeError::SendFailed {
node_addr: *dest_addr,
reason: "session not established".into(),
});
}
};
let counter = session.current_send_counter();
// FSP inner header + plaintext
let inner_plaintext = fsp_prepend_inner_header(timestamp, msg_type, inner_flags, payload);
// Build 12-byte FSP header (no flags — no CP for reports)
let payload_len = inner_plaintext.len() as u16;
let header = build_fsp_header(counter, 0, payload_len);
// Encrypt with AAD
let ciphertext = session.encrypt_with_aad(&inner_plaintext, &header).map_err(|e| {
NodeError::SendFailed {
node_addr: *dest_addr,
reason: format!("session encrypt failed: {}", e),
}
})?;
// Assemble: header(12) + ciphertext (no coords)
let mut fsp_payload = Vec::with_capacity(FSP_HEADER_SIZE + ciphertext.len());
fsp_payload.extend_from_slice(&header);
fsp_payload.extend_from_slice(&ciphertext);
let my_addr = *self.node_addr();
let mut datagram = SessionDatagram::new(my_addr, *dest_addr, fsp_payload)
.with_ttl(self.config.node.session.default_ttl);
self.send_session_datagram(&mut datagram).await?;
// Record in MMP sender state and touch
if let Some(entry) = self.sessions.get_mut(dest_addr) {
if let Some(mmp) = entry.mmp_mut() {
mmp.sender.record_sent(counter, timestamp, ciphertext.len());
}
entry.touch(now_ms);
}
Ok(())
@@ -512,11 +867,11 @@ impl Node {
/// Route and send a SessionDatagram through the mesh.
///
/// Finds the next hop for the destination and sends the datagram
/// as an encrypted link message.
/// Finds the next hop for the destination, seeds path_mtu from the
/// first-hop transport MTU, and sends as an encrypted link message.
async fn send_session_datagram(
&mut self,
datagram: &SessionDatagram,
datagram: &mut SessionDatagram,
) -> Result<(), NodeError> {
let next_hop_addr = match self.find_next_hop(&datagram.dest_addr) {
Some(peer) => *peer.node_addr(),
@@ -528,6 +883,23 @@ impl Node {
}
};
// Seed path_mtu from the first-hop transport MTU (same as forwarding path)
if let Some(peer) = self.peers.get(&next_hop_addr)
&& let Some(tid) = peer.transport_id()
&& let Some(transport) = self.transports.get(&tid)
{
datagram.path_mtu = datagram.path_mtu.min(transport.mtu());
}
// Source-side: seed our PathMtuState.current_mtu from the outbound
// transport MTU so it doesn't stay at u16::MAX until the destination
// sends a PathMtuNotification back.
if let Some(entry) = self.sessions.get_mut(&datagram.dest_addr)
&& let Some(mmp) = entry.mmp_mut()
{
mmp.path_mtu.seed_source_mtu(datagram.path_mtu);
}
let encoded = datagram.encode();
self.send_encrypted_link_message(&next_hop_addr, &encoded).await
}
src/node/handlers/timeout.rs
+6
View File
@@ -98,6 +98,12 @@ impl Node {
.collect();
for addr in idle {
// Log MMP teardown metrics before removing the session
if let Some(entry) = self.sessions.get(&addr)
&& let Some(mmp) = entry.mmp()
{
Self::log_session_mmp_teardown(&addr, mmp);
}
self.sessions.remove(&addr);
self.pending_tun_packets.remove(&addr);
debug!(
src/node/mod.rs
+1
View File
@@ -11,6 +11,7 @@ mod retry;
mod rate_limit;
mod routing_error_rate_limit;
pub(crate) mod session;
pub(crate) mod session_wire;
pub(crate) mod wire;
mod tree;
#[cfg(test)]
src/node/session.rs
+53 -1
View File
@@ -4,6 +4,8 @@
//! Sessions are established via SessionSetup/SessionAck handshake
//! messages carried inside SessionDatagram envelopes through the mesh.
use crate::config::SessionMmpConfig;
use crate::mmp::MmpSessionState;
use crate::noise::{HandshakeState, NoiseSession};
use crate::NodeAddr;
use secp256k1::PublicKey;
@@ -54,10 +56,19 @@ pub(crate) struct SessionEntry {
created_at: u64,
/// Last activity timestamp (Unix milliseconds).
last_activity: u64,
/// Remaining DataPackets that should include COORDS_PRESENT.
/// When the session transitioned to Established (Unix milliseconds).
/// Used to compute session-relative timestamps for the FSP inner header.
/// Set to 0 until the session is established.
session_start_ms: u64,
/// Remaining data packets that should include COORDS_PRESENT.
/// Initialized from config when session becomes Established;
/// reset on CoordsRequired receipt.
coords_warmup_remaining: u8,
/// Whether this node initiated the Noise IK handshake.
/// Used for spin bit role assignment in session-layer MMP.
is_initiator: bool,
/// Session-layer MMP state. Initialized on Established transition.
mmp: Option<MmpSessionState>,
}
impl SessionEntry {
@@ -67,6 +78,7 @@ impl SessionEntry {
remote_pubkey: PublicKey,
state: EndToEndState,
now_ms: u64,
is_initiator: bool,
) -> Self {
Self {
remote_addr,
@@ -74,7 +86,10 @@ impl SessionEntry {
state: Some(state),
created_at: now_ms,
last_activity: now_ms,
session_start_ms: 0,
coords_warmup_remaining: 0,
is_initiator,
mmp: None,
}
}
@@ -132,4 +147,41 @@ impl SessionEntry {
pub(crate) fn set_coords_warmup_remaining(&mut self, value: u8) {
self.coords_warmup_remaining = value;
}
/// Mark the session as started (transition to Established).
///
/// Records the current time as the session start for computing
/// session-relative timestamps in the FSP inner header.
pub(crate) fn mark_established(&mut self, now_ms: u64) {
self.session_start_ms = now_ms;
}
/// Compute a session-relative timestamp for the FSP inner header.
///
/// Returns `(now_ms - session_start_ms)` truncated to u32.
/// Wraps naturally at ~49.7 days, which is fine for relative timing.
pub(crate) fn session_timestamp(&self, now_ms: u64) -> u32 {
now_ms.wrapping_sub(self.session_start_ms) as u32
}
/// Whether this node initiated the Noise IK handshake.
#[cfg_attr(not(test), allow(dead_code))]
pub(crate) fn is_initiator(&self) -> bool {
self.is_initiator
}
/// Get a reference to the session-layer MMP state, if initialized.
pub(crate) fn mmp(&self) -> Option<&MmpSessionState> {
self.mmp.as_ref()
}
/// Get a mutable reference to the session-layer MMP state, if initialized.
pub(crate) fn mmp_mut(&mut self) -> Option<&mut MmpSessionState> {
self.mmp.as_mut()
}
/// Initialize session-layer MMP state (called on Established transition).
pub(crate) fn init_mmp(&mut self, config: &SessionMmpConfig) {
self.mmp = Some(MmpSessionState::new(config, self.is_initiator));
}
}
src/node/session_wire.rs
+582
View File
@@ -0,0 +1,582 @@
//! FSP Wire Format Parsing and Serialization
//!
//! Defines the FIPS session-layer wire format (FSP) for packet dispatch.
//! All FSP messages begin with a 4-byte common prefix followed by phase-specific
//! fields. Encrypted messages use a 12-byte cleartext header as AAD for AEAD,
//! and a 6-byte encrypted inner header containing timestamps and message type.
//!
//! ## Common Prefix (4 bytes)
//!
//! ```text
//! [ver+phase:1][flags:1][payload_len:2 LE]
//! ```
//!
//! ## Message Classes
//!
//! | Phase | U Flag | Type | Description |
//! |-------|--------|------------------|-----------------------------------|
//! | 0x0 | 0 | Encrypted | Post-handshake encrypted data |
//! | 0x0 | 1 | Plaintext error | CoordsRequired, PathBroken |
//! | 0x1 | - | Handshake msg1 | SessionSetup (Noise IK msg1) |
//! | 0x2 | - | Handshake msg2 | SessionAck (Noise IK msg2) |
use crate::protocol::{ProtocolError, decode_optional_coords};
use crate::tree::TreeCoordinate;
// ============================================================================
// Constants
// ============================================================================
/// FSP protocol version (4 high bits of byte 0).
pub const FSP_VERSION: u8 = 0;
/// Phase value for established (encrypted or plaintext error) messages.
pub const FSP_PHASE_ESTABLISHED: u8 = 0x0;
/// Phase value for SessionSetup (Noise IK message 1).
pub const FSP_PHASE_MSG1: u8 = 0x1;
/// Phase value for SessionAck (Noise IK message 2).
pub const FSP_PHASE_MSG2: u8 = 0x2;
/// Size of the common packet prefix (all FSP message types).
pub const FSP_COMMON_PREFIX_SIZE: usize = 4;
/// Size of the full encrypted message header (prefix + counter).
pub const FSP_HEADER_SIZE: usize = 12;
/// Size of the encrypted inner header (timestamp + msg_type + inner_flags).
pub const FSP_INNER_HEADER_SIZE: usize = 6;
/// AEAD authentication tag size (ChaCha20-Poly1305).
const TAG_SIZE: usize = 16;
/// Minimum size for an encrypted FSP message: header + tag (no plaintext).
pub const FSP_ENCRYPTED_MIN_SIZE: usize = FSP_HEADER_SIZE + TAG_SIZE; // 28 bytes
// Cleartext flag bit constants (byte 1 of common prefix, phase 0x0 only).
/// Coords Present — source and destination coordinates follow the header.
pub const FSP_FLAG_CP: u8 = 0x01;
/// Key Epoch — selects active key during rekeying.
#[allow(dead_code)]
pub const FSP_FLAG_K: u8 = 0x02;
/// Unencrypted — payload is plaintext (error signals).
pub const FSP_FLAG_U: u8 = 0x04;
// Inner flag bit constants (byte 5 of decrypted inner header).
/// Spin bit for end-to-end RTT measurement (inside AEAD).
#[allow(dead_code)]
pub const FSP_INNER_FLAG_SP: u8 = 0x01;
// ============================================================================
// Common Prefix
// ============================================================================
/// Parsed FSP common packet prefix (first 4 bytes of every FSP message).
///
/// Wire format:
/// ```text
/// [ver(4bits)+phase(4bits)][flags:1][payload_len:2 LE]
/// ```
#[derive(Clone, Debug)]
pub struct FspCommonPrefix {
/// Protocol version (high nibble of byte 0).
#[cfg_attr(not(test), allow(dead_code))]
pub version: u8,
/// Session lifecycle phase (low nibble of byte 0).
pub phase: u8,
/// Per-message signal flags.
pub flags: u8,
/// Length of payload following the phase-specific header.
#[cfg_attr(not(test), allow(dead_code))]
pub payload_len: u16,
}
impl FspCommonPrefix {
/// Parse a common prefix from the first 4 bytes of FSP message data.
pub fn parse(data: &[u8]) -> Option<Self> {
if data.len() < FSP_COMMON_PREFIX_SIZE {
return None;
}
let version = data[0] >> 4;
let phase = data[0] & 0x0F;
let flags = data[1];
let payload_len = u16::from_le_bytes([data[2], data[3]]);
Some(Self {
version,
phase,
flags,
payload_len,
})
}
/// Check if the Unencrypted flag is set.
pub fn is_unencrypted(&self) -> bool {
self.flags & FSP_FLAG_U != 0
}
/// Check if the Coords Present flag is set.
pub fn has_coords(&self) -> bool {
self.flags & FSP_FLAG_CP != 0
}
/// Encode the ver+phase byte.
fn ver_phase_byte(version: u8, phase: u8) -> u8 {
(version << 4) | (phase & 0x0F)
}
}
// ============================================================================
// Encrypted Message Header
// ============================================================================
/// Parsed FSP encrypted message header (phase 0x0, U flag clear).
///
/// Wire format (12 bytes):
/// ```text
/// [ver+phase:1][flags:1][payload_len:2 LE][counter:8 LE]
/// ```
///
/// The full 12-byte header is used as AAD for the AEAD construction.
/// No receiver_idx — unlike FLP, FSP is end-to-end (dispatched by src_addr
/// from the SessionDatagram envelope, not by index).
#[derive(Clone, Debug)]
pub struct FspEncryptedHeader {
/// Per-message flags (CP, K).
pub flags: u8,
/// Length of encrypted payload (excluding AEAD tag).
#[cfg_attr(not(test), allow(dead_code))]
pub payload_len: u16,
/// Monotonic counter used as AEAD nonce.
pub counter: u64,
/// Raw 12-byte header for use as AEAD AAD.
pub header_bytes: [u8; FSP_HEADER_SIZE],
}
impl FspEncryptedHeader {
/// Parse an encrypted message header from FSP message data.
///
/// Returns None if the data is too short or has wrong version/phase,
/// or if the U flag is set (plaintext messages use a different path).
pub fn parse(data: &[u8]) -> Option<Self> {
if data.len() < FSP_ENCRYPTED_MIN_SIZE {
return None;
}
let version = data[0] >> 4;
let phase = data[0] & 0x0F;
if version != FSP_VERSION || phase != FSP_PHASE_ESTABLISHED {
return None;
}
let flags = data[1];
// U flag means plaintext — not an encrypted message
if flags & FSP_FLAG_U != 0 {
return None;
}
let payload_len = u16::from_le_bytes([data[2], data[3]]);
let counter = u64::from_le_bytes([
data[4], data[5], data[6], data[7],
data[8], data[9], data[10], data[11],
]);
let mut header_bytes = [0u8; FSP_HEADER_SIZE];
header_bytes.copy_from_slice(&data[..FSP_HEADER_SIZE]);
Some(Self {
flags,
payload_len,
counter,
header_bytes,
})
}
/// Check if the Coords Present flag is set.
pub fn has_coords(&self) -> bool {
self.flags & FSP_FLAG_CP != 0
}
/// Offset where ciphertext (or coords if CP) begins in the original data.
#[cfg_attr(not(test), allow(dead_code))]
pub fn data_offset(&self) -> usize {
FSP_HEADER_SIZE
}
}
// ============================================================================
// Serialization Helpers
// ============================================================================
/// Build the 12-byte cleartext header for an encrypted FSP message.
///
/// Returns the header bytes for use as AEAD AAD.
pub fn build_fsp_header(
counter: u64,
flags: u8,
payload_len: u16,
) -> [u8; FSP_HEADER_SIZE] {
let mut header = [0u8; FSP_HEADER_SIZE];
header[0] = FspCommonPrefix::ver_phase_byte(FSP_VERSION, FSP_PHASE_ESTABLISHED);
header[1] = flags;
header[2..4].copy_from_slice(&payload_len.to_le_bytes());
header[4..12].copy_from_slice(&counter.to_le_bytes());
header
}
/// Assemble a wire-format encrypted FSP message.
///
/// Format: `[header:12][ciphertext+tag]`
#[cfg_attr(not(test), allow(dead_code))]
pub fn build_fsp_encrypted(header: &[u8; FSP_HEADER_SIZE], ciphertext: &[u8]) -> Vec<u8> {
let mut packet = Vec::with_capacity(FSP_HEADER_SIZE + ciphertext.len());
packet.extend_from_slice(header);
packet.extend_from_slice(ciphertext);
packet
}
/// Build a 4-byte common prefix for a handshake message.
///
/// `phase` should be `FSP_PHASE_MSG1` or `FSP_PHASE_MSG2`.
/// Flags are zero during handshake.
#[cfg_attr(not(test), allow(dead_code))]
pub fn build_fsp_handshake_prefix(phase: u8, payload_len: u16) -> [u8; FSP_COMMON_PREFIX_SIZE] {
let mut prefix = [0u8; FSP_COMMON_PREFIX_SIZE];
prefix[0] = FspCommonPrefix::ver_phase_byte(FSP_VERSION, phase);
prefix[1] = 0x00; // flags must be zero during handshake
prefix[2..4].copy_from_slice(&payload_len.to_le_bytes());
prefix
}
/// Build a 4-byte common prefix for a plaintext error signal.
///
/// Sets phase 0x0 and U flag.
#[cfg_attr(not(test), allow(dead_code))]
pub fn build_fsp_error_prefix(payload_len: u16) -> [u8; FSP_COMMON_PREFIX_SIZE] {
let mut prefix = [0u8; FSP_COMMON_PREFIX_SIZE];
prefix[0] = FspCommonPrefix::ver_phase_byte(FSP_VERSION, FSP_PHASE_ESTABLISHED);
prefix[1] = FSP_FLAG_U;
prefix[2..4].copy_from_slice(&payload_len.to_le_bytes());
prefix
}
// ============================================================================
// Inner Header Helpers
// ============================================================================
/// Prepend the 6-byte FSP inner header to a message payload.
///
/// Inner header: `[timestamp:4 LE][msg_type:1][inner_flags:1]`
///
/// The caller provides the message-type-specific payload (e.g., application
/// data for msg_type 0x10, report fields for SenderReport). This function
/// prepends the inner header.
pub fn fsp_prepend_inner_header(
timestamp_ms: u32,
msg_type: u8,
inner_flags: u8,
payload: &[u8],
) -> Vec<u8> {
let mut buf = Vec::with_capacity(FSP_INNER_HEADER_SIZE + payload.len());
buf.extend_from_slice(&timestamp_ms.to_le_bytes());
buf.push(msg_type);
buf.push(inner_flags);
buf.extend_from_slice(payload);
buf
}
/// Strip the 6-byte FSP inner header from a decrypted payload.
///
/// Returns `(timestamp, msg_type, inner_flags, &rest)` or None if too short.
pub fn fsp_strip_inner_header(plaintext: &[u8]) -> Option<(u32, u8, u8, &[u8])> {
if plaintext.len() < FSP_INNER_HEADER_SIZE {
return None;
}
let timestamp = u32::from_le_bytes([
plaintext[0], plaintext[1], plaintext[2], plaintext[3],
]);
let msg_type = plaintext[4];
let inner_flags = plaintext[5];
Some((timestamp, msg_type, inner_flags, &plaintext[FSP_INNER_HEADER_SIZE..]))
}
// ============================================================================
// Coordinate Parsing (for transit nodes and receive path)
// ============================================================================
/// Parse source and destination coordinates from the cleartext section
/// of an encrypted FSP message when the CP flag is set.
///
/// Coordinates appear between the 12-byte header and the ciphertext:
/// `[src_coords_count:2 LE][src_coords:16×n][dest_coords_count:2 LE][dest_coords:16×m]`
///
/// Returns `(src_coords, dest_coords, bytes_consumed)`.
pub fn parse_encrypted_coords(
data: &[u8],
) -> Result<(Option<TreeCoordinate>, Option<TreeCoordinate>, usize), ProtocolError> {
let (src_coords, src_consumed) = decode_optional_coords(data)?;
let (dest_coords, dest_consumed) = decode_optional_coords(&data[src_consumed..])?;
Ok((src_coords, dest_coords, src_consumed + dest_consumed))
}
// ============================================================================
// Tests
// ============================================================================
#[cfg(test)]
mod tests {
use super::*;
// ===== Size Constant Tests =====
#[test]
fn test_wire_sizes() {
assert_eq!(FSP_COMMON_PREFIX_SIZE, 4);
assert_eq!(FSP_HEADER_SIZE, 12);
assert_eq!(FSP_INNER_HEADER_SIZE, 6);
assert_eq!(FSP_ENCRYPTED_MIN_SIZE, 28); // 12 + 16
}
// ===== Common Prefix Tests =====
#[test]
fn test_common_prefix_parse_established() {
let data = [0x00, 0x01, 0x40, 0x00]; // ver=0, phase=0, flags=CP, payload_len=64
let prefix = FspCommonPrefix::parse(&data).unwrap();
assert_eq!(prefix.version, 0);
assert_eq!(prefix.phase, FSP_PHASE_ESTABLISHED);
assert_eq!(prefix.flags, FSP_FLAG_CP);
assert_eq!(prefix.payload_len, 64);
assert!(prefix.has_coords());
assert!(!prefix.is_unencrypted());
}
#[test]
fn test_common_prefix_parse_handshake() {
let data = [0x01, 0x00, 0x50, 0x00]; // ver=0, phase=1, flags=0, payload_len=80
let prefix = FspCommonPrefix::parse(&data).unwrap();
assert_eq!(prefix.version, 0);
assert_eq!(prefix.phase, FSP_PHASE_MSG1);
assert_eq!(prefix.flags, 0);
assert_eq!(prefix.payload_len, 80);
}
#[test]
fn test_common_prefix_parse_error_signal() {
let data = [0x00, FSP_FLAG_U, 0x22, 0x00]; // ver=0, phase=0, U flag, payload_len=34
let prefix = FspCommonPrefix::parse(&data).unwrap();
assert_eq!(prefix.phase, FSP_PHASE_ESTABLISHED);
assert!(prefix.is_unencrypted());
assert_eq!(prefix.payload_len, 34);
}
#[test]
fn test_common_prefix_too_short() {
assert!(FspCommonPrefix::parse(&[0, 0, 0]).is_none());
}
// ===== Encrypted Header Tests =====
#[test]
fn test_encrypted_header_parse() {
let counter = 42u64;
let flags = FSP_FLAG_CP;
let payload_len = 100u16;
let header = build_fsp_header(counter, flags, payload_len);
// Build a minimal packet: header + 16 bytes of fake ciphertext (tag)
let mut packet = Vec::from(header);
packet.extend_from_slice(&[0xaa; TAG_SIZE]);
let parsed = FspEncryptedHeader::parse(&packet).unwrap();
assert_eq!(parsed.counter, 42);
assert_eq!(parsed.flags, FSP_FLAG_CP);
assert_eq!(parsed.payload_len, 100);
assert!(parsed.has_coords());
assert_eq!(parsed.header_bytes, header);
assert_eq!(parsed.data_offset(), FSP_HEADER_SIZE);
}
#[test]
fn test_encrypted_header_too_short() {
let packet = vec![0x00; FSP_ENCRYPTED_MIN_SIZE - 1];
assert!(FspEncryptedHeader::parse(&packet).is_none());
}
#[test]
fn test_encrypted_header_wrong_phase() {
let mut packet = vec![0x00; FSP_ENCRYPTED_MIN_SIZE];
packet[0] = 0x01; // phase 1 (msg1), not established
assert!(FspEncryptedHeader::parse(&packet).is_none());
}
#[test]
fn test_encrypted_header_wrong_version() {
let mut packet = vec![0x00; FSP_ENCRYPTED_MIN_SIZE];
packet[0] = 0x10; // version 1, phase 0
assert!(FspEncryptedHeader::parse(&packet).is_none());
}
#[test]
fn test_encrypted_header_u_flag_rejected() {
let mut packet = vec![0x00; FSP_ENCRYPTED_MIN_SIZE];
packet[1] = FSP_FLAG_U; // U flag set → not encrypted
assert!(FspEncryptedHeader::parse(&packet).is_none());
}
// ===== Build Header Tests =====
#[test]
fn test_build_fsp_header() {
let header = build_fsp_header(1000, FSP_FLAG_CP, 200);
assert_eq!(header[0], 0x00); // ver=0, phase=0
assert_eq!(header[1], FSP_FLAG_CP);
assert_eq!(u16::from_le_bytes([header[2], header[3]]), 200);
assert_eq!(
u64::from_le_bytes([
header[4], header[5], header[6], header[7],
header[8], header[9], header[10], header[11],
]),
1000
);
}
#[test]
fn test_build_fsp_encrypted() {
let header = build_fsp_header(0, 0, 10);
let ciphertext = vec![0xCC; 26]; // 10 payload + 16 tag
let packet = build_fsp_encrypted(&header, &ciphertext);
assert_eq!(packet.len(), FSP_HEADER_SIZE + 26);
assert_eq!(&packet[..FSP_HEADER_SIZE], &header);
assert_eq!(&packet[FSP_HEADER_SIZE..], &ciphertext[..]);
}
// ===== Handshake Prefix Tests =====
#[test]
fn test_build_fsp_handshake_prefix_msg1() {
let prefix = build_fsp_handshake_prefix(FSP_PHASE_MSG1, 100);
assert_eq!(prefix[0], 0x01); // ver=0, phase=1
assert_eq!(prefix[1], 0x00); // flags zero
assert_eq!(u16::from_le_bytes([prefix[2], prefix[3]]), 100);
let parsed = FspCommonPrefix::parse(&prefix).unwrap();
assert_eq!(parsed.phase, FSP_PHASE_MSG1);
}
#[test]
fn test_build_fsp_handshake_prefix_msg2() {
let prefix = build_fsp_handshake_prefix(FSP_PHASE_MSG2, 50);
assert_eq!(prefix[0], 0x02); // ver=0, phase=2
assert_eq!(prefix[1], 0x00);
assert_eq!(u16::from_le_bytes([prefix[2], prefix[3]]), 50);
}
// ===== Error Prefix Tests =====
#[test]
fn test_build_fsp_error_prefix() {
let prefix = build_fsp_error_prefix(34);
assert_eq!(prefix[0], 0x00); // ver=0, phase=0
assert_eq!(prefix[1], FSP_FLAG_U);
assert_eq!(u16::from_le_bytes([prefix[2], prefix[3]]), 34);
let parsed = FspCommonPrefix::parse(&prefix).unwrap();
assert!(parsed.is_unencrypted());
assert_eq!(parsed.phase, FSP_PHASE_ESTABLISHED);
}
// ===== Inner Header Tests =====
#[test]
fn test_inner_header_prepend_strip() {
let timestamp: u32 = 12345;
let msg_type: u8 = 0x10;
let inner_flags: u8 = 0x01; // SP bit
let payload = vec![0xAA, 0xBB, 0xCC];
let with_header = fsp_prepend_inner_header(timestamp, msg_type, inner_flags, &payload);
assert_eq!(with_header.len(), FSP_INNER_HEADER_SIZE + 3);
let (ts, mt, flags, rest) = fsp_strip_inner_header(&with_header).unwrap();
assert_eq!(ts, 12345);
assert_eq!(mt, 0x10);
assert_eq!(flags, 0x01);
assert_eq!(rest, &payload[..]);
}
#[test]
fn test_inner_header_empty_payload() {
let with_header = fsp_prepend_inner_header(0, 0x13, 0, &[]);
assert_eq!(with_header.len(), FSP_INNER_HEADER_SIZE);
let (ts, mt, flags, rest) = fsp_strip_inner_header(&with_header).unwrap();
assert_eq!(ts, 0);
assert_eq!(mt, 0x13);
assert_eq!(flags, 0);
assert!(rest.is_empty());
}
#[test]
fn test_inner_header_too_short() {
assert!(fsp_strip_inner_header(&[0, 0, 0, 0, 0]).is_none()); // needs 6 bytes
assert!(fsp_strip_inner_header(&[]).is_none());
}
// ===== Flag Constants Tests =====
#[test]
fn test_flag_bits_distinct() {
// Cleartext flags don't overlap
assert_eq!(FSP_FLAG_CP & FSP_FLAG_K, 0);
assert_eq!(FSP_FLAG_CP & FSP_FLAG_U, 0);
assert_eq!(FSP_FLAG_K & FSP_FLAG_U, 0);
}
#[test]
fn test_header_roundtrip() {
let counter = 0xDEADBEEF_12345678u64;
let flags = FSP_FLAG_CP | FSP_FLAG_K;
let payload_len = 1234u16;
let header = build_fsp_header(counter, flags, payload_len);
let ciphertext = vec![0xFF; payload_len as usize + TAG_SIZE];
let packet = build_fsp_encrypted(&header, &ciphertext);
let parsed = FspEncryptedHeader::parse(&packet).unwrap();
assert_eq!(parsed.counter, counter);
assert_eq!(parsed.flags, flags);
assert_eq!(parsed.payload_len, payload_len);
assert!(parsed.has_coords());
assert_eq!(parsed.header_bytes, header);
}
#[test]
fn test_all_message_types_through_prefix() {
// Encrypted (phase 0, no U)
let prefix = FspCommonPrefix::parse(&[0x00, 0x00, 0x10, 0x00]).unwrap();
assert_eq!(prefix.phase, 0);
assert!(!prefix.is_unencrypted());
// Error signal (phase 0, U set)
let prefix = FspCommonPrefix::parse(&[0x00, FSP_FLAG_U, 0x22, 0x00]).unwrap();
assert_eq!(prefix.phase, 0);
assert!(prefix.is_unencrypted());
// SessionSetup (phase 1)
let prefix = FspCommonPrefix::parse(&[0x01, 0x00, 0x50, 0x00]).unwrap();
assert_eq!(prefix.phase, 1);
// SessionAck (phase 2)
let prefix = FspCommonPrefix::parse(&[0x02, 0x00, 0x21, 0x00]).unwrap();
assert_eq!(prefix.phase, 2);
}
}
src/node/tests/forwarding.rs
+20 -13
View File
@@ -5,7 +5,8 @@
//! multi-hop forwarding through live node topologies.
use super::*;
use crate::protocol::{DataPacket, SessionAck, SessionDatagram, SessionSetup};
use crate::node::session_wire::{build_fsp_header, FSP_FLAG_CP};
use crate::protocol::{SessionAck, SessionDatagram, SessionSetup, encode_coords};
use crate::tree::TreeCoordinate;
use spanning_tree::{
cleanup_nodes, process_available_packets, run_tree_test, verify_tree_convergence,
@@ -184,7 +185,7 @@ async fn test_coord_cache_warming_session_ack() {
}
#[tokio::test]
async fn test_coord_cache_warming_data_packet_with_coords() {
async fn test_coord_cache_warming_encrypted_msg_with_coords() {
let mut node = make_node();
let from = make_node_addr(0xAA);
let src_addr = make_node_addr(0x01);
@@ -194,9 +195,13 @@ async fn test_coord_cache_warming_data_packet_with_coords() {
let src_coords = TreeCoordinate::from_addrs(vec![src_addr, root_addr]).unwrap();
let dest_coords = TreeCoordinate::from_addrs(vec![dest_addr, root_addr]).unwrap();
let data = DataPacket::new(0, vec![1, 2, 3, 4])
.with_coords(src_coords.clone(), dest_coords.clone());
let data_payload = data.encode();
// Build FSP encrypted message with CP flag: header(12) + coords + fake_ciphertext
let header = build_fsp_header(0, FSP_FLAG_CP, 20);
let mut data_payload = Vec::new();
data_payload.extend_from_slice(&header);
encode_coords(&src_coords, &mut data_payload);
encode_coords(&dest_coords, &mut data_payload);
data_payload.extend_from_slice(&[0xCC; 36]); // fake ciphertext (20 payload + 16 tag)
let dg = SessionDatagram::new(src_addr, dest_addr, data_payload);
let encoded = dg.encode();
@@ -213,24 +218,26 @@ async fn test_coord_cache_warming_data_packet_with_coords() {
assert!(
node.coord_cache().get(&src_addr, now_ms).is_some(),
"src coords not cached from DataPacket"
"src coords not cached from encrypted message"
);
assert!(
node.coord_cache().get(&dest_addr, now_ms).is_some(),
"dest coords not cached from DataPacket"
"dest coords not cached from encrypted message"
);
}
#[tokio::test]
async fn test_coord_cache_warming_opaque_data_packet() {
async fn test_coord_cache_warming_encrypted_msg_no_coords() {
let mut node = make_node();
let from = make_node_addr(0xAA);
let src_addr = make_node_addr(0x01);
let dest_addr = make_node_addr(0x02);
// DataPacket without COORDS_PRESENT — no coords to cache
let data = DataPacket::new(0, vec![1, 2, 3, 4]);
let data_payload = data.encode();
// Build FSP encrypted message without CP flag: header(12) + fake_ciphertext
let header = build_fsp_header(0, 0, 20);
let mut data_payload = Vec::new();
data_payload.extend_from_slice(&header);
data_payload.extend_from_slice(&[0xCC; 36]); // fake ciphertext (20 payload + 16 tag)
let dg = SessionDatagram::new(src_addr, dest_addr, data_payload);
let encoded = dg.encode();
@@ -244,11 +251,11 @@ async fn test_coord_cache_warming_opaque_data_packet() {
assert!(
node.coord_cache().get(&src_addr, now_ms).is_none(),
"Should not cache coords from opaque DataPacket"
"Should not cache coords from message without CP flag"
);
assert!(
node.coord_cache().get(&dest_addr, now_ms).is_none(),
"Should not cache coords from opaque DataPacket"
"Should not cache coords from message without CP flag"
);
}
src/node/tests/session.rs
+14 -3
View File
@@ -60,6 +60,7 @@ fn test_session_entry_new_initiating() {
identity_b.pubkey_full(),
EndToEndState::Initiating(handshake),
1000,
true,
);
assert!(entry.state().is_initiating());
@@ -86,6 +87,7 @@ fn test_session_entry_touch() {
identity_b.pubkey_full(),
EndToEndState::Initiating(handshake),
1000,
true,
);
entry.touch(2000);
@@ -111,6 +113,7 @@ fn test_session_table_operations() {
identity_b.pubkey_full(),
EndToEndState::Initiating(handshake),
1000,
true,
);
node.sessions.insert(dest_addr, entry);
@@ -223,10 +226,10 @@ async fn test_session_direct_peer_data_transfer() {
.await
.expect("send_session_data failed");
// Process packets: DataPacket arrives at Node 1
// Process packets: encrypted data arrives at Node 1
tokio::time::sleep(Duration::from_millis(20)).await;
let count = process_available_packets(&mut nodes).await;
assert!(count > 0, "Expected DataPacket to arrive");
assert!(count > 0, "Expected encrypted data to arrive");
// Node 1's session should now be Established (was Responding, transitions on first data)
assert!(nodes[1]
@@ -958,7 +961,7 @@ async fn test_tun_outbound_established_session() {
nodes[0].node.handle_tun_outbound(ipv6_packet.clone()).await;
// Process packets: encrypted DataPacket → Node 1
// Process packets: encrypted data → Node 1
tokio::time::sleep(Duration::from_millis(20)).await;
process_available_packets(&mut nodes).await;
@@ -1176,6 +1179,7 @@ fn test_purge_idle_sessions_removes_expired() {
remote.pubkey_full(),
EndToEndState::Established(session),
1000, // created at t=1000ms
true,
);
node.sessions.insert(remote_addr, entry);
@@ -1201,6 +1205,7 @@ fn test_purge_idle_sessions_keeps_active() {
remote.pubkey_full(),
EndToEndState::Established(session),
1000,
true,
);
// Touch at t=80s — recent activity
@@ -1232,6 +1237,7 @@ fn test_purge_idle_sessions_ignores_initiating() {
remote.pubkey_full(),
EndToEndState::Initiating(handshake),
1000,
true,
);
node.sessions.insert(remote_addr, entry);
@@ -1255,6 +1261,7 @@ fn test_purge_idle_sessions_cleans_pending_packets() {
remote.pubkey_full(),
EndToEndState::Established(session),
1000,
true,
);
node.sessions.insert(remote_addr, entry);
@@ -1288,6 +1295,7 @@ fn test_purge_idle_sessions_disabled_when_zero() {
remote.pubkey_full(),
EndToEndState::Established(session),
1000,
true,
);
node.sessions.insert(remote_addr, entry);
@@ -1320,6 +1328,7 @@ fn test_coords_warmup_counter_default_zero_on_new() {
identity_b.pubkey_full(),
EndToEndState::Initiating(handshake),
1000,
true,
);
assert_eq!(entry.coords_warmup_remaining(), 0,
@@ -1338,6 +1347,7 @@ fn test_coords_warmup_counter_set_and_get() {
remote.pubkey_full(),
EndToEndState::Established(session),
1000,
true,
);
assert_eq!(entry.coords_warmup_remaining(), 0);
@@ -1361,6 +1371,7 @@ fn test_coords_warmup_counter_decrement() {
remote.pubkey_full(),
EndToEndState::Established(session),
1000,
true,
);
entry.set_coords_warmup_remaining(3);