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

proto/fsp: extract FSP session protocol into sans-IO layout, retire src/protocol

Browse Source 
Migrate the FSP end-to-end session subsystem into src/proto/fsp/ following the
established sans-IO shape, and retire the src/protocol grab-bag now that FSP was
its last occupant.

Relocate the FSP session wire (node/session_wire.rs plus the FSP message types
from protocol/session.rs) into proto/fsp/wire.rs. Hoist the pure decision logic
into proto/fsp/core.rs over plain-data SessionSnapshots returning an ordered
FspAction list the shell drives: session-rekey policy, msg3-resend
classification, post-decrypt epoch reaction, setup/dual-init tie-break,
coords/path-MTU emit-policy, bounded pending-queue, and IPv6 ECN. The
crypto-owning SessionEntry stays shell-side in node/session.rs (matching the FMP
ActivePeer pattern); proto/fsp is wire + core + limits only, with no proto->noise
dependency and no crypto.

Move the coords helpers to proto/stp/ (they serialize TreeCoordinate), and split
SessionMessageType: the encrypted-inner 0x10-0x1F variants stay in proto/fsp/wire.rs
while the 0x20-0x2F routing signals become a new RoutingSignalType in
proto/routing/wire.rs. Migrate the session-MMP shell adapter, which continues to
drive proto/mmp/.

Retire src/protocol: LinkMessageType and SessionDatagram move to a new shared
proto/link.rs, ProtocolError becomes proto::Error (relocated verbatim), the
deprecated MessageType alias and the unimported PROTOCOL_VERSION are dropped, and
src/protocol/ is deleted along with its lib.rs module declaration.

Behavior-neutral: wire bytes unchanged, oracle tests pass unedited except
mod-path relocation; adds rekey/epoch characterization tests and pure
poll/emit-policy core tests.
This commit is contained in:
Johnathan Corgan
2026-07-08 04:57:00 +00:00
parent 4ed674ea8b
commit 4ad5940114
37 changed files with 2848 additions and 1741 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/node/encrypt_worker.rs
+5 -3
View File
@@ -50,8 +50,8 @@
// warnings rather than gate every function individually.
#![cfg_attr(not(unix), allow(dead_code))]
use crate::node::session_wire::FSP_HEADER_SIZE;
use crate::node::wire::ESTABLISHED_HEADER_SIZE;
use crate::proto::fsp::wire::FSP_HEADER_SIZE;
use crate::transport::udp::socket::AsyncUdpSocket;
#[cfg(not(target_os = "macos"))]
use crossbeam_channel::{Receiver, SendError, Sender, TrySendError, bounded};
@@ -1904,10 +1904,12 @@ mod unix_tests {
#[test]
fn pipelined_send_wire_layout_roundtrips_canonical_decoders() {
use crate::NodeAddr;
use crate::node::session_wire::build_fsp_header;
use crate::node::wire::{EncryptedHeader, FLAG_KEY_EPOCH, build_established_header};
use crate::noise::TAG_SIZE;
use crate::protocol::{LinkMessageType, SESSION_DATAGRAM_HEADER_SIZE, SessionDatagramRef};
use crate::proto::fsp::wire::build_fsp_header;
use crate::proto::link::{
LinkMessageType, SESSION_DATAGRAM_HEADER_SIZE, SessionDatagramRef,
};
use crate::utils::index::SessionIndex;
let rt = tokio::runtime::Builder::new_current_thread()
src/node/handlers/forwarding.rs
+4 -3
View File
@@ -7,13 +7,14 @@
use crate::NodeAddr;
use crate::node::reject::ForwardingReject;
use crate::node::session_wire::{
use crate::node::{Node, NodeError, NodeRoutingView};
use crate::proto::fsp::wire::{
FSP_COMMON_PREFIX_SIZE, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED, FSP_PHASE_MSG1, FSP_PHASE_MSG2,
FspCommonPrefix, parse_encrypted_coords,
};
use crate::node::{Node, NodeError, NodeRoutingView};
use crate::proto::fsp::{SessionAck, SessionSetup};
use crate::proto::link::{SessionDatagram, SessionDatagramRef};
use crate::proto::routing::{DropReason, NextHop, RouteAction, RouteOutcome};
use crate::protocol::{SessionAck, SessionDatagram, SessionDatagramRef, SessionSetup};
use std::time::{Duration, Instant};
use tracing::{debug, warn};
src/node/handlers/mmp.rs
+4 -222
View File
@@ -8,12 +8,11 @@ use crate::NodeAddr;
use crate::node::Node;
use crate::node::reject::{MmpReject, RejectReason, TreeReject};
use crate::node::tree::sign_declaration;
use crate::proto::link::LinkMessageType;
use crate::proto::mmp::{
BackoffUpdate, LinkReportKind, LinkReportSnapshot, MmpAction, MmpSessionState,
PathMtuNotification, PeerLivenessSnapshot, ReceiverReport, RrLog, SendResult, SenderReport,
SessionReceiverReport, SessionReportKind, SessionReportSnapshot, SessionSenderReport,
LinkReportKind, LinkReportSnapshot, MmpAction, PeerLivenessSnapshot, ReceiverReport, RrLog,
SenderReport,
};
use crate::protocol::{LinkMessageType, SessionMessageType};
use std::time::{Duration, Instant};
use tracing::{debug, info, trace, warn};
@@ -58,7 +57,7 @@ pub(super) fn log_rr_outcome(rr: &ReceiverReport, our_timestamp_ms: u32, log: Rr
}
/// Format bytes/sec as human-readable throughput.
fn format_throughput(bps: f64) -> String {
pub(in crate::node) fn format_throughput(bps: f64) -> String {
if bps == 0.0 {
"n/a".to_string()
} else if bps >= 1_000_000.0 {
@@ -418,223 +417,6 @@ impl Node {
);
}
// === 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_ms = crate::mmp::mono_ms();
// Build one report-gating snapshot per session, resolving every timing
// read shell-side into a `bool`. The snapshots own only
// `NodeAddr`/`MmpMode`/`bool`, so the session-iteration borrow is released
// before the pure decision runs and the driving loop mutates the
// reporting state / performs the sends.
let snapshots: Vec<SessionReportSnapshot> = self
.sessions
.iter()
.filter_map(|(dest_addr, entry)| {
let mmp = entry.mmp()?;
Some(SessionReportSnapshot {
dest: *dest_addr,
mode: mmp.mode(),
sr_due: mmp.sender.should_send_report(now_ms),
rr_due: mmp.receiver.should_send_report(now_ms),
mtu_due: mmp.path_mtu.should_send_notification(now_ms),
log_due: mmp.should_log(now_ms),
})
})
.collect();
let actions = self.mmp.plan_session_reports(&snapshots);
// Drive the planned actions in phase-grouped order (all logs, then the
// sends in per-session SR/RR/MTU order). Logs run first because the
// session operator log reads cumulative_packets_sent, which each send
// advances (send_session_msg -> sender.record_sent); the pre-refactor
// handler logged during its collect pass, before any send. Each build
// (`build_report`/`build_notification`, which advance interval/
// notification state) runs only on its SendSessionReport action, exactly
// as the pre-refactor collect pass did. Per-destination success/failure
// is collected for the backoff dedup + failure-log suppression.
let mut send_results: Vec<SendResult> = Vec::new();
for action in actions {
match action {
MmpAction::LogSession { dest } => {
// Resolve the display name exactly as the pre-refactor loop
// did (alias, else short_npub from the session's remote key).
let session_name = self.peer_aliases.get(&dest).cloned().unwrap_or_else(|| {
self.sessions
.get(&dest)
.map(|entry| {
let (xonly, _) = entry.remote_pubkey().x_only_public_key();
crate::PeerIdentity::from_pubkey(xonly).short_npub()
})
.unwrap_or_default()
});
if let Some(mmp) = self.sessions.get_mut(&dest).and_then(|e| e.mmp_mut()) {
Self::log_session_mmp_metrics(&session_name, mmp);
mmp.mark_logged(now_ms);
}
}
MmpAction::SendSessionReport { dest, kind } => {
let built = self
.sessions
.get_mut(&dest)
.and_then(|entry| entry.mmp_mut())
.and_then(|mmp| match kind {
SessionReportKind::Sender => {
mmp.sender.build_report(now_ms).map(|sr| {
(
SessionMessageType::SenderReport.to_byte(),
SessionSenderReport::from(&sr).encode(),
)
})
}
SessionReportKind::Receiver => {
mmp.receiver.build_report(now_ms).map(|rr| {
(
SessionMessageType::ReceiverReport.to_byte(),
SessionReceiverReport::from(&rr).encode(),
)
})
}
SessionReportKind::PathMtu => {
mmp.path_mtu.build_notification(now_ms).map(|mtu_value| {
(
SessionMessageType::PathMtuNotification.to_byte(),
PathMtuNotification::new(mtu_value).encode(),
)
})
}
});
let Some((msg_type, body)) = built else {
continue;
};
match self.send_session_msg(&dest, msg_type, &body).await {
Ok(()) => send_results.push(SendResult { dest, ok: true }),
Err(e) => {
// Peek at current failure count for log suppression
// (unchanged by the backoff apply, which runs later).
let failures = self
.sessions
.get(&dest)
.and_then(|entry| entry.mmp())
.map(|mmp| mmp.sender.consecutive_send_failures())
.unwrap_or(0);
if failures < 3 {
debug!(
dest = %self.peer_display_name(&dest),
msg_type,
error = %e,
"Failed to send session MMP report"
);
} else if failures == 3 {
debug!(
dest = %self.peer_display_name(&dest),
"Suppressing further session MMP send failure logs"
);
}
// failures > 3: silently suppressed
send_results.push(SendResult { dest, ok: false });
}
}
}
MmpAction::ReapPeer { .. }
| MmpAction::Heartbeat { .. }
| MmpAction::SendLinkReport { .. }
| MmpAction::LogLink { .. } => {}
}
}
// Deduplicate send results per destination (any-ok -> success, all-fail
// -> failure) and apply the backoff state transition for each dest.
for update in self.mmp.plan_backoff(&send_results) {
match update {
BackoffUpdate::Success { dest } => {
if let Some(mmp) = self.sessions.get_mut(&dest).and_then(|e| e.mmp_mut()) {
let prev = mmp.sender.record_send_success();
if prev > 3 {
debug!(
dest = %self.peer_display_name(&dest),
consecutive_failures = prev,
"Resumed session MMP reporting"
);
}
}
}
BackoffUpdate::Failure { dest } => {
if let Some(mmp) = self.sessions.get_mut(&dest).and_then(|e| e.mmp_mut()) {
mmp.sender.record_send_failure();
}
}
}
}
}
/// Emit periodic session MMP metrics.
fn log_session_mmp_metrics(session_name: &str, mmp: &MmpSessionState) {
let m = &mmp.metrics;
let rtt_str = if m.rtt_trend.initialized() {
format!("{:.1}ms", m.rtt_trend.long() / 1000.0)
} else {
"n/a".to_string()
};
let loss_str = if m.loss_trend.initialized() {
format!("{:.1}%", m.loss_trend.long() * 100.0)
} else {
"n/a".to_string()
};
let jitter_ms = mmp.receiver.jitter_us() as f64 / 1000.0;
debug!(
session = %session_name,
rtt = %rtt_str,
loss = %loss_str,
jitter = format_args!("{:.1}ms", jitter_ms),
goodput = %format_throughput(m.goodput_bps()),
mtu = mmp.path_mtu.last_observed_mtu(),
tx_pkts = mmp.sender.cumulative_packets_sent(),
rx_pkts = mmp.receiver.cumulative_packets_recv(),
"MMP session metrics"
);
}
/// Emit a teardown log summarizing lifetime session MMP metrics.
pub(in crate::node) fn log_session_mmp_teardown(session_name: &str, mmp: &MmpSessionState) {
let m = &mmp.metrics;
let jitter_ms = mmp.receiver.jitter_us() as f64 / 1000.0;
let rtt_str = match m.srtt_ms() {
Some(rtt) => format!("{:.1}ms", rtt),
None => "n/a".to_string(),
};
let loss_str = format!("{:.1}%", m.loss_rate() * 100.0);
debug!(
session = %session_name,
rtt = %rtt_str,
loss = %loss_str,
jitter = format_args!("{:.1}ms", jitter_ms),
etx = format_args!("{:.2}", m.etx),
goodput = %format_throughput(m.goodput_bps()),
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(),
"MMP session teardown"
);
}
/// Send heartbeats and remove dead peers.
///
/// Called from the tick handler. Sends a 1-byte heartbeat to each peer
src/node/handlers/rekey.rs
+131 -153
View File
@@ -10,26 +10,22 @@ use crate::node::Node;
use crate::node::wire::build_msg1;
use crate::noise::HandshakeState;
use crate::proto::fmp::{ConnAction, LifecycleView, PeerSnapshot, RekeyCfg, RekeyResendSnapshot};
use crate::protocol::{SessionDatagram, SessionSetup};
use crate::proto::fsp::{
FspAction, RekeyMsg3ResendSnapshot, SessionSetup, SessionSnapshot, cutover_timer_elapsed,
};
use crate::proto::link::SessionDatagram;
use tracing::{debug, trace, warn};
/// Keep previous session alive for this long after cutover.
///
/// FMP-scoped copy for `check_rekey`; the FSP session-rekey timing bounds live
/// in `crate::proto::fsp::limits`.
const DRAIN_WINDOW_SECS: u64 = 10;
/// Suppress local rekey initiation for this long after receiving
/// a peer's rekey msg1.
/// a peer's rekey msg1. FMP-scoped copy for `check_rekey`.
const REKEY_DAMPENING_SECS: u64 = 30;
/// Liveness bound on how long the FSP rekey initiator holds the
/// `current` + `pending` state before cutting over to the new epoch.
///
/// This is NOT safety-critical: overlapping-epoch trial-decrypt covers
/// any skew between the two endpoints' cutovers. The timer only bounds
/// how long the initiator advertises the old K-bit. An opportunistic
/// early cutover also fires if the initiator authenticates a peer frame
/// against its own `pending` session (the responder cut over first).
const FSP_CUTOVER_DELAY_MS: u64 = 2000;
impl Node {
/// Periodic rekey check. Called from the tick loop.
///
@@ -358,66 +354,77 @@ impl Node {
let ttl = self.config().node.session.default_ttl;
let my_addr = *self.node_addr();
// Collect rekey initiators whose msg3 retransmission is due.
let mut to_resend: Vec<(NodeAddr, Vec<u8>)> = Vec::new();
let mut to_abandon: Vec<NodeAddr> = Vec::new();
for (node_addr, entry) in &self.sessions {
// Only the rekey initiator retains a msg3 payload.
let payload = match entry.rekey_msg3_payload() {
Some(p) => p,
None => continue,
};
if entry.rekey_msg3_next_resend_ms() == 0 || now_ms < entry.rekey_msg3_next_resend_ms()
{
continue;
}
if entry.rekey_msg3_resend_count() >= max_resends {
to_abandon.push(*node_addr);
continue;
}
to_resend.push((*node_addr, payload.to_vec()));
}
// Abandon rekey cycles that exhausted their retransmission budget.
for node_addr in to_abandon {
if let Some(entry) = self.sessions.get_mut(&node_addr) {
entry.abandon_rekey();
}
debug!(
peer = %self.peer_display_name(&node_addr),
"FSP rekey aborted: msg3 unconfirmed after max retransmissions, abandoning cycle"
);
}
// Retransmit msg3 for cycles still within budget.
for (node_addr, payload) in to_resend {
let mut datagram = SessionDatagram::new(my_addr, node_addr, payload).with_ttl(ttl);
let sent = match self.send_session_datagram(&mut datagram).await {
Ok(_) => true,
Err(e) => {
// The shell snapshots each session retaining a msg3 payload (resend-due
// predicate resolved here); the core classifies abandon-vs-resend,
// abandons first.
let candidates = self.rekey_msg3_resend_snapshots(now_ms);
for action in self.fsp.poll_rekey_msg3_resends(candidates, max_resends) {
match action {
FspAction::AbandonRekey { addr } => {
if let Some(entry) = self.sessions.get_mut(&addr) {
entry.abandon_rekey();
}
debug!(
peer = %self.peer_display_name(&node_addr),
error = %e,
"FSP rekey msg3 retransmission failed"
peer = %self.peer_display_name(&addr),
"FSP rekey aborted: msg3 unconfirmed after max retransmissions, abandoning cycle"
);
false
}
};
FspAction::ResendSessionMsg3 { addr } => {
let payload = match self
.sessions
.get(&addr)
.and_then(|e| e.rekey_msg3_payload())
{
Some(p) => p.to_vec(),
None => continue,
};
let mut datagram = SessionDatagram::new(my_addr, addr, payload).with_ttl(ttl);
let sent = match self.send_session_datagram(&mut datagram).await {
Ok(_) => true,
Err(e) => {
debug!(
peer = %self.peer_display_name(&addr),
error = %e,
"FSP rekey msg3 retransmission failed"
);
false
}
};
if sent && let Some(entry) = self.sessions.get_mut(&node_addr) {
let count = entry.rekey_msg3_resend_count() + 1;
let next = now_ms + (interval_ms as f64 * backoff.powi(count as i32)) as u64;
entry.record_rekey_msg3_resend(next);
trace!(
peer = %self.peer_display_name(&node_addr),
resend = count,
"Resent FSP rekey msg3"
);
if sent && let Some(entry) = self.sessions.get_mut(&addr) {
let count = entry.rekey_msg3_resend_count() + 1;
let next =
now_ms + (interval_ms as f64 * backoff.powi(count as i32)) as u64;
entry.record_rekey_msg3_resend(next);
trace!(
peer = %self.peer_display_name(&addr),
resend = count,
"Resent FSP rekey msg3"
);
}
}
#[allow(unreachable_patterns)]
_ => {}
}
}
}
/// Snapshot every session retaining a rekey-msg3 payload for the
/// retransmission decision, pre-evaluating the resend-due predicate against
/// `now_ms` so the core reads no clock.
fn rekey_msg3_resend_snapshots(&self, now_ms: u64) -> Vec<RekeyMsg3ResendSnapshot> {
self.sessions
.iter()
.filter(|(_, entry)| entry.rekey_msg3_payload().is_some())
.map(|(node_addr, entry)| RekeyMsg3ResendSnapshot {
addr: *node_addr,
resend_count: entry.rekey_msg3_resend_count(),
resend_due: entry.rekey_msg3_next_resend_ms() != 0
&& now_ms >= entry.rekey_msg3_next_resend_ms(),
})
.collect()
}
/// Periodic session (FSP) rekey check. Called from the tick loop.
///
/// For each established session:
@@ -435,100 +442,71 @@ impl Node {
return;
}
let rekey_after_secs = self.config().node.rekey.after_secs;
let rekey_after_messages = self.config().node.rekey.after_messages;
let cfg = crate::proto::fsp::RekeyCfg {
after_secs: self.config().node.rekey.after_secs,
after_messages: self.config().node.rekey.after_messages,
};
let now_ms = Self::now_ms();
let drain_ms = DRAIN_WINDOW_SECS * 1000;
let dampening_ms = REKEY_DAMPENING_SECS * 1000;
let mut sessions_to_cutover: Vec<NodeAddr> = Vec::new();
let mut sessions_to_drain: Vec<NodeAddr> = Vec::new();
let mut sessions_to_rekey: Vec<NodeAddr> = Vec::new();
for (node_addr, entry) in &self.sessions {
if !entry.is_established() {
continue;
}
// 1. Initiator-side cutover (option A): completed rekey,
// pending session ready, liveness timer elapsed. This is
// an unconditional timer, NOT gated on responder progress —
// overlapping-epoch trial-decrypt covers the cutover skew,
// so flipping the K-bit here is always safe. An
// opportunistic early cutover also happens in
// `handle_encrypted_session_msg` if the initiator
// authenticates a peer frame against its own `pending`.
if entry.pending_new_session().is_some()
&& !entry.has_rekey_in_progress()
&& entry.is_rekey_initiator()
&& now_ms.saturating_sub(entry.rekey_completed_ms()) >= FSP_CUTOVER_DELAY_MS
{
sessions_to_cutover.push(*node_addr);
continue;
}
// 2. Drain window expiry
if entry.is_draining() && entry.drain_expired(now_ms, drain_ms) {
sessions_to_drain.push(*node_addr);
}
// 3. Rekey trigger
if entry.has_rekey_in_progress() {
continue;
}
if entry.pending_new_session().is_some() {
continue; // Pending session present, awaiting cutover
}
if entry.rekey_msg3_payload().is_some() {
// Initiator already cut over on its liveness timer but is
// still retransmitting msg3 to a responder not yet
// confirmed on the new epoch. Don't start another rekey
// until the current cycle's msg3 is delivered or abandoned.
continue;
}
if entry.is_rekey_dampened(now_ms, dampening_ms) {
continue;
}
let elapsed_secs = now_ms.saturating_sub(entry.session_start_ms()) / 1000;
let counter = entry.send_counter();
// Apply per-session symmetric jitter to desynchronize
// dual-initiation in symmetric-start meshes.
let effective_after_secs =
rekey_after_secs.saturating_add_signed(entry.rekey_jitter_secs());
if elapsed_secs >= effective_after_secs || counter >= rekey_after_messages {
sessions_to_rekey.push(*node_addr);
// The shell snapshots each established session's rekey ages/flags
// (every clock read resolved here); the core decides
// cutover/drain/trigger with no clock, phase-grouped to preserve the
// pre-refactor execution order.
let snapshots = self.session_rekey_snapshots(now_ms);
for action in self.fsp.poll_rekey(snapshots, &cfg) {
match action {
FspAction::CutOver { addr } => {
if let Some(entry) = self.sessions.get_mut(&addr)
&& entry.cutover_to_new_session(now_ms)
{
debug!(
peer = %self.peer_display_name(&addr),
"FSP rekey cutover complete (initiator), K-bit flipped"
);
}
}
FspAction::CompleteDrain { addr } => {
if let Some(entry) = self.sessions.get_mut(&addr) {
entry.complete_drain();
trace!(
peer = %self.peer_display_name(&addr),
"FSP drain complete, previous session erased"
);
}
}
FspAction::InitiateRekey { addr } => {
self.initiate_session_rekey(&addr).await;
}
#[allow(unreachable_patterns)]
_ => {}
}
}
}
// Execute cutover for initiator side
for node_addr in sessions_to_cutover {
if let Some(entry) = self.sessions.get_mut(&node_addr)
&& entry.cutover_to_new_session(now_ms)
{
debug!(
peer = %self.peer_display_name(&node_addr),
"FSP rekey cutover complete (initiator), K-bit flipped"
);
}
}
// Execute drain completion
for node_addr in sessions_to_drain {
if let Some(entry) = self.sessions.get_mut(&node_addr) {
entry.complete_drain();
trace!(
peer = %self.peer_display_name(&node_addr),
"FSP drain complete, previous session erased"
);
}
}
// Initiate new rekeys
for node_addr in sessions_to_rekey {
self.initiate_session_rekey(&node_addr).await;
}
/// Snapshot every established session for the FSP rekey decision,
/// pre-computing its monotonic age and timer predicates so the pure core
/// applies the thresholds without reading a clock (see [`SessionSnapshot`]).
fn session_rekey_snapshots(&self, now_ms: u64) -> Vec<SessionSnapshot> {
let drain_ms = crate::proto::fsp::limits::DRAIN_WINDOW_SECS * 1000;
let dampening_ms = crate::proto::fsp::limits::REKEY_DAMPENING_SECS * 1000;
self.sessions
.iter()
.filter(|(_, entry)| entry.is_established())
.map(|(node_addr, entry)| SessionSnapshot {
addr: *node_addr,
has_pending: entry.pending_new_session().is_some(),
rekey_in_progress: entry.has_rekey_in_progress(),
is_rekey_initiator: entry.is_rekey_initiator(),
cutover_timer_elapsed: cutover_timer_elapsed(now_ms, entry.rekey_completed_ms()),
is_draining: entry.is_draining(),
drain_expired: entry.drain_expired(now_ms, drain_ms),
has_rekey_msg3_payload: entry.rekey_msg3_payload().is_some(),
is_dampened: entry.is_rekey_dampened(now_ms, dampening_ms),
elapsed_secs: now_ms.saturating_sub(entry.session_start_ms()) / 1000,
counter: entry.send_counter(),
jitter_secs: entry.rekey_jitter_secs(),
})
.collect()
}
/// Initiate an FSP session rekey.
src/node/handlers/session.rs
+366 -132
View File
@@ -6,14 +6,9 @@
//! encrypted data, and error signals (CoordsRequired, PathBroken).
use crate::NodeAddr;
use crate::node::handlers::mmp::format_throughput;
use crate::node::reject::{RejectReason, SessionReject};
use crate::node::session::{EndToEndState, EpochSlot, SessionEntry};
use crate::node::session_wire::{
FSP_COMMON_PREFIX_SIZE, FSP_FLAG_CP, FSP_FLAG_K, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED,
FSP_PHASE_MSG1, FSP_PHASE_MSG2, FSP_PHASE_MSG3, FSP_PORT_HEADER_SIZE, FSP_PORT_IPV6_SHIM,
FspCommonPrefix, FspEncryptedHeader, build_fsp_header, fsp_prepend_inner_header,
fsp_strip_inner_header, parse_encrypted_coords,
};
#[cfg(unix)]
use crate::node::wire::{
ESTABLISHED_HEADER_SIZE, FLAG_KEY_EPOCH, FLAG_SP, build_established_header,
@@ -22,19 +17,28 @@ use crate::node::{Node, NodeError};
use crate::noise::{
HandshakeState, XK_HANDSHAKE_MSG1_SIZE, XK_HANDSHAKE_MSG2_SIZE, XK_HANDSHAKE_MSG3_SIZE,
};
use crate::proto::mmp::{MAX_SESSION_REPORT_INTERVAL_MS, MIN_SESSION_REPORT_INTERVAL_MS};
use crate::proto::fsp::wire::{
FSP_COMMON_PREFIX_SIZE, FSP_FLAG_CP, FSP_FLAG_K, FSP_HEADER_SIZE, FSP_PHASE_ESTABLISHED,
FSP_PHASE_MSG1, FSP_PHASE_MSG2, FSP_PHASE_MSG3, FSP_PORT_HEADER_SIZE, FSP_PORT_IPV6_SHIM,
FspCommonPrefix, FspEncryptedHeader, build_fsp_header, fsp_prepend_inner_header,
fsp_strip_inner_header, parse_encrypted_coords,
};
use crate::proto::fsp::{
DecryptSlot, EpochReaction, FspAction, FspInnerFlags, SessionAck, SessionMessageType,
SessionMsg3, SessionSetup, mark_ipv6_ecn_ce,
};
#[cfg(unix)]
use crate::proto::link::LinkMessageType;
#[cfg(unix)]
use crate::proto::link::SESSION_DATAGRAM_HEADER_SIZE;
use crate::proto::link::SessionDatagram;
use crate::proto::mmp::{
PathMtuNotification, ReceiverReport, SessionReceiverReport, SessionSenderReport,
BackoffUpdate, MmpAction, MmpSessionState, PathMtuNotification, ReceiverReport, SendResult,
SessionReceiverReport, SessionReportKind, SessionReportSnapshot, SessionSenderReport,
};
use crate::proto::routing::{CoordsRequired, MtuExceeded, PathBroken};
#[cfg(unix)]
use crate::protocol::LinkMessageType;
#[cfg(unix)]
use crate::protocol::SESSION_DATAGRAM_HEADER_SIZE;
use crate::protocol::{
FspInnerFlags, SessionAck, SessionDatagram, SessionMessageType, SessionMsg3, SessionSetup,
};
use crate::protocol::{coords_wire_size, encode_coords};
use crate::proto::mmp::{MAX_SESSION_REPORT_INTERVAL_MS, MIN_SESSION_REPORT_INTERVAL_MS};
use crate::proto::routing::{CoordsRequired, MtuExceeded, PathBroken, RoutingSignalType};
use crate::proto::stp::{coords_wire_size, encode_coords};
#[cfg(unix)]
use crate::transport::TransportHandle;
use crate::upper::icmp::FIPS_OVERHEAD;
@@ -105,14 +109,14 @@ impl Node {
}
let error_type = inner[0];
let error_body = &inner[1..];
match SessionMessageType::from_byte(error_type) {
Some(SessionMessageType::CoordsRequired) => {
match RoutingSignalType::from_byte(error_type) {
Some(RoutingSignalType::CoordsRequired) => {
self.handle_coords_required(error_body).await;
}
Some(SessionMessageType::PathBroken) => {
Some(RoutingSignalType::PathBroken) => {
self.handle_path_broken(error_body).await;
}
Some(SessionMessageType::MtuExceeded) => {
Some(RoutingSignalType::MtuExceeded) => {
self.handle_mtu_exceeded(error_body).await;
}
_ => {
@@ -167,11 +171,14 @@ impl Node {
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);
let my_addr = *self.node_addr();
for action in
self.fsp
.plan_cache_coords(*src_addr, my_addr, src_coords, dest_coords)
{
if let FspAction::CacheCoords { addr, coords } = action {
self.coord_cache.insert(addr, coords, now_ms);
}
}
ciphertext_offset += bytes_consumed;
}
@@ -249,44 +256,55 @@ impl Node {
}
};
// React to the epoch the frame decrypted against.
match slot {
EpochSlot::Pending => {
// A frame that authenticates against `pending` is itself
// the cutover signal — proof the peer derived the new
// session and moved to it. Promote now: current →
// previous, pending → current, flip the K-bit. The
// header K-bit is no longer the gating event; the
// authenticated decrypt is.
// React to the epoch the frame decrypted against. The shell opened
// the frame; the core classifies the post-decrypt reaction over the
// plain-data slot + session flags, and the shell applies the
// `SessionEntry` mutation.
let decrypt_slot = match slot {
EpochSlot::Current => DecryptSlot::Current,
EpochSlot::Pending => DecryptSlot::Pending,
EpochSlot::Previous => DecryptSlot::Previous,
};
match self.fsp.classify_epoch(
decrypt_slot,
entry.rekey_msg3_payload().is_some(),
entry.pending_new_session().is_some(),
) {
EpochReaction::PromoteConfirming => {
// A frame that authenticates against `pending` is itself the
// cutover signal — proof the peer derived the new session and
// moved to it. The peer received msg3, so confirm it on the new
// epoch (stop retransmitting) before `handle_peer_kbit_flip`
// consumes the pending session, then promote.
info!(
peer = %self.peer_display_name(src_addr),
"Peer FSP new-epoch frame authenticated, FSP rekey cutover complete, promoting new session"
);
// The peer derived the new session, so it received msg3:
// confirm it on the new epoch and stop retransmitting.
// `handle_peer_kbit_flip` consumes the pending session,
// so confirm first.
if entry.rekey_msg3_payload().is_some() {
entry.confirm_peer_new_epoch();
}
entry.confirm_peer_new_epoch();
entry.handle_peer_kbit_flip(now_ms);
}
EpochSlot::Current => {
// If we still retain a msg3 retransmission payload but no
// longer hold a `pending` session, we are the rekey
// initiator that already cut over on its own timer:
// `current` is now the new epoch, so a frame decrypting
// against it confirms the responder reached the new
// epoch. Stop retransmitting msg3.
if entry.rekey_msg3_payload().is_some() && entry.pending_new_session().is_none() {
entry.confirm_peer_new_epoch();
}
EpochReaction::Promote => {
// Promote now: current → previous, pending → current, flip the
// K-bit. The header K-bit is only a hint; the authenticated
// decrypt is the gating event.
info!(
peer = %self.peer_display_name(src_addr),
"Peer FSP new-epoch frame authenticated, FSP rekey cutover complete, promoting new session"
);
entry.handle_peer_kbit_flip(now_ms);
}
EpochSlot::Previous => {
// The peer is still on the old epoch. `fsp_trial_decrypt`
// already refreshed the drain deadline so the `previous`
// slot is not retired while the peer keeps using it —
// no further state change here, just deliver.
EpochReaction::ConfirmResponder => {
// We are the rekey initiator that already cut over on its own
// timer: `current` is now the new epoch, so a frame decrypting
// against it confirms the responder reached it. Stop
// retransmitting msg3.
entry.confirm_peer_new_epoch();
}
EpochReaction::None => {
// Steady-state `current`, or an old-epoch `previous` straggler:
// `fsp_trial_decrypt` already refreshed the drain deadline so
// the `previous` slot is not retired while the peer keeps using
// it — no further state change, just deliver.
}
}
@@ -445,7 +463,7 @@ impl Node {
if let Some(existing) = self.sessions.get(src_addr) {
if existing.is_initiating() {
// Simultaneous initiation: smaller NodeAddr wins as initiator
if self.identity().node_addr() < src_addr {
if crate::proto::fsp::initiation_winner(self.identity().node_addr(), src_addr) {
// We win — drop their setup, they'll process ours
debug!(
src = %self.peer_display_name(src_addr),
@@ -483,7 +501,10 @@ impl Node {
// simultaneously. Apply tie-breaker — smaller NodeAddr
// wins as initiator (same as initial session setup).
if rekey_in_progress {
if self.identity().node_addr() < src_addr {
if crate::proto::fsp::initiation_winner(
self.identity().node_addr(),
src_addr,
) {
// We win as initiator — drop their msg1.
debug!(
src = %self.peer_display_name(src_addr),
@@ -920,6 +941,221 @@ impl Node {
// === Session-layer MMP report handlers ===
/// 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_ms = crate::mmp::mono_ms();
// Build one report-gating snapshot per session, resolving every timing
// read shell-side into a `bool`. The snapshots own only
// `NodeAddr`/`MmpMode`/`bool`, so the session-iteration borrow is released
// before the pure decision runs and the driving loop mutates the
// reporting state / performs the sends.
let snapshots: Vec<SessionReportSnapshot> = self
.sessions
.iter()
.filter_map(|(dest_addr, entry)| {
let mmp = entry.mmp()?;
Some(SessionReportSnapshot {
dest: *dest_addr,
mode: mmp.mode(),
sr_due: mmp.sender.should_send_report(now_ms),
rr_due: mmp.receiver.should_send_report(now_ms),
mtu_due: mmp.path_mtu.should_send_notification(now_ms),
log_due: mmp.should_log(now_ms),
})
})
.collect();
let actions = self.mmp.plan_session_reports(&snapshots);
// Drive the planned actions in phase-grouped order (all logs, then the
// sends in per-session SR/RR/MTU order). Logs run first because the
// session operator log reads cumulative_packets_sent, which each send
// advances (send_session_msg -> sender.record_sent); the pre-refactor
// handler logged during its collect pass, before any send. Each build
// (`build_report`/`build_notification`, which advance interval/
// notification state) runs only on its SendSessionReport action, exactly
// as the pre-refactor collect pass did. Per-destination success/failure
// is collected for the backoff dedup + failure-log suppression.
let mut send_results: Vec<SendResult> = Vec::new();
for action in actions {
match action {
MmpAction::LogSession { dest } => {
// Resolve the display name exactly as the pre-refactor loop
// did (alias, else short_npub from the session's remote key).
let session_name = self.peer_aliases.get(&dest).cloned().unwrap_or_else(|| {
self.sessions
.get(&dest)
.map(|entry| {
let (xonly, _) = entry.remote_pubkey().x_only_public_key();
crate::PeerIdentity::from_pubkey(xonly).short_npub()
})
.unwrap_or_default()
});
if let Some(mmp) = self.sessions.get_mut(&dest).and_then(|e| e.mmp_mut()) {
Self::log_session_mmp_metrics(&session_name, mmp);
mmp.mark_logged(now_ms);
}
}
MmpAction::SendSessionReport { dest, kind } => {
let built = self
.sessions
.get_mut(&dest)
.and_then(|entry| entry.mmp_mut())
.and_then(|mmp| match kind {
SessionReportKind::Sender => {
mmp.sender.build_report(now_ms).map(|sr| {
(
SessionMessageType::SenderReport.to_byte(),
SessionSenderReport::from(&sr).encode(),
)
})
}
SessionReportKind::Receiver => {
mmp.receiver.build_report(now_ms).map(|rr| {
(
SessionMessageType::ReceiverReport.to_byte(),
SessionReceiverReport::from(&rr).encode(),
)
})
}
SessionReportKind::PathMtu => {
mmp.path_mtu.build_notification(now_ms).map(|mtu_value| {
(
SessionMessageType::PathMtuNotification.to_byte(),
PathMtuNotification::new(mtu_value).encode(),
)
})
}
});
let Some((msg_type, body)) = built else {
continue;
};
match self.send_session_msg(&dest, msg_type, &body).await {
Ok(()) => send_results.push(SendResult { dest, ok: true }),
Err(e) => {
// Peek at current failure count for log suppression
// (unchanged by the backoff apply, which runs later).
let failures = self
.sessions
.get(&dest)
.and_then(|entry| entry.mmp())
.map(|mmp| mmp.sender.consecutive_send_failures())
.unwrap_or(0);
if failures < 3 {
debug!(
dest = %self.peer_display_name(&dest),
msg_type,
error = %e,
"Failed to send session MMP report"
);
} else if failures == 3 {
debug!(
dest = %self.peer_display_name(&dest),
"Suppressing further session MMP send failure logs"
);
}
// failures > 3: silently suppressed
send_results.push(SendResult { dest, ok: false });
}
}
}
MmpAction::ReapPeer { .. }
| MmpAction::Heartbeat { .. }
| MmpAction::SendLinkReport { .. }
| MmpAction::LogLink { .. } => {}
}
}
// Deduplicate send results per destination (any-ok -> success, all-fail
// -> failure) and apply the backoff state transition for each dest.
for update in self.mmp.plan_backoff(&send_results) {
match update {
BackoffUpdate::Success { dest } => {
if let Some(mmp) = self.sessions.get_mut(&dest).and_then(|e| e.mmp_mut()) {
let prev = mmp.sender.record_send_success();
if prev > 3 {
debug!(
dest = %self.peer_display_name(&dest),
consecutive_failures = prev,
"Resumed session MMP reporting"
);
}
}
}
BackoffUpdate::Failure { dest } => {
if let Some(mmp) = self.sessions.get_mut(&dest).and_then(|e| e.mmp_mut()) {
mmp.sender.record_send_failure();
}
}
}
}
}
/// Emit periodic session MMP metrics.
fn log_session_mmp_metrics(session_name: &str, mmp: &MmpSessionState) {
let m = &mmp.metrics;
let rtt_str = if m.rtt_trend.initialized() {
format!("{:.1}ms", m.rtt_trend.long() / 1000.0)
} else {
"n/a".to_string()
};
let loss_str = if m.loss_trend.initialized() {
format!("{:.1}%", m.loss_trend.long() * 100.0)
} else {
"n/a".to_string()
};
let jitter_ms = mmp.receiver.jitter_us() as f64 / 1000.0;
debug!(
session = %session_name,
rtt = %rtt_str,
loss = %loss_str,
jitter = format_args!("{:.1}ms", jitter_ms),
goodput = %format_throughput(m.goodput_bps()),
mtu = mmp.path_mtu.last_observed_mtu(),
tx_pkts = mmp.sender.cumulative_packets_sent(),
rx_pkts = mmp.receiver.cumulative_packets_recv(),
"MMP session metrics"
);
}
/// Emit a teardown log summarizing lifetime session MMP metrics.
pub(in crate::node) fn log_session_mmp_teardown(session_name: &str, mmp: &MmpSessionState) {
let m = &mmp.metrics;
let jitter_ms = mmp.receiver.jitter_us() as f64 / 1000.0;
let rtt_str = match m.srtt_ms() {
Some(rtt) => format!("{:.1}ms", rtt),
None => "n/a".to_string(),
};
let loss_str = format!("{:.1}%", m.loss_rate() * 100.0);
debug!(
session = %session_name,
rtt = %rtt_str,
loss = %loss_str,
jitter = format_args!("{:.1}ms", jitter_ms),
etx = format_args!("{:.2}", m.etx),
goodput = %format_throughput(m.goodput_bps()),
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(),
"MMP session teardown"
);
}
/// Handle an incoming session-layer SenderReport (msg_type 0x11).
///
/// Informational only — the peer is telling us about what they sent.
@@ -1069,28 +1305,34 @@ impl Node {
// tighter of existing-or-new — never loosen the clamp.
let fips_addr = crate::FipsAddress::from_node_addr(src_addr);
match self.path_mtu_lookup.write() {
Ok(mut map) => match map.get(&fips_addr).copied() {
Some(existing) if existing <= new_mtu => {
Ok(mut map) => {
// Read existing, decide, and apply the write under one guard so
// the keep-tighter update stays atomic.
let prior = map.get(&fips_addr).copied();
let actions = self.fsp.plan_path_mtu_tighten(fips_addr, prior, new_mtu);
if actions.is_empty() {
debug!(
dest = %peer_name,
fips_addr = %fips_addr,
new_mtu,
existing,
existing = prior.unwrap_or(new_mtu),
"PathMtuNotification: keeping tighter existing path_mtu_lookup value"
);
}
other => {
map.insert(fips_addr, new_mtu);
debug!(
dest = %peer_name,
fips_addr = %fips_addr,
new_mtu,
prior = ?other,
map_len = map.len(),
"PathMtuNotification: tightened path_mtu_lookup"
);
for action in actions {
if let FspAction::TightenPathMtuLookup { fips_addr, mtu } = action {
map.insert(fips_addr, mtu);
debug!(
dest = %peer_name,
fips_addr = %fips_addr,
new_mtu,
prior = ?prior,
map_len = map.len(),
"PathMtuNotification: tightened path_mtu_lookup"
);
}
}
},
}
Err(e) => {
warn!(
dest = %peer_name,
@@ -1145,12 +1387,19 @@ impl Node {
// Only trigger discovery if we have the target's identity cached —
// otherwise we can't verify the LookupResponse proof.
if self.has_cached_identity(&msg.dest_addr) {
self.maybe_initiate_lookup(&msg.dest_addr).await;
} else {
let has_cached_identity = self.has_cached_identity(&msg.dest_addr);
let actions = self
.fsp
.plan_coords_required_lookup(msg.dest_addr, has_cached_identity);
if actions.is_empty() {
debug!(dest = %msg.dest_addr,
"Skipping discovery after CoordsRequired: no cached identity for target");
}
for action in actions {
if let FspAction::InitiateLookup { dest } = action {
self.maybe_initiate_lookup(&dest).await;
}
}
// Reset coords warmup counter so the next N packets also include
// COORDS_PRESENT, re-warming transit caches along the path.
@@ -1204,16 +1453,26 @@ impl Node {
"PathBroken response rate-limited, skipping standalone CoordsWarmup");
}
// Invalidate stale cached coordinates
self.coord_cache.remove(&msg.dest_addr);
// Trigger re-discovery to get fresh coordinates, but only if we have
// the target's identity cached — otherwise we can't verify the
// LookupResponse proof. This avoids a race when the XK responder
// receives PathBroken before msg3 completes (identity unknown).
if self.has_cached_identity(&msg.dest_addr) {
self.maybe_initiate_lookup(&msg.dest_addr).await;
} else {
// Invalidate stale cached coordinates, then (only if the target's
// identity is cached — else the LookupResponse proof cannot be verified,
// e.g. when the XK responder receives PathBroken before msg3 completes)
// trigger re-discovery. The core emits invalidate-then-lookup in order.
let has_cached_identity = self.has_cached_identity(&msg.dest_addr);
let actions = self
.fsp
.plan_path_broken(msg.dest_addr, has_cached_identity);
for action in actions {
match action {
FspAction::InvalidateCoords { addr } => {
self.coord_cache.remove(&addr);
}
FspAction::InitiateLookup { dest } => {
self.maybe_initiate_lookup(&dest).await;
}
_ => {}
}
}
if !has_cached_identity {
debug!(dest = %msg.dest_addr,
"Skipping discovery after PathBroken: no cached identity for target");
}
@@ -1283,28 +1542,34 @@ impl Node {
// tighter of existing-or-new — never loosen the clamp.
let fips_addr = crate::FipsAddress::from_node_addr(&msg.dest_addr);
match self.path_mtu_lookup.write() {
Ok(mut map) => match map.get(&fips_addr).copied() {
Some(existing) if existing <= msg.mtu => {
Ok(mut map) => {
// Read existing, decide, and apply the write under one guard so
// the keep-tighter update stays atomic.
let prior = map.get(&fips_addr).copied();
let actions = self.fsp.plan_path_mtu_tighten(fips_addr, prior, msg.mtu);
if actions.is_empty() {
debug!(
dest = %peer_name,
fips_addr = %fips_addr,
bottleneck_mtu = msg.mtu,
existing,
existing = prior.unwrap_or(msg.mtu),
"Reactive MtuExceeded: keeping tighter existing path_mtu_lookup value"
);
}
other => {
map.insert(fips_addr, msg.mtu);
debug!(
dest = %peer_name,
fips_addr = %fips_addr,
bottleneck_mtu = msg.mtu,
prior = ?other,
map_len = map.len(),
"Reactive MtuExceeded: tightened path_mtu_lookup"
);
for action in actions {
if let FspAction::TightenPathMtuLookup { fips_addr, mtu } = action {
map.insert(fips_addr, mtu);
debug!(
dest = %peer_name,
fips_addr = %fips_addr,
bottleneck_mtu = msg.mtu,
prior = ?prior,
map_len = map.len(),
"Reactive MtuExceeded: tightened path_mtu_lookup"
);
}
}
},
}
Err(e) => {
warn!(
dest = %peer_name,
@@ -2243,10 +2508,7 @@ impl Node {
let per_dest = self.config().node.session.pending_packets_per_dest;
let queue = self.pending_tun_packets.entry(dest_addr).or_default();
if queue.len() >= per_dest {
queue.pop_front(); // Drop oldest
}
queue.push_back(packet);
crate::proto::fsp::push_bounded_pending(queue, packet, per_dest);
}
/// Flush pending packets for a destination whose session just reached Established.
@@ -2297,31 +2559,3 @@ impl Node {
}
}
}
/// Mark ECN-CE in an IPv6 packet's Traffic Class field.
///
/// IPv6 Traffic Class occupies bits across bytes 0 and 1:
/// byte[0] bits[3:0] = TC[7:4]
/// byte[1] bits[7:4] = TC[3:0]
/// ECN is TC[1:0]. Only marks CE (0b11) if the packet is ECN-capable
/// (ECT(0) or ECT(1)). Packets with ECN=0b00 (Not-ECT) are never marked
/// per RFC 3168.
///
/// No checksum update needed: IPv6 has no header checksum, and the Traffic
/// Class field is not part of the TCP/UDP pseudo-header.
pub(in crate::node) fn mark_ipv6_ecn_ce(packet: &mut [u8]) {
if packet.len() < 2 {
return;
}
// Extract 8-bit Traffic Class from IPv6 header bytes 0-1
let tc = ((packet[0] & 0x0F) << 4) | (packet[1] >> 4);
let ecn = tc & 0x03;
// Only mark CE on ECN-capable packets (ECT(0)=0b10 or ECT(1)=0b01)
if ecn == 0 {
return;
}
// Set both ECN bits to 1 (CE = 0b11)
let new_tc = tc | 0x03;
packet[0] = (packet[0] & 0xF0) | (new_tc >> 4);
packet[1] = (new_tc << 4) | (packet[1] & 0x0F);
}
src/node/handlers/timeout.rs
+1 -1
View File
@@ -248,7 +248,7 @@ impl Node {
.collect();
for (dest_addr, payload) in candidates {
use crate::protocol::SessionDatagram;
use crate::proto::link::SessionDatagram;
let mut datagram = SessionDatagram::new(my_addr, dest_addr, payload).with_ttl(ttl);
let sent = match self.send_session_datagram(&mut datagram).await {
src/node/mod.rs
+6 -1
View File
@@ -19,7 +19,6 @@ pub(crate) mod reject;
mod reloadable;
mod retry;
pub(crate) mod session;
pub(crate) mod session_wire;
pub(crate) mod stats;
pub(crate) mod stats_history;
#[cfg(test)]
@@ -46,6 +45,7 @@ use crate::peer::{ActivePeer, PeerConnection};
use crate::proto::bloom::{BloomFilter, BloomState};
use crate::proto::discovery::{Discovery, DiscoveryBackoff, DiscoveryForwardRateLimiter};
use crate::proto::fmp::Fmp;
use crate::proto::fsp::Fsp;
use crate::proto::mmp::Mmp;
use crate::proto::routing::{self, Router, RoutingErrorRateLimiter};
use crate::proto::stp::TreeState;
@@ -421,6 +421,9 @@ pub struct Node {
/// FMP connection-lifecycle decision anchor (stateless; drives the
/// tick-poll maintain/teardown decisions).
fmp: Fmp,
/// FSP session-lifecycle decision anchor (stateless; drives the rekey /
/// epoch-reaction decisions).
fsp: Fsp,
/// MMP reporting decision anchor (stateless; drives the report-fan-out /
/// liveness / heartbeat decisions).
mmp: Mmp,
@@ -670,6 +673,7 @@ impl Node {
icmp_rate_limiter: IcmpRateLimiter::new(),
routing: Router::new(),
fmp: Fmp::new(),
fsp: Fsp::new(),
mmp: Mmp::new(),
coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval_ms(
coords_response_interval_ms,
@@ -834,6 +838,7 @@ impl Node {
icmp_rate_limiter: IcmpRateLimiter::new(),
routing: Router::new(),
fmp: Fmp::new(),
fsp: Fsp::new(),
mmp: Mmp::new(),
coords_response_rate_limiter: RoutingErrorRateLimiter::with_interval_ms(
coords_response_interval_ms,
src/node/session.rs
+152 -1
View File
@@ -801,8 +801,8 @@ impl SessionEntry {
#[cfg(test)]
mod overlapping_epoch_tests {
use super::*;
use crate::node::session_wire::{FSP_FLAG_K, build_fsp_header};
use crate::noise::HandshakeState;
use crate::proto::fsp::wire::{FSP_FLAG_K, build_fsp_header};
use secp256k1::{Keypair, Secp256k1, SecretKey};
/// Deterministic keypair from a single seed byte.
@@ -1265,4 +1265,155 @@ mod overlapping_epoch_tests {
"window must expire on the plain wall-clock timer when peer is off the old epoch"
);
}
// ========================================================================
// Rekey-policy characterization (pins `check_session_rekey`'s decision
// boundaries before the `Fsp::poll_rekey` hoist — these thresholds have no
// other test module; see plan §10).
// ========================================================================
/// The initiator liveness-cutover delay used by `check_session_rekey`
/// (`FSP_CUTOVER_DELAY_MS`). Mirrored here as the characterization anchor.
const CUTOVER_DELAY_MS: u64 = 2000;
/// Build an established entry that has completed a rekey as initiator and
/// holds a pending session awaiting the K-bit cutover.
fn entry_pending_cutover(rekey_completed_ms: u64) -> SessionEntry {
let (_cur_send, cur_recv) = xk_pair(1, 2);
let (_new_send, new_recv) = xk_pair(3, 4);
let mut entry = entry_with_current(cur_recv);
// Mark ourselves the rekey initiator, then land the completed session
// as pending (clears rekey_state, so has_rekey_in_progress() == false).
entry.set_rekey_state(HandshakeState::new_xk_responder(keypair(7)), true);
entry.set_pending_session(new_recv);
entry.set_rekey_completed_ms(rekey_completed_ms);
entry
}
// The initiator-side cutover predicate: pending session present, no rekey
// in progress, we are the initiator, and the liveness timer has elapsed.
#[test]
fn rekey_cutover_predicate_boundary() {
let completed = 1_000u64;
let entry = entry_pending_cutover(completed);
assert!(entry.pending_new_session().is_some());
assert!(!entry.has_rekey_in_progress());
assert!(entry.is_rekey_initiator());
// Not yet eligible one ms before the delay elapses.
let just_before = completed + CUTOVER_DELAY_MS - 1;
assert!(
just_before.saturating_sub(entry.rekey_completed_ms()) < CUTOVER_DELAY_MS,
"cutover must not fire before the liveness delay"
);
// Eligible exactly at the delay.
let at = completed + CUTOVER_DELAY_MS;
assert!(
at.saturating_sub(entry.rekey_completed_ms()) >= CUTOVER_DELAY_MS,
"cutover fires once the liveness delay has elapsed"
);
}
// Rekey-trigger threshold: elapsed time (with symmetric jitter applied)
// OR the send counter crossing its configured bound.
#[test]
fn rekey_trigger_threshold_arithmetic() {
let after_secs = 100u64;
let after_messages = 1_000u64;
// Jitter is always within [-REKEY_JITTER_SECS, +REKEY_JITTER_SECS].
let (_s, recv) = xk_pair(1, 2);
let entry = entry_with_current(recv);
let jitter = entry.rekey_jitter_secs();
assert!(
jitter.abs() <= REKEY_JITTER_SECS,
"jitter within configured bound"
);
// Effective time threshold applies the symmetric jitter.
let effective_after = after_secs.saturating_add_signed(jitter);
// Reproduce the policy's OR predicate directly.
let triggers =
|elapsed: u64, counter: u64| elapsed >= effective_after || counter >= after_messages;
// Time arm: fires at/after the effective threshold, not below it.
assert!(!triggers(effective_after - 1, 0), "below time threshold");
assert!(triggers(effective_after, 0), "at time threshold");
// Counter arm: fires independently of elapsed time.
assert!(!triggers(0, after_messages - 1), "below counter threshold");
assert!(triggers(0, after_messages), "at counter threshold");
}
// Dampening boundary: within `dampening_ms` of the peer's rekey msg1, local
// initiation is suppressed; at/after the window it is not.
#[test]
fn rekey_dampening_boundary() {
let (_s, recv) = xk_pair(1, 2);
let mut entry = entry_with_current(recv);
const DAMP_MS: u64 = 30_000;
// No peer rekey recorded → never dampened.
assert!(!entry.is_rekey_dampened(50_000, DAMP_MS));
entry.record_peer_rekey(10_000);
assert!(
entry.is_rekey_dampened(10_000 + DAMP_MS - 1, DAMP_MS),
"dampened within the window"
);
assert!(
!entry.is_rekey_dampened(10_000 + DAMP_MS, DAMP_MS),
"not dampened once the window has elapsed"
);
}
// Epoch-reaction: a frame authenticating against `pending` while a msg3
// retransmission is retained confirms the peer on the new epoch (clears the
// msg3 payload) and then promotes.
#[test]
fn epoch_reaction_pending_confirms_then_promotes() {
let (mut p_send, p_recv) = xk_pair(3, 4);
let (_cur_send, cur_recv) = xk_pair(1, 2);
let mut entry = entry_with_current(cur_recv);
let k_before = entry.current_k_bit();
entry.set_pending_session(p_recv);
entry.set_rekey_msg3_payload(vec![0xAB; 8], 5_000);
assert!(entry.rekey_msg3_payload().is_some());
let (ct, counter, hdr) = seal(&mut p_send, b"new-epoch", !k_before);
let (_pt, slot) = entry
.fsp_trial_decrypt(&ct, counter, &hdr, !k_before, 2_000)
.expect("pending frame decrypts");
assert_eq!(slot, EpochSlot::Pending);
// Reaction order: confirm (while pending still held) then promote.
entry.confirm_peer_new_epoch();
assert!(entry.rekey_msg3_payload().is_none());
entry.handle_peer_kbit_flip(2_000);
assert!(entry.pending_new_session().is_none());
assert_ne!(entry.current_k_bit(), k_before);
}
// Epoch-reaction: as the initiator that already cut over on its own timer
// (msg3 retained, no pending), a frame authenticating against `current`
// confirms the responder reached the new epoch.
#[test]
fn epoch_reaction_current_confirms_responder() {
let (mut cur_send, cur_recv) = xk_pair(1, 2);
let mut entry = entry_with_current(cur_recv);
entry.set_rekey_msg3_payload(vec![0xCD; 8], 5_000);
assert!(entry.pending_new_session().is_none());
assert!(entry.rekey_msg3_payload().is_some());
let (ct, counter, hdr) = seal(&mut cur_send, b"steady", false);
let (_pt, slot) = entry
.fsp_trial_decrypt(&ct, counter, &hdr, false, 2_000)
.expect("current frame decrypts");
assert_eq!(slot, EpochSlot::Current);
// The Current-with-retained-msg3-and-no-pending arm confirms.
entry.confirm_peer_new_epoch();
assert!(entry.rekey_msg3_payload().is_none());
}
}
src/node/session_wire.rs
-618
View File
@@ -1,618 +0,0 @@
//! 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]
//! ```
//!
//! ## DataPacket Port Multiplexing
//!
//! DataPacket (msg_type 0x10) payloads inside the AEAD envelope carry a 4-byte
//! port header for service dispatch:
//!
//! ```text
//! [src_port:2 LE][dst_port:2 LE][service payload...]
//! ```
//!
//! Port 256 (0x100) = IPv6 shim with header compression.
//!
//! ## 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 XK msg1) |
//! | 0x2 | - | Handshake msg2 | SessionAck (Noise XK msg2) |
//! | 0x3 | - | Handshake msg3 | SessionMsg3 (Noise XK msg3) |
use crate::proto::stp::TreeCoordinate;
use crate::protocol::{ProtocolError, decode_optional_coords};
// ============================================================================
// 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 handshake message 2).
pub const FSP_PHASE_MSG2: u8 = 0x2;
/// Phase value for XK message 3 (initiator's encrypted static).
pub const FSP_PHASE_MSG3: u8 = 0x3;
/// 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
// FSP DataPacket port header constants.
/// Size of the FSP DataPacket port header (src_port + dst_port).
pub const FSP_PORT_HEADER_SIZE: usize = 4;
/// FSP port: IPv6 shim service.
pub const FSP_PORT_IPV6_SHIM: u16 = 256;
// 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 FMP, 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`, `FSP_PHASE_MSG2`, or `FSP_PHASE_MSG3`.
/// 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);
}
#[test]
fn test_build_fsp_handshake_prefix_msg3() {
let prefix = build_fsp_handshake_prefix(FSP_PHASE_MSG3, 73);
assert_eq!(prefix[0], 0x03); // ver=0, phase=3
assert_eq!(prefix[1], 0x00); // flags zero
assert_eq!(u16::from_le_bytes([prefix[2], prefix[3]]), 73);
let parsed = FspCommonPrefix::parse(&prefix).unwrap();
assert_eq!(parsed.phase, FSP_PHASE_MSG3);
}
// ===== 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);
// SessionMsg3 (phase 3)
let prefix = FspCommonPrefix::parse(&[0x03, 0x00, 0x49, 0x00]).unwrap();
assert_eq!(prefix.phase, 3);
}
}
src/node/tests/forwarding.rs
+5 -3
View File
@@ -5,9 +5,11 @@
//! multi-hop forwarding through live node topologies.
use super::*;
use crate::node::session_wire::{FSP_FLAG_CP, build_fsp_header};
use crate::proto::fsp::wire::{FSP_FLAG_CP, build_fsp_header};
use crate::proto::fsp::{SessionAck, SessionSetup};
use crate::proto::link::SessionDatagram;
use crate::proto::stp::TreeCoordinate;
use crate::protocol::{SessionAck, SessionDatagram, SessionSetup, encode_coords};
use crate::proto::stp::encode_coords;
use spanning_tree::{
TestNode, cleanup_nodes, process_available_packets, run_tree_test, verify_tree_convergence,
};
@@ -591,7 +593,7 @@ async fn test_forwarding_with_cache_warming_enables_routing() {
// ============================================================================
use crate::node::TransportDropState;
use crate::node::handlers::session::mark_ipv6_ecn_ce;
use crate::proto::fsp::mark_ipv6_ecn_ce;
use crate::transport::TransportId;
/// Build a minimal IPv6 header (40 bytes) with specified ECN bits.
src/node/tests/session.rs
+2 -1
View File
@@ -6,7 +6,8 @@ use crate::node::tests::spanning_tree::{
TestNode, cleanup_nodes, generate_random_edges, lock_large_network_test,
process_available_packets, run_tree_test, run_tree_test_with_mtus, verify_tree_convergence,
};
use crate::protocol::{SessionAck, SessionDatagram};
use crate::proto::fsp::SessionAck;
use crate::proto::link::SessionDatagram;
/// Populate all nodes' coordinate caches with each other's coords.
///