vpncloud/src/cloud.rs

931 lines
34 KiB
Rust
Raw Normal View History

// VpnCloud - Peer-to-Peer VPN
2020-05-28 07:03:48 +00:00
// Copyright (C) 2015-2020 Dennis Schwerdel
// This software is licensed under GPL-3 or newer (see LICENSE.md)
2019-12-04 08:32:35 +00:00
use std::{
cmp::min,
collections::HashMap,
fmt,
2019-12-19 15:09:52 +00:00
fs::{self, File},
2019-12-04 08:32:35 +00:00
hash::BuildHasherDefault,
2019-12-29 12:03:48 +00:00
io::{self, Seek, SeekFrom, Write},
2019-12-04 08:32:35 +00:00
marker::PhantomData,
2019-12-19 15:09:52 +00:00
net::{SocketAddr, ToSocketAddrs},
path::Path
2019-12-04 08:32:35 +00:00
};
2015-11-19 15:34:20 +00:00
2016-03-29 08:45:54 +00:00
use fnv::FnvHasher;
2019-01-01 23:35:14 +00:00
use rand::{prelude::*, random, thread_rng};
2015-11-19 15:34:20 +00:00
2019-12-04 08:32:35 +00:00
use super::{
beacon::BeaconSerializer,
config::{Config, DEFAULT_PEER_TIMEOUT},
2019-12-04 08:32:35 +00:00
crypto::Crypto,
device::Device,
net::Socket,
poll::{WaitImpl, WaitResult},
port_forwarding::PortForwarding,
traffic::TrafficStats,
types::{Error, HeaderMagic, NodeId, Protocol, Range, Table},
udpmessage::{decode, encode, Message},
2020-05-29 08:04:14 +00:00
util::{addr_nice, resolve, CtrlC, Duration, Time, TimeSource}
2019-12-04 08:32:35 +00:00
};
2015-11-19 15:34:20 +00:00
2019-01-09 16:45:12 +00:00
pub type Hash = BuildHasherDefault<FnvHasher>;
2016-03-29 08:45:54 +00:00
const MAX_RECONNECT_INTERVAL: u16 = 3600;
const RESOLVE_INTERVAL: Time = 300;
2019-01-09 16:45:12 +00:00
pub const STATS_INTERVAL: Time = 60;
2019-01-09 16:45:12 +00:00
struct PeerData {
timeout: Time,
peer_timeout: u16,
2019-01-09 16:45:12 +00:00
node_id: NodeId,
2019-12-04 08:32:35 +00:00
alt_addrs: Vec<SocketAddr>
2019-01-09 16:45:12 +00:00
}
2019-02-26 17:36:54 +00:00
pub struct PeerList<TS: TimeSource> {
2015-11-19 15:34:20 +00:00
timeout: Duration,
2019-01-09 16:45:12 +00:00
peers: HashMap<SocketAddr, PeerData, Hash>,
2016-05-02 07:05:34 +00:00
nodes: HashMap<NodeId, SocketAddr, Hash>,
2019-02-24 19:01:32 +00:00
addresses: HashMap<SocketAddr, NodeId, Hash>,
_dummy_ts: PhantomData<TS>
2015-11-19 15:34:20 +00:00
}
2019-02-24 19:01:32 +00:00
impl<TS: TimeSource> PeerList<TS> {
fn new(timeout: Duration) -> PeerList<TS> {
2019-12-04 08:32:35 +00:00
PeerList {
2016-05-02 07:05:34 +00:00
peers: HashMap::default(),
2019-01-01 23:35:14 +00:00
timeout,
2016-05-02 07:05:34 +00:00
nodes: HashMap::default(),
2019-02-24 19:01:32 +00:00
addresses: HashMap::default(),
_dummy_ts: PhantomData
2016-05-02 07:05:34 +00:00
}
2015-11-19 15:34:20 +00:00
}
2015-11-20 11:09:07 +00:00
fn timeout(&mut self) -> Vec<SocketAddr> {
2019-02-24 19:01:32 +00:00
let now = TS::now();
2015-11-19 15:34:20 +00:00
let mut del: Vec<SocketAddr> = Vec::new();
2019-01-09 16:45:12 +00:00
for (&addr, ref data) in &self.peers {
if data.timeout < now {
2015-11-19 15:34:20 +00:00
del.push(addr);
}
}
2015-11-20 11:09:07 +00:00
for addr in &del {
2020-05-29 09:51:04 +00:00
info!("Forgot peer: {}", addr_nice(*addr));
2019-01-09 16:45:12 +00:00
if let Some(data) = self.peers.remove(addr) {
self.nodes.remove(&data.node_id);
2016-05-02 07:05:34 +00:00
self.addresses.remove(addr);
2019-01-09 16:45:12 +00:00
for addr in &data.alt_addrs {
2016-05-02 07:05:34 +00:00
self.addresses.remove(addr);
}
}
2015-11-19 15:34:20 +00:00
}
2015-11-20 11:09:07 +00:00
del
2015-11-19 15:34:20 +00:00
}
pub fn min_peer_timeout(&self) -> u16 {
self.peers.iter().map(|p| p.1.peer_timeout).min().unwrap_or(DEFAULT_PEER_TIMEOUT)
}
2016-06-11 14:08:57 +00:00
#[inline]
2019-02-26 17:36:54 +00:00
pub fn contains_addr(&self, addr: &SocketAddr) -> bool {
2019-02-14 22:39:16 +00:00
self.addresses.contains_key(addr)
2016-05-02 07:05:34 +00:00
}
#[inline]
2019-12-04 08:32:35 +00:00
pub fn is_connected<Addr: ToSocketAddrs + fmt::Debug>(&self, addr: Addr) -> Result<bool, Error> {
2019-03-01 22:12:19 +00:00
for addr in resolve(&addr)? {
2016-06-11 14:08:57 +00:00
if self.contains_addr(&addr) {
2019-12-04 08:32:35 +00:00
return Ok(true)
}
}
Ok(false)
}
2016-06-11 14:08:57 +00:00
#[inline]
2019-02-26 17:36:54 +00:00
pub fn contains_node(&self, node_id: &NodeId) -> bool {
2016-05-02 07:05:34 +00:00
self.nodes.contains_key(node_id)
2015-11-19 15:34:20 +00:00
}
2016-05-02 07:05:34 +00:00
2015-11-20 12:34:54 +00:00
#[inline]
fn add(&mut self, node_id: NodeId, addr: SocketAddr, peer_timeout: u16) {
2016-05-02 07:05:34 +00:00
if self.nodes.insert(node_id, addr).is_none() {
2020-05-29 09:51:04 +00:00
info!("New peer: {}", addr_nice(addr));
let mut alt_addrs = vec![];
if let SocketAddr::V6(v6_addr) = addr {
if let Some(ipv4) = v6_addr.ip().to_ipv4() {
let v4_addr = SocketAddr::from((ipv4, v6_addr.port()));
alt_addrs.push(v4_addr);
self.addresses.insert(v4_addr, node_id);
}
}
2019-01-09 16:45:12 +00:00
self.peers.insert(addr, PeerData {
2019-02-24 19:01:32 +00:00
timeout: TS::now() + Time::from(self.timeout),
2019-01-09 16:45:12 +00:00
node_id,
alt_addrs,
peer_timeout
2019-01-09 16:45:12 +00:00
});
2019-02-14 22:39:16 +00:00
self.addresses.insert(addr, node_id);
2016-05-02 07:05:34 +00:00
}
}
2016-08-12 06:31:21 +00:00
#[inline]
fn refresh(&mut self, addr: &SocketAddr) {
2019-01-09 16:45:12 +00:00
if let Some(ref mut data) = self.peers.get_mut(addr) {
2019-12-04 08:32:35 +00:00
data.timeout = TS::now() + Time::from(self.timeout);
2016-08-12 06:31:21 +00:00
}
}
2016-05-02 07:05:34 +00:00
#[inline]
2019-02-14 22:39:16 +00:00
fn make_primary(&mut self, node_id: NodeId, addr: SocketAddr) {
if self.peers.contains_key(&addr) {
return
2015-11-19 15:34:20 +00:00
}
2019-02-14 22:39:16 +00:00
let old_addr = match self.nodes.remove(&node_id) {
Some(old_addr) => old_addr,
None => return error!("Node not connected")
};
self.nodes.insert(node_id, addr);
let mut peer = match self.peers.remove(&old_addr) {
Some(peer) => peer,
None => return error!("Main address for node is not connected")
};
peer.alt_addrs.retain(|i| i != &addr);
peer.alt_addrs.push(old_addr);
self.peers.insert(addr, peer);
2019-02-19 21:04:21 +00:00
self.addresses.insert(addr, node_id);
2015-11-19 15:34:20 +00:00
}
2019-02-15 21:42:13 +00:00
#[inline]
2019-02-26 17:36:54 +00:00
pub fn get_node_id(&self, addr: &SocketAddr) -> Option<NodeId> {
2019-03-01 22:25:42 +00:00
self.addresses.get(addr).cloned()
2019-02-15 21:42:13 +00:00
}
2015-11-20 12:34:54 +00:00
#[inline]
2019-02-26 17:36:54 +00:00
pub fn as_vec(&self) -> Vec<SocketAddr> {
2019-02-14 22:39:16 +00:00
self.addresses.keys().cloned().collect()
2015-11-19 15:34:20 +00:00
}
2016-06-11 14:08:57 +00:00
#[inline]
2019-02-26 17:36:54 +00:00
pub fn len(&self) -> usize {
2015-11-20 12:34:54 +00:00
self.peers.len()
}
2016-06-11 14:08:57 +00:00
#[inline]
#[allow(dead_code)]
2019-02-26 17:36:54 +00:00
pub fn is_empty(&self) -> bool {
2016-06-11 14:08:57 +00:00
self.peers.is_empty()
}
2015-11-20 12:34:54 +00:00
#[inline]
2015-12-04 10:25:14 +00:00
fn subset(&self, size: usize) -> Vec<SocketAddr> {
2019-02-14 22:39:16 +00:00
self.peers.keys().choose_multiple(&mut thread_rng(), size).into_iter().cloned().collect()
2015-11-20 12:34:54 +00:00
}
#[inline]
2015-11-19 15:34:20 +00:00
fn remove(&mut self, addr: &SocketAddr) {
2019-01-09 16:45:12 +00:00
if let Some(data) = self.peers.remove(addr) {
2020-05-29 09:51:04 +00:00
info!("Removed peer: {}", addr_nice(*addr));
2019-01-09 16:45:12 +00:00
self.nodes.remove(&data.node_id);
2016-05-02 07:05:34 +00:00
self.addresses.remove(addr);
2019-01-09 16:45:12 +00:00
for addr in data.alt_addrs {
2016-05-02 07:05:34 +00:00
self.addresses.remove(&addr);
}
2015-11-19 15:34:20 +00:00
}
}
2019-01-09 16:45:12 +00:00
#[inline]
fn write_out<W: Write>(&self, out: &mut W) -> Result<(), io::Error> {
2020-05-29 08:04:14 +00:00
writeln!(out, "peers:")?;
2019-02-24 19:01:32 +00:00
let now = TS::now();
2019-01-09 16:45:12 +00:00
for (addr, data) in &self.peers {
2020-05-29 08:04:14 +00:00
writeln!(out, " - \"{}\": {{ ttl_secs: {} }}", addr_nice(*addr), data.timeout - now)?;
2019-01-09 16:45:12 +00:00
}
Ok(())
}
2015-11-19 15:34:20 +00:00
}
#[derive(Clone)]
pub struct ReconnectEntry {
address: String,
2016-08-29 13:20:32 +00:00
resolved: Vec<SocketAddr>,
next_resolve: Time,
tries: u16,
timeout: u16,
next: Time
}
2015-11-22 16:28:04 +00:00
2016-11-23 10:27:29 +00:00
2019-02-26 00:21:15 +00:00
pub struct GenericCloud<D: Device, P: Protocol, T: Table, S: Socket, TS: TimeSource> {
2019-02-19 21:04:21 +00:00
config: Config,
magic: HeaderMagic,
node_id: NodeId,
2019-02-24 19:01:32 +00:00
peers: PeerList<TS>,
2015-11-22 23:49:58 +00:00
addresses: Vec<Range>,
2015-11-22 21:00:34 +00:00
learning: bool,
2015-11-22 21:45:04 +00:00
broadcast: bool,
reconnect_peers: Vec<ReconnectEntry>,
2019-02-19 21:04:21 +00:00
own_addresses: Vec<SocketAddr>,
table: T,
socket: S,
2019-02-26 00:21:15 +00:00
device: D,
2015-11-23 14:40:04 +00:00
crypto: Crypto,
2015-11-25 20:55:30 +00:00
next_peerlist: Time,
peer_timeout_publish: u16,
update_freq: u16,
2019-12-04 08:32:35 +00:00
buffer_out: [u8; 64 * 1024],
stats_file: Option<File>,
2020-05-29 09:51:04 +00:00
statsd_server: Option<String>,
2015-11-25 20:55:30 +00:00
next_housekeep: Time,
2019-01-09 16:45:12 +00:00
next_stats_out: Time,
2019-02-19 21:04:21 +00:00
next_beacon: Time,
port_forwarding: Option<PortForwarding>,
2019-01-09 16:45:12 +00:00
traffic: TrafficStats,
2019-02-24 19:01:32 +00:00
beacon_serializer: BeaconSerializer<TS>,
2015-11-23 00:40:47 +00:00
_dummy_p: PhantomData<P>,
2019-02-24 19:01:32 +00:00
_dummy_ts: PhantomData<TS>
2015-11-19 16:11:59 +00:00
}
2019-02-26 00:21:15 +00:00
impl<D: Device, P: Protocol, T: Table, S: Socket, TS: TimeSource> GenericCloud<D, P, T, S, TS> {
2019-03-01 22:25:42 +00:00
#[allow(clippy::too_many_arguments)]
2019-12-04 08:32:35 +00:00
pub fn new(
config: &Config, device: D, table: T, learning: bool, broadcast: bool, addresses: Vec<Range>, crypto: Crypto,
port_forwarding: Option<PortForwarding>, stats_file: Option<File>
2019-12-04 08:32:35 +00:00
) -> Self
{
let socket = match S::listen(config.listen) {
2015-11-19 15:34:20 +00:00
Ok(socket) => socket,
Err(err) => fail!("Failed to open socket {}: {}", config.listen, err)
2015-11-19 15:34:20 +00:00
};
2019-02-24 19:01:32 +00:00
let now = TS::now();
let update_freq = config.get_keepalive() as u16;
2019-12-04 08:32:35 +00:00
let mut res = GenericCloud {
2019-01-10 18:36:50 +00:00
magic: config.get_magic(),
node_id: random(),
peers: PeerList::new(config.peer_timeout),
2019-01-01 23:35:14 +00:00
addresses,
learning,
broadcast,
2015-11-20 11:09:07 +00:00
reconnect_peers: Vec::new(),
2019-02-19 21:04:21 +00:00
own_addresses: Vec::new(),
peer_timeout_publish: config.peer_timeout as u16,
2019-01-01 23:35:14 +00:00
table,
socket,
2019-01-01 23:35:14 +00:00
device,
2019-02-24 19:01:32 +00:00
next_peerlist: now,
update_freq,
stats_file,
2020-05-29 09:51:04 +00:00
statsd_server: config.statsd_server.clone(),
2019-12-04 08:32:35 +00:00
buffer_out: [0; 64 * 1024],
2019-02-24 19:01:32 +00:00
next_housekeep: now,
next_stats_out: now + STATS_INTERVAL,
next_beacon: now,
2019-01-01 23:35:14 +00:00
port_forwarding,
2019-02-14 22:39:16 +00:00
traffic: TrafficStats::default(),
2019-02-19 17:42:50 +00:00
beacon_serializer: BeaconSerializer::new(&config.get_magic(), crypto.get_key()),
crypto,
2019-02-19 21:04:21 +00:00
config: config.clone(),
2015-11-23 00:40:47 +00:00
_dummy_p: PhantomData,
2019-02-24 19:01:32 +00:00
_dummy_ts: PhantomData
2019-02-21 21:41:36 +00:00
};
res.initialize();
2019-03-01 22:25:42 +00:00
res
2015-11-19 15:34:20 +00:00
}
2015-12-22 21:45:52 +00:00
#[inline]
2015-11-23 18:06:25 +00:00
pub fn ifname(&self) -> &str {
self.device.ifname()
}
2016-06-27 13:43:30 +00:00
/// Sends the message to all peers
///
/// # Errors
/// Returns an `Error::SocketError` when the underlying system call fails or only part of the
/// message could be sent (can this even happen?).
/// Some messages could have been sent.
2015-12-22 21:45:52 +00:00
#[inline]
fn broadcast_msg(&mut self, msg: &mut Message) -> Result<(), Error> {
debug!("Broadcasting {:?}", msg);
2016-06-27 13:43:30 +00:00
// Encrypt and encode once and send several times
let msg_data = encode(msg, &mut self.buffer_out, self.magic, &mut self.crypto);
for addr in self.peers.peers.keys() {
2019-01-09 16:45:12 +00:00
self.traffic.count_out_traffic(*addr, msg_data.len());
match self.socket.send(msg_data, *addr) {
2015-12-22 21:45:52 +00:00
Ok(written) if written == msg_data.len() => Ok(()),
2019-12-04 08:32:35 +00:00
Ok(_) => {
Err(Error::Socket("Sent out truncated packet", io::Error::new(io::ErrorKind::Other, "truncated")))
}
2016-08-08 20:24:04 +00:00
Err(e) => Err(Error::Socket("IOError when sending", e))
2019-03-01 22:12:19 +00:00
}?
2015-12-22 21:45:52 +00:00
}
Ok(())
}
2016-06-27 13:43:30 +00:00
/// Sends a message to one peer
///
/// # Errors
/// Returns an `Error::SocketError` when the underlying system call fails or only part of the
/// message could be sent (can this even happen?).
2015-12-22 21:45:52 +00:00
#[inline]
fn send_msg(&mut self, addr: SocketAddr, msg: &mut Message) -> Result<(), Error> {
2015-11-19 15:34:20 +00:00
debug!("Sending {:?} to {}", msg, addr);
2016-06-27 13:43:30 +00:00
// Encrypt and encode
let msg_data = encode(msg, &mut self.buffer_out, self.magic, &mut self.crypto);
2019-01-09 16:45:12 +00:00
self.traffic.count_out_traffic(addr, msg_data.len());
match self.socket.send(msg_data, addr) {
2015-12-13 21:03:06 +00:00
Ok(written) if written == msg_data.len() => Ok(()),
2016-07-06 20:35:42 +00:00
Ok(_) => Err(Error::Socket("Sent out truncated packet", io::Error::new(io::ErrorKind::Other, "truncated"))),
2016-08-08 20:24:04 +00:00
Err(e) => Err(Error::Socket("IOError when sending", e))
2015-11-19 15:34:20 +00:00
}
}
2016-06-27 13:43:30 +00:00
/// Returns the self-perceived addresses (IPv4 and IPv6) of this node
///
/// Note that those addresses could be private addresses that are not reachable by other nodes,
/// or only some other nodes inside the same network.
///
/// # Errors
/// Returns an IOError if the underlying system call fails
2016-02-08 19:37:06 +00:00
#[allow(dead_code)]
pub fn address(&self) -> io::Result<SocketAddr> {
Ok(self.socket.address()?)
2016-02-08 19:37:06 +00:00
}
2016-06-27 13:43:30 +00:00
/// Returns the number of peers
2016-02-08 19:37:06 +00:00
#[allow(dead_code)]
pub fn peer_count(&self) -> usize {
self.peers.len()
}
2016-06-27 13:43:30 +00:00
/// Adds a peer to the reconnect list
///
/// This method adds a peer to the list of nodes to reconnect to. A periodic task will try to
/// connect to the peer if it is not already connected.
pub fn add_reconnect_peer(&mut self, add: String) {
2019-02-24 19:01:32 +00:00
let now = TS::now();
2019-12-04 08:32:35 +00:00
let resolved = match resolve(&add as &str) {
2019-03-03 13:56:59 +00:00
Ok(addrs) => addrs,
Err(err) => {
warn!("Failed to resolve {}: {:?}", add, err);
vec![]
}
};
self.reconnect_peers.push(ReconnectEntry {
address: add,
tries: 0,
timeout: 1,
2019-12-04 08:32:35 +00:00
resolved,
2019-02-24 19:01:32 +00:00
next_resolve: now,
next: now
})
}
2019-02-19 21:04:21 +00:00
/// Returns whether the address is of this node
2016-06-27 13:43:30 +00:00
///
/// # Errors
/// Returns an `Error::SocketError` if the given address is a name that failed to resolve to
/// actual addresses.
2019-12-04 08:32:35 +00:00
fn is_own_address<Addr: ToSocketAddrs + fmt::Debug>(&self, addr: Addr) -> Result<bool, Error> {
2019-03-01 22:12:19 +00:00
for addr in resolve(&addr)? {
2019-02-19 21:04:21 +00:00
if self.own_addresses.contains(&addr) {
2019-12-04 08:32:35 +00:00
return Ok(true)
2015-11-20 11:09:07 +00:00
}
}
Ok(false)
}
2016-06-27 13:43:30 +00:00
/// Connects to a node given by its address
///
/// This method connects to node by sending a `Message::Init` to it. If `addr` is a name that
/// resolves to multiple addresses, one message is sent to each of them.
/// If the node is already a connected peer or the address is blacklisted, no message is sent.
///
/// # Errors
/// This method returns `Error::NameError` if the address is a name that fails to resolve.
2019-12-04 08:32:35 +00:00
pub fn connect<Addr: ToSocketAddrs + fmt::Debug + Clone>(&mut self, addr: Addr) -> Result<(), Error> {
2019-03-01 22:12:19 +00:00
if self.peers.is_connected(addr.clone())? || self.is_own_address(addr.clone())? {
return Ok(())
}
2016-08-29 13:20:32 +00:00
debug!("Connecting to {:?}", addr);
2015-12-22 21:44:25 +00:00
let subnets = self.addresses.clone();
2016-06-11 14:08:57 +00:00
let node_id = self.node_id;
2016-06-27 13:43:30 +00:00
// Send a message to each resolved address
2019-03-01 22:12:19 +00:00
for a in resolve(&addr)? {
2016-06-27 13:43:30 +00:00
// Ignore error this time
let mut msg = Message::Init(0, node_id, subnets.clone(), self.peer_timeout_publish);
2016-06-27 13:43:30 +00:00
self.send_msg(a, &mut msg).ok();
2015-12-22 21:44:25 +00:00
}
Ok(())
2015-11-19 15:34:20 +00:00
}
2019-02-19 21:04:21 +00:00
/// Connects to a node given by its address
///
/// This method connects to node by sending a `Message::Init` to it. If `addr` is a name that
/// resolves to multiple addresses, one message is sent to each of them.
/// If the node is already a connected peer or the address is blacklisted, no message is sent.
///
/// # Errors
/// This method returns `Error::NameError` if the address is a name that fails to resolve.
fn connect_sock(&mut self, addr: SocketAddr) -> Result<(), Error> {
if self.peers.contains_addr(&addr) || self.own_addresses.contains(&addr) {
return Ok(())
}
debug!("Connecting to {:?}", addr);
let subnets = self.addresses.clone();
let node_id = self.node_id;
2019-12-20 12:54:58 +00:00
let mut msg = Message::Init(0, node_id, subnets, self.peer_timeout_publish);
2019-02-19 21:04:21 +00:00
self.send_msg(addr, &mut msg)
}
2016-06-27 13:43:30 +00:00
/// Run all periodic housekeeping tasks
///
/// This method executes several tasks:
/// - Remove peers that have timed out
/// - Remove switch table entries that have timed out
/// - Periodically send the peers list to all peers
/// - Periodically reconnect to peers in the reconnect list
///
/// # Errors
/// This method returns errors if sending a message fails or resolving an address fails.
2015-11-20 08:11:54 +00:00
fn housekeep(&mut self) -> Result<(), Error> {
for peer in self.peers.timeout() {
self.table.remove_all(&peer);
}
2015-11-21 17:09:13 +00:00
self.table.housekeep();
// Periodically extend the port-forwarding
if let Some(ref mut pfw) = self.port_forwarding {
pfw.check_extend();
}
2016-06-27 13:43:30 +00:00
// Periodically send peer list to peers
2019-02-24 19:01:32 +00:00
let now = TS::now();
if self.next_peerlist <= now {
2015-11-19 15:34:20 +00:00
debug!("Send peer list to all peers");
2015-11-20 12:34:54 +00:00
let mut peer_num = self.peers.len();
2016-06-27 13:43:30 +00:00
// If the number of peers is high, send only a fraction of the full peer list to
2017-01-11 13:31:28 +00:00
// reduce the management traffic. The number of peers to send is limited by 20.
peer_num = min(peer_num, 20);
2016-06-27 13:43:30 +00:00
// Select that many peers...
2015-12-04 10:25:14 +00:00
let peers = self.peers.subset(peer_num);
2016-06-27 13:43:30 +00:00
// ...and send them to all peers
2015-12-13 21:03:06 +00:00
let mut msg = Message::Peers(peers);
2019-03-01 22:12:19 +00:00
self.broadcast_msg(&mut msg)?;
2016-06-27 13:43:30 +00:00
// Reschedule for next update
let interval = min(self.update_freq as u16, self.peers.min_peer_timeout());
self.next_peerlist = now + Time::from(interval);
}
2016-06-27 13:43:30 +00:00
// Connect to those reconnect_peers that are due
for entry in self.reconnect_peers.clone() {
if entry.next > now {
continue
}
2019-03-01 22:12:19 +00:00
self.connect(&entry.resolved as &[SocketAddr])?;
2015-11-19 15:34:20 +00:00
}
for entry in &mut self.reconnect_peers {
2016-06-27 13:43:30 +00:00
// Schedule for next second if node is connected
2019-03-01 22:12:19 +00:00
if self.peers.is_connected(&entry.resolved as &[SocketAddr])? {
entry.tries = 0;
entry.timeout = 1;
entry.next = now + 1;
continue
}
2016-08-29 13:20:32 +00:00
// Resolve entries anew
if entry.next_resolve <= now {
if let Ok(addrs) = resolve(&entry.address as &str) {
entry.resolved = addrs;
}
entry.next_resolve = now + RESOLVE_INTERVAL;
2016-08-29 13:20:32 +00:00
}
2016-06-27 13:43:30 +00:00
// Ignore if next attempt is already in the future
if entry.next > now {
continue
}
2016-06-27 13:43:30 +00:00
// Exponential backoff: every 10 tries, the interval doubles
entry.tries += 1;
if entry.tries > 10 {
entry.tries = 0;
entry.timeout *= 2;
}
2016-06-27 13:43:30 +00:00
// Maximum interval is one hour
if entry.timeout > MAX_RECONNECT_INTERVAL {
entry.timeout = MAX_RECONNECT_INTERVAL;
}
2016-06-27 13:43:30 +00:00
// Schedule next connection attempt
2019-01-01 23:35:14 +00:00
entry.next = now + Time::from(entry.timeout);
2015-11-20 11:09:07 +00:00
}
2019-01-09 16:45:12 +00:00
if self.next_stats_out < now {
// Write out the statistics
2019-03-01 22:12:19 +00:00
self.write_out_stats().map_err(|err| Error::File("Failed to write stats file", err))?;
2020-05-29 09:51:04 +00:00
self.send_stats_to_statsd()?;
2019-01-09 16:45:12 +00:00
self.next_stats_out = now + STATS_INTERVAL;
self.traffic.period(Some(5));
2019-01-09 16:45:12 +00:00
}
2019-02-19 21:04:21 +00:00
if let Some(peers) = self.beacon_serializer.get_cmd_results() {
debug!("Loaded beacon with peers: {:?}", peers);
for peer in peers {
2019-03-01 22:12:19 +00:00
self.connect_sock(peer)?;
2019-02-19 21:04:21 +00:00
}
}
if self.next_beacon < now {
2019-03-01 22:12:19 +00:00
self.store_beacon()?;
self.load_beacon()?;
2019-02-19 21:04:21 +00:00
self.next_beacon = now + Time::from(self.config.beacon_interval);
}
Ok(())
}
/// Stores the beacon
fn store_beacon(&mut self) -> Result<(), Error> {
if let Some(ref path) = self.config.beacon_store {
2019-12-04 08:32:35 +00:00
let peers: Vec<_> = self.own_addresses.choose_multiple(&mut thread_rng(), 3).cloned().collect();
2019-02-19 21:04:21 +00:00
if path.starts_with('|') {
2019-12-04 08:32:35 +00:00
self.beacon_serializer
.write_to_cmd(&peers, &path[1..])
.map_err(|e| Error::Beacon("Failed to call beacon command", e))?;
2019-02-19 21:04:21 +00:00
} else {
2019-12-04 08:32:35 +00:00
self.beacon_serializer
.write_to_file(&peers, &path)
.map_err(|e| Error::Beacon("Failed to write beacon to file", e))?;
2019-02-19 21:04:21 +00:00
}
}
Ok(())
}
/// Loads the beacon
fn load_beacon(&mut self) -> Result<(), Error> {
let peers;
if let Some(ref path) = self.config.beacon_load {
if path.starts_with('|') {
2019-12-04 08:32:35 +00:00
self.beacon_serializer
.read_from_cmd(&path[1..], Some(50))
.map_err(|e| Error::Beacon("Failed to call beacon command", e))?;
2019-02-19 21:04:21 +00:00
return Ok(())
} else {
2019-12-04 08:32:35 +00:00
peers = self
.beacon_serializer
.read_from_file(&path, Some(50))
.map_err(|e| Error::Beacon("Failed to read beacon from file", e))?;
2019-02-19 21:04:21 +00:00
}
} else {
return Ok(())
}
debug!("Loaded beacon with peers: {:?}", peers);
for peer in peers {
2019-03-01 22:12:19 +00:00
self.connect_sock(peer)?;
2019-02-19 21:04:21 +00:00
}
2019-01-09 16:45:12 +00:00
Ok(())
}
2020-05-29 09:51:04 +00:00
/// Writes out the statistics to a file
2019-01-09 16:45:12 +00:00
fn write_out_stats(&mut self) -> Result<(), io::Error> {
if let Some(ref mut f) = self.stats_file {
debug!("Writing out stats");
2019-12-29 12:03:48 +00:00
f.seek(SeekFrom::Start(0))?;
f.set_len(0)?;
self.peers.write_out(f)?;
writeln!(f)?;
self.table.write_out(f)?;
writeln!(f)?;
self.traffic.write_out(f)?;
writeln!(f)?;
2019-12-04 08:32:35 +00:00
}
2015-11-19 15:34:20 +00:00
Ok(())
}
2020-05-29 09:51:04 +00:00
/// Sends the statistics to a statsd endpoint
fn send_stats_to_statsd(&mut self) -> Result<(), Error> {
if let Some(ref endpoint) = self.statsd_server {
let peer_traffic = self.traffic.total_peer_traffic();
let payload_traffic = self.traffic.total_payload_traffic();
let dropped = &self.traffic.dropped;
let msg = format!(
"peer_count:{}|g\ntable_entries:{}|g\n\
traffic.protocol.inbound.bytes:{}\n\
traffic.protocol.inbound.packets:{}\n\
traffic.protocol.outbound.bytes:{}\n\
traffic.protocol.outbound.packets:{}\n\
traffic.payload.inbound.bytes:{}\n\
traffic.payload.inbound.packets:{}\n\
traffic.payload.outbound.bytes:{}\n\
traffic.payload.outbound.packets:{}\n\
invalid_protocol_traffic.bytes:{}\n\
invalid_protocol_traffic.packets:{}\n\
dropped_payload.bytes:{}\n\
dropped_payload.packets:{}",
self.peers.len(),
self.table.len(),
peer_traffic.in_bytes,
peer_traffic.in_packets,
peer_traffic.out_bytes,
peer_traffic.out_packets,
payload_traffic.in_bytes,
payload_traffic.in_packets,
payload_traffic.out_bytes,
payload_traffic.out_packets,
dropped.in_bytes,
dropped.in_packets,
dropped.out_bytes,
dropped.out_packets
);
let msg_data = msg.as_bytes();
let addrs = resolve(endpoint)?;
if let Some(addr) = addrs.first() {
match self.socket.send(msg_data, *addr) {
Ok(written) if written == msg_data.len() => Ok(()),
Ok(_) => {
Err(Error::Socket(
"Sent out truncated packet",
io::Error::new(io::ErrorKind::Other, "truncated")
))
}
Err(e) => Err(Error::Socket("IOError when sending", e))
}?
} else {
error!("Failed to resolve statsd server {}", endpoint);
}
}
Ok(())
}
2016-06-27 13:43:30 +00:00
/// Handles payload data coming in from the local network device
///
/// This method takes payload data received from the local device and parses it to obtain the
/// destination address. Then it checks the lookup table to get the peer for that destination
/// address. If a peer is found, the message is sent to it, otherwise the message is either
/// broadcast to all peers or dropped (depending on mode).
///
/// The parameter `payload` contains the payload data starting at position `start` and ending
/// at `end`. It is important that the buffer has enough space before the payload data to
/// prepend a header of max 64 bytes and enough space after the payload data to append a mac of
/// max 64 bytes.
///
/// # Errors
/// This method fails
/// - with `Error::ParseError` if the payload data failed to parse
/// - with `Error::SocketError` if sending a message fails
2016-02-08 19:37:06 +00:00
pub fn handle_interface_data(&mut self, payload: &mut [u8], start: usize, end: usize) -> Result<(), Error> {
2019-03-01 22:12:19 +00:00
let (src, dst) = P::parse(&payload[start..end])?;
2019-12-04 08:32:35 +00:00
debug!("Read data from interface: src: {}, dst: {}, {} bytes", src, dst, end - start);
self.traffic.count_out_payload(dst, src, end - start);
2015-11-22 17:05:15 +00:00
match self.table.lookup(&dst) {
2019-12-04 08:32:35 +00:00
Some(addr) => {
// Peer found for destination
2015-11-26 21:16:51 +00:00
debug!("Found destination for {} => {}", dst, addr);
2019-03-01 22:12:19 +00:00
self.send_msg(addr, &mut Message::Data(payload, start, end))?;
if !self.peers.contains_addr(&addr) {
2019-01-09 16:45:12 +00:00
// If the peer is not actually connected, remove the entry in the table and try
2016-06-27 13:43:30 +00:00
// to reconnect.
2020-05-29 09:51:04 +00:00
warn!("Destination for {} not found in peers: {}", dst, addr_nice(addr));
self.table.remove(&dst);
2019-03-01 22:12:19 +00:00
self.connect_sock(addr)?;
2015-11-22 21:00:34 +00:00
}
2019-12-04 08:32:35 +00:00
}
2015-11-19 15:34:20 +00:00
None => {
2016-06-27 13:43:30 +00:00
if self.broadcast {
debug!("No destination for {} found, broadcasting", dst);
let mut msg = Message::Data(payload, start, end);
2019-03-01 22:12:19 +00:00
self.broadcast_msg(&mut msg)?;
2016-06-27 13:43:30 +00:00
} else {
2015-11-26 21:16:51 +00:00
debug!("No destination for {} found, dropping", dst);
2020-05-29 08:04:14 +00:00
self.traffic.count_dropped_payload(end - start);
2015-11-22 21:45:04 +00:00
}
2015-11-19 15:34:20 +00:00
}
}
Ok(())
}
2016-06-27 13:43:30 +00:00
/// Handles a message received from the network
///
/// This method handles messages from the network, i.e. from peers. `peer` contains the sender
/// of the message and `msg` contains the message.
2016-06-27 13:43:30 +00:00
///
/// Then this method will check the message type and will handle each message type differently.
///
/// # `Message::Data` messages
/// This message type contains payload data and therefore this path is optimized for speed.
///
/// The payload of data messages is written to the local network device and if the node is in
/// a learning mode it will associate the sender peer with the source address.
///
/// # `Message::Peers` messages
/// If this message is received, the local node will use all the node addresses in the message
/// as well as the senders address to connect to.
///
/// # `Message::Init` messages
/// This message is used in the peer connection handshake.
///
/// To make sure, the node does not connect to itself, it will compare the remote `node_id` to
/// the local one. If the id is the same, it will ignore the message and blacklist the address
/// so that it won't be used in the future.
///
/// If the message is coming from a different node, the nodes address is added to the peer list
/// and its claimed addresses are associated with it.
///
/// If the `stage` of the message is 1, a `Message::Init` message with `stage=1` is sent in
/// reply, together with a peer list.
///
/// # `Message::Close` message
/// If this message is received, the sender is removed from the peer list and its claimed
/// addresses are removed from the table.
pub fn handle_net_message(&mut self, peer: SocketAddr, msg: Message) -> Result<(), Error> {
debug!("Received {:?} from {}", msg, peer);
2015-11-19 15:34:20 +00:00
match msg {
2015-12-13 21:03:06 +00:00
Message::Data(payload, start, end) => {
2019-03-01 22:12:19 +00:00
let (src, dst) = P::parse(&payload[start..end])?;
2019-12-04 08:32:35 +00:00
debug!("Writing data to device: {} bytes", end - start);
self.traffic.count_in_payload(src, dst, end - start);
2016-07-06 20:35:42 +00:00
if let Err(e) = self.device.write(&mut payload[..end], start) {
error!("Failed to send via device: {}", e);
2019-12-04 08:32:35 +00:00
return Err(e)
2015-11-19 15:34:20 +00:00
}
2015-11-22 21:00:34 +00:00
if self.learning {
2016-06-27 13:43:30 +00:00
// Learn single address
2015-11-22 23:49:58 +00:00
self.table.learn(src, None, peer);
2015-11-22 21:00:34 +00:00
}
2016-06-27 13:43:30 +00:00
// Not adding peer in this case to increase performance
2019-12-04 08:32:35 +00:00
}
2015-11-20 17:09:51 +00:00
Message::Peers(peers) => {
2016-06-27 13:43:30 +00:00
// Connect to sender if not connected
if !self.peers.contains_addr(&peer) {
2019-03-01 22:12:19 +00:00
self.connect_sock(peer)?;
}
2019-02-15 21:42:13 +00:00
if let Some(node_id) = self.peers.get_node_id(&peer) {
self.peers.make_primary(node_id, peer);
}
2016-06-27 13:43:30 +00:00
// Connect to all peers in the message
2015-11-19 15:34:20 +00:00
for p in &peers {
2019-03-01 22:12:19 +00:00
self.connect_sock(*p)?;
2015-11-19 15:34:20 +00:00
}
2016-08-12 06:31:21 +00:00
// Refresh peer
self.peers.refresh(&peer);
2019-12-04 08:32:35 +00:00
}
Message::Init(stage, node_id, ranges, peer_timeout) => {
2016-06-27 13:43:30 +00:00
// Avoid connecting to self
if node_id == self.node_id {
2019-02-19 21:04:21 +00:00
self.own_addresses.push(peer);
return Ok(())
}
2016-06-27 13:43:30 +00:00
// Add sender as peer or as alternative address to existing peer
2016-05-02 07:05:34 +00:00
if self.peers.contains_node(&node_id) {
2019-02-14 22:39:16 +00:00
self.peers.make_primary(node_id, peer);
2016-05-02 07:05:34 +00:00
} else {
self.peers.add(node_id, peer, peer_timeout);
2020-02-20 21:52:40 +00:00
if self.learning && !ranges.is_empty() {
2020-05-29 09:51:04 +00:00
warn!("Ignoring claimed addresses received from {} in learning mode.", addr_nice(peer));
2020-02-20 21:52:40 +00:00
} else {
for range in ranges {
self.table.learn(range.base, Some(range.prefix_len), peer);
}
2016-05-02 07:05:34 +00:00
}
2015-11-22 21:00:34 +00:00
}
2016-06-27 13:43:30 +00:00
// Reply with stage=1 if stage is 0
2015-11-26 09:52:58 +00:00
if stage == 0 {
let own_addrs = self.addresses.clone();
2016-06-11 14:08:57 +00:00
let own_node_id = self.node_id;
self.send_msg(
peer,
&mut Message::Init(stage + 1, own_node_id, own_addrs, self.peer_timeout_publish)
)?;
2015-11-26 09:52:58 +00:00
}
2019-02-26 17:36:54 +00:00
// Send peers in any case
let peers = self.peers.as_vec();
2019-03-01 22:12:19 +00:00
self.send_msg(peer, &mut Message::Peers(peers))?;
2019-12-04 08:32:35 +00:00
}
2015-11-20 17:09:51 +00:00
Message::Close => {
2015-11-20 09:59:01 +00:00
self.peers.remove(&peer);
self.table.remove_all(&peer);
2015-11-20 09:59:01 +00:00
}
2015-11-19 15:34:20 +00:00
}
Ok(())
}
2019-02-21 21:41:36 +00:00
fn initialize(&mut self) {
2019-02-19 21:04:21 +00:00
match self.address() {
Err(err) => error!("Failed to obtain local addresses: {}", err),
Ok(addr) => self.own_addresses.push(addr)
2019-02-19 21:04:21 +00:00
}
2019-02-21 21:41:36 +00:00
}
fn handle_socket_data(&mut self, src: SocketAddr, data: &mut [u8]) {
let size = data.len();
2019-03-01 22:25:42 +00:00
if let Err(e) = decode(data, self.magic, &self.crypto).and_then(|msg| {
2019-02-21 21:41:36 +00:00
self.traffic.count_in_traffic(src, size);
self.handle_net_message(src, msg)
}) {
2020-05-29 09:51:04 +00:00
error!("Error: {}, from: {}", e, addr_nice(src));
2020-05-29 08:04:14 +00:00
self.traffic.count_invalid_protocol(size);
2019-01-10 18:36:50 +00:00
}
2019-02-21 21:41:36 +00:00
}
fn handle_socket_event(&mut self, buffer: &mut [u8]) {
let (size, src) = try_fail!(self.socket.receive(buffer), "Failed to read from network socket: {}");
2019-02-21 21:41:36 +00:00
self.handle_socket_data(src, &mut buffer[..size])
}
fn handle_device_event(&mut self, buffer: &mut [u8]) {
let mut start = 64;
let (offset, size) = try_fail!(self.device.read(&mut buffer[start..]), "Failed to read from tap device: {}");
start += offset;
2019-12-04 08:32:35 +00:00
if let Err(e) = self.handle_interface_data(buffer, start, start + size) {
2019-02-21 21:41:36 +00:00
error!("Error: {}", e);
}
}
/// The main method of the node
///
/// This method will use epoll to wait in the sockets and the device at the same time.
/// It will read from the sockets, decode and decrypt the message and then call the
/// `handle_net_message` method. It will also read from the device and call
/// `handle_interface_data` for each packet read.
/// Also, this method will call `housekeep` every second.
pub fn run(&mut self) {
let ctrlc = CtrlC::new();
let waiter = try_fail!(WaitImpl::new(&self.socket, &self.device, 1000), "Failed to setup poll: {}");
2019-12-04 08:32:35 +00:00
let mut buffer = [0; 64 * 1024];
let mut poll_error = false;
2019-02-21 21:41:36 +00:00
for evt in waiter {
match evt {
WaitResult::Error(err) => {
if poll_error {
fail!("Poll wait failed again: {}", err);
}
error!("Poll wait failed: {}, retrying...", err);
poll_error = true;
2019-12-04 08:32:35 +00:00
}
WaitResult::Timeout => {}
WaitResult::Socket => self.handle_socket_event(&mut buffer),
2019-02-21 21:41:36 +00:00
WaitResult::Device => self.handle_device_event(&mut buffer)
2015-11-19 15:34:20 +00:00
}
2019-02-24 19:01:32 +00:00
if self.next_housekeep < TS::now() {
poll_error = false;
2019-02-21 21:41:36 +00:00
if ctrlc.was_pressed() {
break
2015-11-25 20:55:30 +00:00
}
2016-06-29 06:43:39 +00:00
if let Err(e) = self.housekeep() {
error!("Error: {}", e)
2015-11-20 09:59:01 +00:00
}
2019-02-24 19:01:32 +00:00
self.next_housekeep = TS::now() + 1
2015-11-20 09:59:01 +00:00
}
2015-11-19 19:51:53 +00:00
}
2015-11-25 13:31:05 +00:00
info!("Shutting down...");
2015-12-22 21:45:52 +00:00
self.broadcast_msg(&mut Message::Close).ok();
2019-12-19 15:09:52 +00:00
if let Some(ref path) = self.config.beacon_store {
let path = Path::new(path);
if path.exists() {
info!("Removing beacon file");
if let Err(e) = fs::remove_file(path) {
error!("Failed to remove beacon file: {}", e)
}
}
}
2015-11-19 19:51:53 +00:00
}
2015-11-19 15:34:20 +00:00
}
2019-02-26 00:21:15 +00:00
2019-12-04 08:32:35 +00:00
#[cfg(test)] use super::device::MockDevice;
2019-02-26 00:21:15 +00:00
#[cfg(test)] use super::net::MockSocket;
2019-02-26 17:36:54 +00:00
#[cfg(test)] use super::util::MockTimeSource;
2019-02-26 00:21:15 +00:00
#[cfg(test)]
2019-02-26 17:36:54 +00:00
impl<P: Protocol, T: Table> GenericCloud<MockDevice, P, T, MockSocket, MockTimeSource> {
pub fn socket(&mut self) -> &mut MockSocket {
&mut self.socket
2019-02-26 17:36:54 +00:00
}
2019-02-26 00:21:15 +00:00
2019-02-26 17:36:54 +00:00
pub fn device(&mut self) -> &mut MockDevice {
&mut self.device
}
pub fn trigger_socket_event(&mut self) {
2019-12-04 08:32:35 +00:00
let mut buffer = [0; 64 * 1024];
self.handle_socket_event(&mut buffer);
2019-02-26 17:36:54 +00:00
}
pub fn trigger_device_event(&mut self) {
2019-12-04 08:32:35 +00:00
let mut buffer = [0; 64 * 1024];
2019-02-26 17:36:54 +00:00
self.handle_device_event(&mut buffer);
}
2019-03-03 13:56:59 +00:00
pub fn trigger_housekeep(&mut self) {
assert!(self.housekeep().is_ok())
}
2019-02-26 17:36:54 +00:00
pub fn node_id(&self) -> NodeId {
self.node_id
}
pub fn peers(&self) -> &PeerList<MockTimeSource> {
&self.peers
}
pub fn own_addresses(&self) -> &[SocketAddr] {
&self.own_addresses
}
pub fn decode_message<'a>(&self, msg: &'a mut [u8]) -> Result<Message<'a>, Error> {
decode(msg, self.magic, &self.crypto)
}
}