impl Node {
pub(in crate::node) async fn handle_session_payload(
&mut self,
delivery: LocalSessionPayload<'_>,
) {
let src_addr = *delivery.source_addr();
let payload = delivery.payload();
let prefix = match FspCommonPrefix::parse(payload) {
Some(p) => p,
None => {
debug!(
len = payload.len(),
"Session payload too short for FSP prefix"
);
return;
}
};
let inner = &payload[FSP_COMMON_PREFIX_SIZE..];
match prefix.phase {
FSP_PHASE_MSG1 => {
self.handle_session_setup(&src_addr, inner).await;
}
FSP_PHASE_MSG2 => {
self.handle_session_ack(&src_addr, inner).await;
}
FSP_PHASE_MSG3 => {
self.handle_session_msg3(&src_addr, inner).await;
}
FSP_PHASE_ESTABLISHED if prefix.is_unencrypted() => {
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;
}
Some(SessionMessageType::MtuExceeded) => {
self.handle_mtu_exceeded(error_body).await;
}
_ => {
debug!(error_type, "Unknown plaintext error signal type");
}
}
}
FSP_PHASE_ESTABLISHED => {
debug!(
src = %self.peer_display_name(&src_addr),
"Dropping established FSP payload outside packet_mover2 receive path"
);
return;
}
_ => {
debug!(phase = prefix.phase, "Unknown FSP phase");
}
}
}
pub(in crate::node) async fn process_packet_mover2_authenticated_sessions(
&mut self,
ingress_batch: Vec<crate::packet_mover2::PacketMover2FspSessionIngress>,
) -> usize {
let mut processed = 0usize;
let mut endpoint_deliveries = Vec::new();
let mut endpoint_commit = SessionReceiveBatchCommit::default();
for ingress in ingress_batch {
let Some(dispatch) = self.packet_mover2_authenticated_session_dispatch(ingress) else {
continue;
};
if dispatch.is_endpoint_data() {
let deliveries =
dispatch.dispatch_endpoint_data_batched(self, &mut endpoint_commit);
processed = processed.saturating_add(deliveries.len());
endpoint_deliveries.extend(deliveries);
continue;
}
self.flush_packet_mover2_endpoint_session_batch(
&mut endpoint_deliveries,
&mut endpoint_commit,
)
.await;
dispatch.dispatch(self).await;
processed = processed.saturating_add(1);
}
self.flush_packet_mover2_endpoint_session_batch(&mut endpoint_deliveries, &mut endpoint_commit)
.await;
processed
}
pub(in crate::node) async fn process_packet_mover2_compact_endpoint_data(
&mut self,
endpoint_bulks: Vec<crate::packet_mover2::PacketMover2EndpointDataBulk>,
) -> usize {
if endpoint_bulks.is_empty() {
return 0;
}
let message_count = endpoint_bulks
.iter()
.map(crate::packet_mover2::PacketMover2EndpointDataBulk::len)
.sum::<usize>();
let direct_packet_runs = endpoint_bulks
.iter()
.map(crate::packet_mover2::PacketMover2EndpointDataBulk::direct_packet_run_count)
.sum::<usize>();
let direct_sink = if direct_packet_runs > 0 {
match self.packet_mover2_endpoint_direct_sink() {
Some(sink) => Some(sink),
None => {
debug!(
messages = message_count,
"Dropping PM2 endpoint-data bulk without direct sink"
);
return 0;
}
}
} else {
None
};
let mut endpoint_commit = SessionReceiveBatchCommit::default();
let mut direct_packet_batches = Vec::with_capacity(endpoint_bulks.len());
for bulk in endpoint_bulks {
for run in bulk.commit_runs() {
let commit = run.commit();
let source_addr = commit.source_addr();
let previous_hop_addr = commit.previous_hop_addr();
if self.promote_packet_mover2_authenticated_pending_fsp_epoch(
&source_addr,
commit.received_k_bit(),
) {
debug!(
src = %self.peer_display_name(&source_addr),
received_k_bit = commit.received_k_bit(),
run_len = run.len(),
"FSP rekey cutover complete after PM2 compact endpoint-data receive commit"
);
}
self.learn_reverse_route(source_addr, previous_hop_addr);
endpoint_commit.push_receive_completion(SessionReceiveCompletion {
source_addr,
previous_hop_addr,
direct_path: commit.direct_path(),
});
}
if bulk.direct_packet_run_count() > 0 {
direct_packet_batches.push(bulk.into_direct_packet_batch());
}
}
let pending_flush_destinations = endpoint_commit.finish(self);
let count = direct_packet_batches
.iter()
.map(crate::node::FipsEndpointDirectPacketBatch::len)
.sum::<usize>();
if count > 0 {
let direct_sink = direct_sink.expect("direct sink is required when packet runs remain");
for batch in direct_packet_batches {
if direct_sink.deliver_direct_packet_batch(batch).is_err() {
crate::perf_profile::record_event_count(
crate::perf_profile::Event::EndpointEventBulkDropped,
count as u64,
);
break;
}
}
}
for dest_addr in pending_flush_destinations {
self.flush_pending_packets(&dest_addr).await;
}
message_count
}
fn packet_mover2_endpoint_direct_sink(&self) -> Option<crate::node::EndpointDirectSink> {
self.endpoint_events
.sender()
.and_then(|sender| sender.direct_sink().cloned())
}
fn packet_mover2_authenticated_session_dispatch(
&mut self,
ingress: crate::packet_mover2::PacketMover2FspSessionIngress,
) -> Option<AuthenticatedSessionDispatch> {
let received_k_bit = ingress.received_k_bit();
let (
source_addr,
source_peer,
previous_hop_addr,
ce_flag,
_activity_tick,
timestamp_ms,
msg_type,
inner_flags,
plaintext,
) = ingress.into_parts();
let body_len = plaintext
.len()
.saturating_sub(crate::node::session_wire::FSP_INNER_HEADER_SIZE);
debug!(
src = %self.peer_display_name(&source_addr),
previous_hop = %self.peer_display_name(&previous_hop_addr),
msg_type,
msg_kind = ?SessionMessageType::from_byte(msg_type),
plaintext_len = plaintext.len(),
body_len,
endpoint_data = msg_type == SessionMessageType::EndpointData.to_byte()
|| msg_type == SessionMessageType::EndpointDataBulk.to_byte(),
"Dispatching packet mover2 authenticated session"
);
if self.promote_packet_mover2_authenticated_pending_fsp_epoch(
&source_addr,
received_k_bit,
) {
debug!(
src = %self.peer_display_name(&source_addr),
received_k_bit,
"FSP rekey cutover complete after PM2 authenticated pending epoch"
);
}
let message =
AuthenticatedSessionMessage::new(source_peer, plaintext, msg_type, inner_flags, timestamp_ms);
Some(AuthenticatedSessionDispatch::new(
source_addr,
previous_hop_addr,
ce_flag,
message,
))
}
async fn flush_packet_mover2_endpoint_session_batch(
&mut self,
endpoint_deliveries: &mut Vec<EndpointDataDelivery>,
endpoint_commit: &mut SessionReceiveBatchCommit,
) {
if endpoint_deliveries.is_empty()
&& endpoint_commit.previous_hops.is_empty()
&& endpoint_commit.direct_sources.is_empty()
&& endpoint_commit.retry_peers.is_empty()
&& endpoint_commit.pending_flush_sources.is_empty()
{
return;
}
let pending_flush_destinations = std::mem::take(endpoint_commit).finish(self);
if !endpoint_deliveries.is_empty() {
self.deliver_endpoint_data_batch(std::mem::take(endpoint_deliveries));
}
for dest_addr in pending_flush_destinations {
self.flush_pending_packets(&dest_addr).await;
}
}
pub(in crate::node) fn record_authenticated_fmp_receive_facts(
&mut self,
fmp: crate::node::AuthenticatedFmpReceiveFacts<'_>,
previous_hop: Option<&NodeAddr>,
) {
let now = Instant::now();
let source_addr = fmp.source_node_addr();
let arrived_from_source = previous_hop.is_none_or(|hop| hop == source_addr);
let path_bookkeeping_allowed = self.authenticated_packet_path_allows_bookkeeping(
source_addr,
fmp.transport_id,
fmp.remote_addr,
fmp.packet_timestamp_ms,
) && arrived_from_source;
if path_bookkeeping_allowed {
let _ = self.packet_mover2.record_authenticated_fmp_mmp_receive(
source_addr,
fmp.fmp_counter,
fmp.inner_timestamp_ms,
fmp.packet_len,
fmp.fmp_flags & FLAG_CE != 0,
fmp.fmp_flags & FLAG_SP != 0,
now,
);
}
let bookkeeping = self.peers.record_authenticated_fmp_receive(
source_addr,
fmp.transport_id,
&fmp.remote_addr,
fmp.packet_timestamp_ms,
fmp.packet_len,
fmp.fmp_counter,
fmp.inner_timestamp_ms,
fmp.fmp_flags & FLAG_CE != 0,
fmp.fmp_flags & FLAG_SP != 0,
now,
path_bookkeeping_allowed,
);
if let Some(update) = bookkeeping {
if update.path_bookkeeping_recorded {
self.clear_retry_unless_direct_refresh_needed(source_addr);
}
if update.address_changed {
self.sync_packet_mover2_fmp_owner(source_addr);
}
}
}
pub(in crate::node) async fn handle_packet_mover2_fsp_decrypt_failure(
&mut self,
source_addr: NodeAddr,
counter: u64,
received_k_bit: bool,
) -> bool {
self.handle_reported_fsp_decrypt_failure(
source_addr,
counter,
received_k_bit,
"packet_mover2",
)
.await
}
async fn handle_reported_fsp_decrypt_failure(
&mut self,
src_addr: NodeAddr,
counter: u64,
received_k_bit: bool,
source: &'static str,
) -> bool {
let now_ms = Self::now_ms();
let owner_activity = self.packet_mover2.fsp_owner_activity(&src_addr);
let authenticated_inbound_age_ms =
owner_activity.and_then(|activity| activity.last_rx_age_ms(now_ms));
if owner_activity.is_some_and(|activity| {
activity.should_ignore_stale_epoch_decrypt_failure(received_k_bit)
}) {
trace!(
src = %self.peer_display_name(&src_addr),
counter,
source,
"Ignoring FSP AEAD failure from stale previous key epoch during PM2-owned drain"
);
return true;
}
let Some(entry) = self.sessions.get(&src_addr) else {
debug!(
src = %self.peer_display_name(&src_addr),
counter,
source,
"FSP AEAD failure for unknown session"
);
return false;
};
let entry_can_recover = entry.is_established()
&& !entry.has_rekey_in_progress()
&& entry.pending_new_session().is_none();
let Some(consecutive) = self.packet_mover2.record_fsp_decrypt_failure(src_addr) else {
debug!(
src = %self.peer_display_name(&src_addr),
counter,
source,
"FSP AEAD failure for missing packet_mover2 owner"
);
return false;
};
let recover_session =
should_start_decrypt_failure_rekey(entry_can_recover, consecutive, authenticated_inbound_age_ms);
debug!(
src = %self.peer_display_name(&src_addr),
counter,
consecutive_failures = consecutive,
source,
"FSP AEAD decryption failed"
);
if recover_session {
warn!(
peer = %self.peer_display_name(&src_addr),
consecutive_failures = consecutive,
"Session AEAD failures exceeded threshold; starting recovery rekey"
);
if !self.initiate_session_rekey(&src_addr).await {
debug!(
peer = %self.peer_display_name(&src_addr),
source,
"Failed to start recovery rekey after FSP decrypt-failure threshold"
);
}
}
true
}
async fn handle_mesh_traversal_offer(&mut self, src_addr: &NodeAddr, body: &[u8]) {
let Some(bootstrap) = self.nostr_discovery.clone() else {
trace!(
src = %self.peer_display_name(src_addr),
"Ignoring mesh traversal offer without Nostr discovery runtime"
);
return;
};
if self.configured_peer(src_addr).is_none() {
debug!(
src = %self.peer_display_name(src_addr),
"Ignoring mesh traversal offer from unconfigured peer"
);
return;
}
let Some(sender_npub) = self.npub_for_node_addr(src_addr) else {
debug!(
src = %self.peer_display_name(src_addr),
"Ignoring mesh traversal offer without known sender npub"
);
return;
};
let offer = match serde_json::from_slice::<TraversalOffer>(body) {
Ok(offer) => offer,
Err(error) => {
debug!(
src = %self.peer_display_name(src_addr),
error = %error,
"Malformed mesh traversal offer"
);
return;
}
};
if offer.sender_npub != sender_npub {
debug!(
src = %self.peer_display_name(src_addr),
claimed = %offer.sender_npub,
actual = %sender_npub,
"Ignoring mesh traversal offer with sender mismatch"
);
return;
}
bootstrap
.receive_mesh_traversal_offer(offer, sender_npub)
.await;
}
async fn handle_mesh_traversal_answer(&mut self, src_addr: &NodeAddr, body: &[u8]) {
let Some(bootstrap) = self.nostr_discovery.clone() else {
trace!(
src = %self.peer_display_name(src_addr),
"Ignoring mesh traversal answer without Nostr discovery runtime"
);
return;
};
if self.configured_peer(src_addr).is_none() {
debug!(
src = %self.peer_display_name(src_addr),
"Ignoring mesh traversal answer from unconfigured peer"
);
return;
}
let Some(sender_npub) = self.npub_for_node_addr(src_addr) else {
debug!(
src = %self.peer_display_name(src_addr),
"Ignoring mesh traversal answer without known sender npub"
);
return;
};
let answer = match serde_json::from_slice::<TraversalAnswer>(body) {
Ok(answer) => answer,
Err(error) => {
debug!(
src = %self.peer_display_name(src_addr),
error = %error,
"Malformed mesh traversal answer"
);
return;
}
};
if answer.sender_npub != sender_npub {
debug!(
src = %self.peer_display_name(src_addr),
claimed = %answer.sender_npub,
actual = %sender_npub,
"Ignoring mesh traversal answer with sender mismatch"
);
return;
}
bootstrap
.receive_mesh_traversal_answer(answer, sender_npub)
.await;
}
}