network: start work on NAT implementation

This commit is contained in:
Alex Zenla
2024-02-09 13:06:00 +00:00
parent e8a3eba57d
commit 360506bbdd
18 changed files with 483 additions and 1993 deletions

View File

@ -16,14 +16,12 @@ futures = { workspace = true }
libc = { workspace = true }
udp-stream = { workspace = true }
smoltcp = { workspace = true }
etherparse = { workspace = true }
async-trait = { workspace = true }
[dependencies.advmac]
path = "../libs/advmac"
[dependencies.ipstack]
path = "../libs/ipstack"
features = ["log"]
[lib]
path = "src/lib.rs"

View File

@ -1,36 +1,75 @@
use crate::raw_socket::{AsyncRawSocket, RawSocket};
use crate::chandev::ChannelDevice;
use crate::nat::NatRouter;
use crate::proxynat::ProxyNatHandlerFactory;
use crate::raw_socket::AsyncRawSocket;
use advmac::MacAddr6;
use anyhow::{anyhow, Result};
use futures::channel::oneshot;
use futures::{try_join, TryStreamExt};
use ipstack::stream::IpStackStream;
use log::{debug, warn};
use futures::TryStreamExt;
use log::warn;
use smoltcp::iface::{Config, Interface, SocketSet};
use smoltcp::time::Instant;
use smoltcp::wire::{HardwareAddress, IpCidr};
use std::os::fd::AsRawFd;
use std::str::FromStr;
use std::thread;
use std::time::Duration;
use tokio::net::TcpStream;
use udp_stream::UdpStream;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::select;
use tokio::sync::mpsc::{channel, Receiver};
pub trait NetworkSlice {
async fn run(&self) -> Result<()>;
#[derive(Clone)]
pub struct NetworkBackend {
network: String,
interface: String,
}
pub struct NetworkBackend {
pub interface: String,
local: LocalNetworkSlice,
internet: InternetNetworkSlice,
enum NetworkStackSelect<'a> {
Receive(&'a [u8]),
Send(Option<Vec<u8>>),
}
struct NetworkStack<'a> {
tx: Receiver<Vec<u8>>,
kdev: AsyncRawSocket,
udev: ChannelDevice,
interface: Interface,
sockets: SocketSet<'a>,
router: NatRouter,
}
impl NetworkStack<'_> {
async fn poll(&mut self, receive_buffer: &mut [u8]) -> Result<()> {
let what = select! {
x = self.tx.recv() => NetworkStackSelect::Send(x),
x = self.kdev.read(receive_buffer) => NetworkStackSelect::Receive(&receive_buffer[0..x?]),
};
match what {
NetworkStackSelect::Send(packet) => {
if let Some(packet) = packet {
self.kdev.write_all(&packet).await?
}
}
NetworkStackSelect::Receive(packet) => {
if let Err(error) = self.router.process(packet).await {
warn!("router failed to process packet: {}", error);
}
self.udev.rx = Some(packet.to_vec());
let timestamp = Instant::now();
self.interface
.poll(timestamp, &mut self.udev, &mut self.sockets);
}
}
Ok(())
}
}
impl NetworkBackend {
pub fn new(network: &str, interface: &str) -> Result<Self> {
Ok(Self {
network: network.to_string(),
interface: interface.to_string(),
local: LocalNetworkSlice::new(network, interface)?,
internet: InternetNetworkSlice::new(interface)?,
})
}
@ -56,116 +95,41 @@ impl NetworkBackend {
Ok(())
}
pub async fn run(&mut self) -> Result<()> {
try_join!(self.local.run(), self.internet.run()).map(|_| ())
}
}
#[derive(Clone)]
struct LocalNetworkSlice {
network: String,
interface: String,
}
impl LocalNetworkSlice {
fn new(network: &str, interface: &str) -> Result<Self> {
Ok(Self {
network: network.to_string(),
interface: interface.to_string(),
})
pub async fn run(&self) -> Result<()> {
let mut stack = self.create_network_stack()?;
let mut buffer = vec![0u8; 1500];
loop {
stack.poll(&mut buffer).await?;
}
}
fn run_blocking(&self) -> Result<()> {
fn create_network_stack(&self) -> Result<NetworkStack> {
let proxy = Box::new(ProxyNatHandlerFactory::new());
let address = IpCidr::from_str(&self.network)
.map_err(|_| anyhow!("failed to parse cidr: {}", self.network))?;
let addresses: Vec<IpCidr> = vec![address];
let mut socket = RawSocket::new(&self.interface)?;
let kdev = AsyncRawSocket::bind(&self.interface)?;
let (sender, receiver) = channel::<Vec<u8>>(4);
let mut udev = ChannelDevice::new(1500, sender);
let mac = MacAddr6::random();
let mac = HardwareAddress::Ethernet(smoltcp::wire::EthernetAddress(mac.to_array()));
let mac = smoltcp::wire::EthernetAddress(mac.to_array());
let nat = NatRouter::new(proxy, mac);
let mac = HardwareAddress::Ethernet(mac);
let config = Config::new(mac);
let mut iface = Interface::new(config, &mut socket, Instant::now());
let mut iface = Interface::new(config, &mut udev, Instant::now());
iface.update_ip_addrs(|addrs| {
addrs
.extend_from_slice(&addresses)
.expect("failed to set ip addresses");
});
let mut sockets = SocketSet::new(vec![]);
let fd = socket.as_raw_fd();
loop {
let timestamp = Instant::now();
iface.poll(timestamp, &mut socket, &mut sockets);
smoltcp::phy::wait(fd, iface.poll_delay(timestamp, &sockets))?;
}
}
}
impl NetworkSlice for LocalNetworkSlice {
async fn run(&self) -> Result<()> {
let (tx, rx) = oneshot::channel();
let me = self.clone();
thread::spawn(move || {
let _ = tx.send(me.run_blocking());
});
rx.await?
}
}
struct InternetNetworkSlice {
interface: String,
}
impl InternetNetworkSlice {
pub fn new(interface: &str) -> Result<Self> {
Ok(Self {
interface: interface.to_string(),
let sockets = SocketSet::new(vec![]);
Ok(NetworkStack {
tx: receiver,
kdev,
udev,
interface: iface,
sockets,
router: nat,
})
}
async fn process_stream(stream: IpStackStream) {
match stream {
IpStackStream::Tcp(mut tcp) => {
debug!("tcp: {}", tcp.peer_addr());
if let Ok(mut stream) = TcpStream::connect(tcp.peer_addr()).await {
let _ = tokio::io::copy_bidirectional(&mut tcp, &mut stream).await;
} else {
warn!("failed to connect to tcp address: {}", tcp.peer_addr());
}
}
IpStackStream::Udp(mut udp) => {
debug!("udp: {}", udp.peer_addr());
if let Ok(mut stream) = UdpStream::connect(udp.peer_addr()).await {
let _ = tokio::io::copy_bidirectional(&mut stream, &mut udp).await;
} else {
warn!("failed to connect to udp address: {}", udp.peer_addr());
}
}
IpStackStream::UnknownTransport(u) => {
debug!("unknown transport: {}", u.dst_addr());
}
IpStackStream::UnknownNetwork(packet) => {
debug!("unknown network: {:?}", packet);
}
}
}
}
impl NetworkSlice for InternetNetworkSlice {
async fn run(&self) -> Result<()> {
let mut config = ipstack::IpStackConfig::default();
config.mtu(1500);
config.tcp_timeout(std::time::Duration::from_secs(60));
config.udp_timeout(std::time::Duration::from_secs(10));
let socket = AsyncRawSocket::bind(&self.interface)?;
let mut stack = ipstack::IpStack::new(config, socket);
while let Ok(stream) = stack.accept().await {
tokio::spawn(InternetNetworkSlice::process_stream(stream));
}
Ok(())
}
}

73
network/src/chandev.rs Normal file
View File

@ -0,0 +1,73 @@
use log::warn;
// Referenced https://github.com/vi/wgslirpy/blob/master/crates/libwgslirpy/src/channelized_smoltcp_device.rs
use smoltcp::phy::{Checksum, Device};
use tokio::sync::mpsc::Sender;
pub struct ChannelDevice {
pub mtu: usize,
pub tx: Sender<Vec<u8>>,
pub rx: Option<Vec<u8>>,
}
impl ChannelDevice {
pub fn new(mtu: usize, tx: Sender<Vec<u8>>) -> Self {
Self { mtu, tx, rx: None }
}
}
pub struct RxToken(pub Vec<u8>);
impl Device for ChannelDevice {
type RxToken<'a> = RxToken where Self: 'a;
type TxToken<'a> = &'a mut ChannelDevice where Self: 'a;
fn receive(
&mut self,
_timestamp: smoltcp::time::Instant,
) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> {
self.rx.take().map(|x| (RxToken(x), self))
}
fn transmit(&mut self, _timestamp: smoltcp::time::Instant) -> Option<Self::TxToken<'_>> {
if self.tx.capacity() == 0 {
warn!("ran out of transmission capacity");
return None;
}
Some(self)
}
fn capabilities(&self) -> smoltcp::phy::DeviceCapabilities {
let mut capabilities = smoltcp::phy::DeviceCapabilities::default();
capabilities.medium = smoltcp::phy::Medium::Ethernet;
capabilities.max_transmission_unit = self.mtu;
capabilities.checksum = smoltcp::phy::ChecksumCapabilities::ignored();
capabilities.checksum.tcp = Checksum::Tx;
capabilities.checksum.ipv4 = Checksum::Tx;
capabilities.checksum.icmpv4 = Checksum::Tx;
capabilities.checksum.icmpv6 = Checksum::Tx;
capabilities
}
}
impl smoltcp::phy::RxToken for RxToken {
fn consume<R, F>(mut self, f: F) -> R
where
F: FnOnce(&mut [u8]) -> R,
{
f(&mut self.0[..])
}
}
impl<'a> smoltcp::phy::TxToken for &'a mut ChannelDevice {
fn consume<R, F>(self, len: usize, f: F) -> R
where
F: FnOnce(&mut [u8]) -> R,
{
let mut buffer = vec![0u8; len];
let result = f(&mut buffer[..]);
if let Err(error) = self.tx.try_send(buffer) {
warn!("failed to transmit packet: {}", error);
}
result
}
}

View File

@ -9,6 +9,9 @@ use tokio::time::sleep;
use crate::backend::NetworkBackend;
mod backend;
mod chandev;
mod nat;
mod proxynat;
mod raw_socket;
pub struct NetworkService {

189
network/src/nat.rs Normal file
View File

@ -0,0 +1,189 @@
// Referenced https://github.com/vi/wgslirpy/blob/master/crates/libwgslirpy/src/router.rs as a very interesting way to implement NAT.
// hypha will heavily change how the original code functions however. NatKey was a very useful example of what we need to store in a NAT map.
use anyhow::Result;
use async_trait::async_trait;
use etherparse::IpNumber;
use etherparse::IpPayloadSlice;
use etherparse::Ipv4Slice;
use etherparse::LinkSlice;
use etherparse::NetSlice;
use etherparse::SlicedPacket;
use etherparse::TcpHeaderSlice;
use etherparse::UdpHeaderSlice;
use smoltcp::wire::EthernetAddress;
use smoltcp::wire::IpAddress;
use smoltcp::wire::IpEndpoint;
use std::collections::hash_map::Entry;
use std::collections::HashMap;
use std::fmt::Display;
#[derive(Debug, Copy, Clone, Eq, PartialEq, PartialOrd, Ord, Hash)]
pub enum NatKey {
Tcp {
client: IpEndpoint,
external: IpEndpoint,
},
Udp {
client: IpEndpoint,
external: IpEndpoint,
},
Ping {
client: IpAddress,
external: IpAddress,
},
}
impl Display for NatKey {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
NatKey::Tcp { client, external } => write!(f, "TCP {client} -> {external}"),
NatKey::Udp { client, external } => write!(f, "UDP {client} -> {external}"),
NatKey::Ping { client, external } => write!(f, "Ping {client} -> {external}"),
}
}
}
#[async_trait]
pub trait NatHandler: Send {
async fn receive(&self, packet: &[u8]) -> Result<()>;
}
pub struct NatTable {
inner: HashMap<NatKey, Box<dyn NatHandler>>,
}
impl NatTable {
pub fn new() -> Self {
Self {
inner: HashMap::new(),
}
}
}
#[async_trait]
pub trait NatHandlerFactory: Send {
async fn nat(&self, key: NatKey) -> Option<Box<dyn NatHandler>>;
}
pub struct NatRouter {
_mac: EthernetAddress,
factory: Box<dyn NatHandlerFactory>,
table: NatTable,
}
impl NatRouter {
pub fn new(factory: Box<dyn NatHandlerFactory>, mac: EthernetAddress) -> Self {
Self {
_mac: mac,
factory,
table: NatTable::new(),
}
}
pub async fn process(&mut self, data: &[u8]) -> Result<()> {
let packet = SlicedPacket::from_ethernet(data)?;
let Some(ref link) = packet.link else {
return Ok(());
};
let LinkSlice::Ethernet2(ref ether) = link else {
return Ok(());
};
let _mac = EthernetAddress(ether.destination());
let Some(ref net) = packet.net else {
return Ok(());
};
match net {
NetSlice::Ipv4(ipv4) => {
self.process_ipv4(data, ipv4).await?;
}
_ => {
return Ok(());
}
}
Ok(())
}
pub async fn process_ipv4<'a>(&mut self, data: &[u8], ipv4: &Ipv4Slice<'a>) -> Result<()> {
let source_addr = IpAddress::Ipv4(ipv4.header().source_addr().into());
let dest_addr = IpAddress::Ipv4(ipv4.header().destination_addr().into());
match ipv4.header().protocol() {
IpNumber::TCP => {
self.process_tcp(data, source_addr, dest_addr, ipv4.payload())
.await?;
}
IpNumber::UDP => {
self.process_udp(data, source_addr, dest_addr, ipv4.payload())
.await?;
}
_ => {}
}
Ok(())
}
pub async fn process_tcp<'a>(
&mut self,
data: &'a [u8],
source_addr: IpAddress,
dest_addr: IpAddress,
payload: &IpPayloadSlice<'a>,
) -> Result<()> {
let header = TcpHeaderSlice::from_slice(payload.payload)?;
let source = IpEndpoint::new(source_addr, header.source_port());
let dest = IpEndpoint::new(dest_addr, header.destination_port());
let key = NatKey::Tcp {
client: source,
external: dest,
};
self.process_nat(data, key).await?;
Ok(())
}
pub async fn process_udp<'a>(
&mut self,
data: &'a [u8],
source_addr: IpAddress,
dest_addr: IpAddress,
payload: &IpPayloadSlice<'a>,
) -> Result<()> {
let header = UdpHeaderSlice::from_slice(payload.payload)?;
let source = IpEndpoint::new(source_addr, header.source_port());
let dest = IpEndpoint::new(dest_addr, header.destination_port());
let key = NatKey::Udp {
client: source,
external: dest,
};
self.process_nat(data, key).await?;
Ok(())
}
pub async fn process_nat(&mut self, data: &[u8], key: NatKey) -> Result<()> {
let handler: Option<&mut Box<dyn NatHandler>> = match self.table.inner.entry(key) {
Entry::Occupied(entry) => Some(entry.into_mut()),
Entry::Vacant(entry) => {
if let Some(handler) = self.factory.nat(key).await {
Some(entry.insert(handler))
} else {
None
}
}
};
if let Some(handler) = handler {
handler.receive(data).await?;
}
Ok(())
}
}

135
network/src/proxynat.rs Normal file
View File

@ -0,0 +1,135 @@
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use anyhow::{anyhow, Result};
use async_trait::async_trait;
use etherparse::{SlicedPacket, UdpSlice};
use log::{debug, warn};
use smoltcp::{
phy::{Checksum, ChecksumCapabilities},
wire::{IpAddress, IpEndpoint},
};
use tokio::{
io::{AsyncReadExt, AsyncWriteExt},
select,
sync::mpsc::channel,
};
use tokio::{sync::mpsc::Receiver, sync::mpsc::Sender};
use udp_stream::UdpStream;
use crate::nat::{NatHandler, NatHandlerFactory, NatKey};
pub struct ProxyNatHandlerFactory {}
struct ProxyUdpHandler {
external: IpEndpoint,
sender: Sender<Vec<u8>>,
}
impl ProxyNatHandlerFactory {
pub fn new() -> Self {
Self {}
}
}
#[async_trait]
impl NatHandlerFactory for ProxyNatHandlerFactory {
async fn nat(&self, key: NatKey) -> Option<Box<dyn NatHandler>> {
debug!("creating proxy nat entry for key: {}", key);
match key {
NatKey::Udp {
client: _,
external,
} => {
let (sender, receiver) = channel::<Vec<u8>>(4);
let mut handler = ProxyUdpHandler { external, sender };
if let Err(error) = handler.spawn(receiver).await {
warn!("unable to spawn udp proxy handler: {}", error);
None
} else {
Some(Box::new(handler))
}
}
_ => None,
}
}
}
#[async_trait]
impl NatHandler for ProxyUdpHandler {
async fn receive(&self, data: &[u8]) -> Result<()> {
self.sender.try_send(data.to_vec())?;
Ok(())
}
}
enum ProxySelect {
External(usize),
Internal(Vec<u8>),
Closed,
}
impl ProxyUdpHandler {
async fn spawn(&mut self, receiver: Receiver<Vec<u8>>) -> Result<()> {
let external_addr = match self.external.addr {
IpAddress::Ipv4(addr) => SocketAddr::new(
IpAddr::V4(Ipv4Addr::new(addr.0[0], addr.0[1], addr.0[2], addr.0[3])),
self.external.port,
),
IpAddress::Ipv6(_) => return Err(anyhow!("IPv6 unsupported")),
};
let socket = UdpStream::connect(external_addr).await?;
tokio::spawn(async move {
if let Err(error) = ProxyUdpHandler::process(socket, receiver).await {
warn!("processing of udp proxy failed: {}", error);
}
});
Ok(())
}
async fn process(mut socket: UdpStream, mut receiver: Receiver<Vec<u8>>) -> Result<()> {
let mut checksum = ChecksumCapabilities::ignored();
checksum.udp = Checksum::Tx;
checksum.ipv4 = Checksum::Tx;
checksum.tcp = Checksum::Tx;
let mut external_buffer = vec![0u8; 2048];
loop {
let selection = select! {
x = receiver.recv() => if let Some(data) = x {
ProxySelect::Internal(data)
} else {
ProxySelect::Closed
},
x = socket.read(&mut external_buffer) => ProxySelect::External(x?),
};
match selection {
ProxySelect::External(size) => {
let data = &external_buffer[0..size];
debug!("UDP from external: {:?}", data);
}
ProxySelect::Internal(data) => {
debug!("udp socket to handle data: {:?}", data);
let packet = SlicedPacket::from_ethernet(&data)?;
let Some(ref net) = packet.net else {
continue;
};
let Some(ip) = net.ip_payload_ref() else {
continue;
};
let udp = UdpSlice::from_slice(ip.payload)?;
debug!("UDP from internal: {:?}", udp.payload());
socket.write_all(udp.payload()).await?;
}
ProxySelect::Closed => warn!("UDP socket closed"),
}
}
}
}

View File

@ -1,12 +1,7 @@
use anyhow::Result;
use futures::ready;
use log::debug;
use smoltcp::phy::{Device, DeviceCapabilities, Medium};
use smoltcp::time::Instant;
use std::cell::RefCell;
use std::os::unix::io::{AsRawFd, RawFd};
use std::pin::Pin;
use std::rc::Rc;
use std::task::{Context, Poll};
use std::{io, mem};
use tokio::io::unix::AsyncFd;
@ -121,107 +116,6 @@ impl Drop for RawSocketHandle {
}
}
#[derive(Debug)]
pub struct RawSocket {
lower: Rc<RefCell<RawSocketHandle>>,
mtu: usize,
}
impl AsRawFd for RawSocket {
fn as_raw_fd(&self) -> RawFd {
self.lower.borrow().as_raw_fd()
}
}
impl RawSocket {
pub fn new(name: &str) -> io::Result<RawSocket> {
let mut lower = RawSocketHandle::new(name)?;
lower.bind_interface()?;
let mtu = lower.mtu;
Ok(RawSocket {
lower: Rc::new(RefCell::new(lower)),
mtu,
})
}
}
impl Device for RawSocket {
type RxToken<'a> = RxToken
where
Self: 'a;
type TxToken<'a> = TxToken
where
Self: 'a;
fn capabilities(&self) -> DeviceCapabilities {
let mut capabilities = DeviceCapabilities::default();
capabilities.medium = Medium::Ethernet;
capabilities.max_transmission_unit = self.mtu;
capabilities
}
fn receive(&mut self, _timestamp: Instant) -> Option<(Self::RxToken<'_>, Self::TxToken<'_>)> {
let lower = self.lower.borrow_mut();
let mut buffer = vec![0; self.mtu];
match lower.recv(&mut buffer[..]) {
Ok(size) => {
buffer.resize(size, 0);
let rx = RxToken { buffer };
let tx = TxToken {
lower: self.lower.clone(),
};
Some((rx, tx))
}
Err(err) if err.kind() == io::ErrorKind::WouldBlock => None,
Err(err) => panic!("{}", err),
}
}
fn transmit(&mut self, _timestamp: Instant) -> Option<Self::TxToken<'_>> {
Some(TxToken {
lower: self.lower.clone(),
})
}
}
#[doc(hidden)]
pub struct RxToken {
buffer: Vec<u8>,
}
impl smoltcp::phy::RxToken for RxToken {
fn consume<R, F>(mut self, f: F) -> R
where
F: FnOnce(&mut [u8]) -> R,
{
f(&mut self.buffer[..])
}
}
#[doc(hidden)]
pub struct TxToken {
lower: Rc<RefCell<RawSocketHandle>>,
}
impl smoltcp::phy::TxToken for TxToken {
fn consume<R, F>(self, len: usize, f: F) -> R
where
F: FnOnce(&mut [u8]) -> R,
{
let lower = self.lower.borrow_mut();
let mut buffer = vec![0; len];
let result = f(&mut buffer);
match lower.send(&buffer[..]) {
Ok(_) => {}
Err(err) if err.kind() == io::ErrorKind::WouldBlock => {
debug!("phy: tx failed due to WouldBlock")
}
Err(err) => panic!("{}", err),
}
result
}
}
#[repr(C)]
#[derive(Debug)]
struct Ifreq {