aboutsummaryrefslogtreecommitdiffstats
path: root/src/router/device.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/router/device.rs')
-rw-r--r--src/router/device.rs150
1 files changed, 125 insertions, 25 deletions
diff --git a/src/router/device.rs b/src/router/device.rs
index 1296be6..ec63111 100644
--- a/src/router/device.rs
+++ b/src/router/device.rs
@@ -1,24 +1,33 @@
-use arraydeque::{ArrayDeque, Wrapping};
+use arraydeque::{ArrayDeque, Saturating, Wrapping};
+use lifeguard::{Pool, Recycled};
use treebitmap::IpLookupTable;
-use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
-use std::sync::atomic::{AtomicPtr, AtomicU64};
-use std::sync::{Arc, Mutex};
-use std::time::Instant;
+use std::collections::HashMap;
+use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::sync::atomic::{AtomicPtr, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, Weak};
+use std::time::{Duration, Instant};
+
+use spin::RwLock;
use super::super::types::KeyPair;
use super::anti_replay::AntiReplay;
+use std::u64;
+
+const REJECT_AFTER_MESSAGES: u64 = u64::MAX - (1 << 4);
const MAX_STAGED_PACKETS: usize = 128;
-pub struct Device {
- ipv4: IpLookupTable<Ipv4Addr, Arc<Peer>>,
- ipv6: IpLookupTable<Ipv6Addr, Arc<Peer>>,
+pub struct Device<'a> {
+ recv: RwLock<HashMap<u32, Arc<Peer<'a>>>>, // map receiver id -> peer
+ ipv4: IpLookupTable<Ipv4Addr, Arc<Peer<'a>>>, // ipv4 trie
+ ipv6: IpLookupTable<Ipv6Addr, Arc<Peer<'a>>>, // ipv6 trie
+ pool: Pool<Vec<u8>>, // message buffer pool
}
struct KeyState(KeyPair, AntiReplay);
-struct EncryptState {
+struct EncryptionState {
key: [u8; 32], // encryption key
id: u64, // sender id
nonce: AtomicU64, // next available nonce
@@ -32,28 +41,119 @@ struct KeyWheel {
previous: AtomicPtr<Arc<Option<KeyState>>>, // old key state (used for decryption)
}
-pub struct Peer {
+pub struct Peer<'a> {
+ inorder: Mutex<ArrayDeque<[Option<Recycled<'a, Vec<u8>>>; MAX_STAGED_PACKETS], Saturating>>, // inorder queue
staged_packets: Mutex<ArrayDeque<[Vec<u8>; MAX_STAGED_PACKETS], Wrapping>>, // packets awaiting handshake
rx_bytes: AtomicU64, // received bytes
tx_bytes: AtomicU64, // transmitted bytes
keys: KeyWheel, // key-wheel
- encryption: AtomicPtr<Arc<EncryptState>>, // current encryption key (starts expired)
+ ekey: AtomicPtr<Arc<EncryptionState>>, // encryption state
+ endpoint: AtomicPtr<Arc<Option<SocketAddr>>>,
}
-pub struct PeerRef();
+impl<'a> Peer<'a> {
+ pub fn set_endpoint(&self, endpoint: SocketAddr) {
+ self.endpoint
+ .store(&mut Arc::new(Some(endpoint)), Ordering::Relaxed)
+ }
-impl Device {
- pub fn new() -> Device {
- unimplemented!();
+ pub fn add_keypair(&self, keypair: KeyPair) {
+ let confirmed = keypair.confirmed;
+ let mut st_new = Arc::new(Some(KeyState(keypair, AntiReplay::new())));
+ let st_previous = self.keys.previous.load(Ordering::Relaxed);
+ if confirmed {
+ // previous <- current
+ self.keys.previous.compare_and_swap(
+ st_previous,
+ self.keys.current.load(Ordering::Relaxed),
+ Ordering::Relaxed,
+ );
+
+ // current <- new
+ self.keys.next.store(&mut st_new, Ordering::Relaxed)
+ } else {
+ // previous <- next
+ self.keys.previous.compare_and_swap(
+ st_previous,
+ self.keys.next.load(Ordering::Relaxed),
+ Ordering::Relaxed,
+ );
+
+ // next <- new
+ self.keys.next.store(&mut st_new, Ordering::Relaxed)
+ }
+ }
+
+ pub fn rx_bytes(&self) -> u64 {
+ self.rx_bytes.load(Ordering::Relaxed)
+ }
+
+ pub fn tx_bytes(&self) -> u64 {
+ self.tx_bytes.load(Ordering::Relaxed)
+ }
+}
+
+impl<'a> Device<'a> {
+ pub fn new() -> Device<'a> {
+ Device {
+ recv: RwLock::new(HashMap::new()),
+ ipv4: IpLookupTable::new(),
+ ipv6: IpLookupTable::new(),
+ pool: Pool::with_size_and_max(0, MAX_STAGED_PACKETS * 2),
+ }
+ }
+
+ pub fn subnets(&self, peer: Arc<Peer<'a>>) -> Vec<(IpAddr, u32)> {
+ let mut subnets = Vec::new();
+
+ // extract ipv4 entries
+ for subnet in self.ipv4.iter() {
+ let (ip, masklen, p) = subnet;
+ if Arc::ptr_eq(&peer, p) {
+ subnets.push((IpAddr::V4(ip), masklen))
+ }
+ }
+
+ // extract ipv6 entries
+ for subnet in self.ipv6.iter() {
+ let (ip, masklen, p) = subnet;
+ if Arc::ptr_eq(&peer, p) {
+ subnets.push((IpAddr::V6(ip), masklen))
+ }
+ }
+
+ subnets
}
/// Adds a new peer to the device
///
/// # Returns
///
- /// An opaque value representing the peer.
- pub fn add(&self) -> PeerRef {
- unimplemented!();
+ /// A atomic ref. counted peer (with liftime matching the device)
+ pub fn add(&mut self) -> Arc<Peer<'a>> {
+ Arc::new(Peer {
+ inorder: Mutex::new(ArrayDeque::new()),
+ staged_packets: Mutex::new(ArrayDeque::new()),
+ rx_bytes: AtomicU64::new(0),
+ tx_bytes: AtomicU64::new(0),
+ keys: KeyWheel {
+ next: AtomicPtr::new(&mut Arc::new(None)),
+ current: AtomicPtr::new(&mut Arc::new(None)),
+ previous: AtomicPtr::new(&mut Arc::new(None)),
+ },
+ // long expired encryption key
+ ekey: AtomicPtr::new(&mut Arc::new(EncryptionState {
+ key: [0u8; 32],
+ id: 0,
+ nonce: AtomicU64::new(REJECT_AFTER_MESSAGES),
+ death: Instant::now() - Duration::from_secs(31536000),
+ })),
+ endpoint: AtomicPtr::new(&mut Arc::new(None)),
+ })
+ }
+
+ pub fn get_buffer(&self) -> Recycled<Vec<u8>> {
+ self.pool.new()
}
/// Cryptkey routes and sends a plaintext message (IP packet)
@@ -68,7 +168,7 @@ impl Device {
/// This indicates that a handshake should be initated (see the handshake module).
/// If this occurs the packet is copied to an internal buffer
/// and retransmission can be attempted using send_run_queue
- pub fn send(&self, pt_msg: &mut [u8]) -> Option<PeerRef> {
+ pub fn send(&self, pt_msg: &mut [u8]) -> Arc<Peer> {
unimplemented!();
}
@@ -80,7 +180,7 @@ impl Device {
///
/// - peer: Reference to the destination peer
/// - msg: Message to transmit
- pub fn send_raw(&self, peer: PeerRef, msg: &mut [u8]) {
+ pub fn send_raw(&self, peer: Arc<Peer>, msg: &mut [u8]) {
unimplemented!();
}
@@ -89,7 +189,7 @@ impl Device {
/// # Arguments
///
/// - peer: Reference for the peer to flush
- pub fn flush_queue(&self, peer: PeerRef) {
+ pub fn flush_queue(&self, peer: Arc<Peer>) {
unimplemented!();
}
@@ -101,7 +201,7 @@ impl Device {
///
/// A boolean indicating whether packages where sent.
/// Note: This is used for implicit confirmation of handshakes.
- pub fn send_run_queue(&self, peer: PeerRef) -> bool {
+ pub fn send_run_queue(&self, peer: Arc<Peer>) -> bool {
unimplemented!();
}
@@ -119,15 +219,15 @@ impl Device {
/// # Arguments
///
/// - peer: The peer to retrieve the endpoint for
- pub fn get_endpoint(&self, peer: PeerRef) -> SocketAddr {
+ pub fn get_endpoint(&self, peer: Arc<Peer>) -> SocketAddr {
unimplemented!();
}
- pub fn set_endpoint(&self, peer: PeerRef, endpoint: SocketAddr) {
+ pub fn set_endpoint(&self, peer: Arc<Peer>, endpoint: SocketAddr) {
unimplemented!();
}
- pub fn new_keypair(&self, peer: PeerRef, keypair: KeyPair) {
+ pub fn new_keypair(&self, peer: Arc<Peer>, keypair: KeyPair) {
unimplemented!();
}
}