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 dataplane receive path"
);
}
_ => {
debug!(phase = prefix.phase, "Unknown FSP phase");
}
}
}
pub(in crate::node) async fn process_dataplane_authenticated_sessions(
&mut self,
ingress_batch: Vec<crate::dataplane::DataplaneFspSessionIngress>,
) -> usize {
let mut processed = 0usize;
let mut endpoint_deliveries = Vec::new();
let mut endpoint_commit = SessionReceiveBatchCommit::default();
let mut tun_packets = Vec::new();
let mut tun_commit = SessionReceiveBatchCommit::default();
for ingress in ingress_batch {
let Some(dispatch) = self.dataplane_authenticated_session_dispatch(ingress) else {
continue;
};
if dispatch.is_endpoint_data() {
self.flush_dataplane_tun_session_batch(&mut tun_packets, &mut tun_commit)
.await;
let deliveries =
dispatch.dispatch_endpoint_data_batched(self, &mut endpoint_commit);
processed = processed.saturating_add(deliveries.len());
endpoint_deliveries.extend(deliveries);
continue;
}
if dispatch.is_ipv6_shim_data_packet() {
self.flush_dataplane_endpoint_session_batch(
&mut endpoint_deliveries,
&mut endpoint_commit,
)
.await;
dispatch.dispatch_ipv6_shim_batched(self, &mut tun_packets, &mut tun_commit);
processed = processed.saturating_add(1);
continue;
}
self.flush_dataplane_endpoint_session_batch(
&mut endpoint_deliveries,
&mut endpoint_commit,
)
.await;
self.flush_dataplane_tun_session_batch(&mut tun_packets, &mut tun_commit)
.await;
dispatch.dispatch(self).await;
processed = processed.saturating_add(1);
}
self.flush_dataplane_endpoint_session_batch(&mut endpoint_deliveries, &mut endpoint_commit)
.await;
self.flush_dataplane_tun_session_batch(&mut tun_packets, &mut tun_commit)
.await;
processed
}
pub(in crate::node) async fn process_dataplane_authenticated_ingress(
&mut self,
ingress_batch: crate::dataplane::DataplaneFspAuthenticatedIngress,
) -> usize {
let mut processed = 0usize;
let mut endpoint_batches = Vec::new();
let mut session_ingress = Vec::new();
let (runs, endpoint_data_batches, sessions) = ingress_batch.into_parts();
let mut endpoint_data_batches = endpoint_data_batches.into_iter();
let mut sessions = sessions.into_iter();
for run in runs {
match run {
crate::dataplane::DataplaneFspAuthenticatedIngressRun::EndpointDataBatch => {
processed = processed.saturating_add(
self.process_dataplane_authenticated_sessions(std::mem::take(
&mut session_ingress,
))
.await,
);
endpoint_batches.push(
endpoint_data_batches
.next()
.expect("endpoint-data run has a batch"),
);
}
crate::dataplane::DataplaneFspAuthenticatedIngressRun::Sessions { count } => {
processed = processed.saturating_add(
self.process_dataplane_compact_endpoint_data(std::mem::take(
&mut endpoint_batches,
))
.await,
);
for _ in 0..count {
session_ingress
.push(sessions.next().expect("session run has session ingress"));
}
}
}
}
debug_assert!(
endpoint_data_batches.next().is_none(),
"authenticated ingress runs consumed all endpoint-data batches"
);
debug_assert!(
sessions.next().is_none(),
"authenticated ingress runs consumed all session ingress"
);
processed = processed.saturating_add(
self.process_dataplane_compact_endpoint_data(endpoint_batches)
.await,
);
processed = processed.saturating_add(
self.process_dataplane_authenticated_sessions(session_ingress)
.await,
);
processed
}
pub(in crate::node) async fn process_dataplane_compact_endpoint_data(
&mut self,
endpoint_batches: Vec<crate::dataplane::DataplaneEndpointDataBatch>,
) -> usize {
if endpoint_batches.is_empty() {
return 0;
}
let message_count = endpoint_batches
.iter()
.map(crate::dataplane::DataplaneEndpointDataBatch::len)
.sum::<usize>();
let mut endpoint_commit = SessionReceiveBatchCommit::default();
for batch in &endpoint_batches {
for run in batch.commit_runs() {
let commit = run.commit();
let source_addr = commit.source_addr();
let previous_hop_addr = commit.previous_hop_addr();
if self.promote_dataplane_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 dataplane 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(),
});
}
}
let pending_flush_destinations = endpoint_commit.finish(self);
for dest_addr in pending_flush_destinations {
self.flush_pending_packets(&dest_addr).await;
}
message_count
}
async fn flush_dataplane_tun_session_batch(
&mut self,
packets: &mut Vec<crate::transport::PacketBuffer>,
commit: &mut SessionReceiveBatchCommit,
) {
if packets.is_empty() && commit.is_empty() {
return;
}
if let Some(tun_tx) = &self.tun_tx {
if !packets.is_empty() {
let _t = crate::perf_profile::Timer::start(crate::perf_profile::Stage::TunWrite);
let dropped = tun_tx.send_batch(packets.drain(..));
if dropped != 0 {
debug!(
dropped,
"Failed to deliver decompressed IPv6 packet batch to TUN"
);
}
}
} else {
packets.clear();
}
let pending_flush_destinations = std::mem::take(commit).finish(self);
for dest_addr in pending_flush_destinations {
self.flush_pending_packets(&dest_addr).await;
}
}
fn dataplane_authenticated_session_dispatch(
&mut self,
ingress: crate::dataplane::DataplaneFspSessionIngress,
) -> 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(),
"Dispatching dataplane authenticated session"
);
if self.promote_dataplane_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 dataplane authenticated pending epoch"
);
}
let message = AuthenticatedSessionMessage::new(source_peer, plaintext, msg_type);
Some(AuthenticatedSessionDispatch::new(
source_addr,
previous_hop_addr,
ce_flag,
message,
))
}
async fn flush_dataplane_endpoint_session_batch(
&mut self,
endpoint_deliveries: &mut Vec<EndpointDataDelivery>,
endpoint_commit: &mut SessionReceiveBatchCommit,
) {
if endpoint_deliveries.is_empty() && endpoint_commit.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;
let liveness_bookkeeping_allowed = arrived_from_source;
let received_k_bit = fmp.fmp_flags & crate::node::wire::FLAG_KEY_EPOCH != 0;
let _ = self.promote_dataplane_authenticated_pending_fmp_epoch(source_addr, received_k_bit);
if liveness_bookkeeping_allowed {
let _ = self.dataplane.record_authenticated_fmp_mmp_receive(
crate::dataplane::DataplaneAuthenticatedFmpMmpReceive::new(
*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(
fmp,
liveness_bookkeeping_allowed,
path_bookkeeping_allowed,
);
if let Some(update) = bookkeeping {
if update.path_bookkeeping_recorded || update.liveness_bookkeeping_recorded {
self.clear_retry_unless_direct_refresh_needed(source_addr);
}
if update.address_changed {
self.sync_dataplane_fmp_owner(source_addr);
}
}
}
pub(in crate::node) async fn handle_dataplane_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,
"dataplane",
)
.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.dataplane.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 dataplane-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.dataplane.record_fsp_decrypt_failure(src_addr) else {
debug!(
src = %self.peer_display_name(&src_addr),
counter,
source,
"FSP AEAD failure for missing dataplane 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;
}
}