clean code

.github/workflows/rust.yml

@@ -1,10 +1,7 @@
 name: Push or PR
 
 on:
-  push:
-    branches: [ "master" ]
-  pull_request:
-    branches: [ "master" ]
+  [push, pull_request]
 
 env:
   CARGO_TERM_COLOR: always

Cargo.toml

@@ -24,7 +24,7 @@ tokio-util = { version = "0.7", features = ["codec"], optional = true }
 ioctl = { version = "0.8", package = "ioctl-sys" }
 
 [target.'cfg(target_os = "windows")'.dependencies]
-wintun = { version = "0.3", features = ["panic_on_unsent_packets"] }
+wintun = { version = "0.4", features = ["panic_on_unsent_packets"] }
 
 [dev-dependencies]
 futures = "0.3"

README.md

@@ -1,5 +1,9 @@
-TUN interfaces [![Crates.io](https://img.shields.io/crates/v/tun.svg)](https://crates.io/crates/tun) ![tun](https://docs.rs/tun/badge.svg) ![WTFPL](http://img.shields.io/badge/license-WTFPL-blue.svg)
+TUN interfaces 
 ==============
+[![Crates.io](https://img.shields.io/crates/v/tun.svg)](https://crates.io/crates/tun)
+![tun](https://docs.rs/tun/badge.svg)
+![WTFPL](http://img.shields.io/badge/license-WTFPL-blue.svg)
+
 This crate allows the creation and usage of TUN interfaces, the aim is to make this cross-platform.
 
 Usage
@@ -8,7 +12,7 @@ First, add the following to your `Cargo.toml`:
 
 ```toml
 [dependencies]
-tun = "0.6.1"
+tun = "0.6"
 ```
 
 Next, add this to your crate root:
@@ -21,7 +25,7 @@ If you want to use the TUN interface with mio/tokio, you need to enable the `asy
 
 ```toml
 [dependencies]
-tun = { version = "0.6.1", features = ["async"] }
+tun = { version = "0.6", features = ["async"] }
 ```
 
 Example
@@ -43,6 +47,7 @@ fn main() {
 	#[cfg(target_os = "linux")]
 	config.platform(|config| {
 		config.packet_information(true);
+		config.apply_settings(true);
 	});
 
 	let mut dev = tun::create(&config).unwrap();
@@ -57,7 +62,7 @@ fn main() {
 
 Platforms
 =========
-Not every platform is supported.
+Recently, `tun` supports Linux, Android, macOS, iOS and Windows.
 
 Linux
 -----

examples/ping-tun.rs

@@ -29,11 +29,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     #[cfg(target_os = "linux")]
     config.platform(|config| {
         config.packet_information(true);
+        config.apply_settings(true);
     });
 
     #[cfg(target_os = "windows")]
     config.platform(|config| {
-        config.initialize();
+        config.initialize(Some(9099482345783245345345_u128));
     });
 
     let dev = tun::create_as_async(&config)?;

examples/read-async.rs

@@ -27,6 +27,7 @@ async fn main() {
     #[cfg(target_os = "linux")]
     config.platform(|config| {
         config.packet_information(true);
+        config.apply_settings(true);
     });
 
     let dev = tun::create_as_async(&config).unwrap();

examples/read.rs

@@ -26,6 +26,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     #[cfg(target_os = "linux")]
     config.platform(|config| {
         config.packet_information(true);
+        config.apply_settings(true);
     });
 
     let mut dev = tun::create(&config)?;

src/async/codec.rs

@@ -14,7 +14,6 @@
 
 use std::io;
 
-use byteorder::{NativeEndian, NetworkEndian, WriteBytesExt};
 use bytes::{BufMut, Bytes, BytesMut};
 use tokio_util::codec::{Decoder, Encoder};
 
@@ -54,6 +53,7 @@ impl PacketProtocol {
     }
 
     #[cfg(target_os = "windows")]
+    #[allow(dead_code)]
     fn into_pi_field(self) -> Result<u16, io::Error> {
         unimplemented!()
     }
@@ -95,8 +95,8 @@ pub struct TunPacketCodec(bool, i32);
 impl TunPacketCodec {
     /// Create a new `TunPacketCodec` specifying whether the underlying
     ///  tunnel Device has enabled the packet information header.
-    pub fn new(pi: bool, mtu: i32) -> TunPacketCodec {
-        TunPacketCodec(pi, mtu)
+    pub fn new(packet_information: bool, mtu: i32) -> TunPacketCodec {
+        TunPacketCodec(packet_information, mtu)
     }
 }
 
@@ -132,19 +132,24 @@ impl Encoder<TunPacket> for TunPacketCodec {
     type Error = io::Error;
 
     fn encode(&mut self, item: TunPacket, dst: &mut BytesMut) -> Result<(), Self::Error> {
-        dst.reserve(item.get_bytes().len() + 4);
+        dst.reserve(item.get_bytes().len() + if self.0 { 4 } else { 0 });
         match item {
-            TunPacket(proto, bytes) if self.0 => {
-                // build the packet information header comprising of 2 u16
-                // fields: flags and protocol.
-                let mut buf = Vec::<u8>::with_capacity(4);
-
-                // flags is always 0
-                buf.write_u16::<NativeEndian>(0)?;
-                // write the protocol as network byte order
-                buf.write_u16::<NetworkEndian>(proto.into_pi_field()?)?;
-
-                dst.put_slice(&buf);
+            TunPacket(_proto, bytes) if self.0 => {
+                #[cfg(unix)]
+                {
+                    use byteorder::{NativeEndian, NetworkEndian, WriteBytesExt};
+
+                    // build the packet information header comprising of 2 u16
+                    // fields: flags and protocol.
+                    let mut buf = Vec::<u8>::with_capacity(4);
+
+                    // flags is always 0
+                    buf.write_u16::<NativeEndian>(0)?;
+                    // write the protocol as network byte order
+                    buf.write_u16::<NetworkEndian>(_proto.into_pi_field()?)?;
+
+                    dst.put_slice(&buf);
+                }
                 dst.put(bytes);
             }
             TunPacket(_, bytes) => dst.put(bytes),

src/async/device.rs

@@ -51,8 +51,9 @@ impl AsyncDevice {
 
     /// Consumes this AsyncDevice and return a Framed object (unified Stream and Sink interface)
     pub fn into_framed(mut self) -> Framed<Self, TunPacketCodec> {
-        let pi = self.get_mut().has_packet_information();
-        let codec = TunPacketCodec::new(pi, self.inner.get_ref().mtu().unwrap_or(1504));
+        let packet_information = self.get_mut().has_packet_information();
+        let mtu = self.inner.get_ref().mtu().unwrap_or(1504);
+        let codec = TunPacketCodec::new(packet_information, mtu);
         Framed::new(self, codec)
     }
 }
@@ -147,8 +148,8 @@ impl AsyncQueue {
 
     /// Consumes this AsyncQueue and return a Framed object (unified Stream and Sink interface)
     pub fn into_framed(mut self) -> Framed<Self, TunPacketCodec> {
-        let pi = self.get_mut().has_packet_information();
-        let codec = TunPacketCodec::new(pi, 1504);
+        let packet_information = self.get_mut().has_packet_information();
+        let codec = TunPacketCodec::new(packet_information, 1504);
         Framed::new(self, codec)
     }
 }

src/async/win/device.rs

@@ -15,7 +15,7 @@
 use core::pin::Pin;
 use core::task::{Context, Poll};
 use std::io;
-use std::io::{Error, Write};
+use std::io::Error;
 
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio_util::codec::Framed;
@@ -26,12 +26,17 @@ use crate::r#async::codec::*;
 
 pub struct AsyncDevice {
     inner: Device,
+    session: WinSession,
 }
 
 impl AsyncDevice {
     /// Create a new `AsyncDevice` wrapping around a `Device`.
     pub fn new(device: Device) -> io::Result<AsyncDevice> {
-        Ok(AsyncDevice { inner: device })
+        let session = WinSession::new(device.queue.get_session())?;
+        Ok(AsyncDevice {
+            inner: device,
+            session,
+        })
     }
     /// Returns a shared reference to the underlying Device object
     pub fn get_ref(&self) -> &Device {
@@ -56,50 +61,41 @@ impl AsyncRead for AsyncDevice {
         cx: &mut Context<'_>,
         buf: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
-        let rbuf = buf.initialize_unfilled();
-        match Pin::new(&mut self.inner).poll_read(cx, rbuf) {
-            Poll::Ready(Ok(n)) => {
-                buf.advance(n);
-                Poll::Ready(Ok(()))
-            }
-            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
-            Poll::Pending => Poll::Pending,
-        }
+        Pin::new(&mut self.session).poll_read(cx, buf)
     }
 }
 
 impl AsyncWrite for AsyncDevice {
     fn poll_write(
         mut self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
+        cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<Result<usize, Error>> {
-        match self.inner.write(buf) {
-            Ok(n) => Poll::Ready(Ok(n)),
-            Err(e) => Poll::Ready(Err(e)),
-        }
+        Pin::new(&mut self.session).poll_write(cx, buf)
     }
 
-    fn poll_flush(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
-        match self.inner.flush() {
-            Ok(n) => Poll::Ready(Ok(n)),
-            Err(e) => Poll::Ready(Err(e)),
-        }
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        Pin::new(&mut self.session).poll_flush(cx)
     }
 
-    fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
-        Poll::Ready(Ok(()))
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        Pin::new(&mut self.session).poll_shutdown(cx)
     }
 }
 
 pub struct AsyncQueue {
     inner: Queue,
+    session: WinSession,
 }
 
 impl AsyncQueue {
     /// Create a new `AsyncQueue` wrapping around a `Queue`.
     pub fn new(queue: Queue) -> io::Result<AsyncQueue> {
-        Ok(AsyncQueue { inner: queue })
+        let session = WinSession::new(queue.get_session())?;
+        Ok(AsyncQueue {
+            inner: queue,
+            session,
+        })
     }
     /// Returns a shared reference to the underlying Queue object
     pub fn get_ref(&self) -> &Queue {
@@ -124,29 +120,87 @@ impl AsyncRead for AsyncQueue {
         cx: &mut Context<'_>,
         buf: &mut ReadBuf<'_>,
     ) -> Poll<io::Result<()>> {
-        let rbuf = buf.initialize_unfilled();
-        match Pin::new(&mut self.inner).poll_read(cx, rbuf) {
-            Poll::Ready(Ok(n)) => {
-                buf.advance(n);
-                Poll::Ready(Ok(()))
-            }
-            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
-            Poll::Pending => Poll::Pending,
-        }
+        Pin::new(&mut self.session).poll_read(cx, buf)
     }
 }
 
 impl AsyncWrite for AsyncQueue {
     fn poll_write(
         mut self: Pin<&mut Self>,
-        _cx: &mut Context<'_>,
+        cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<Result<usize, Error>> {
-        match self.inner.write(buf) {
-            Ok(n) => Poll::Ready(Ok(n)),
-            Err(e) => Poll::Ready(Err(e)),
+        Pin::new(&mut self.session).poll_write(cx, buf)
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        Pin::new(&mut self.session).poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
+        Pin::new(&mut self.session).poll_shutdown(cx)
+    }
+}
+
+struct WinSession {
+    session: std::sync::Arc<wintun::Session>,
+    receiver: tokio::sync::mpsc::UnboundedReceiver<Vec<u8>>,
+    _task: std::thread::JoinHandle<()>,
+}
+
+impl WinSession {
+    fn new(session: std::sync::Arc<wintun::Session>) -> Result<WinSession, io::Error> {
+        let session_reader = session.clone();
+        let (receiver_tx, receiver_rx) = tokio::sync::mpsc::unbounded_channel::<Vec<u8>>();
+        let task = std::thread::spawn(move || loop {
+            match session_reader.receive_blocking() {
+                Ok(packet) => {
+                    if let Err(err) = receiver_tx.send(packet.bytes().to_vec()) {
+                        log::error!("{}", err);
+                    }
+                }
+                Err(err) => {
+                    log::info!("{}", err);
+                    break;
+                }
+            }
+        });
+
+        Ok(WinSession {
+            session,
+            receiver: receiver_rx,
+            _task: task,
+        })
+    }
+}
+
+impl AsyncRead for WinSession {
+    fn poll_read(
+        mut self: Pin<&mut Self>,
+        cx: &mut Context<'_>,
+        buf: &mut ReadBuf<'_>,
+    ) -> Poll<io::Result<()>> {
+        match std::task::ready!(self.receiver.poll_recv(cx)) {
+            Some(bytes) => {
+                buf.put_slice(&bytes);
+                std::task::Poll::Ready(Ok(()))
+            }
+            None => std::task::Poll::Ready(Ok(())),
         }
     }
+}
+
+impl AsyncWrite for WinSession {
+    fn poll_write(
+        self: Pin<&mut Self>,
+        _cx: &mut Context<'_>,
+        buf: &[u8],
+    ) -> Poll<Result<usize, Error>> {
+        let mut write_pack = self.session.allocate_send_packet(buf.len() as u16)?;
+        write_pack.bytes_mut().copy_from_slice(buf.as_ref());
+        self.session.send_packet(write_pack);
+        std::task::Poll::Ready(Ok(buf.len()))
+    }
 
     fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Error>> {
         Poll::Ready(Ok(()))

src/error.rs

@@ -54,4 +54,13 @@ pub enum Error {
     WintunError(#[from] wintun::Error),
 }
 
+impl From<Error> for io::Error {
+    fn from(value: Error) -> Self {
+        match value {
+            Error::Io(err) => err,
+            _ => io::Error::new(io::ErrorKind::Other, value),
+        }
+    }
+}
+
 pub type Result<T, E = Error> = ::std::result::Result<T, E>;