No longer connecting to self, flush on write

Cargo.lock

@@ -1,6 +1,6 @@
 [root]
 name = "vpncloud"
-version = "0.3.0"
+version = "0.3.1"
 dependencies = [
  "aligned_alloc 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "docopt 0.6.78 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -10,6 +10,7 @@ dependencies = [
  "log 0.3.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "nix 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "pkg-config 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "rand 0.3.12 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustc-serialize 0.3.16 (registry+https://github.com/rust-lang/crates.io-index)",
  "signal 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "time 0.1.34 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -143,6 +144,16 @@ name = "pkg-config"
 version = "0.3.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
+[[package]]
+name = "rand"
+version = "0.3.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "advapi32-sys 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "libc 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.2.5 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "regex"
 version = "0.1.43"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vpncloud"
-version = "0.3.0"
+version = "0.3.1"
 authors = ["Dennis Schwerdel <schwerdel@informatik.uni-kl.de>"]
 build = "build.rs"
 license = "GPL-3.0"
@@ -20,6 +20,7 @@ signal = "0.1"
 nix = "0.4"
 libc = "0.2"
 aligned_alloc = "0.1.1"
+rand = "0.3"
 
 [build-dependencies]
 gcc = "0.3"

README.md

@@ -32,6 +32,7 @@ This is what works:
 * Automatic reconnecting when connections are lost
 * Non-native forwarding modes, e.g. IP based learning switch and prefix routed Ethernet networks.
 * High throughput and low additional latency (see [performance page](performance.md))
+* Support for tunneled VLans (TAP device)
 
 However there are some open issues:
 

src/cloud.rs

@@ -11,8 +11,9 @@ use epoll;
 use nix::sys::signal::{SIGTERM, SIGQUIT, SIGINT};
 use signal::trap::Trap;
 use time::SteadyTime;
+use rand::random;
 
-use super::types::{Table, Protocol, Range, Error, NetworkId};
+use super::types::{Table, Protocol, Range, Error, NetworkId, NodeId};
 use super::device::Device;
 use super::udpmessage::{encode, decode, Options, Message};
 use super::crypto::Crypto;
@@ -87,11 +88,13 @@ impl PeerList {
 
 
 pub struct GenericCloud<P: Protocol> {
+    node_id: NodeId,
     peers: PeerList,
     addresses: Vec<Range>,
     learning: bool,
     broadcast: bool,
     reconnect_peers: Vec<SocketAddr>,
+    blacklist_peers: Vec<SocketAddr>,
     table: Box<Table>,
     socket: UdpSocket,
     device: Device,
@@ -115,11 +118,13 @@ impl<P: Protocol> GenericCloud<P> {
         let mut options = Options::default();
         options.network_id = network_id;
         GenericCloud{
+            node_id: random(),
             peers: PeerList::new(peer_timeout),
             addresses: addresses,
             learning: learning,
             broadcast: broadcast,
             reconnect_peers: Vec::new(),
+            blacklist_peers: Vec::new(),
             table: table,
             socket: socket,
             device: device,
@@ -153,7 +158,7 @@ impl<P: Protocol> GenericCloud<P> {
     pub fn connect<Addr: ToSocketAddrs+fmt::Display>(&mut self, addr: Addr, reconnect: bool) -> Result<(), Error> {
         if let Ok(mut addrs) = addr.to_socket_addrs() {
             while let Some(addr) = addrs.next() {
-                if self.peers.contains(&addr) {
+                if self.peers.contains(&addr) || self.blacklist_peers.contains(&addr) {
                     return Ok(());
                 }
             }
@@ -164,7 +169,8 @@ impl<P: Protocol> GenericCloud<P> {
             self.reconnect_peers.push(addr);
         }
         let addrs = self.addresses.clone();
-        self.send_msg(addr, &Message::Init(0, addrs))
+        let node_id = self.node_id.clone();
+        self.send_msg(addr, &Message::Init(0, node_id, addrs))
     }
 
     fn housekeep(&mut self) -> Result<(), Error> {
@@ -245,20 +251,25 @@ impl<P: Protocol> GenericCloud<P> {
             Message::Peers(peers) => {
                 self.peers.add(&peer);
                 for p in &peers {
-                    if ! self.peers.contains(p) {
+                    if ! self.peers.contains(p) && ! self.blacklist_peers.contains(p) {
                         try!(self.connect(p, false));
                     }
                 }
             },
-            Message::Init(stage, ranges) => {
+            Message::Init(stage, node_id, ranges) => {
+                if node_id == self.node_id {
+                    self.blacklist_peers.push(peer);
+                    return Ok(())
+                }
                 self.peers.add(&peer);
-                let peers = self.peers.as_vec();
-                let own_addrs = self.addresses.clone();
                 for range in ranges {
                     self.table.learn(range.base, Some(range.prefix_len), peer.clone());
                 }
                 if stage == 0 {
-                    try!(self.send_msg(peer, &Message::Init(stage+1, own_addrs)));
+                    let peers = self.peers.as_vec();
+                    let own_addrs = self.addresses.clone();
+                    let own_node_id = self.node_id.clone();
+                    try!(self.send_msg(peer, &Message::Init(stage+1, own_node_id, own_addrs)));
                     try!(self.send_msg(peer, &Message::Peers(peers)));
                 }
             },

src/device.rs

@@ -46,7 +46,7 @@ impl Device {
     #[inline]
     pub fn write(&mut self, data: &[u8]) -> Result<(), Error> {
         match self.fd.write_all(&data) {
-            Ok(_) => Ok(()),
+            Ok(_) => self.fd.flush().map_err(|_| Error::TunTapDevError("Flush error")),
             Err(_) => Err(Error::TunTapDevError("Write error"))
         }
     }

src/main.rs

@@ -8,6 +8,7 @@ extern crate signal;
 extern crate nix;
 extern crate libc;
 extern crate aligned_alloc;
+extern crate rand;
 #[cfg(feature = "bench")] extern crate test;
 
 #[macro_use] mod util;

src/tests.rs

@@ -79,11 +79,15 @@ fn udpmessage_init() {
     let mut crypto = Crypto::None;
     let addrs = vec![Range{base: Address{data: [0,1,2,3,0,0,0,0,0,0,0,0,0,0,0,0], len: 4}, prefix_len: 24},
         Range{base: Address{data: [0,1,2,3,4,5,0,0,0,0,0,0,0,0,0,0], len: 6}, prefix_len: 16}];
-    let msg = Message::Init(0, addrs);
+    let node_id = [0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15];
+    let msg = Message::Init(0, node_id, addrs);
     let mut buf = [0; 1024];
     let size = encode(&mut options, &msg, &mut buf[..], &mut crypto);
-    assert_eq!(size, 24);
+    assert_eq!(size, 40);
-    assert_eq!(&buf[..size], &[118,112,110,1,0,0,0,2,0,2,4,0,1,2,3,24,6,0,1,2,3,4,5,16]);
+    let should = [118,112,110,1,0,0,0,2,0,0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,2,4,0,1,2,3,24,6,0,1,2,3,4,5,16];
+    for i in 0..size {
+        assert_eq!(buf[i], should[i]);
+    }
     let (options2, msg2) = decode(&mut buf[..size], &mut crypto).unwrap();
     assert_eq!(options, options2);
     assert_eq!(msg, msg2);
@@ -243,10 +247,10 @@ fn message_fmt() {
         SocketAddr::from_str("5.6.7.8:12345").unwrap(),
         SocketAddr::from_str("[0001:0203:0405:0607:0809:0a0b:0c0d:0e0f]:6789").unwrap()])),
         "Peers [1.2.3.4:123, 5.6.7.8:12345, [1:203:405:607:809:a0b:c0d:e0f]:6789]");
-    assert_eq!(format!("{:?}", Message::Init(0, vec![
+    assert_eq!(format!("{:?}", Message::Init(0, [0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15], vec![
         Range{base: Address{data: [0,1,2,3,0,0,0,0,0,0,0,0,0,0,0,0], len: 4}, prefix_len: 24},
         Range{base: Address{data: [0,1,2,3,4,5,0,0,0,0,0,0,0,0,0,0], len: 6}, prefix_len: 16}
-        ])), "Init(stage=0, [0.1.2.3/24, 00:01:02:03:04:05/16])");
+        ])), "Init(stage=0, node_id=000102030405060708090a0b0c0d0e0f, [0.1.2.3/24, 00:01:02:03:04:05/16])");
     assert_eq!(format!("{:?}", Message::Close), "Close");
 }
 

src/types.rs

@@ -3,9 +3,13 @@ use std::hash::Hasher;
 use std::fmt;
 use std::str::FromStr;
 
-use super::util::Encoder;
+use super::util::{bytes_to_hex, Encoder};
+
+pub const NODE_ID_BYTES: usize = 16;
 
 pub type NetworkId = u64;
+pub type NodeId = [u8; NODE_ID_BYTES];
+
 
 #[derive(PartialOrd, Eq, Ord, Clone, Hash)]
 pub struct Address {
@@ -80,7 +84,7 @@ impl fmt::Display for Address {
             },
             16 => write!(formatter, "{:02x}{:02x}:{:02x}{:02x}:{:02x}{:02x}:{:02x}{:02x}:{:02x}{:02x}:{:02x}{:02x}:{:02x}{:02x}:{:02x}{:02x}",
                 d[0], d[1], d[2], d[3], d[4], d[5], d[6], d[7], d[8], d[9], d[10], d[11], d[12], d[13], d[14], d[15]),
-            _ => write!(formatter, "{:?}", d)
+            _ => write!(formatter, "{}", bytes_to_hex(d))
         }
     }
 }

src/udpmessage.rs

@@ -1,8 +1,8 @@
 use std::fmt;
 use std::net::{SocketAddr, SocketAddrV4, Ipv4Addr, SocketAddrV6, Ipv6Addr};
 
-use super::types::{Error, NetworkId, Range};
+use super::types::{NodeId, Error, NetworkId, Range, NODE_ID_BYTES};
-use super::util::{Encoder, memcopy};
+use super::util::{bytes_to_hex, Encoder, memcopy};
 use super::crypto::Crypto;
 
 const MAGIC: [u8; 3] = [0x76, 0x70, 0x6e];
@@ -70,7 +70,7 @@ pub struct Options {
 pub enum Message<'a> {
     Data(&'a[u8]),
     Peers(Vec<SocketAddr>),
-    Init(u8, Vec<Range>),
+    Init(u8, NodeId, Vec<Range>),
     Close,
 }
 
@@ -90,7 +90,7 @@ impl<'a> fmt::Debug for Message<'a> {
                 }
                 write!(formatter, "]")
             },
-            &Message::Init(stage, ref data) => write!(formatter, "Init(stage={}, {:?})", stage, data),
+            &Message::Init(stage, ref node_id, ref peers) => write!(formatter, "Init(stage={}, node_id={}, {:?})", stage, bytes_to_hex(node_id), peers),
             &Message::Close => write!(formatter, "Close"),
         }
     }
@@ -171,11 +171,16 @@ pub fn decode<'a>(data: &'a mut [u8], crypto: &mut Crypto) -> Result<(Options, M
             Message::Peers(peers)
         },
         2 => {
-            if end < pos + 2 {
+            if end < pos + 2 + NODE_ID_BYTES {
                 return Err(Error::ParseError("Init data too short"));
             }
             let stage = data[pos];
             pos += 1;
+            let mut node_id = [0; NODE_ID_BYTES];
+            for i in 0..NODE_ID_BYTES {
+                node_id[i] = data[pos+i];
+            }
+            pos += NODE_ID_BYTES;
             let count = data[pos] as usize;
             pos += 1;
             let mut addrs = Vec::with_capacity(count);
@@ -184,7 +189,7 @@ pub fn decode<'a>(data: &'a mut [u8], crypto: &mut Crypto) -> Result<(Options, M
                 pos += read;
                 addrs.push(range);
             }
-            Message::Init(stage, addrs)
+            Message::Init(stage, node_id, addrs)
         },
         3 => Message::Close,
         _ => return Err(Error::ParseError("Unknown message type"))
@@ -198,7 +203,7 @@ pub fn encode(options: &Options, msg: &Message, buf: &mut [u8], crypto: &mut Cry
     header.msgtype = match msg {
         &Message::Data(_) => 0,
         &Message::Peers(_) => 1,
-        &Message::Init(_, _) => 2,
+        &Message::Init(_, _, _) => 2,
         &Message::Close => 3
     };
     header.crypto_method = crypto.method();
@@ -252,10 +257,14 @@ pub fn encode(options: &Options, msg: &Message, buf: &mut [u8], crypto: &mut Cry
                 pos += 2;
             };
         },
-        &Message::Init(stage, ref ranges) => {
+        &Message::Init(stage, ref node_id, ref ranges) => {
-            assert!(buf.len() >= pos + 2);
+            assert!(buf.len() >= pos + 2 + NODE_ID_BYTES);
             buf[pos] = stage;
             pos += 1;
+            for i in 0..NODE_ID_BYTES {
+                buf[pos+i] = node_id[i];
+            }
+            pos += NODE_ID_BYTES;
             assert!(ranges.len() <= 255);
             buf[pos] = ranges.len() as u8;
             pos += 1;

src/util.rs

@@ -24,6 +24,19 @@ pub fn memcopy(src: &[u8], dst: &mut[u8]) {
     unsafe { ptr::copy_nonoverlapping(src.as_ptr(), dst.as_mut_ptr(), src.len()) };
 }
 
+const HEX_CHARS: &'static [u8] = b"0123456789abcdef";
+
+pub fn bytes_to_hex(bytes: &[u8]) -> String {
+    let mut v = Vec::with_capacity(bytes.len() * 2);
+    for &byte in bytes {
+        v.push(HEX_CHARS[(byte >> 4) as usize]);
+        v.push(HEX_CHARS[(byte & 0xf) as usize]);
+    }
+    unsafe {
+        String::from_utf8_unchecked(v)
+    }
+}
+
 
 pub struct Encoder;