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;