From 7776c0f4ca21432704cc6edce572755316b23659 Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Fri, 13 Feb 2026 04:19:30 +0000 Subject: [PATCH] =?UTF-8?q?Data=20plane=20integration:=20TUN=20outbound=20?= =?UTF-8?q?=E2=86=92=20session=20encryption=20=E2=86=92=20mesh=20delivery?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the TUN reader to the session layer for end-to-end IPv6 packet delivery through the mesh. TUN reader forwards FIPS-destined packets (fd::/8) to Node via tokio::sync::mpsc channel. Identity cache maps FipsAddress prefix bytes to NodeAddr + PublicKey for reverse address resolution, populated at peer promotion and session creation. Outbound handler validates IPv6, looks up destination identity, routes through established session or initiates new one with pending packet queue (16/dest, 256 dests max). Queue flushed on session establishment. ICMPv6 Destination Unreachable for unknown destinations. 6 new tests including 3-node forwarded TUN data and pending queue flush. 410 tests pass, clean build. --- src/identity.rs | 2 +- src/node/handlers/handshake.rs | 2 + src/node/handlers/rx_loop.rs | 17 ++ src/node/handlers/session.rs | 119 ++++++++++++++ src/node/lifecycle.rs | 6 +- src/node/mod.rs | 41 ++++- src/node/tests/session.rs | 275 +++++++++++++++++++++++++++++++++ src/tun.rs | 42 +++-- 8 files changed, 489 insertions(+), 15 deletions(-) diff --git a/src/identity.rs b/src/identity.rs index 4e374d3..31989df 100644 --- a/src/identity.rs +++ b/src/identity.rs @@ -22,7 +22,7 @@ const NSEC_HRP: Hrp = Hrp::parse_unchecked("nsec"); const AUTH_DOMAIN: &[u8] = b"fips-auth-v1"; /// FIPS address prefix (IPv6 ULA range). -const FIPS_ADDRESS_PREFIX: u8 = 0xfd; +pub const FIPS_ADDRESS_PREFIX: u8 = 0xfd; /// Errors that can occur in identity operations. #[derive(Debug, Error)] diff --git a/src/node/handlers/handshake.rs b/src/node/handlers/handshake.rs index b623abe..a471e44 100644 --- a/src/node/handlers/handshake.rs +++ b/src/node/handlers/handshake.rs @@ -555,6 +555,7 @@ impl Node { self.peers_by_index .insert((transport_id, our_index.as_u32()), peer_node_addr); self.retry_pending.remove(&peer_node_addr); + self.register_identity(peer_node_addr, verified_identity.pubkey_full()); info!( node_addr = %peer_node_addr, @@ -633,6 +634,7 @@ impl Node { self.peers_by_index .insert((transport_id, our_index.as_u32()), peer_node_addr); self.retry_pending.remove(&peer_node_addr); + self.register_identity(peer_node_addr, verified_identity.pubkey_full()); info!( node_addr = %peer_node_addr, diff --git a/src/node/handlers/rx_loop.rs b/src/node/handlers/rx_loop.rs index 59b6c22..6b0da69 100644 --- a/src/node/handlers/rx_loop.rs +++ b/src/node/handlers/rx_loop.rs @@ -15,6 +15,9 @@ impl Node { /// - 0x01: Handshake message 1 (initiator -> responder) /// - 0x02: Handshake message 2 (responder -> initiator) /// + /// Also processes outbound IPv6 packets from the TUN reader for session + /// encapsulation and routing through the mesh. + /// /// Also runs a periodic tick (1s) to clean up stale handshake connections /// that never received a response. This prevents resource leaks when peers /// are unreachable. @@ -25,6 +28,17 @@ impl Node { let mut packet_rx = self.packet_rx.take() .ok_or(NodeError::NotStarted)?; + // Take the TUN outbound receiver, or create a dummy channel that never + // produces messages (when TUN is disabled). Holding the sender prevents + // the channel from closing. + let (mut tun_outbound_rx, _tun_guard) = match self.tun_outbound_rx.take() { + Some(rx) => (rx, None), + None => { + let (tx, rx) = tokio::sync::mpsc::channel(1); + (rx, Some(tx)) + } + }; + let mut tick = tokio::time::interval(Duration::from_secs(1)); info!("RX event loop started"); @@ -37,6 +51,9 @@ impl Node { None => break, // channel closed } } + Some(ipv6_packet) = tun_outbound_rx.recv() => { + self.handle_tun_outbound(ipv6_packet).await; + } _ = tick.tick() => { self.check_timeouts(); let now_ms = std::time::SystemTime::now() diff --git a/src/node/handlers/session.rs b/src/node/handlers/session.rs index 75af5df..f0e6888 100644 --- a/src/node/handlers/session.rs +++ b/src/node/handlers/session.rs @@ -127,6 +127,9 @@ impl Node { } }; + // Register the initiator's identity for future TUN → session routing + self.register_identity(*src_addr, remote_pubkey); + // Generate msg2 let msg2 = match handshake.write_message_2() { Ok(m) => m, @@ -214,6 +217,9 @@ impl Node { let now_ms = Self::now_ms(); self.coord_cache.insert(*src_addr, ack.src_coords, now_ms); + // Flush any queued outbound packets for this destination + self.flush_pending_packets(src_addr).await; + debug!(src = %src_addr, "Session established (initiator)"); } @@ -307,6 +313,10 @@ impl Node { "DataPacket decrypted (no TUN interface, plaintext dropped)" ); } + + // Flush any pending outbound packets (e.g., simultaneous initiation + // where responder also had queued outbound packets) + self.flush_pending_packets(src_addr).await; } /// Handle a CoordsRequired error signal from a transit router. @@ -393,6 +403,9 @@ impl Node { // Route toward destination self.send_session_datagram(&datagram).await?; + // Register destination identity for TUN → session routing + self.register_identity(dest_addr, dest_pubkey); + // Store session entry let now_ms = Self::now_ms(); let entry = SessionEntry::new(dest_addr, dest_pubkey, EndToEndState::Initiating(handshake), now_ms); @@ -496,4 +509,110 @@ impl Node { .map(|d| d.as_millis() as u64) .unwrap_or(0) } + + // === TUN Outbound (Data Plane) === + + /// Maximum pending packets per destination during session establishment. + const MAX_PENDING_PER_DEST: usize = 16; + /// Maximum destinations with pending packets. + const MAX_PENDING_DESTINATIONS: usize = 256; + + /// Handle an outbound IPv6 packet from the TUN reader. + /// + /// Extracts the destination FipsAddress, looks up the NodeAddr and PublicKey + /// from the identity cache, and either sends through an established session + /// or initiates a new one (queuing the packet until established). + pub(in crate::node) async fn handle_tun_outbound(&mut self, ipv6_packet: Vec) { + // Validate IPv6 header + if ipv6_packet.len() < 40 || ipv6_packet[0] >> 4 != 6 { + return; + } + + // Extract destination FipsAddress prefix (IPv6 dest bytes 1-15) + // IPv6 header: bytes 24-39 are dest addr, so prefix = bytes 25-39 + let mut prefix = [0u8; 15]; + prefix.copy_from_slice(&ipv6_packet[25..40]); + + // Look up in identity cache + let (dest_addr, dest_pubkey) = match self.lookup_by_fips_prefix(&prefix) { + Some(&(addr, pk)) => (addr, pk), + None => { + self.send_icmpv6_dest_unreachable(&ipv6_packet); + return; + } + }; + + // Check for established session + if let Some(entry) = self.sessions.get(&dest_addr) { + if entry.is_established() { + if let Err(e) = self.send_session_data(&dest_addr, &ipv6_packet).await { + debug!(dest = %dest_addr, error = %e, "Failed to send TUN packet via session"); + } + return; + } + // Session exists but not yet established — queue the packet + self.queue_pending_packet(dest_addr, ipv6_packet); + return; + } + + // No session: initiate one and queue the packet + if let Err(e) = self.initiate_session(dest_addr, dest_pubkey).await { + debug!(dest = %dest_addr, error = %e, "Failed to initiate session for TUN packet"); + self.send_icmpv6_dest_unreachable(&ipv6_packet); + return; + } + self.queue_pending_packet(dest_addr, ipv6_packet); + } + + /// Send ICMPv6 Destination Unreachable back through TUN. + fn send_icmpv6_dest_unreachable(&self, original_packet: &[u8]) { + use crate::icmp::{build_dest_unreachable, should_send_icmp_error, DestUnreachableCode}; + use crate::FipsAddress; + + if !should_send_icmp_error(original_packet) { + return; + } + + let our_ipv6 = FipsAddress::from_node_addr(self.node_addr()).to_ipv6(); + if let Some(response) = build_dest_unreachable( + original_packet, + DestUnreachableCode::NoRoute, + our_ipv6, + ) && let Some(tun_tx) = &self.tun_tx { + let _ = tun_tx.send(response); + } + } + + /// Queue a packet while waiting for session establishment. + fn queue_pending_packet(&mut self, dest_addr: NodeAddr, packet: Vec) { + // Reject if we already have too many pending destinations + if !self.pending_tun_packets.contains_key(&dest_addr) + && self.pending_tun_packets.len() >= Self::MAX_PENDING_DESTINATIONS + { + return; + } + + let queue = self + .pending_tun_packets + .entry(dest_addr) + .or_default(); + if queue.len() >= Self::MAX_PENDING_PER_DEST { + queue.pop_front(); // Drop oldest + } + queue.push_back(packet); + } + + /// Flush pending packets for a destination whose session just reached Established. + async fn flush_pending_packets(&mut self, dest_addr: &NodeAddr) { + let packets = match self.pending_tun_packets.remove(dest_addr) { + Some(q) => q, + None => return, + }; + for packet in packets { + if let Err(e) = self.send_session_data(dest_addr, &packet).await { + debug!(dest = %dest_addr, error = %e, "Failed to send queued TUN packet"); + break; + } + } + } } diff --git a/src/node/lifecycle.rs b/src/node/lifecycle.rs index c800e55..468ea1e 100644 --- a/src/node/lifecycle.rs +++ b/src/node/lifecycle.rs @@ -288,14 +288,18 @@ impl Node { // Clone tun_tx for the reader let reader_tun_tx = tun_tx.clone(); + // Create outbound channel for TUN reader → Node + let (outbound_tx, outbound_rx) = tokio::sync::mpsc::channel(1024); + // Spawn reader thread let reader_handle = thread::spawn(move || { - run_tun_reader(device, mtu, our_addr, reader_tun_tx); + run_tun_reader(device, mtu, our_addr, reader_tun_tx, outbound_tx); }); self.tun_state = TunState::Active; self.tun_name = Some(name); self.tun_tx = Some(tun_tx); + self.tun_outbound_rx = Some(outbound_rx); self.tun_reader_handle = Some(reader_handle); self.tun_writer_handle = Some(writer_handle); } diff --git a/src/node/mod.rs b/src/node/mod.rs index 83f1034..fbfc6ca 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -24,10 +24,10 @@ use crate::transport::{ }; use crate::transport::udp::UdpTransport; use crate::tree::TreeState; -use crate::tun::{TunError, TunState, TunTx}; +use crate::tun::{TunError, TunOutboundRx, TunState, TunTx}; use crate::wire::build_encrypted; use crate::{Config, ConfigError, Identity, IdentityError, NodeAddr}; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::fmt; use std::thread::JoinHandle; use thiserror::Error; @@ -248,6 +248,16 @@ pub struct Node { /// Keyed by remote NodeAddr. sessions: HashMap, + // === Identity Cache === + /// Maps FipsAddress prefix bytes (bytes 1-15) to (NodeAddr, PublicKey). + /// Enables reverse lookup from IPv6 destination to session/routing identity. + identity_cache: HashMap<[u8; 15], (NodeAddr, secp256k1::PublicKey)>, + + // === Pending TUN Packets === + /// Packets queued while waiting for session establishment. + /// Keyed by destination NodeAddr, bounded per-dest and total. + pending_tun_packets: HashMap>>, + // === Resource Limits === /// Maximum connections (0 = unlimited). max_connections: usize, @@ -269,6 +279,8 @@ pub struct Node { tun_name: Option, /// TUN packet sender channel. tun_tx: Option, + /// Receiver for outbound packets from the TUN reader. + tun_outbound_rx: Option, /// TUN reader thread handle. tun_reader_handle: Option>, /// TUN writer thread handle. @@ -343,6 +355,8 @@ impl Node { connections: HashMap::new(), peers: HashMap::new(), sessions: HashMap::new(), + identity_cache: HashMap::new(), + pending_tun_packets: HashMap::new(), max_connections: 256, max_peers: 128, max_links: 256, @@ -351,6 +365,7 @@ impl Node { tun_state, tun_name: None, tun_tx: None, + tun_outbound_rx: None, tun_reader_handle: None, tun_writer_handle: None, index_allocator: IndexAllocator::new(), @@ -395,6 +410,8 @@ impl Node { connections: HashMap::new(), peers: HashMap::new(), sessions: HashMap::new(), + identity_cache: HashMap::new(), + pending_tun_packets: HashMap::new(), max_connections: 256, max_peers: 128, max_links: 256, @@ -403,6 +420,7 @@ impl Node { tun_state, tun_name: None, tun_tx: None, + tun_outbound_rx: None, tun_reader_handle: None, tun_writer_handle: None, index_allocator: IndexAllocator::new(), @@ -785,6 +803,25 @@ impl Node { self.sessions.len() } + // === Identity Cache === + + /// Register a node in the identity cache for FipsAddress → NodeAddr lookup. + pub(crate) fn register_identity(&mut self, node_addr: NodeAddr, pubkey: secp256k1::PublicKey) { + let mut prefix = [0u8; 15]; + prefix.copy_from_slice(&node_addr.as_bytes()[0..15]); + self.identity_cache.insert(prefix, (node_addr, pubkey)); + } + + /// Look up a destination by FipsAddress prefix (bytes 1-15 of the IPv6 address). + pub(crate) fn lookup_by_fips_prefix(&self, prefix: &[u8; 15]) -> Option<&(NodeAddr, secp256k1::PublicKey)> { + self.identity_cache.get(prefix) + } + + /// Number of identity cache entries. + pub fn identity_cache_len(&self) -> usize { + self.identity_cache.len() + } + // === Routing === /// Find next hop for a destination node address. diff --git a/src/node/tests/session.rs b/src/node/tests/session.rs index 990eb16..db92631 100644 --- a/src/node/tests/session.rs +++ b/src/node/tests/session.rs @@ -870,3 +870,278 @@ async fn test_session_100_nodes() { cleanup_nodes(&mut nodes).await; } + +// ============================================================================ +// Data plane integration tests: TUN → session → link → TUN +// ============================================================================ + +/// Build a minimal valid IPv6 packet with given source and destination addresses. +fn build_ipv6_packet(src: &crate::FipsAddress, dst: &crate::FipsAddress, payload: &[u8]) -> Vec { + let payload_len = payload.len() as u16; + let mut packet = vec![0u8; 40 + payload.len()]; + // Version (6) + traffic class high nibble + packet[0] = 0x60; + // Payload length (u16 BE) + packet[4] = (payload_len >> 8) as u8; + packet[5] = (payload_len & 0xff) as u8; + // Next header: 59 = No Next Header + packet[6] = 59; + // Hop limit + packet[7] = 64; + // Source address (bytes 8-23) + packet[8..24].copy_from_slice(src.as_bytes()); + // Destination address (bytes 24-39) + packet[24..40].copy_from_slice(dst.as_bytes()); + // Payload + packet[40..].copy_from_slice(payload); + packet +} + +#[test] +fn test_identity_cache_populated_on_promote() { + use crate::peer::PromotionResult; + + let mut node = make_node(); + let transport_id = TransportId::new(1); + let link_id = LinkId::new(1); + + let (conn, peer_identity) = make_completed_connection( + &mut node, + link_id, + transport_id, + 1000, + ); + + node.add_connection(conn).unwrap(); + + // Promote + let result = node.promote_connection(link_id, peer_identity, 2000).unwrap(); + assert!(matches!(result, PromotionResult::Promoted(_))); + + // Identity cache should contain the peer + let peer_addr = *peer_identity.node_addr(); + let mut prefix = [0u8; 15]; + prefix.copy_from_slice(&peer_addr.as_bytes()[0..15]); + let cached = node.lookup_by_fips_prefix(&prefix); + assert!(cached.is_some(), "Identity cache should contain promoted peer"); + let (cached_addr, cached_pk) = cached.unwrap(); + assert_eq!(*cached_addr, peer_addr); + assert_eq!(*cached_pk, peer_identity.pubkey_full()); +} + +#[tokio::test] +async fn test_tun_outbound_established_session() { + // Two directly connected nodes, session established. + // Inject IPv6 packet via handle_tun_outbound on Node 0, + // verify plaintext arrives at Node 1's tun_tx. + let edges = vec![(0, 1)]; + let mut nodes = run_tree_test(2, &edges, false).await; + verify_tree_convergence(&nodes); + populate_all_coord_caches(&mut nodes); + + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + let node1_pubkey = nodes[1].node.identity().pubkey_full(); + + let src_fips = crate::FipsAddress::from_node_addr(&node0_addr); + let dst_fips = crate::FipsAddress::from_node_addr(&node1_addr); + + // Establish session + nodes[0].node.initiate_session(node1_addr, node1_pubkey).await.unwrap(); + tokio::time::sleep(Duration::from_millis(20)).await; + process_available_packets(&mut nodes).await; // Setup → Node 1 + tokio::time::sleep(Duration::from_millis(20)).await; + process_available_packets(&mut nodes).await; // Ack → Node 0 + + assert!(nodes[0].node.get_session(&node1_addr).unwrap().state().is_established()); + + // Install TUN receiver on Node 1 + let (tun_tx, tun_rx) = std::sync::mpsc::channel(); + nodes[1].node.tun_tx = Some(tun_tx); + + // Build and inject an IPv6 packet + let test_payload = b"data-plane-test-12345"; + let ipv6_packet = build_ipv6_packet(&src_fips, &dst_fips, test_payload); + + nodes[0].node.handle_tun_outbound(ipv6_packet.clone()).await; + + // Process packets: encrypted DataPacket → Node 1 + tokio::time::sleep(Duration::from_millis(20)).await; + process_available_packets(&mut nodes).await; + + // Verify plaintext arrived at Node 1's TUN + let delivered: Vec> = std::iter::from_fn(|| tun_rx.try_recv().ok()).collect(); + assert_eq!(delivered.len(), 1, "Exactly one packet should be delivered"); + assert_eq!(delivered[0], ipv6_packet, "Delivered packet should match original"); + + cleanup_nodes(&mut nodes).await; +} + +#[tokio::test] +async fn test_tun_outbound_triggers_session_initiation() { + // Two connected nodes, no session yet. + // Inject a TUN packet — should trigger session initiation, + // queue the packet, and deliver after handshake completes. + let edges = vec![(0, 1)]; + let mut nodes = run_tree_test(2, &edges, false).await; + verify_tree_convergence(&nodes); + populate_all_coord_caches(&mut nodes); + + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + + let src_fips = crate::FipsAddress::from_node_addr(&node0_addr); + let dst_fips = crate::FipsAddress::from_node_addr(&node1_addr); + + // No session yet + assert_eq!(nodes[0].node.session_count(), 0); + + // Install TUN receiver on Node 1 + let (tun_tx, tun_rx) = std::sync::mpsc::channel(); + nodes[1].node.tun_tx = Some(tun_tx); + + // Build and inject an IPv6 packet (identity cache populated at peer promotion) + let test_payload = b"trigger-session-test"; + let ipv6_packet = build_ipv6_packet(&src_fips, &dst_fips, test_payload); + + nodes[0].node.handle_tun_outbound(ipv6_packet.clone()).await; + + // Session should now be initiating + assert_eq!(nodes[0].node.session_count(), 1); + assert!(nodes[0].node.get_session(&node1_addr).unwrap().state().is_initiating()); + + // Drain packets until session established and queued packet delivered + drain_to_quiescence(&mut nodes).await; + + // Session should be established on Node 0 + assert!(nodes[0].node.get_session(&node1_addr).unwrap().state().is_established()); + + // Verify the queued packet was delivered to Node 1 + let delivered: Vec> = std::iter::from_fn(|| tun_rx.try_recv().ok()).collect(); + assert_eq!(delivered.len(), 1, "Queued packet should be delivered after handshake"); + assert_eq!(delivered[0], ipv6_packet); + + cleanup_nodes(&mut nodes).await; +} + +#[tokio::test] +async fn test_tun_outbound_unknown_destination() { + // Inject a packet for an unknown destination — should get ICMPv6 back + let edges = vec![(0, 1)]; + let mut nodes = run_tree_test(2, &edges, false).await; + verify_tree_convergence(&nodes); + + // Install TUN receiver on Node 0 (for ICMPv6 response) + let (tun_tx, tun_rx) = std::sync::mpsc::channel(); + nodes[0].node.tun_tx = Some(tun_tx); + + let src_fips = crate::FipsAddress::from_node_addr(nodes[0].node.node_addr()); + + // Build a packet to an unknown FIPS address (not in identity cache) + let unknown_addr = NodeAddr::from_bytes([0xAA; 16]); + let unknown_fips = crate::FipsAddress::from_node_addr(&unknown_addr); + let ipv6_packet = build_ipv6_packet(&src_fips, &unknown_fips, b"unknown"); + + nodes[0].node.handle_tun_outbound(ipv6_packet).await; + + // Should receive ICMPv6 Destination Unreachable back on TUN + let delivered: Vec> = std::iter::from_fn(|| tun_rx.try_recv().ok()).collect(); + assert_eq!(delivered.len(), 1, "Should receive ICMPv6 Destination Unreachable"); + // Verify it's an ICMPv6 Destination Unreachable (type 1, code 0) + // ICMPv6 header starts at byte 40, type at byte 40, code at byte 41 + assert!(delivered[0].len() >= 48, "ICMPv6 response too short"); + assert_eq!(delivered[0][6], 58, "Next header should be ICMPv6 (58)"); + assert_eq!(delivered[0][40], 1, "ICMPv6 type should be Destination Unreachable (1)"); + assert_eq!(delivered[0][41], 0, "ICMPv6 code should be No Route (0)"); + + cleanup_nodes(&mut nodes).await; +} + +#[tokio::test] +async fn test_tun_outbound_3node_forwarded() { + // A—B—C: TUN packet from A destined for C, forwarded through B + let edges = vec![(0, 1), (1, 2)]; + let mut nodes = run_tree_test(3, &edges, false).await; + verify_tree_convergence(&nodes); + populate_all_coord_caches(&mut nodes); + + let node0_addr = *nodes[0].node.node_addr(); + let node2_addr = *nodes[2].node.node_addr(); + + let src_fips = crate::FipsAddress::from_node_addr(&node0_addr); + let dst_fips = crate::FipsAddress::from_node_addr(&node2_addr); + + // Register Node 2's identity in Node 0's cache + // (In production, this would come from the discovery protocol or DNS priming) + let node2_pubkey = nodes[2].node.identity().pubkey_full(); + nodes[0].node.register_identity(node2_addr, node2_pubkey); + + // Install TUN receiver on Node 2 + let (tun_tx, tun_rx) = std::sync::mpsc::channel(); + nodes[2].node.tun_tx = Some(tun_tx); + + // Build and inject an IPv6 packet (triggers session initiation to Node 2) + let test_payload = b"forwarded-data-plane"; + let ipv6_packet = build_ipv6_packet(&src_fips, &dst_fips, test_payload); + + nodes[0].node.handle_tun_outbound(ipv6_packet.clone()).await; + + // Drain packets: handshake + queued data delivery + drain_to_quiescence(&mut nodes).await; + + // Session should be established + assert!(nodes[0].node.get_session(&node2_addr).unwrap().state().is_established()); + + // Verify packet delivered to Node 2 + let delivered: Vec> = std::iter::from_fn(|| tun_rx.try_recv().ok()).collect(); + assert_eq!(delivered.len(), 1, "Packet should be delivered to Node 2"); + assert_eq!(delivered[0], ipv6_packet); + + cleanup_nodes(&mut nodes).await; +} + +#[tokio::test] +async fn test_tun_outbound_pending_queue_flush() { + // Send multiple packets before session exists — all should be delivered + let edges = vec![(0, 1)]; + let mut nodes = run_tree_test(2, &edges, false).await; + verify_tree_convergence(&nodes); + populate_all_coord_caches(&mut nodes); + + let node0_addr = *nodes[0].node.node_addr(); + let node1_addr = *nodes[1].node.node_addr(); + + let src_fips = crate::FipsAddress::from_node_addr(&node0_addr); + let dst_fips = crate::FipsAddress::from_node_addr(&node1_addr); + + // Install TUN receiver on Node 1 + let (tun_tx, tun_rx) = std::sync::mpsc::channel(); + nodes[1].node.tun_tx = Some(tun_tx); + + // Send 5 packets before any session exists + let mut packets = Vec::new(); + for i in 0..5u8 { + let payload = format!("queued-pkt-{}", i).into_bytes(); + let ipv6_packet = build_ipv6_packet(&src_fips, &dst_fips, &payload); + packets.push(ipv6_packet.clone()); + nodes[0].node.handle_tun_outbound(ipv6_packet).await; + } + + // First packet triggers session initiation, rest are queued + assert_eq!(nodes[0].node.session_count(), 1); + assert!(nodes[0].node.get_session(&node1_addr).unwrap().state().is_initiating()); + + // Drain until session established and queued packets flushed + drain_to_quiescence(&mut nodes).await; + + assert!(nodes[0].node.get_session(&node1_addr).unwrap().state().is_established()); + + // All 5 packets should have been delivered + let delivered: Vec> = std::iter::from_fn(|| tun_rx.try_recv().ok()).collect(); + assert_eq!(delivered.len(), 5, "All 5 queued packets should be delivered"); + for (i, pkt) in delivered.iter().enumerate() { + assert_eq!(*pkt, packets[i], "Packet {} should match", i); + } + + cleanup_nodes(&mut nodes).await; +} diff --git a/src/tun.rs b/src/tun.rs index 89d9b9b..7eb99e1 100644 --- a/src/tun.rs +++ b/src/tun.rs @@ -19,6 +19,11 @@ use tun::Layer; /// Channel sender for packets to be written to TUN. pub type TunTx = mpsc::Sender>; +/// Channel sender for outbound packets from TUN reader to Node. +pub type TunOutboundTx = tokio::sync::mpsc::Sender>; +/// Channel receiver for outbound packets (consumed by Node's RX loop). +pub type TunOutboundRx = tokio::sync::mpsc::Receiver>; + /// Errors that can occur with TUN operations. #[derive(Debug, Error)] pub enum TunError { @@ -223,8 +228,10 @@ impl TunWriter { /// TUN packet reader loop. /// -/// Reads packets from the TUN device, logs them, and sends ICMPv6 -/// Destination Unreachable responses for packets we can't route. +/// Reads IPv6 packets from the TUN device. Packets destined for FIPS addresses +/// (fd::/8) are forwarded to the Node via the outbound channel for session +/// encapsulation and routing. Non-FIPS packets receive ICMPv6 Destination +/// Unreachable responses. /// /// This is designed to run in a dedicated thread since TUN reads are blocking. /// The loop exits when the TUN interface is deleted (EFAULT) or an unrecoverable @@ -234,6 +241,7 @@ pub fn run_tun_reader( mtu: u16, our_addr: FipsAddress, tun_tx: TunTx, + outbound_tx: TunOutboundTx, ) { use crate::icmp::{build_dest_unreachable, should_send_icmp_error, DestUnreachableCode}; @@ -248,18 +256,30 @@ pub fn run_tun_reader( let packet = &buf[..n]; log_ipv6_packet(packet); - // Currently no routing capability - send ICMPv6 Destination Unreachable - // for all packets that qualify for an error response - if should_send_icmp_error(packet) { - if let Some(response) = build_dest_unreachable( - packet, - DestUnreachableCode::NoRoute, - our_addr.to_ipv6(), - ) { + // Must be a valid IPv6 packet + if packet.len() < 40 || packet[0] >> 4 != 6 { + continue; + } + + // Check if destination is a FIPS address (fd::/8 prefix) + if packet[24] == crate::identity::FIPS_ADDRESS_PREFIX { + // Forward to Node for session encapsulation and routing + if outbound_tx.blocking_send(packet.to_vec()).is_err() { + break; // Channel closed, shutdown + } + } else { + // Non-FIPS destination: send ICMPv6 Destination Unreachable + if should_send_icmp_error(packet) + && let Some(response) = build_dest_unreachable( + packet, + DestUnreachableCode::NoRoute, + our_addr.to_ipv6(), + ) + { debug!( name = %name, len = response.len(), - "Sending ICMPv6 Destination Unreachable" + "Sending ICMPv6 Destination Unreachable (non-FIPS destination)" ); if tun_tx.send(response).is_err() { break;