Update locking to be interior to PeerPcb

remove timeouts

Apple/NetworkExtension/BurrowIpc.swift

@@ -113,7 +113,7 @@ final class BurrowIpc {
         return data
     }
 
-    func request<U: Decodable>(_ request: Request, type: U.Type) async throws -> U {
+    func request<U: Decodable>(_ request: any Request, type: U.Type) async throws -> U {
         do {
             var data: Data = try JSONEncoder().encode(request)
             data.append(contentsOf: [10])

Apple/NetworkExtension/DataTypes.swift

@@ -7,16 +7,40 @@ enum BurrowError: Error {
     case resultIsNone
 }
 
-protocol Request: Codable {
+protocol Request: Codable where T: Codable{
+    associatedtype T
     var id: UInt { get set }
-    var command: String { get set }
+    var command: T { get set }
 }
 
-struct BurrowRequest: Request {
+struct BurrowSingleCommand: Request {
     var id: UInt
     var command: String
 }
 
+struct BurrowRequest<T>: Request where T: Codable{
+    var id: UInt
+    var command: T
+}
+
+struct BurrowStartRequest: Codable {
+    struct TunOptions: Codable{
+        let name: String?
+        let no_pi: Bool
+        let tun_excl: Bool
+        let seek_utun: Int?
+        let address: String?
+    }
+    struct StartOptions: Codable{
+        let tun: TunOptions
+    }
+    let Start: StartOptions
+}
+
+func start_req_fd(id: UInt, fd: Int) -> BurrowRequest<BurrowStartRequest> {
+    return BurrowRequest(id: id, command: BurrowStartRequest(Start: BurrowStartRequest.StartOptions(tun: BurrowStartRequest.TunOptions(name: nil, no_pi: false, tun_excl: false, seek_utun: fd, address: nil))))
+}
+
 struct Response<T>: Decodable where T: Decodable {
     var id: UInt
     var result: T

Apple/NetworkExtension/PacketTunnelProvider.swift

@@ -6,7 +6,7 @@ class PacketTunnelProvider: NEPacketTunnelProvider {
     let logger = Logger(subsystem: "com.hackclub.burrow", category: "frontend")
     var client: BurrowIpc?
     var osInitialized = false
-    override func startTunnel(options: [String: NSObject]?, completionHandler: @escaping (Error?) -> Void) {
+    override func startTunnel(options: [String : NSObject]? = nil) async throws {
         logger.log("Starting tunnel")
         if !osInitialized {
             libburrow.initialize_oslog()
@@ -15,28 +15,35 @@ class PacketTunnelProvider: NEPacketTunnelProvider {
         libburrow.start_srv()
         client = BurrowIpc(logger: logger)
         logger.info("Started server")
-        Task {
+        do {
-            do {
+            let command = BurrowSingleCommand(id: 0, command: "ServerConfig")
-                let command = BurrowRequest(id: 0, command: "ServerConfig")
+            guard let data = try await client?.request(command, type: Response<BurrowResult<ServerConfigData>>.self)
-                guard let data = try await client?.request(command, type: Response<BurrowResult<ServerConfigData>>.self)
+            else {
-                else {
+                throw BurrowError.cantParseResult
-                    throw BurrowError.cantParseResult
-                }
-                let encoded = try JSONEncoder().encode(data.result)
-                self.logger.log("Received final data: \(String(decoding: encoded, as: UTF8.self))")
-                guard let serverconfig = data.result.Ok else {
-                    throw BurrowError.resultIsError
-                }
-                guard let tunNs = self.generateTunSettings(from: serverconfig) else {
-                    throw BurrowError.addrDoesntExist
-                }
-                try await self.setTunnelNetworkSettings(tunNs)
-                self.logger.info("Set remote tunnel address to \(tunNs.tunnelRemoteAddress)")
-                completionHandler(nil)
-            } catch {
-                self.logger.error("An error occurred: \(error)")
-                completionHandler(error)
             }
+            let encoded = try JSONEncoder().encode(data.result)
+            self.logger.log("Received final data: \(String(decoding: encoded, as: UTF8.self))")
+            guard let serverconfig = data.result.Ok else {
+                throw BurrowError.resultIsError
+            }
+            guard let tunNs = self.generateTunSettings(from: serverconfig) else {
+                throw BurrowError.addrDoesntExist
+            }
+            try await self.setTunnelNetworkSettings(tunNs)
+            self.logger.info("Set remote tunnel address to \(tunNs.tunnelRemoteAddress)")
+
+            //                let tunFd = self.packetFlow.value(forKeyPath: "socket.fileDescriptor") as! Int;
+            //                self.logger.info("Found File Descriptor: \(tunFd)")
+            let start_command = start_req_fd(id: 1, fd: 0)
+            guard let data = try await client?.request(start_command, type: Response<BurrowResult<String>>.self)
+            else {
+                throw BurrowError.cantParseResult
+            }
+            let encoded_startres = try JSONEncoder().encode(data.result)
+            self.logger.log("Received start server response: \(String(decoding: encoded_startres, as: UTF8.self))")
+        } catch {
+            self.logger.error("An error occurred: \(error)")
+            throw error
         }
     }
     private func generateTunSettings(from: ServerConfigData) -> NETunnelNetworkSettings? {
@@ -50,17 +57,16 @@ class PacketTunnelProvider: NEPacketTunnelProvider {
         logger.log("Initialized ipv4 settings: \(nst.ipv4Settings)")
         return nst
     }
-    override func stopTunnel(with reason: NEProviderStopReason, completionHandler: @escaping () -> Void) {
+    override func stopTunnel(with reason: NEProviderStopReason) async {
-        completionHandler()
+
     }
-    override func handleAppMessage(_ messageData: Data, completionHandler: ((Data?) -> Void)?) {
+    override func handleAppMessage(_ messageData: Data) async -> Data? {
-        if let handler = completionHandler {
+        messageData
-            handler(messageData)
-        }
     }
-    override func sleep(completionHandler: @escaping () -> Void) {
+    override func sleep() async {
-        completionHandler()
+
     }
     override func wake() {
+
     }
 }

burrow/src/apple.rs

@@ -1,15 +1,13 @@
-use tracing::{debug, Subscriber};
+use tracing::debug;
-use tracing::instrument::WithSubscriber;
 use tracing_oslog::OsLogger;
-use tracing_subscriber::FmtSubscriber;
 use tracing_subscriber::layer::SubscriberExt;
 
 pub use crate::daemon::start_srv;
 
 #[no_mangle]
 pub extern "C" fn initialize_oslog() {
-    let collector = tracing_subscriber::registry()
+    let collector =
-        .with(OsLogger::new("com.hackclub.burrow", "backend"));
+        tracing_subscriber::registry().with(OsLogger::new("com.hackclub.burrow", "backend"));
     tracing::subscriber::set_global_default(collector).unwrap();
     debug!("Initialized oslog tracing in libburrow rust FFI");
 }

burrow/src/daemon/command.rs

@@ -12,21 +12,22 @@ pub enum DaemonCommand {
 
 #[derive(Debug, Clone, Default, Serialize, Deserialize, JsonSchema)]
 pub struct DaemonStartOptions {
-    pub(super) tun: TunOptions,
+    pub tun: TunOptions,
 }
 
 #[test]
 fn test_daemoncommand_serialization() {
+    insta::assert_snapshot!(serde_json::to_string(&DaemonCommand::Start(
+        DaemonStartOptions::default()
+    ))
+    .unwrap());
     insta::assert_snapshot!(
-        serde_json::to_string(&DaemonCommand::Start(DaemonStartOptions::default())).unwrap()
+        serde_json::to_string(&DaemonCommand::Start(DaemonStartOptions {
+            tun: TunOptions { ..TunOptions::default() }
+        }))
+        .unwrap()
     );
-    insta::assert_snapshot!(
+    insta::assert_snapshot!(serde_json::to_string(&DaemonCommand::ServerInfo).unwrap());
-        serde_json::to_string(&DaemonCommand::ServerInfo).unwrap()
+    insta::assert_snapshot!(serde_json::to_string(&DaemonCommand::Stop).unwrap());
-    );
+    insta::assert_snapshot!(serde_json::to_string(&DaemonCommand::ServerConfig).unwrap())
-    insta::assert_snapshot!(
-        serde_json::to_string(&DaemonCommand::Stop).unwrap()
-    );
-    insta::assert_snapshot!(
-        serde_json::to_string(&DaemonCommand::ServerConfig).unwrap()
-    )
 }

burrow/src/daemon/instance.rs

@@ -1,50 +1,105 @@
+use std::sync::Arc;
+
+use anyhow::Result;
+use tokio::{sync::RwLock, task::JoinHandle};
 use tracing::{debug, info, warn};
-use DaemonResponse;
+use tun::tokio::TunInterface;
-use tun::TunInterface;
+
-use crate::daemon::response::{DaemonResponseData, ServerConfig, ServerInfo};
+use crate::{
-use super::*;
+    daemon::{
+        command::DaemonCommand,
+        response::{DaemonResponse, DaemonResponseData, ServerConfig, ServerInfo},
+    },
+    wireguard::Interface,
+};
+
+enum RunState {
+    Running(JoinHandle<Result<()>>),
+    Idle,
+}
 
 pub struct DaemonInstance {
     rx: async_channel::Receiver<DaemonCommand>,
     sx: async_channel::Sender<DaemonResponse>,
-    tun_interface: Option<TunInterface>,
+    tun_interface: Option<Arc<RwLock<TunInterface>>>,
+    wg_interface: Arc<RwLock<Interface>>,
+    wg_state: RunState,
 }
 
 impl DaemonInstance {
-    pub fn new(rx: async_channel::Receiver<DaemonCommand>, sx: async_channel::Sender<DaemonResponse>) -> Self {
+    pub fn new(
+        rx: async_channel::Receiver<DaemonCommand>,
+        sx: async_channel::Sender<DaemonResponse>,
+        wg_interface: Arc<RwLock<Interface>>,
+    ) -> Self {
         Self {
             rx,
             sx,
+            wg_interface,
             tun_interface: None,
+            wg_state: RunState::Idle,
         }
     }
 
+    pub fn set_tun_interface(&mut self, tun_interface: Arc<RwLock<TunInterface>>) {
+        self.tun_interface = Some(tun_interface);
+    }
+
     async fn proc_command(&mut self, command: DaemonCommand) -> Result<DaemonResponseData> {
         info!("Daemon got command: {:?}", command);
         match command {
             DaemonCommand::Start(st) => {
-                if self.tun_interface.is_none() {
+                match self.wg_state {
-                    debug!("Daemon attempting start tun interface.");
+                    RunState::Running(_) => {
-                    self.tun_interface = Some(st.tun.open()?);
+                        warn!("Got start, but tun interface already up.");
-                    info!("Daemon started tun interface");
+                    }
-                } else {
+                    RunState::Idle => {
-                    warn!("Got start, but tun interface already up.");
+                        let raw = tun::TunInterface::retrieve().unwrap();
+                        debug!("TunInterface retrieved: {:?}", raw.name()?);
+
+                        let retrieved = TunInterface::new(raw)?;
+                        let tun_if = Arc::new(RwLock::new(retrieved));
+                        // let tun_if = Arc::new(RwLock::new(TunInterface::new(st.tun.open()?)?));
+
+                        debug!("Setting tun_interface");
+                        self.tun_interface = Some(tun_if.clone());
+                        debug!("tun_interface set: {:?}", self.tun_interface);
+
+                        debug!("Setting tun on wg_interface");
+                        self.wg_interface.write().await.set_tun(tun_if);
+                        debug!("tun set on wg_interface");
+
+                        debug!("Cloning wg_interface");
+                        let tmp_wg = self.wg_interface.clone();
+                        debug!("wg_interface cloned");
+
+                        debug!("Spawning run task");
+                        let run_task = tokio::spawn(async move {
+                            debug!("Running wg_interface");
+                            let twlock = tmp_wg.read().await;
+                            debug!("wg_interface read lock acquired");
+                            twlock.run().await
+                        });
+                        debug!("Run task spawned: {:?}", run_task);
+
+                        debug!("Setting wg_state to Running");
+                        self.wg_state = RunState::Running(run_task);
+                        debug!("wg_state set to Running");
+
+                        info!("Daemon started tun interface");
+                    }
                 }
                 Ok(DaemonResponseData::None)
             }
-            DaemonCommand::ServerInfo => {
+            DaemonCommand::ServerInfo => match &self.tun_interface {
-                match &self.tun_interface {
+                None => Ok(DaemonResponseData::None),
-                    None => {Ok(DaemonResponseData::None)}
+                Some(ti) => {
-                    Some(ti) => {
+                    info!("{:?}", ti);
-                        info!("{:?}", ti);
+                    Ok(DaemonResponseData::ServerInfo(ServerInfo::try_from(
-                        Ok(
+                        ti.read().await.inner.get_ref(),
-                            DaemonResponseData::ServerInfo(
+                    )?))
-                                ServerInfo::try_from(ti)?
-                            )
-                        )
-                    }
                 }
-            }
+            },
             DaemonCommand::Stop => {
                 if self.tun_interface.is_some() {
                     self.tun_interface = None;
@@ -61,6 +116,7 @@ impl DaemonInstance {
     }
 
     pub async fn run(&mut self) -> Result<()> {
+        tracing::info!("BEGIN");
         while let Ok(command) = self.rx.recv().await {
             let response = self.proc_command(command).await;
             info!("Daemon response: {:?}", response);

burrow/src/daemon/mod.rs

@@ -1,5 +1,7 @@
-use std::net::{Ipv4Addr, SocketAddr, ToSocketAddrs};
+use std::{
-
+    net::{IpAddr, Ipv4Addr, SocketAddr},
+    sync::Arc,
+};
 
 mod command;
 mod instance;
@@ -10,15 +12,18 @@ use anyhow::{Error, Result};
 use base64::{engine::general_purpose, Engine as _};
 pub use command::{DaemonCommand, DaemonStartOptions};
 use fehler::throws;
-use ip_network::{IpNetwork, Ipv4Network};
 use instance::DaemonInstance;
-use crate::wireguard::{StaticSecret, Peer, Interface, PublicKey};
+use ip_network::{IpNetwork, Ipv4Network};
-pub use net::DaemonClient;
-
 #[cfg(target_vendor = "apple")]
 pub use net::start_srv;
+pub use net::DaemonClient;
+pub use response::{DaemonResponse, DaemonResponseData, ServerInfo};
+use tokio::sync::RwLock;
 
-pub use response::{DaemonResponseData, DaemonResponse, ServerInfo};
+use crate::{
+    daemon::net::listen,
+    wireguard::{Interface, Peer, PublicKey, StaticSecret},
+};
 
 #[throws]
 fn parse_key(string: &str) -> [u8; 32] {
@@ -43,18 +48,13 @@ fn parse_public_key(string: &str) -> PublicKey {
 pub async fn daemon_main() -> Result<()> {
     let (commands_tx, commands_rx) = async_channel::unbounded();
     let (response_tx, response_rx) = async_channel::unbounded();
-    let mut inst = DaemonInstance::new(commands_rx, response_tx);
-
-    let mut _tun = tun::TunInterface::new()?;
-    _tun.set_ipv4_addr(Ipv4Addr::from([192, 168, 1, 10]))?;
-    _tun.set_timeout(Some(std::time::Duration::from_secs(1)))?;
-    let tun = tun::tokio::TunInterface::new(_tun)?;
 
     let private_key = parse_secret_key("GNqIAOCRxjl/cicZyvkvpTklgQuUmGUIEkH7IXF/sEE=")?;
     let public_key = parse_public_key("uy75leriJay0+oHLhRMpV+A5xAQ0hCJ+q7Ww81AOvT4=")?;
     let preshared_key = Some(parse_key("s7lx/mg+reVEMnGnqeyYOQkzD86n2+gYnx1M9ygi08k=")?);
-    let endpoint = "wg.burrow.rs:51820".to_socket_addrs()?.next().unwrap();
+    let endpoint = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(198, 18, 6, 180)), 51820); // DNS lookup under macos fails, somehow
-    let iface = Interface::new(tun, vec![Peer {
+
+    let iface = Interface::new(vec![Peer {
         endpoint,
         private_key,
         public_key,
@@ -62,6 +62,25 @@ pub async fn daemon_main() -> Result<()> {
         allowed_ips: vec![IpNetwork::V4(Ipv4Network::DEFAULT_ROUTE)],
     }])?;
 
-    iface.run().await;
+    let mut inst: DaemonInstance =
+        DaemonInstance::new(commands_rx, response_tx, Arc::new(RwLock::new(iface)));
+
+    tracing::info!("Starting daemon jobs...");
+
+    let inst_job = tokio::spawn(async move {
+        let res = inst.run().await;
+        if let Err(e) = res {
+            tracing::error!("Error when running instance: {}", e);
+        }
+    });
+
+    let listen_job = tokio::spawn(async move {
+        let res = listen(commands_tx, response_rx).await;
+        if let Err(e) = res {
+            tracing::error!("Error when listening: {}", e);
+        }
+    });
+
+    tokio::try_join!(inst_job, listen_job).map(|_| ());
     Ok(())
 }

burrow/src/daemon/net/apple.rs

@@ -1,10 +1,13 @@
 use std::thread;
+
 use tokio::runtime::Runtime;
-use tracing::error;
+use tracing::{error, info};
+
 use crate::daemon::{daemon_main, DaemonClient};
 
 #[no_mangle]
-pub extern "C" fn start_srv(){
+pub extern "C" fn start_srv() {
+    info!("Rust: Starting server");
     let _handle = thread::spawn(move || {
         let rt = Runtime::new().unwrap();
         rt.block_on(async {
@@ -16,8 +19,11 @@ pub extern "C" fn start_srv(){
     let rt = Runtime::new().unwrap();
     rt.block_on(async {
         loop {
-            if let Ok(_) = DaemonClient::new().await{
+            match DaemonClient::new().await {
-                break
+                Ok(_) => break,
+                Err(_e) => {
+                    // error!("Error when connecting to daemon: {}", e)
+                }
             }
         }
     });

burrow/src/daemon/net/mod.rs

@@ -29,4 +29,3 @@ pub struct DaemonRequest {
     pub id: u32,
     pub command: DaemonCommand,
 }
-

burrow/src/daemon/net/systemd.rs

@@ -1,13 +1,23 @@
-pub async fn listen(cmd_tx: async_channel::Sender<DaemonCommand>, rsp_rx: async_channel::Receiver<DaemonResponse>) -> Result<()> {
+pub async fn listen(
-    if !libsystemd::daemon::booted() || listen_with_systemd(cmd_tx.clone(), rsp_rx.clone()).await.is_err() {
+    cmd_tx: async_channel::Sender<DaemonCommand>,
+    rsp_rx: async_channel::Receiver<DaemonResponse>,
+) -> Result<()> {
+    if !libsystemd::daemon::booted()
+        || listen_with_systemd(cmd_tx.clone(), rsp_rx.clone())
+            .await
+            .is_err()
+    {
         unix::listen(cmd_tx, rsp_rx).await?;
     }
     Ok(())
 }
 
-async fn listen_with_systemd(cmd_tx: async_channel::Sender<DaemonCommand>, rsp_rx: async_channel::Receiver<DaemonResponse>) -> Result<()> {
+async fn listen_with_systemd(
+    cmd_tx: async_channel::Sender<DaemonCommand>,
+    rsp_rx: async_channel::Receiver<DaemonResponse>,
+) -> Result<()> {
     let fds = libsystemd::activation::receive_descriptors(false)?;
-    super::unix::listen_with_optional_fd(cmd_tx, rsp_rx,Some(fds[0].clone().into_raw_fd())).await
+    super::unix::listen_with_optional_fd(cmd_tx, rsp_rx, Some(fds[0].clone().into_raw_fd())).await
 }
 
 pub type DaemonClient = unix::DaemonClient;

burrow/src/daemon/net/unix.rs

@@ -1,22 +1,20 @@
-use super::*;
 use std::{
-   ascii, io, os::{
+    io,
+    os::{
         fd::{FromRawFd, RawFd},
         unix::net::UnixListener as StdUnixListener,
     },
-    path::Path};
+    path::{Path, PathBuf},
-use std::hash::Hash;
+};
-use std::path::PathBuf;
-use anyhow::anyhow;
-use log::log;
-use tracing::info;
 
-use anyhow::Result;
+use anyhow::{anyhow, Result};
 use tokio::{
     io::{AsyncBufReadExt, AsyncWriteExt, BufReader},
     net::{UnixListener, UnixStream},
 };
-use tracing::debug;
+use tracing::{debug, info};
+
+use super::*;
 use crate::daemon::{DaemonCommand, DaemonResponse, DaemonResponseData};
 
 #[cfg(not(target_vendor = "apple"))]
@@ -26,28 +24,33 @@ const UNIX_SOCKET_PATH: &str = "/run/burrow.sock";
 const UNIX_SOCKET_PATH: &str = "burrow.sock";
 
 #[cfg(target_os = "macos")]
-fn fetch_socket_path() -> Option<PathBuf>{
+fn fetch_socket_path() -> Option<PathBuf> {
     let tries = vec![
         "burrow.sock".to_string(),
-        format!("{}/Library/Containers/com.hackclub.burrow.network/Data/burrow.sock",
+        format!(
-                std::env::var("HOME").unwrap_or_default())
+            "{}/Library/Containers/com.hackclub.burrow.network/Data/burrow.sock",
-                .to_string(),
+            std::env::var("HOME").unwrap_or_default()
+        )
+        .to_string(),
     ];
-    for path in tries{
+    for path in tries {
         let path = PathBuf::from(path);
-        if path.exists(){
+        if path.exists() {
-            return Some(path);
+            return Some(path)
         }
     }
     None
 }
 
 #[cfg(not(target_os = "macos"))]
-fn fetch_socket_path() -> Option<PathBuf>{
+fn fetch_socket_path() -> Option<PathBuf> {
     Some(Path::new(UNIX_SOCKET_PATH).to_path_buf())
 }
 
-pub async fn listen(cmd_tx: async_channel::Sender<DaemonCommand>, rsp_rx: async_channel::Receiver<DaemonResponse>) -> Result<()> {
+pub async fn listen(
+    cmd_tx: async_channel::Sender<DaemonCommand>,
+    rsp_rx: async_channel::Receiver<DaemonResponse>,
+) -> Result<()> {
     listen_with_optional_fd(cmd_tx, rsp_rx, None).await
 }
 
@@ -69,14 +72,12 @@ pub(crate) async fn listen_with_optional_fd(
         listener
     } else {
         //  Won't help all that much, if we use the async version of fs.
-        if let Some(par) = path.parent(){
+        if let Some(par) = path.parent() {
-            std::fs::create_dir_all(
+            std::fs::create_dir_all(par)?;
-                par
-            )?;
         }
-        match std::fs::remove_file(path){
+        match std::fs::remove_file(path) {
-            Err(e) if e.kind()==io::ErrorKind::NotFound => {Ok(())}
+            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(()),
-            stuff => stuff
+            stuff => stuff,
         }?;
         info!("Relative path: {}", path.to_string_lossy());
         UnixListener::bind(path)?
@@ -98,18 +99,18 @@ pub(crate) async fn listen_with_optional_fd(
             while let Ok(Some(line)) = lines.next_line().await {
                 info!("Got line: {}", line);
                 debug!("Line raw data: {:?}", line.as_bytes());
-                let mut res : DaemonResponse = DaemonResponseData::None.into();
+                let mut res: DaemonResponse = DaemonResponseData::None.into();
                 let req = match serde_json::from_str::<DaemonRequest>(&line) {
                     Ok(req) => Some(req),
                     Err(e) => {
                         res.result = Err(e.to_string());
+                        tracing::error!("Failed to parse request: {}", e);
                         None
                     }
                 };
                 let mut res = serde_json::to_string(&res).unwrap();
                 res.push('\n');
 
-
                 if let Some(req) = req {
                     cmd_tx.send(req.command).await.unwrap();
                     let res = rsp_rxc.recv().await.unwrap().with_id(req.id);
@@ -117,6 +118,8 @@ pub(crate) async fn listen_with_optional_fd(
                     retres.push('\n');
                     info!("Sending response: {}", retres);
                     write_stream.write_all(retres.as_bytes()).await.unwrap();
+                } else {
+                    write_stream.write_all(res.as_bytes()).await.unwrap();
                 }
             }
         });
@@ -129,8 +132,7 @@ pub struct DaemonClient {
 
 impl DaemonClient {
     pub async fn new() -> Result<Self> {
-        let path = fetch_socket_path()
+        let path = fetch_socket_path().ok_or(anyhow!("Failed to find socket path"))?;
-            .ok_or(anyhow!("Failed to find socket path"))?;
         // debug!("found path: {:?}", path);
         let connection = UnixStream::connect(path).await?;
         debug!("connected to socket");

burrow/src/daemon/net/windows.rs

@@ -1,6 +1,9 @@
 use super::*;
 
-pub async fn listen(_cmd_tx: async_channel::Sender<DaemonCommand>, _rsp_rx: async_channel::Receiver<DaemonResponse>) -> Result<()> {
+pub async fn listen(
+    _cmd_tx: async_channel::Sender<DaemonCommand>,
+    _rsp_rx: async_channel::Receiver<DaemonResponse>,
+) -> Result<()> {
     unimplemented!("This platform does not currently support daemon mode.")
 }
 

burrow/src/daemon/response.rs

@@ -1,4 +1,3 @@
-use anyhow::anyhow;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use tun::TunInterface;
@@ -7,30 +6,27 @@ use tun::TunInterface;
 pub struct DaemonResponse {
     //  Error types can't be serialized, so this is the second best option.
     pub result: Result<DaemonResponseData, String>,
-    pub id: u32
+    pub id: u32,
 }
 
-impl DaemonResponse{
+impl DaemonResponse {
-    pub fn new(result: Result<DaemonResponseData, impl ToString>) -> Self{
+    pub fn new(result: Result<DaemonResponseData, impl ToString>) -> Self {
-        Self{
+        Self {
             result: result.map_err(|e| e.to_string()),
-            id: 0
+            id: 0,
         }
     }
 }
 
-impl Into<DaemonResponse> for DaemonResponseData{
+impl Into<DaemonResponse> for DaemonResponseData {
-    fn into(self) -> DaemonResponse{
+    fn into(self) -> DaemonResponse {
         DaemonResponse::new(Ok::<DaemonResponseData, String>(self))
     }
 }
 
-impl DaemonResponse{
+impl DaemonResponse {
-    pub fn with_id(self, id: u32) -> Self{
+    pub fn with_id(self, id: u32) -> Self {
-        Self {
+        Self { id, ..self }
-            id,
-            ..self
-        }
     }
 }
 
@@ -38,24 +34,22 @@ impl DaemonResponse{
 pub struct ServerInfo {
     pub name: Option<String>,
     pub ip: Option<String>,
-    pub mtu: Option<i32>
+    pub mtu: Option<i32>,
 }
 
-impl TryFrom<&TunInterface> for ServerInfo{
+impl TryFrom<&TunInterface> for ServerInfo {
     type Error = anyhow::Error;
 
-    #[cfg(any(target_os="linux",target_vendor="apple"))]
+    #[cfg(any(target_os = "linux", target_vendor = "apple"))]
     fn try_from(server: &TunInterface) -> anyhow::Result<Self> {
-        Ok(
+        Ok(ServerInfo {
-            ServerInfo{
+            name: server.name().ok(),
-                name: server.name().ok(),
+            ip: server.ipv4_addr().ok().map(|ip| ip.to_string()),
-                ip: server.ipv4_addr().ok().map(|ip| ip.to_string()),
+            mtu: server.mtu().ok(),
-                mtu: server.mtu().ok()
+        })
-            }
-        )
     }
 
-    #[cfg(not(any(target_os="linux",target_vendor="apple")))]
+    #[cfg(not(any(target_os = "linux", target_vendor = "apple")))]
     fn try_from(server: &TunInterface) -> anyhow::Result<Self> {
         Err(anyhow!("Not implemented in this platform"))
     }
@@ -65,45 +59,55 @@ impl TryFrom<&TunInterface> for ServerInfo{
 pub struct ServerConfig {
     pub address: Option<String>,
     pub name: Option<String>,
-    pub mtu: Option<i32>
+    pub mtu: Option<i32>,
 }
 
 impl Default for ServerConfig {
     fn default() -> Self {
-        Self{
+        Self {
-            address: Some("10.0.0.1".to_string()), // Dummy remote address
+            address: Some("10.13.13.2".to_string()), // Dummy remote address
             name: None,
-            mtu: None
+            mtu: None,
         }
     }
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
-pub enum DaemonResponseData{
+pub enum DaemonResponseData {
     ServerInfo(ServerInfo),
     ServerConfig(ServerConfig),
-    None
+    None,
 }
 
 #[test]
-fn test_response_serialization() -> anyhow::Result<()>{
+fn test_response_serialization() -> anyhow::Result<()> {
-    insta::assert_snapshot!(
+    insta::assert_snapshot!(serde_json::to_string(&DaemonResponse::new(Ok::<
-        serde_json::to_string(&DaemonResponse::new(Ok::<DaemonResponseData, String>(DaemonResponseData::None)))?
+        DaemonResponseData,
-    );
+        String,
-    insta::assert_snapshot!(
+    >(
-        serde_json::to_string(&DaemonResponse::new(Ok::<DaemonResponseData, String>(DaemonResponseData::ServerInfo(ServerInfo{
+        DaemonResponseData::None
+    )))?);
+    insta::assert_snapshot!(serde_json::to_string(&DaemonResponse::new(Ok::<
+        DaemonResponseData,
+        String,
+    >(
+        DaemonResponseData::ServerInfo(ServerInfo {
             name: Some("burrow".to_string()),
             ip: None,
             mtu: Some(1500)
-        }))))?
+        })
-    );
+    )))?);
-    insta::assert_snapshot!(
+    insta::assert_snapshot!(serde_json::to_string(&DaemonResponse::new(Err::<
-        serde_json::to_string(&DaemonResponse::new(Err::<DaemonResponseData, String>("error".to_string())))?
+        DaemonResponseData,
-    );
+        String,
-    insta::assert_snapshot!(
+    >(
-        serde_json::to_string(&DaemonResponse::new(Ok::<DaemonResponseData, String>(DaemonResponseData::ServerConfig(
+        "error".to_string()
-            ServerConfig::default()
+    )))?);
-        ))))?
+    insta::assert_snapshot!(serde_json::to_string(&DaemonResponse::new(Ok::<
-    );
+        DaemonResponseData,
+        String,
+    >(
+        DaemonResponseData::ServerConfig(ServerConfig::default())
+    )))?);
     Ok(())
 }

burrow/src/daemon/snapshots/burrow__daemon__command__daemoncommand_serialization-2.snap

@@ -1,5 +1,5 @@
 ---
 source: burrow/src/daemon/command.rs
-expression: "serde_json::to_string(&DaemonCommand::ServerInfo).unwrap()"
+expression: "serde_json::to_string(&DaemonCommand::Start(DaemonStartOptions {\n                    tun: TunOptions { seek_utun: true, ..TunOptions::default() },\n                })).unwrap()"
 ---
-"ServerInfo"
+{"Start":{"tun":{"name":null,"no_pi":false,"tun_excl":false,"seek_utun":true,"address":null}}}

burrow/src/daemon/snapshots/burrow__daemon__command__daemoncommand_serialization-3.snap

@@ -1,5 +1,5 @@
 ---
 source: burrow/src/daemon/command.rs
-expression: "serde_json::to_string(&DaemonCommand::Stop).unwrap()"
+expression: "serde_json::to_string(&DaemonCommand::ServerInfo).unwrap()"
 ---
-"Stop"
+"ServerInfo"

burrow/src/daemon/snapshots/burrow__daemon__command__daemoncommand_serialization-4.snap

@@ -1,5 +1,5 @@
 ---
 source: burrow/src/daemon/command.rs
-expression: "serde_json::to_string(&DaemonCommand::ServerConfig).unwrap()"
+expression: "serde_json::to_string(&DaemonCommand::Stop).unwrap()"
 ---
-"ServerConfig"
+"Stop"

burrow/src/daemon/snapshots/burrow__daemon__command__daemoncommand_serialization-5.snap

@@ -0,0 +1,5 @@
+---
+source: burrow/src/daemon/command.rs
+expression: "serde_json::to_string(&DaemonCommand::ServerConfig).unwrap()"
+---
+"ServerConfig"

burrow/src/daemon/snapshots/burrow__daemon__command__daemoncommand_serialization.snap

@@ -2,4 +2,4 @@
 source: burrow/src/daemon/command.rs
 expression: "serde_json::to_string(&DaemonCommand::Start(DaemonStartOptions::default())).unwrap()"
 ---
-{"Start":{"tun":{"name":null,"no_pi":null,"tun_excl":null}}}
+{"Start":{"tun":{"name":null,"no_pi":false,"tun_excl":false,"seek_utun":false,"address":null}}}

burrow/src/daemon/snapshots/burrow__daemon__response__response_serialization-4.snap

@@ -2,4 +2,4 @@
 source: burrow/src/daemon/response.rs
 expression: "serde_json::to_string(&DaemonResponse::new(Ok::<DaemonResponseData,\n                        String>(DaemonResponseData::ServerConfig(ServerConfig::default()))))?"
 ---
-{"result":{"Ok":{"ServerConfig":{"address":"10.0.0.1","name":null,"mtu":null}}},"id":0}
+{"result":{"Ok":{"ServerConfig":{"address":"10.13.13.2","name":null,"mtu":null}}},"id":0}

burrow/src/ensureroot.rs

@@ -1,40 +0,0 @@
-use tracing::instrument;
-
-// Check capabilities on Linux
-#[cfg(target_os = "linux")]
-#[instrument]
-pub fn ensure_root() {
-    use caps::{has_cap, CapSet, Capability};
-
-    let cap_net_admin = Capability::CAP_NET_ADMIN;
-    if let Ok(has_cap) = has_cap(None, CapSet::Effective, cap_net_admin) {
-        if !has_cap {
-            eprintln!(
-                "This action needs the CAP_NET_ADMIN permission. Did you mean to run it as root?"
-            );
-            std::process::exit(77);
-        }
-    } else {
-        eprintln!("Failed to check capabilities. Please file a bug report!");
-        std::process::exit(71);
-    }
-}
-
-// Check for root user on macOS
-#[cfg(target_vendor = "apple")]
-#[instrument]
-pub fn ensure_root() {
-    use nix::unistd::Uid;
-
-    let current_uid = Uid::current();
-    if !current_uid.is_root() {
-        eprintln!("This action must be run as root!");
-        std::process::exit(77);
-    }
-}
-
-#[cfg(target_family = "windows")]
-#[instrument]
-pub fn ensure_root() {
-    todo!()
-}

burrow/src/lib.rs

@@ -1,44 +1,16 @@
-pub mod ensureroot;
 pub mod wireguard;
 
-use anyhow::Result;
-
-#[cfg(any(target_os = "linux", target_vendor = "apple"))]
-use std::{
-    mem,
-    os::fd::{AsRawFd, FromRawFd},
-};
-
-use tun::TunInterface;
-
-// TODO Separate start and retrieve functions
-
 mod daemon;
-pub use daemon::{DaemonCommand, DaemonResponseData, DaemonStartOptions, DaemonResponse, ServerInfo};
+pub use daemon::{
+    DaemonCommand,
+    DaemonResponse,
+    DaemonResponseData,
+    DaemonStartOptions,
+    ServerInfo,
+};
 
 #[cfg(target_vendor = "apple")]
 mod apple;
 
 #[cfg(target_vendor = "apple")]
 pub use apple::*;
-
-#[cfg(any(target_os = "linux", target_vendor = "apple"))]
-#[no_mangle]
-pub extern "C" fn retrieve() -> i32 {
-    let iface2 = (1..100)
-        .filter_map(|i| {
-            let iface = unsafe { TunInterface::from_raw_fd(i) };
-            match iface.name() {
-                Ok(_name) => Some(iface),
-                Err(_) => {
-                    mem::forget(iface);
-                    None
-                }
-            }
-        })
-        .next();
-    match iface2 {
-        Some(iface) => iface.as_raw_fd(),
-        None => -1,
-    }
-}

burrow/src/main.rs

@@ -1,23 +1,19 @@
-use std::mem;
-#[cfg(any(target_os = "linux", target_vendor = "apple"))]
-use std::os::fd::FromRawFd;
-
 use anyhow::{Context, Result};
 #[cfg(any(target_os = "linux", target_vendor = "apple"))]
-use burrow::retrieve;
 use clap::{Args, Parser, Subcommand};
-use tracing::{instrument, Level};
+use tracing::instrument;
 use tracing_log::LogTracer;
 use tracing_oslog::OsLogger;
-use tracing_subscriber::{prelude::*, FmtSubscriber, EnvFilter};
+use tracing_subscriber::{prelude::*, EnvFilter, FmtSubscriber};
 #[cfg(any(target_os = "linux", target_vendor = "apple"))]
 use tun::TunInterface;
 
-
 mod daemon;
 mod wireguard;
 
 use daemon::{DaemonClient, DaemonCommand, DaemonStartOptions};
+use tun::TunOptions;
+
 use crate::daemon::DaemonResponseData;
 
 #[derive(Parser)]
@@ -65,7 +61,9 @@ struct DaemonArgs {}
 async fn try_start() -> Result<()> {
     let mut client = DaemonClient::new().await?;
     client
-        .send_command(DaemonCommand::Start(DaemonStartOptions::default()))
+        .send_command(DaemonCommand::Start(DaemonStartOptions {
+            tun: TunOptions::new().address("10.13.13.2"),
+        }))
         .await
         .map(|_| ())
 }
@@ -87,9 +85,8 @@ async fn try_retrieve() -> Result<()> {
         }
     }
 
-    burrow::ensureroot::ensure_root();
+    let iface2 = TunInterface::retrieve().ok_or(anyhow::anyhow!("No interface found"))?;
-    let iface2 = retrieve();
+    tracing::info!("{:?}", iface2);
-    tracing::info!("{}", iface2);
     Ok(())
 }
 
@@ -104,9 +101,10 @@ async fn initialize_tracing() -> Result<()> {
                 FmtSubscriber::builder()
                     .with_line_number(true)
                     .with_env_filter(EnvFilter::from_default_env())
-                    .finish()
+                    .finish(),
             );
-            tracing::subscriber::set_global_default(logger).context("Failed to set the global tracing subscriber")?;
+            tracing::subscriber::set_global_default(logger)
+                .context("Failed to set the global tracing subscriber")?;
         }
     }
 
@@ -121,7 +119,7 @@ async fn try_stop() -> Result<()> {
 }
 
 #[cfg(any(target_os = "linux", target_vendor = "apple"))]
-async fn try_serverinfo() -> Result<()>{
+async fn try_serverinfo() -> Result<()> {
     let mut client = DaemonClient::new().await?;
     let res = client.send_command(DaemonCommand::ServerInfo).await?;
     match res.result {
@@ -131,7 +129,9 @@ async fn try_serverinfo() -> Result<()>{
         Ok(DaemonResponseData::None) => {
             println!("Server not started.")
         }
-        Ok(res) => {println!("Unexpected Response: {:?}", res)}
+        Ok(res) => {
+            println!("Unexpected Response: {:?}", res)
+        }
         Err(e) => {
             println!("Error when retrieving from server: {}", e)
         }
@@ -140,7 +140,7 @@ async fn try_serverinfo() -> Result<()>{
 }
 
 #[cfg(any(target_os = "linux", target_vendor = "apple"))]
-async fn try_serverconfig() -> Result<()>{
+async fn try_serverconfig() -> Result<()> {
     let mut client = DaemonClient::new().await?;
     let res = client.send_command(DaemonCommand::ServerConfig).await?;
     match res.result {
@@ -150,7 +150,9 @@ async fn try_serverconfig() -> Result<()>{
         Ok(DaemonResponseData::None) => {
             println!("Server not started.")
         }
-        Ok(res) => {println!("Unexpected Response: {:?}", res)}
+        Ok(res) => {
+            println!("Unexpected Response: {:?}", res)
+        }
         Err(e) => {
             println!("Error when retrieving from server: {}", e)
         }
@@ -201,12 +203,8 @@ async fn main() -> Result<()> {
             try_stop().await?;
         }
         Commands::Daemon(_) => daemon::daemon_main().await?,
-        Commands::ServerInfo => {
+        Commands::ServerInfo => try_serverinfo().await?,
-            try_serverinfo().await?
+        Commands::ServerConfig => try_serverconfig().await?,
-        }
-        Commands::ServerConfig => {
-            try_serverconfig().await?
-        }
     }
 
     Ok(())

burrow/src/wireguard/iface.rs

@@ -1,23 +1,15 @@
-use std::{net::IpAddr, rc::Rc};
+use std::{net::IpAddr, sync::Arc, time::Duration};
-use std::sync::Arc;
-use std::time::Duration;
 
 use anyhow::Error;
 use async_trait::async_trait;
 use fehler::throws;
+use futures::{future::join_all, FutureExt};
 use ip_network_table::IpNetworkTable;
-use log::log;
+use tokio::{sync::RwLock, task::JoinHandle, time::timeout};
-use tokio::{
+use tracing::{debug, error};
-    join,
-    sync::{Mutex, RwLock},
-    task::{self, JoinHandle},
-};
 use tun::tokio::TunInterface;
-use futures::future::join_all;
-use futures::FutureExt;
-use tokio::time::timeout;
 
-use super::{noise::Tunnel, pcb, Peer, PeerPcb};
+use super::{noise::Tunnel, Peer, PeerPcb};
 
 #[async_trait]
 pub trait PacketInterface {
@@ -37,7 +29,7 @@ impl PacketInterface for tun::tokio::TunInterface {
 }
 
 struct IndexedPcbs {
-    pcbs: Vec<Arc<RwLock<PeerPcb>>>,
+    pcbs: Vec<Arc<PeerPcb>>,
     allowed_ips: IpNetworkTable<usize>,
 }
 
@@ -54,7 +46,7 @@ impl IndexedPcbs {
         for allowed_ip in pcb.allowed_ips.iter() {
             self.allowed_ips.insert(allowed_ip.clone(), idx);
         }
-        self.pcbs.insert(idx, Arc::new(RwLock::new(pcb)));
+        self.pcbs.insert(idx, Arc::new(pcb));
     }
 
     pub fn find(&self, addr: IpAddr) -> Option<usize> {
@@ -63,7 +55,7 @@ impl IndexedPcbs {
     }
 
     pub async fn connect(&self, idx: usize, handle: JoinHandle<()>) {
-        self.pcbs[idx].write().await.handle = Some(handle);
+        self.pcbs[idx].handle.write().await.replace(handle);
     }
 }
 
@@ -77,115 +69,124 @@ impl FromIterator<PeerPcb> for IndexedPcbs {
 }
 
 pub struct Interface {
-    tun: Arc<RwLock<TunInterface>>,
+    tun: Option<Arc<RwLock<TunInterface>>>,
     pcbs: Arc<IndexedPcbs>,
 }
 
 impl Interface {
     #[throws]
-    pub fn new<I: IntoIterator<Item = Peer>>(tun: TunInterface, peers: I) -> Self {
+    pub fn new<I: IntoIterator<Item = Peer>>(peers: I) -> Self {
-        let mut pcbs: IndexedPcbs = peers
+        let pcbs: IndexedPcbs = peers
             .into_iter()
             .map(|peer| PeerPcb::new(peer))
             .collect::<Result<_, _>>()?;
 
-        let tun = Arc::new(RwLock::new(tun));
         let pcbs = Arc::new(pcbs);
-        Self { tun, pcbs }
+        Self { pcbs, tun: None }
     }
 
-    pub async fn run(self) {
+    pub fn set_tun(&mut self, tun: Arc<RwLock<TunInterface>>) {
+        self.tun = Some(tun);
+    }
+
+    pub async fn run(&self) -> anyhow::Result<()> {
+        debug!("RUN: starting interface");
         let pcbs = self.pcbs.clone();
-        let tun = self.tun.clone();
+        let tun = self
+            .tun
+            .clone()
+            .ok_or(anyhow::anyhow!("tun interface does not exist"))?;
         log::info!("starting interface");
 
         let outgoing = async move {
             loop {
-                log::debug!("starting loop...");
+                // tracing::debug!("starting loop...");
                 let mut buf = [0u8; 3000];
 
                 let src = {
-                    log::debug!("awaiting read...");
+                    let src = match timeout(
-                    let src = match timeout(Duration::from_secs(2), tun.write().await.recv(&mut buf[..])).await {
+                        Duration::from_millis(10),
+                        tun.read().await.recv(&mut buf[..]),
+                    )
+                    .await
+                    {
                         Ok(Ok(len)) => &buf[..len],
-                        Ok(Err(e)) => {continue}
+                        Ok(Err(e)) => {
-                        Err(_would_block) => {
+                            error!("failed to read from interface: {}", e);
                             continue
                         }
+                        Err(_would_block) => continue,
                     };
-                    log::debug!("read {} bytes from interface", src.len());
+                    debug!("read {} bytes from interface", src.len());
-                    log::debug!("bytes: {:?}", src);
+                    debug!("bytes: {:?}", src);
                     src
                 };
 
-
                 let dst_addr = match Tunnel::dst_address(src) {
                     Some(addr) => addr,
                     None => {
-                        log::debug!("no destination found");
+                        tracing::debug!("no destination found");
                         continue
-                    },
+                    }
                 };
 
-                log::debug!("dst_addr: {}", dst_addr);
+                tracing::debug!("dst_addr: {}", dst_addr);
 
                 let Some(idx) = pcbs.find(dst_addr) else {
                     continue
                 };
 
-                log::debug!("found peer:{}", idx);
+                tracing::debug!("found peer:{}", idx);
 
-                match pcbs.pcbs[idx].read().await.send(src).await {
+                match pcbs.pcbs[idx].send(src).await {
                     Ok(..) => {
-                        log::debug!("sent packet to peer {}", dst_addr);
+                        let addr = pcbs.pcbs[idx].endpoint;
+                        tracing::debug!("sent packet to peer {}", addr);
                     }
                     Err(e) => {
                         log::error!("failed to send packet {}", e);
                         continue
-                    },
+                    }
                 };
-
-                // let mut buf = [0u8; 3000];
-                // match pcbs.pcbs[idx].read().await.recv(&mut buf).await {
-                //     Ok(len) => log::debug!("received {} bytes from peer {}", len, dst_addr),
-                //     Err(e) => {
-                //         log::error!("failed to receive packet {}", e);
-                //         continue
-                //     },
-                // }
             }
         };
 
         let mut tsks = vec![];
-        let tun = self.tun.clone();
+        let tun = self
+            .tun
+            .clone()
+            .ok_or(anyhow::anyhow!("tun interface does not exist"))?;
         let outgoing = tokio::task::spawn(outgoing);
         tsks.push(outgoing);
+        debug!("preparing to spawn read tasks");
+
         {
-            let pcbs = self.pcbs;
+            let pcbs = &self.pcbs;
-            for i in 0..pcbs.pcbs.len(){
+            for i in 0..pcbs.pcbs.len() {
-                let mut pcb = pcbs.pcbs[i].clone();
+                debug!("spawning read task for peer {}", i);
+                let pcb = pcbs.pcbs[i].clone();
                 let tun = tun.clone();
                 let tsk = async move {
-                    {
+                    if let Err(e) = pcb.open_if_closed().await {
-                        let r1 = pcb.write().await.open_if_closed().await;
+                        log::error!("failed to open pcb: {}", e);
-                        if let Err(e) = r1 {
+                        return
-                            log::error!("failed to open pcb: {}", e);
-                            return
-                        }
                     }
-                    let r2 = pcb.read().await.run().await;
+                    let r2 = pcb.run(tun).await;
                     if let Err(e) = r2 {
                         log::error!("failed to run pcb: {}", e);
                         return
                     } else {
-                        log::debug!("pcb ran successfully");
+                        tracing::debug!("pcb ran successfully");
                     }
                 };
+                debug!("task made..");
                 tsks.push(tokio::spawn(tsk));
             }
-            log::debug!("spawned read tasks");
+            debug!("spawned read tasks");
         }
-        log::debug!("preparing to join..");
+        debug!("preparing to join..");
         join_all(tsks).await;
+        debug!("joined!");
+        Ok(())
     }
 }

burrow/src/wireguard/noise/handshake.rs

@@ -136,10 +136,6 @@ fn aead_chacha20_open(
 ) -> Result<(), WireGuardError> {
     let mut nonce: [u8; 12] = [0; 12];
     nonce[4..].copy_from_slice(&counter.to_le_bytes());
-
-    log::debug!("TAG A");
-    log::debug!("{:?};{:?};{:?};{:?};{}", key, data, aad, nonce, counter);
-
     aead_chacha20_open_inner(buffer, key, nonce, data, aad)
         .map_err(|_| WireGuardError::InvalidAeadTag)?;
     Ok(())
@@ -683,7 +679,6 @@ impl Handshake {
             aad: &mac1[0..16],
             msg: packet.encrypted_cookie,
         };
-        log::debug!("TAG B");
         let plaintext = XChaCha20Poly1305::new_from_slice(&key)
             .unwrap()
             .decrypt(packet.nonce.into(), payload)

burrow/src/wireguard/noise/mod.rs

@@ -146,7 +146,7 @@ impl Tunnel {
 
         // Checks the type, as well as the reserved zero fields
         let packet_type = u32::from_le_bytes(src[0..4].try_into().unwrap());
-        log::debug!("packet_type: {}", packet_type);
+        tracing::debug!("packet_type: {}", packet_type);
 
         Ok(match (packet_type, src.len()) {
             (HANDSHAKE_INIT, HANDSHAKE_INIT_SZ) => Packet::HandshakeInit(HandshakeInit {

burrow/src/wireguard/noise/rate_limiter.rs

@@ -6,7 +6,6 @@ use std::{
 
 use aead::{generic_array::GenericArray, AeadInPlace, KeyInit};
 use chacha20poly1305::{Key, XChaCha20Poly1305};
-use log::log;
 use parking_lot::Mutex;
 use rand_core::{OsRng, RngCore};
 use ring::constant_time::verify_slices_are_equal;
@@ -174,14 +173,14 @@ impl RateLimiter {
         dst: &'b mut [u8],
     ) -> Result<Packet<'a>, TunnResult<'b>> {
         let packet = Tunnel::parse_incoming_packet(src)?;
-        log::debug!("packet: {:?}", packet);
+        tracing::debug!("packet: {:?}", packet);
 
         // Verify and rate limit handshake messages only
         if let Packet::HandshakeInit(HandshakeInit { sender_idx, .. })
         | Packet::HandshakeResponse(HandshakeResponse { sender_idx, .. }) = packet
         {
-            log::debug!("sender_idx: {}", sender_idx);
+            tracing::debug!("sender_idx: {}", sender_idx);
-            log::debug!("response: {:?}", packet);
+            tracing::debug!("response: {:?}", packet);
             let (msg, macs) = src.split_at(src.len() - 32);
             let (mac1, mac2) = macs.split_at(16);
 

burrow/src/wireguard/noise/session.rs

@@ -253,7 +253,7 @@ impl Session {
         // check the counter without running expensive decryption
         self.receiving_counter_quick_check(packet.counter)?;
 
-        log::debug!("TAG C");
+        tracing::debug!("TAG C");
         let ret = {
             let mut nonce = [0u8; 12];
             nonce[4..12].copy_from_slice(&packet.counter.to_le_bytes());

burrow/src/wireguard/pcb.rs

@@ -1,21 +1,17 @@
-use std::io;
+use std::{
-use std::net::SocketAddr;
+    cell::{Cell, RefCell},
-use std::rc::Rc;
+    net::SocketAddr,
-use std::sync::Arc;
+    sync::Arc,
-use std::time::Duration;
+};
 
 use anyhow::{anyhow, Error};
 use fehler::throws;
 use ip_network::IpNetwork;
-use log::log;
 use rand::random;
-use tokio::{net::UdpSocket, task::JoinHandle};
+use tokio::{net::UdpSocket, sync::RwLock, task::JoinHandle};
-use tokio::sync::{Mutex, RwLock};
+use tun::tokio::TunInterface;
-use tokio::time::timeout;
-use uuid::uuid;
 
 use super::{
-    iface::PacketInterface,
     noise::{TunnResult, Tunnel},
     Peer,
 };
@@ -24,107 +20,101 @@ use super::{
 pub struct PeerPcb {
     pub endpoint: SocketAddr,
     pub allowed_ips: Vec<IpNetwork>,
-    pub handle: Option<JoinHandle<()>>,
+    pub handle: RwLock<Option<JoinHandle<()>>>,
-    socket: Option<UdpSocket>,
+    socket: RwLock<Option<UdpSocket>>,
     tunnel: RwLock<Tunnel>,
 }
 
 impl PeerPcb {
     #[throws]
     pub fn new(peer: Peer) -> Self {
-        let tunnel = RwLock::new(Tunnel::new(peer.private_key, peer.public_key, peer.preshared_key, None, 1, None)
+        let tunnel = RwLock::new(
-            .map_err(|s| anyhow::anyhow!("{}", s))?);
+            Tunnel::new(
-
+                peer.private_key,
+                peer.public_key,
+                peer.preshared_key,
+                None,
+                1,
+                None,
+            )
+            .map_err(|s| anyhow::anyhow!("{}", s))?,
+        );
         Self {
             endpoint: peer.endpoint,
             allowed_ips: peer.allowed_ips,
-            handle: None,
+            handle: RwLock::new(None),
-            socket: None,
+            socket: RwLock::new(None),
             tunnel,
         }
     }
 
-    pub async fn open_if_closed(&mut self) -> Result<(), Error> {
+    pub async fn open_if_closed(&self) -> Result<(), Error> {
-        if self.socket.is_none() {
+        if self.socket.read().await.is_none() {
             let socket = UdpSocket::bind("0.0.0.0:0").await?;
             socket.connect(self.endpoint).await?;
-            self.socket = Some(socket);
+            self.socket.write().await.replace(socket);
         }
         Ok(())
     }
 
-    pub async fn run(&self) -> Result<(), Error> {
+    pub async fn run(&self, tun_interface: Arc<RwLock<TunInterface>>) -> Result<(), Error> {
-        let mut buf = [0u8; 3000];
+        tracing::debug!("starting read loop for pcb... for {:?}", &self);
-        log::debug!("starting read loop for pcb...");
-        loop {
-            tracing::debug!("waiting for packet");
-            let len = self.recv(&mut buf).await?;
-            tracing::debug!("received {} bytes", len);
-        }
-    }
-
-    pub async fn recv(&self, buf: &mut [u8]) -> Result<usize, Error> {
-        log::debug!("starting read loop for pcb... for {:?}", &self);
         let rid: i32 = random();
-        log::debug!("start read loop {}", rid);
+        let mut buf: [u8; 3000] = [0u8; 3000];
-        loop{
+        tracing::debug!("start read loop {}", rid);
-            log::debug!("{}: waiting for packet", rid);
+        loop {
-            let Some(socket) = &self.socket else {
+            tracing::debug!("{}: waiting for packet", rid);
+            let guard = self.socket.read().await;
+            let Some(socket) = guard.as_ref() else {
                 continue
             };
-            let mut res_buf = [0;1500];
+            let mut res_buf = [0; 1500];
-            log::debug!("{} : waiting for readability on {:?}", rid, socket);
+            // tracing::debug!("{} : waiting for readability on {:?}", rid, socket);
-            match timeout(Duration::from_secs(2), socket.readable()).await {
+            let len = match socket.recv(&mut res_buf).await {
+                Ok(l) => l,
                 Err(e) => {
-                    log::debug!("{}: timeout waiting for readability on {:?}", rid, e);
+                    log::error!("{}: error reading from socket: {:?}", rid, e);
                     continue
                 }
-                Ok(Err(e)) => {
-                    log::debug!("{}: error waiting for readability on {:?}", rid, e);
-                    continue
-                }
-                Ok(Ok(_)) => {}
-            };
-            log::debug!("{}: readable!", rid);
-            let Ok(len) = socket.try_recv(&mut res_buf) else {
-                continue
             };
             let mut res_dat = &res_buf[..len];
             tracing::debug!("{}: Decapsulating {} bytes", rid, len);
             tracing::debug!("{:?}", &res_dat);
             loop {
-                match self.tunnel.write().await.decapsulate(None, res_dat, &mut buf[..]) {
+                match self
-                    TunnResult::Done => {
+                    .tunnel
-                        break;
+                    .write()
-                    }
+                    .await
+                    .decapsulate(None, res_dat, &mut buf[..])
+                {
+                    TunnResult::Done => break,
                     TunnResult::Err(e) => {
                         tracing::error!(message = "Decapsulate error", error = ?e);
-                        break;
+                        break
                     }
                     TunnResult::WriteToNetwork(packet) => {
                         tracing::debug!("WriteToNetwork: {:?}", packet);
+                        self.open_if_closed().await?;
                         socket.send(packet).await?;
                         tracing::debug!("WriteToNetwork done");
                         res_dat = &[];
-                        continue;
+                        continue
                     }
                     TunnResult::WriteToTunnelV4(packet, addr) => {
                         tracing::debug!("WriteToTunnelV4: {:?}, {:?}", packet, addr);
-                        continue;
+                        tun_interface.read().await.send(packet).await?;
+                        break
+                    }
+                    TunnResult::WriteToTunnelV6(packet, addr) => {
+                        tracing::debug!("WriteToTunnelV6: {:?}, {:?}", packet, addr);
+                        tun_interface.read().await.send(packet).await?;
+                        break
                     }
-                    e => panic!("Unexpected result from decapsulate: {:?}", e),
                 }
             }
-            return Ok(len)
         }
     }
 
-    pub async fn socket(&mut self) -> Result<&UdpSocket, Error> {
-        self.open_if_closed().await?;
-        Ok(self.socket.as_ref().expect("socket was just opened"))
-    }
-
-
     pub async fn send(&self, src: &[u8]) -> Result<(), Error> {
         let mut dst_buf = [0u8; 3000];
         match self.tunnel.write().await.encapsulate(src, &mut dst_buf[..]) {
@@ -133,7 +123,12 @@ impl PeerPcb {
                 tracing::error!(message = "Encapsulate error", error = ?e)
             }
             TunnResult::WriteToNetwork(packet) => {
-                let socket = &self.socket.as_ref().ok_or(anyhow!("socket not open"))?;
+                self.open_if_closed().await?;
+                let handle = self.socket.read().await;
+                let Some(socket) = handle.as_ref() else {
+                    tracing::error!("No socket for peer");
+                    return Ok(())
+                };
                 tracing::debug!("Our Encapsulated packet: {:?}", packet);
                 socket.send(packet).await?;
             }

burrow/src/wireguard/peer.rs

@@ -1,7 +1,5 @@
 use std::{fmt, net::SocketAddr};
 
-use anyhow::Error;
-use fehler::throws;
 use ip_network::IpNetwork;
 use x25519_dalek::{PublicKey, StaticSecret};
 
@@ -10,7 +8,7 @@ pub struct Peer {
     pub private_key: StaticSecret,
     pub public_key: PublicKey,
     pub allowed_ips: Vec<IpNetwork>,
-    pub preshared_key: Option<[u8; 32]>
+    pub preshared_key: Option<[u8; 32]>,
 }
 
 impl fmt::Debug for Peer {

tun/build.rs

@@ -26,7 +26,7 @@ async fn generate(out_dir: &std::path::Path) -> anyhow::Result<()> {
     println!("cargo:rerun-if-changed={}", binary_path.to_str().unwrap());
 
     if let (Ok(..), Ok(..)) = (File::open(&bindings_path), File::open(&binary_path)) {
-        return Ok(())
+        return Ok(());
     };
 
     let archive = download(out_dir)

tun/src/options.rs

@@ -5,28 +5,48 @@ use fehler::throws;
 use super::TunInterface;
 
 #[derive(Debug, Clone, Default)]
-#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema))]
+#[cfg_attr(
+    feature = "serde",
+    derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)
+)]
 pub struct TunOptions {
     /// (Windows + Linux) Name the tun interface.
-    pub(crate) name: Option<String>,
+    pub name: Option<String>,
     /// (Linux) Don't include packet information.
-    pub(crate) no_pi: Option<()>,
+    pub no_pi: bool,
     /// (Linux) Avoid opening an existing persistant device.
-    pub(crate) tun_excl: Option<()>,
+    pub tun_excl: bool,
+    /// (Linux) The IP address of the tun interface.
+    pub address: Option<String>,
 }
 
 impl TunOptions {
-    pub fn new() -> Self { Self::default() }
+    pub fn new() -> Self {
+        Self::default()
+    }
 
     pub fn name(mut self, name: &str) -> Self {
         self.name = Some(name.to_owned());
         self
     }
 
-    pub fn no_pi(mut self, enable: bool) { self.no_pi = enable.then_some(()); }
+    pub fn no_pi(mut self, enable: bool) -> Self {
+        self.no_pi = enable;
+        self
+    }
 
-    pub fn tun_excl(mut self, enable: bool) { self.tun_excl = enable.then_some(()); }
+    pub fn tun_excl(mut self, enable: bool) -> Self {
+        self.tun_excl = enable;
+        self
+    }
+
+    pub fn address(mut self, address: impl ToString) -> Self {
+        self.address = Some(address.to_string());
+        self
+    }
 
     #[throws]
-    pub fn open(self) -> TunInterface { TunInterface::new_with_options(self)? }
+    pub fn open(self) -> TunInterface {
+        TunInterface::new_with_options(self)?
+    }
 }

tun/src/tokio/mod.rs

@@ -5,15 +5,14 @@ use tracing::instrument;
 
 #[derive(Debug)]
 pub struct TunInterface {
-    inner: AsyncFd<crate::TunInterface>,
+    pub inner: AsyncFd<crate::TunInterface>,
 }
 
 impl TunInterface {
     #[instrument]
-    pub fn new(tun: crate::TunInterface) -> io::Result<Self> {
+    pub fn new(mut tun: crate::TunInterface) -> io::Result<Self> {
-        Ok(Self {
+        tun.set_nonblocking(true)?;
-            inner: AsyncFd::new(tun)?,
+        Ok(Self { inner: AsyncFd::new(tun)? })
-        })
     }
 
     #[instrument]
@@ -27,38 +26,17 @@ impl TunInterface {
         }
     }
 
-    // #[instrument]
+    #[instrument]
-    pub async fn recv(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+    pub async fn recv(&self, buf: &mut [u8]) -> io::Result<usize> {
         loop {
-            log::debug!("TunInterface receiving...");
+            let mut guard = self.inner.readable().await?;
-            let mut guard = self.inner.readable_mut().await?;
+            match guard.try_io(|inner| inner.get_ref().recv(buf)) {
-            log::debug!("Got! readable_mut");
+                Ok(result) => return result,
-            match guard.try_io(|inner| {
-                // log::debug!("Got! {:#?}", inner);
-                let raw_ref = (*inner).get_mut();
-                // log::debug!("Got mut ref! {:#?}", raw_ref);
-                let recved = raw_ref.recv(buf);
-                // log::debug!("Got recved! {:#?}", recved);
-                recved
-            }) {
-                Ok(result) => {
-                    log::debug!("HORRAY");
-                    return result
-                },
                 Err(_would_block) => {
-                    log::debug!("WouldBlock");
+                    tracing::debug!("WouldBlock");
                     continue
-                },
+                }
             }
         }
     }
-
-    #[instrument]
-    pub async fn try_recv(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        let mut guard = self.inner.readable_mut().await?;
-        match guard.try_io(|inner| (*inner).get_mut().recv(buf)) {
-            Ok(result) => Ok(result.unwrap_or_default()),
-            Err(_would_block) => Err(io::Error::new(io::ErrorKind::WouldBlock, "WouldBlock")),
-        }
-    }
 }

tun/src/unix/apple/mod.rs

@@ -1,8 +1,8 @@
 use std::{
     io::{Error, IoSlice},
-    mem,
+    mem::{self, ManuallyDrop},
     net::{Ipv4Addr, SocketAddrV4},
-    os::fd::{AsRawFd, RawFd},
+    os::fd::{AsRawFd, FromRawFd, RawFd},
 };
 
 use byteorder::{ByteOrder, NetworkEndian};
@@ -11,13 +11,14 @@ use libc::{c_char, iovec, writev, AF_INET, AF_INET6};
 use socket2::{Domain, SockAddr, Socket, Type};
 use tracing::{self, instrument};
 
-mod kern_control;
+pub mod kern_control;
-mod sys;
+pub mod sys;
 
 use kern_control::SysControlSocket;
 
 pub use super::queue::TunQueue;
-use super::{ifname_to_string, string_to_ifname, TunOptions};
+use super::{ifname_to_string, string_to_ifname};
+use crate::TunOptions;
 
 #[derive(Debug)]
 pub struct TunInterface {
@@ -33,8 +34,42 @@ impl TunInterface {
 
     #[throws]
     #[instrument]
-    pub fn new_with_options(_: TunOptions) -> TunInterface {
+    pub fn new_with_options(options: TunOptions) -> TunInterface {
-        TunInterface::connect(0)?
+        let ti = TunInterface::connect(0)?;
+        ti.configure(options)?;
+        ti
+    }
+
+    pub fn retrieve() -> Option<TunInterface> {
+        (3..100)
+            .filter_map(|fd| unsafe {
+                let peer_addr = socket2::SockAddr::init(|storage, len| {
+                    *len = mem::size_of::<sys::sockaddr_ctl>() as u32;
+                    libc::getpeername(fd, storage as *mut _, len);
+                    Ok(())
+                })
+                .map(|(_, addr)| (fd, addr));
+                peer_addr.ok()
+            })
+            .filter(|(_fd, addr)| {
+                let ctl_addr = unsafe { &*(addr.as_ptr() as *const libc::sockaddr_ctl) };
+                addr.family() == libc::AF_SYSTEM as u8
+                    && ctl_addr.ss_sysaddr == libc::AF_SYS_CONTROL as u16
+            })
+            .map(|(fd, _)| {
+                let socket = unsafe { socket2::Socket::from_raw_fd(fd) };
+                TunInterface { socket }
+            })
+            .next()
+    }
+
+    #[throws]
+    fn configure(&self, options: TunOptions) {
+        if let Some(addr) = options.address {
+            if let Ok(addr) = addr.parse() {
+                self.set_ipv4_addr(addr)?;
+            }
+        }
     }
 
     #[throws]

tun/src/unix/apple/sys.rs

@@ -2,20 +2,11 @@ use std::mem;
 
 use libc::{c_char, c_int, c_short, c_uint, c_ulong, sockaddr};
 pub use libc::{
-    c_void,
+    c_void, sockaddr_ctl, sockaddr_in, socklen_t, AF_SYSTEM, AF_SYS_CONTROL, IFNAMSIZ,
-    sockaddr_ctl,
-    sockaddr_in,
-    socklen_t,
-    AF_SYSTEM,
-    AF_SYS_CONTROL,
-    IFNAMSIZ,
     SYSPROTO_CONTROL,
 };
 use nix::{
-    ioctl_read_bad,
+    ioctl_read_bad, ioctl_readwrite, ioctl_write_ptr_bad, request_code_readwrite,
-    ioctl_readwrite,
-    ioctl_write_ptr_bad,
-    request_code_readwrite,
     request_code_write,
 };
 

tun/src/unix/linux/mod.rs

@@ -26,7 +26,9 @@ pub struct TunInterface {
 impl TunInterface {
     #[throws]
     #[instrument]
-    pub fn new() -> TunInterface { Self::new_with_options(TunOptions::new())? }
+    pub fn new() -> TunInterface {
+        Self::new_with_options(TunOptions::new())?
+    }
 
     #[throws]
     #[instrument]
@@ -212,5 +214,7 @@ impl TunInterface {
 
     #[throws]
     #[instrument]
-    pub fn send(&self, buf: &[u8]) -> usize { self.socket.send(buf)? }
+    pub fn send(&self, buf: &[u8]) -> usize {
+        self.socket.send(buf)?
+    }
 }

tun/src/unix/mod.rs

@@ -1,12 +1,11 @@
 use std::{
     io::{Error, Read},
+    mem::MaybeUninit,
     os::fd::{AsRawFd, FromRawFd, IntoRawFd, RawFd},
 };
 
 use tracing::instrument;
 
-use super::TunOptions;
-
 mod queue;
 
 #[cfg(target_vendor = "apple")]
@@ -40,21 +39,26 @@ impl IntoRawFd for TunInterface {
     }
 }
 
+unsafe fn assume_init(buf: &[MaybeUninit<u8>]) -> &[u8] {
+    &*(buf as *const [MaybeUninit<u8>] as *const [u8])
+}
+
 impl TunInterface {
     #[throws]
     #[instrument]
-    pub fn recv(&mut self, buf: &mut [u8]) -> usize {
+    pub fn recv(&self, buf: &mut [u8]) -> usize {
-        // there might be a more efficient way to implement this
+        // Use IoVec to read directly into target buffer
-        let tmp_buf = &mut [0u8; 1500];
+        let mut tmp_buf = [MaybeUninit::uninit(); 1500];
-        let len = self.socket.read(tmp_buf)?;
+        let len = self.socket.recv(&mut tmp_buf)?;
-        buf[..len-4].copy_from_slice(&tmp_buf[4..len]);
+        let result_buf = unsafe { assume_init(&tmp_buf[4..len]) };
-        len-4
+        buf[..len - 4].copy_from_slice(&result_buf);
+        len - 4
     }
 
     #[throws]
     #[instrument]
-    pub fn set_timeout(&self, timeout: Option<std::time::Duration>) {
+    pub fn set_nonblocking(&mut self, nb: bool) {
-        self.socket.set_read_timeout(timeout)?;
+        self.socket.set_nonblocking(nb)?;
     }
 }
 

tun/src/windows/mod.rs

@@ -25,7 +25,9 @@ impl Debug for TunInterface {
 
 impl TunInterface {
     #[throws]
-    pub fn new() -> TunInterface { Self::new_with_options(TunOptions::new())? }
+    pub fn new() -> TunInterface {
+        Self::new_with_options(TunOptions::new())?
+    }
 
     #[throws]
     pub(crate) fn new_with_options(options: TunOptions) -> TunInterface {
@@ -37,17 +39,18 @@ impl TunInterface {
         if handle.is_null() {
             unsafe { GetLastError() }.ok()?
         }
-        TunInterface {
+        TunInterface { handle, name: name_owned }
-            handle,
-            name: name_owned,
-        }
     }
 
-    pub fn name(&self) -> String { self.name.clone() }
+    pub fn name(&self) -> String {
+        self.name.clone()
+    }
 }
 
 impl Drop for TunInterface {
-    fn drop(&mut self) { unsafe { sys::WINTUN.WintunCloseAdapter(self.handle) } }
+    fn drop(&mut self) {
+        unsafe { sys::WINTUN.WintunCloseAdapter(self.handle) }
+    }
 }
 
 pub(crate) mod sys {

tun/tests/configure.rs

@@ -5,7 +5,9 @@ use tun::TunInterface;
 
 #[test]
 #[throws]
-fn test_create() { TunInterface::new()?; }
+fn test_create() {
+    TunInterface::new()?;
+}
 
 #[test]
 #[throws]