diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/handshake/device.rs | 6 | ||||
| -rw-r--r-- | src/handshake/mod.rs | 2 | ||||
| -rw-r--r-- | src/handshake/noise.rs | 10 | ||||
| -rw-r--r-- | src/handshake/peer.rs | 10 | ||||
| -rw-r--r-- | src/wireguard.rs | 72 |
5 files changed, 76 insertions, 24 deletions
diff --git a/src/handshake/device.rs b/src/handshake/device.rs index 5396854..638d63f 100644 --- a/src/handshake/device.rs +++ b/src/handshake/device.rs @@ -4,7 +4,7 @@ use std::net::SocketAddr; use std::sync::Mutex; use zerocopy::AsBytes; -use byteorder::{LittleEndian, ByteOrder}; +use byteorder::{ByteOrder, LittleEndian}; use rand::prelude::*; @@ -35,7 +35,7 @@ pub struct Device<T> { */ impl<T> Device<T> where - T: Copy, + T: Clone, { /// Initialize a new handshake state machine /// @@ -270,7 +270,7 @@ where // return unconfirmed keypair and the response as vector Ok(( - Some(peer.identifier), + Some(peer.identifier.clone()), Some(resp.as_bytes().to_owned()), Some(keys), )) diff --git a/src/handshake/mod.rs b/src/handshake/mod.rs index 8452de8..6d017cc 100644 --- a/src/handshake/mod.rs +++ b/src/handshake/mod.rs @@ -18,4 +18,4 @@ mod types; // publicly exposed interface pub use device::Device; -pub use messages::{TYPE_COOKIE_REPLY, TYPE_INITIATION, TYPE_RESPONSE }; +pub use messages::{TYPE_COOKIE_REPLY, TYPE_INITIATION, TYPE_RESPONSE}; diff --git a/src/handshake/noise.rs b/src/handshake/noise.rs index 9fc0eb4..eafb9e9 100644 --- a/src/handshake/noise.rs +++ b/src/handshake/noise.rs @@ -215,7 +215,7 @@ mod tests { } } -pub fn create_initiation<T: Copy, R: RngCore + CryptoRng>( +pub fn create_initiation<T: Clone, R: RngCore + CryptoRng>( rng: &mut R, device: &Device<T>, peer: &Peer<T>, @@ -296,7 +296,7 @@ pub fn create_initiation<T: Copy, R: RngCore + CryptoRng>( }) } -pub fn consume_initiation<'a, T: Copy>( +pub fn consume_initiation<'a, T: Clone>( device: &'a Device<T>, msg: &NoiseInitiation, ) -> Result<(&'a Peer<T>, TemporaryState), HandshakeError> { @@ -370,7 +370,7 @@ pub fn consume_initiation<'a, T: Copy>( }) } -pub fn create_response<T: Copy, R: RngCore + CryptoRng>( +pub fn create_response<T: Clone, R: RngCore + CryptoRng>( rng: &mut R, peer: &Peer<T>, sender: u32, // sending identifier @@ -456,7 +456,7 @@ pub fn create_response<T: Copy, R: RngCore + CryptoRng>( * allow concurrent processing of potential responses to the initiation, * in order to better mitigate DoS from malformed response messages. */ -pub fn consume_response<T: Copy>( +pub fn consume_response<T: Clone>( device: &Device<T>, msg: &NoiseResponse, ) -> Result<Output<T>, HandshakeError> { @@ -530,7 +530,7 @@ pub fn consume_response<T: Copy>( // return confirmed key-pair Ok(( - Some(peer.identifier), + Some(peer.identifier.clone()), None, Some(KeyPair { birth, diff --git a/src/handshake/peer.rs b/src/handshake/peer.rs index 6f0f5af..4c6f2fd 100644 --- a/src/handshake/peer.rs +++ b/src/handshake/peer.rs @@ -55,19 +55,19 @@ pub enum State { impl Drop for State { fn drop(&mut self) { match self { - State::InitiationSent{hs, ck, ..} => { + State::InitiationSent { hs, ck, .. } => { // eph_sk already cleared by dalek-x25519 hs.clear(); ck.clear(); - }, - _ => () + } + _ => (), } } } impl<T> Peer<T> where - T: Copy, + T: Clone, { pub fn new( identifier: T, // external identifier @@ -141,4 +141,4 @@ where *last_initiation_consumption = Some(Instant::now()); Ok(()) } -}
\ No newline at end of file +} diff --git a/src/wireguard.rs b/src/wireguard.rs index 71b981e..2c166b4 100644 --- a/src/wireguard.rs +++ b/src/wireguard.rs @@ -1,13 +1,15 @@ use crate::handshake; use crate::router; -use crate::types::{Bind, Tun}; +use crate::types::{Bind, Endpoint, Tun}; use std::sync::atomic::{AtomicUsize, Ordering}; -use std::sync::mpsc::sync_channel; use std::sync::Arc; use std::thread; use std::time::{Duration, Instant}; +use log::debug; +use rand::rngs::OsRng; + use byteorder::{ByteOrder, LittleEndian}; use crossbeam_channel::bounded; use x25519_dalek::StaticSecret; @@ -16,6 +18,14 @@ const SIZE_HANDSHAKE_QUEUE: usize = 128; const THRESHOLD_UNDER_LOAD: usize = SIZE_HANDSHAKE_QUEUE / 4; const DURATION_UNDER_LOAD: Duration = Duration::from_millis(10_000); +#[derive(Clone)] +pub struct Peer<T: Tun, B: Bind>(Arc<PeerInner<T, B>>); + +pub struct PeerInner<T: Tun, B: Bind> { + peer: router::Peer<Events, T, B>, + timers: Timers, +} + pub struct Timers {} pub struct Events(); @@ -38,7 +48,7 @@ pub struct Wireguard<T: Tun, B: Bind> { impl<T: Tun, B: Bind> Wireguard<T, B> { fn start(&self) {} - fn new(tun: T, bind: B) -> Wireguard<T, B> { + fn new(tun: T, bind: B, sk: StaticSecret) -> Wireguard<T, B> { let router = Arc::new(router::Device::new( num_cpus::get(), tun.clone(), @@ -46,11 +56,14 @@ impl<T: Tun, B: Bind> Wireguard<T, B> { )); let handshake_staged = Arc::new(AtomicUsize::new(0)); + let handshake_device: Arc<handshake::Device<Peer<T, B>>> = + Arc::new(handshake::Device::new(sk)); // start UDP read IO thread let (handshake_tx, handshake_rx) = bounded(128); { let tun = tun.clone(); + let bind = bind.clone(); thread::spawn(move || { let mut under_load = Instant::now() - DURATION_UNDER_LOAD - Duration::from_millis(1000); @@ -81,11 +94,9 @@ impl<T: Tun, B: Bind> Wireguard<T, B> { } // pass source address along if under load - if under_load.elapsed() < DURATION_UNDER_LOAD { - handshake_tx.send((msg, Some(src))).unwrap(); - } else { - handshake_tx.send((msg, None)).unwrap(); - } + handshake_tx + .send((msg, src, under_load.elapsed() < DURATION_UNDER_LOAD)) + .unwrap(); } router::TYPE_TRANSPORT => { // transport message @@ -98,9 +109,50 @@ impl<T: Tun, B: Bind> Wireguard<T, B> { // start handshake workers for _ in 0..num_cpus::get() { + let bind = bind.clone(); let handshake_rx = handshake_rx.clone(); - thread::spawn(move || loop { - let (msg, src) = handshake_rx.recv().unwrap(); // TODO handle error + let handshake_device = handshake_device.clone(); + thread::spawn(move || { + // prepare OsRng instance for this thread + let mut rng = OsRng::new().unwrap(); + + // process elements from the handshake queue + for (msg, src, under_load) in handshake_rx { + // feed message to handshake device + let src_validate = (&src).into_address(); // TODO avoid + match handshake_device.process( + &mut rng, + &msg[..], + if under_load { + Some(&src_validate) + } else { + None + }, + ) { + Ok((identity, msg, keypair)) => { + // send response + if let Some(msg) = msg { + let _ = bind.send(&msg[..], &src).map_err(|e| { + debug!( + "handshake worker, failed to send response, error = {:?}", + e + ) + }); + } + + // update timers + if let Some(identity) = identity { + // add keypair to peer and free any unused ids + if let Some(keypair) = keypair { + for id in identity.0.peer.add_keypair(keypair) { + handshake_device.release(id); + } + } + } + } + Err(e) => debug!("handshake worker, error = {:?}", e), + } + } }); } |