diff --git a/Cargo.lock b/Cargo.lock index 06145df3e1..dc7fa8462b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2956,6 +2956,15 @@ dependencies = [ "libc", ] +[[package]] +name = "etherparse" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "827292ea592108849932ad8e30218f8b1f21c0dfd0696698a18b5d0aed62d990" +dependencies = [ + "arrayvec", +] + [[package]] name = "event-listener" version = "2.5.3" @@ -5862,6 +5871,17 @@ dependencies = [ "libc", ] +[[package]] +name = "nix" +version = "0.27.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2eb04e9c688eff1c89d72b407f168cf79bb9e867a9d3323ed6c01519eb9cc053" +dependencies = [ + "bitflags 2.4.0", + "cfg-if", + "libc", +] + [[package]] name = "no-std-compat" version = "0.4.1" @@ -7517,12 +7537,15 @@ dependencies = [ "base64 0.21.4", "boringtun", "bytes", + "dashmap", + "etherparse", "futures", "log", "nym-task", "tap", "thiserror", "tokio", + "tokio-tun", ] [[package]] @@ -10807,6 +10830,18 @@ dependencies = [ "tokio-stream", ] +[[package]] +name = "tokio-tun" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c4a67d1405a577ba1f4cd61f46608f1db2cadbb6a9549c3fc2eed7f1195393c9" +dependencies = [ + "libc", + "nix 0.27.1", + "thiserror", + "tokio", +] + [[package]] name = "tokio-tungstenite" version = "0.20.1" diff --git a/common/wireguard/Cargo.toml b/common/wireguard/Cargo.toml index ac54018cc1..961be993c5 100644 --- a/common/wireguard/Cargo.toml +++ b/common/wireguard/Cargo.toml @@ -19,9 +19,12 @@ base64 = "0.21.3" #boringtun = "0.6.0" boringtun = { git = "https://github.com/cloudflare/boringtun", rev = "e1d6360d6ab4529fc942a078e4c54df107abe2ba" } bytes = "1.5.0" +dashmap = "5.5.3" +etherparse = "0.13.0" futures = "0.3.28" log.workspace = true nym-task = { path = "../task" } tap.workspace = true thiserror.workspace = true -tokio = { workspace = true, features = ["rt-multi-thread", "net"]} +tokio = { workspace = true, features = ["rt-multi-thread", "net", "io-util"] } +tokio-tun = "0.9.0" diff --git a/common/wireguard/src/event.rs b/common/wireguard/src/event.rs index 4808517434..7d425cd5d6 100644 --- a/common/wireguard/src/event.rs +++ b/common/wireguard/src/event.rs @@ -5,8 +5,6 @@ use bytes::Bytes; #[allow(unused)] #[derive(Debug, Clone)] pub enum Event { - /// Dumb event with no data. - Dumb, /// IP packet received from the WireGuard tunnel that should be passed through to the corresponding virtual device/internet. /// Original implementation also has protocol here since it understands it, but we'll have to infer it downstream WgPacket(Bytes), @@ -17,9 +15,6 @@ pub enum Event { impl Display for Event { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { match self { - Event::Dumb => { - write!(f, "Dumb{{}}") - } Event::WgPacket(data) => { let size = data.len(); write!(f, "WgPacket{{ size={size} }}") diff --git a/common/wireguard/src/lib.rs b/common/wireguard/src/lib.rs index 75119c723f..870da2da04 100644 --- a/common/wireguard/src/lib.rs +++ b/common/wireguard/src/lib.rs @@ -1,12 +1,22 @@ -use std::{collections::HashMap, net::SocketAddr, sync::Arc}; +use std::{ + net::{Ipv4Addr, SocketAddr}, + sync::Arc, +}; use base64::{engine::general_purpose, Engine as _}; use boringtun::x25519; +use dashmap::DashMap; +use etherparse::{InternetSlice, SlicedPacket}; use futures::StreamExt; use log::{error, info}; use nym_task::TaskClient; use tap::TapFallible; -use tokio::{net::UdpSocket, sync::mpsc, task::JoinHandle}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::UdpSocket, + sync::mpsc::{self, UnboundedSender}, + task::JoinHandle, +}; use tun::WireGuardTunnel; use crate::event::Event; @@ -17,8 +27,14 @@ mod error; mod event; mod tun; -//const WG_ADDRESS = "0.0.0.0:51820"; -const WG_ADDRESS: &str = "0.0.0.0:51822"; +// The wireguard UDP listener +const WG_ADDRESS: &str = "0.0.0.0"; +const WG_PORT: u16 = 51822; + +// The interface used to route traffic +const TUN_BASE_NAME: &str = "nymtun"; +const TUN_DEVICE_ADDRESS: &str = "10.0.0.1"; +const TUN_DEVICE_NETMASK: &str = "255.255.255.0"; // The private key of the listener // Corresponding public key: "WM8s8bYegwMa0TJ+xIwhk+dImk2IpDUKslDBCZPizlE=" @@ -27,11 +43,15 @@ const PRIVATE_KEY: &str = "AEqXrLFT4qjYq3wmX0456iv94uM6nDj5ugp6Jedcflg="; // The public keys of the registered peers (clients) const PEERS: &[&str; 1] = &[ // Corresponding private key: "ILeN6gEh6vJ3Ju8RJ3HVswz+sPgkcKtAYTqzQRhTtlo=" - "NCIhkgiqxFx1ckKl3Zuh595DzIFl8mxju1Vg995EZhI=", // "mxV/mw7WZTe+0Msa0kvJHMHERDA/cSskiZWQce+TdEs=", + "NCIhkgiqxFx1ckKl3Zuh595DzIFl8mxju1Vg995EZhI=", + // Another key + // "mxV/mw7WZTe+0Msa0kvJHMHERDA/cSskiZWQce+TdEs=", ]; const MAX_PACKET: usize = 65535; +type ActivePeers = DashMap>; + fn init_static_dev_keys() -> (x25519::StaticSecret, x25519::PublicKey) { // TODO: this is a temporary solution for development let static_private_bytes: [u8; 32] = general_purpose::STANDARD @@ -58,31 +78,109 @@ fn init_static_dev_keys() -> (x25519::StaticSecret, x25519::PublicKey) { } fn start_wg_tunnel( - addr: SocketAddr, + endpoint: SocketAddr, udp: Arc, static_private: x25519::StaticSecret, peer_static_public: x25519::PublicKey, + tunnel_tx: UnboundedSender>, ) -> (JoinHandle, mpsc::UnboundedSender) { - let (mut tunnel, peer_tx) = WireGuardTunnel::new(udp, addr, static_private, peer_static_public); + let (mut tunnel, peer_tx) = + WireGuardTunnel::new(udp, endpoint, static_private, peer_static_public, tunnel_tx); let join_handle = tokio::spawn(async move { tunnel.spin_off().await; - addr + endpoint }); (join_handle, peer_tx) } -pub async fn start_wg_listener( +fn setup_tokio_tun_device(name: &str, address: Ipv4Addr, netmask: Ipv4Addr) -> tokio_tun::Tun { + log::info!("Creating TUN device with: address={address}, netmask={netmask}"); + tokio_tun::Tun::builder() + .name(name) + .tap(false) + .packet_info(false) + .mtu(1350) + .up() + .address(address) + .netmask(netmask) + .try_build() + .expect("Failed to setup tun device, do you have permission?") +} + +fn start_tun_device(_active_peers: Arc) -> UnboundedSender> { + let tun = setup_tokio_tun_device( + format!("{}%d", TUN_BASE_NAME).as_str(), + TUN_DEVICE_ADDRESS.parse().unwrap(), + TUN_DEVICE_NETMASK.parse().unwrap(), + ); + log::info!("Created TUN device: {}", tun.name()); + + let (mut tun_device_rx, mut tun_device_tx) = tokio::io::split(tun); + + // Channels to communicate with the other tasks + let (tun_task_tx, mut tun_task_rx) = mpsc::unbounded_channel::>(); + + tokio::spawn(async move { + let mut buf = [0u8; 1024]; + loop { + tokio::select! { + // Reading from the TUN device + len = tun_device_rx.read(&mut buf) => match len { + Ok(len) => { + let packet = &buf[..len]; + let dst_addr = boringtun::noise::Tunn::dst_address(packet).unwrap(); + + let headers = SlicedPacket::from_ip(packet).unwrap(); + let src_addr = match headers.ip.unwrap() { + InternetSlice::Ipv4(ip, _) => ip.source_addr().to_string(), + InternetSlice::Ipv6(ip, _) => ip.source_addr().to_string(), + }; + log::info!("iface: read Packet({src_addr} -> {dst_addr}, {len} bytes)"); + + // TODO: route packet to the correct peer. + log::info!("...forward packet to the correct peer (NOT YET IMPLEMENTED)"); + }, + Err(err) => { + log::info!("iface: read error: {err}"); + break; + } + }, + + // Writing to the TUN device + Some(data) = tun_task_rx.recv() => { + let headers = SlicedPacket::from_ip(&data).unwrap(); + let (source_addr, destination_addr) = match headers.ip.unwrap() { + InternetSlice::Ipv4(ip, _) => (ip.source_addr(), ip.destination_addr()), + InternetSlice::Ipv6(_, _) => unimplemented!(), + }; + + log::info!( + "iface: write Packet({source_addr} -> {destination_addr}, {} bytes)", + data.len() + ); + // log::info!("iface: writing {} bytes", data.len()); + tun_device_tx.write_all(&data).await.unwrap(); + } + } + } + log::info!("TUN device shutting down"); + }); + tun_task_tx +} + +async fn start_udp_listener( + tun_task_tx: UnboundedSender>, + active_peers: Arc, mut task_client: TaskClient, ) -> Result<(), Box> { - log::info!("Starting wireguard listener on {}", WG_ADDRESS); - let udp_socket = Arc::new(UdpSocket::bind(WG_ADDRESS).await?); + let wg_address = SocketAddr::new(WG_ADDRESS.parse().unwrap(), WG_PORT); + log::info!("Starting wireguard UDP listener on {wg_address}"); + let udp_socket = Arc::new(UdpSocket::bind(wg_address).await?); // Setup some static keys for development let (static_private, peer_static_public) = init_static_dev_keys(); tokio::spawn(async move { - // The set of active tunnels indexed by the peer's address - let mut active_peers: HashMap> = HashMap::new(); // Each tunnel is run in its own task, and the task handle is stored here so we can remove // it from `active_peers` when the tunnel is closed let mut active_peers_task_handles = futures::stream::FuturesUnordered::new(); @@ -91,46 +189,50 @@ pub async fn start_wg_listener( while !task_client.is_shutdown() { tokio::select! { _ = task_client.recv() => { - log::trace!("WireGuard listener: received shutdown"); + log::trace!("WireGuard UDP listener: received shutdown"); break; } // Handle tunnel closing Some(addr) = active_peers_task_handles.next() => { match addr { Ok(addr) => { - info!("WireGuard listener: closed {addr:?}"); + log::info!("Removing peer: {addr:?}"); active_peers.remove(&addr); } Err(err) => { - error!("WireGuard listener: error receiving shutdown from peer: {err}"); + error!("WireGuard UDP listener: error receiving shutdown from peer: {err}"); } } - } + }, // Handle incoming packets Ok((len, addr)) = udp_socket.recv_from(&mut buf) => { - log::info!("Received {} bytes from {}", len, addr); + log::trace!("udp: received {} bytes from {}", len, addr); if let Some(peer_tx) = active_peers.get_mut(&addr) { - log::info!("WireGuard listener: received packet from known peer"); + log::info!("udp: received {len} bytes from {addr} from known peer"); peer_tx.send(Event::WgPacket(buf[..len].to_vec().into())) .tap_err(|err| log::error!("{err}")) .unwrap(); } else { - log::info!("WireGuard listener: received packet from unknown peer, starting tunnel"); + log::info!("udp: received {len} bytes from {addr} from unknown peer, starting tunnel"); let (join_handle, peer_tx) = start_wg_tunnel( addr, udp_socket.clone(), static_private.clone(), - peer_static_public + peer_static_public, + tun_task_tx.clone(), ); peer_tx.send(Event::WgPacket(buf[..len].to_vec().into())) .tap_err(|err| log::error!("{err}")) .unwrap(); + // WIP(JON): active peers should probably be keyed by peer_static_public + // instead. Does this current setup lead to any issues? + log::info!("Adding peer: {addr}"); active_peers.insert(addr, peer_tx); active_peers_task_handles.push(join_handle); } - } + }, } } log::info!("WireGuard listener: shutting down"); @@ -138,3 +240,18 @@ pub async fn start_wg_listener( Ok(()) } + +pub async fn start_wireguard( + task_client: TaskClient, +) -> Result<(), Box> { + // The set of active tunnels indexed by the peer's address + let active_peers: Arc = Arc::new(ActivePeers::new()); + + // Start the tun device that is used to relay traffic outbound + let tun_task_tx = start_tun_device(active_peers.clone()); + + // Start the UDP listener that clients connect to + start_udp_listener(tun_task_tx, active_peers, task_client).await?; + + Ok(()) +} diff --git a/common/wireguard/src/tun.rs b/common/wireguard/src/tun.rs index ae1819b6fa..f277118e15 100644 --- a/common/wireguard/src/tun.rs +++ b/common/wireguard/src/tun.rs @@ -6,6 +6,7 @@ use boringtun::{ x25519, }; use bytes::Bytes; +use etherparse::{InternetSlice, SlicedPacket}; use log::{debug, error, info, warn}; use tap::TapFallible; use tokio::{ @@ -20,13 +21,16 @@ const MAX_PACKET: usize = 65535; pub struct WireGuardTunnel { // Incoming data from the UDP socket received in the main event loop - udp_rx: mpsc::UnboundedReceiver, + peer_rx: mpsc::UnboundedReceiver, // UDP socket used for sending data udp: Arc, // Peer endpoint - addr: SocketAddr, + endpoint: SocketAddr, + + // The source address of the last packet received from the peer + source_addr: Arc>>, // `boringtun` tunnel, used for crypto & WG protocol wg_tunnel: Arc>, @@ -34,6 +38,9 @@ pub struct WireGuardTunnel { // Signal close close_tx: broadcast::Sender<()>, close_rx: broadcast::Receiver<()>, + + // Send data to the task that handles sending data through the tun device + tun_task_tx: mpsc::UnboundedSender>, } impl Drop for WireGuardTunnel { @@ -44,16 +51,17 @@ impl Drop for WireGuardTunnel { } impl WireGuardTunnel { - fn close(&self) { - let _ = self.close_tx.send(()); - } - pub fn new( udp: Arc, - addr: SocketAddr, + endpoint: SocketAddr, static_private: x25519::StaticSecret, peer_static_public: x25519::PublicKey, + tunnel_tx: mpsc::UnboundedSender>, ) -> (Self, mpsc::UnboundedSender) { + let local_addr = udp.local_addr().unwrap(); + let peer_addr = udp.peer_addr(); + log::info!("New wg tunnel: endpoint: {endpoint}, local_addr: {local_addr}, peer_addr: {peer_addr:?}"); + let preshared_key = None; let persistent_keepalive = None; let index = 0; @@ -72,21 +80,27 @@ impl WireGuardTunnel { )); // Channels with incoming data that is received by the main event loop - let (udp_tx, udp_rx) = mpsc::unbounded_channel(); + let (peer_tx, peer_rx) = mpsc::unbounded_channel(); // Signal close tunnel let (close_tx, close_rx) = broadcast::channel(1); let tunnel = WireGuardTunnel { - udp_rx, + peer_rx, udp, - addr, + endpoint, + source_addr: Default::default(), wg_tunnel, close_tx, close_rx, + tun_task_tx: tunnel_tx, }; - (tunnel, udp_tx) + (tunnel, peer_tx) + } + + fn close(&self) { + let _ = self.close_tx.send(()); } pub async fn spin_off(&mut self) { @@ -96,9 +110,9 @@ impl WireGuardTunnel { info!("WireGuard tunnel: received msg to close"); break; }, - packet = self.udp_rx.recv() => match packet { + packet = self.peer_rx.recv() => match packet { Some(packet) => { - info!("WireGuard tunnel received: {packet}"); + info!("event loop: {packet}"); match packet { Event::WgPacket(data) => { let _ = self.consume_wg(&data) @@ -106,7 +120,6 @@ impl WireGuardTunnel { .tap_err(|err| error!("WireGuard tunnel: consume_wg error: {err}")); }, Event::IpPacket(data) => self.consume_eth(&data).await, - _ => {}, } }, None => { @@ -121,7 +134,7 @@ impl WireGuardTunnel { }, } } - info!("WireGuard tunnel ({}): closed", self.addr); + info!("WireGuard tunnel ({}): closed", self.endpoint); } async fn wg_tunnel_lock(&self) -> Result, WgError> { @@ -130,21 +143,35 @@ impl WireGuardTunnel { .map_err(|_| WgError::UnableToGetTunnel) } - async fn consume_wg(&self, data: &[u8]) -> Result<(), WgError> { + fn set_source_addr(&self, source_addr: std::net::Ipv4Addr) { + let to_update = { + let stored_source_addr = self.source_addr.read().unwrap(); + stored_source_addr + .map(|sa| sa != source_addr) + .unwrap_or(true) + }; + if to_update { + log::info!("wg tunnel set_source_addr: {source_addr}"); + *self.source_addr.write().unwrap() = Some(source_addr); + } + } + + async fn consume_wg(&mut self, data: &[u8]) -> Result<(), WgError> { let mut send_buf = [0u8; MAX_PACKET]; - let mut peer = self.wg_tunnel_lock().await?; - match peer.decapsulate(None, data, &mut send_buf) { + let mut tunnel = self.wg_tunnel_lock().await?; + match tunnel.decapsulate(None, data, &mut send_buf) { TunnResult::WriteToNetwork(packet) => { - debug!("WireGuard: writing to network"); - if let Err(err) = self.udp.send_to(packet, self.addr).await { + log::info!("udp: send {} bytes to {}", packet.len(), self.endpoint); + if let Err(err) = self.udp.send_to(packet, self.endpoint).await { error!("Failed to send decapsulation-instructed packet to WireGuard endpoint: {err:?}"); }; // Flush pending queue loop { let mut send_buf = [0u8; MAX_PACKET]; - match peer.decapsulate(None, &[], &mut send_buf) { + match tunnel.decapsulate(None, &[], &mut send_buf) { TunnResult::WriteToNetwork(packet) => { - if let Err(err) = self.udp.send_to(packet, self.addr).await { + log::info!("udp: send {} bytes to {}", packet.len(), self.endpoint); + if let Err(err) = self.udp.send_to(packet, self.endpoint).await { error!("Failed to send decapsulation-instructed packet to WireGuard endpoint: {err:?}"); break; }; @@ -156,12 +183,13 @@ impl WireGuardTunnel { } } TunnResult::WriteToTunnelV4(packet, _) | TunnResult::WriteToTunnelV6(packet, _) => { - debug!("WireGuard: writing to tunnel"); - info!( - "WireGuard endpoint sent IP packet of {} bytes (not yet implemented)", - packet.len() - ); - // TODO + let headers = SlicedPacket::from_ip(packet).unwrap(); + let (source_addr, _destination_addr) = match headers.ip.unwrap() { + InternetSlice::Ipv4(ip, _) => (ip.source_addr(), ip.destination_addr()), + InternetSlice::Ipv6(_, _) => unimplemented!(), + }; + self.set_source_addr(source_addr); + self.tun_task_tx.send(packet.to_vec()).unwrap(); } TunnResult::Done => { debug!("WireGuard: decapsulate done"); @@ -173,9 +201,35 @@ impl WireGuardTunnel { Ok(()) } - async fn consume_eth(&self, _data: &Bytes) { - info!("WireGuard tunnel: consume_eth"); - todo!(); + async fn consume_eth(&self, data: &Bytes) { + info!("consume_eth: raw packet size: {}", data.len()); + let encapsulated_packet = self.encapsulate_packet(data).await; + info!( + "consume_eth: after encapsulate: {}", + encapsulated_packet.len() + ); + + info!("consume_eth: send to {}: {}", self.endpoint, data.len()); + self.udp + .send_to(&encapsulated_packet, self.endpoint) + .await + .unwrap(); + } + + async fn encapsulate_packet(&self, payload: &[u8]) -> Vec { + // TODO: use fixed dst and src buffers that we can reuse + let len = 148.max(payload.len() + 32); + let mut dst = vec![0; len]; + + let mut wg_tunnel = self.wg_tunnel_lock().await.unwrap(); + + match wg_tunnel.encapsulate(payload, &mut dst) { + TunnResult::WriteToNetwork(packet) => packet.to_vec(), + unexpected => { + error!("{:?}", unexpected); + vec![] + } + } } async fn update_wg_timers(&mut self) -> Result<(), WgError> { @@ -190,16 +244,18 @@ impl WireGuardTunnel { async fn handle_routine_tun_result<'a: 'async_recursion>(&self, result: TunnResult<'a>) { match result { TunnResult::WriteToNetwork(packet) => { - info!( - "Sending routine packet of {} bytes to WireGuard endpoint", + log::info!( + "routine: write to network: {}: {}", + self.endpoint, packet.len() ); - if let Err(err) = self.udp.send_to(packet, self.addr).await { - error!("Failed to send routine packet to WireGuard endpoint: {err:?}",); + if let Err(err) = self.udp.send_to(packet, self.endpoint).await { + error!("routine: failed to send packet: {err:?}"); }; } TunnResult::Err(WireGuardError::ConnectionExpired) => { warn!("Wireguard handshake has expired!"); + // WIP(JON): consider just closing the tunnel here let mut buf = vec![0u8; MAX_PACKET]; let Ok(mut peer) = self.wg_tunnel_lock().await else { warn!("Failed to lock WireGuard peer, closing tunnel"); diff --git a/gateway/src/node/mod.rs b/gateway/src/node/mod.rs index 80bd505ccf..6a59b1e3e7 100644 --- a/gateway/src/node/mod.rs +++ b/gateway/src/node/mod.rs @@ -370,7 +370,7 @@ impl Gateway { // Once this is a bit more mature, make this a commandline flag instead of a compile time // flag #[cfg(feature = "wireguard")] - if let Err(err) = nym_wireguard::start_wg_listener(shutdown.subscribe()).await { + if let Err(err) = nym_wireguard::start_wireguard(shutdown.subscribe()).await { // that's a nasty workaround, but anyhow errors are generally nicer, especially on exit bail!("{err}") }