Merge pull request #28 from ipfs-rust/rku/forward-nat

export NAT port forwarding

Author: rkuhn

Cargo.lock

@@ -3,7 +3,7 @@ members = ["cli", "core", "machine", "nat", "router", "."]
 
 [package]
 name = "netsim-embed"
-version = "0.5.2"
+version = "0.6.0"
 authors = ["David Craven <david@craven.ch>"]
 edition = "2018"
 description = "Network simulator."

Cargo.toml

@@ -4,7 +4,7 @@ use std::str::FromStr;
 use thiserror::Error;
 
 /// A range of IPv4 addresses with a common prefix
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, PartialEq, Eq)]
 pub struct Ipv4Range {
     addr: Ipv4Addr,
     bits: u8,

core/src/range.rs

@@ -66,7 +66,7 @@ pub struct Machine<C, E> {
 impl<C, E> Machine<C, E>
 where
     C: Display + Send + 'static,
-    E: FromStr + Send + 'static,
+    E: FromStr + Display + Send + 'static,
     E::Err: std::fmt::Debug + Display + Send + Sync,
 {
     pub async fn new(id: MachineId, plug: Plug, cmd: Command) -> Self {
@@ -117,6 +117,28 @@ impl<C, E> Machine<C, E> {
         self.tx.unbounded_send(cmd).unwrap();
     }
 
+    pub fn drain(&mut self) -> Vec<E> {
+        let mut res = self.buffer.drain(..).collect::<Vec<_>>();
+        if !self.rx.is_terminated() {
+            while let Ok(Some(x)) = self.rx.try_next() {
+                res.push(x);
+            }
+        }
+        res
+    }
+
+    pub fn up(&self) {
+        self.ctrl.unbounded_send(IfaceCtrl::Up).unwrap();
+    }
+
+    pub fn down(&self) {
+        self.ctrl.unbounded_send(IfaceCtrl::Down).unwrap();
+    }
+
+    pub fn namespace(&self) -> Namespace {
+        self.ns
+    }
+
     pub async fn recv(&mut self) -> Option<E> {
         if let Some(ev) = self.buffer.pop_front() {
             Some(ev)
@@ -172,28 +194,6 @@ impl<C, E> Machine<C, E> {
             }
         }
     }
-
-    pub fn drain(&mut self) -> Vec<E> {
-        let mut res = self.buffer.drain(..).collect::<Vec<_>>();
-        if !self.rx.is_terminated() {
-            while let Ok(Some(x)) = self.rx.try_next() {
-                res.push(x);
-            }
-        }
-        res
-    }
-
-    pub fn up(&self) {
-        self.ctrl.unbounded_send(IfaceCtrl::Up).unwrap();
-    }
-
-    pub fn down(&self) {
-        self.ctrl.unbounded_send(IfaceCtrl::Down).unwrap();
-    }
-
-    pub fn namespace(&self) -> Namespace {
-        self.ns
-    }
 }
 
 impl<C, E> Drop for Machine<C, E> {
@@ -215,7 +215,7 @@ fn machine<C, E>(
 ) -> thread::JoinHandle<Result<()>>
 where
     C: Display + Send + 'static,
-    E: FromStr + Send + 'static,
+    E: FromStr + Display + Send + 'static,
     E::Err: std::fmt::Debug + Display + Send + Sync,
 {
     thread::spawn(move || {
@@ -229,6 +229,7 @@ where
 
             let ctrl_task = async {
                 while let Some(ctrl) = ctrl.next().await {
+                    log::debug!("{} CTRL {:?}", id, ctrl);
                     match ctrl {
                         IfaceCtrl::Up => iface.get_ref().put_up()?,
                         IfaceCtrl::Down => iface.get_ref().put_down()?,
@@ -260,7 +261,7 @@ where
                     if buf[0] >> 4 != 4 {
                         continue;
                     }
-                    log::debug!("{} (reader): sending packet", id);
+                    log::trace!("{} (reader): sending packet", id);
                     let mut bytes = buf[..n].to_vec();
                     if let Some(mut packet) = Packet::new(&mut bytes) {
                         packet.set_checksum();
@@ -277,7 +278,7 @@ where
 
             let writer_task = async {
                 while let Some(packet) = rx.next().await {
-                    log::debug!("{} (writer): received packet", id);
+                    log::trace!("{} (writer): received packet", id);
                     // can error if the interface is down
                     if let Ok(n) = iface.write_with(|iface| iface.send(&packet)).await {
                         if n == 0 {
@@ -303,7 +304,7 @@ where
                 let mut buf = Vec::with_capacity(4096);
                 while let Some(cmd) = cmd.next().await {
                     buf.clear();
-                    log::trace!("{}", cmd);
+                    log::debug!("{} {}", id, cmd);
                     writeln!(buf, "{}", cmd)?;
                     stdin.write_all(&buf).await?;
                 }
@@ -321,6 +322,7 @@ where
                             Ok(ev) => ev,
                             Err(err) => return Err(Error::new(ErrorKind::Other, err.to_string())),
                         };
+                        log::debug!("{} {}", id, ev);
                         if event.unbounded_send(ev).is_err() {
                             break;
                         }

machine/src/lib.rs

@@ -17,6 +17,7 @@ enum RouterCtrl {
 
 #[derive(Debug)]
 pub struct Ipv4Router {
+    #[allow(unused)]
     addr: Ipv4Addr,
     ctrl: mpsc::UnboundedSender<RouterCtrl>,
 }
@@ -129,9 +130,9 @@ fn forward_packet(
         for route in routes {
             if route.dest().contains(dest) || dest.is_broadcast() || dest.is_multicast() {
                 if !*en {
-                    log::debug!("router {}: route {:?} disabled", addr, route);
+                    log::trace!("router {}: route {:?} disabled", addr, route);
                 } else {
-                    log::debug!("router {}: routing packet on route {:?}", addr, route);
+                    log::trace!("router {}: routing packet on route {:?}", addr, route);
                     let _ = tx.unbounded_send(bytes.clone());
                     forwarded = true;
                 }

router/src/lib.rs

@@ -2,12 +2,12 @@ use async_process::Command;
 use futures::prelude::*;
 pub use libpacket::*;
 use netsim_embed_core::*;
-pub use netsim_embed_core::{DelayBuffer, Ipv4Range};
+pub use netsim_embed_core::{DelayBuffer, Ipv4Range, Protocol};
 pub use netsim_embed_machine::{unshare_user, Machine, MachineId, Namespace};
 use netsim_embed_nat::*;
 use netsim_embed_router::*;
 use std::fmt::Display;
-use std::net::Ipv4Addr;
+use std::net::{Ipv4Addr, SocketAddrV4};
 use std::str::FromStr;
 
 pub fn run<F>(f: F)
@@ -52,7 +52,7 @@ impl<C, E> Default for Netsim<C, E> {
 impl<C, E> Netsim<C, E>
 where
     C: Display + Send + 'static,
-    E: FromStr + Send + 'static,
+    E: FromStr + Display + Send + 'static,
     E::Err: std::fmt::Debug + Display + Send + Sync,
 {
     pub fn new() -> Self {
@@ -171,6 +171,9 @@ where
         nat.set_symmetric(config.symmetric);
         nat.set_blacklist_unrecognized_addrs(config.blacklist_unrecognized_addrs);
         nat.set_restrict_endpoints(config.restrict_endpoints);
+        for (protocol, port, local_addr) in config.forward_ports {
+            nat.forward_port(port, local_addr, protocol);
+        }
         async_global_executor::spawn(nat).detach();
         self.networks[public_net.0].router.add_connection(
             private_net.id(),
@@ -219,21 +222,11 @@ impl Network {
     }
 }
 
-#[derive(Clone, Copy, Debug)]
+#[derive(Clone, Debug, Default)]
 pub struct NatConfig {
     pub hair_pinning: bool,
     pub symmetric: bool,
     pub blacklist_unrecognized_addrs: bool,
     pub restrict_endpoints: bool,
-}
-
-impl Default for NatConfig {
-    fn default() -> Self {
-        Self {
-            hair_pinning: false,
-            symmetric: false,
-            blacklist_unrecognized_addrs: false,
-            restrict_endpoints: false,
-        }
-    }
+    pub forward_ports: Vec<(Protocol, u16, SocketAddrV4)>,
 }