vpncloud/src/cloud.rs

860 lines
32 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::{max, min},
2019-12-04 08:32:35 +00:00
collections::HashMap,
fmt,
2019-12-19 15:09:52 +00:00
fs::{self, File},
2019-12-04 08:32:35 +00:00
hash::BuildHasherDefault,
2020-09-24 17:48:13 +00:00
io::{self, Cursor, Seek, SeekFrom, Write},
2019-12-04 08:32:35 +00:00
marker::PhantomData,
2019-12-19 15:09:52 +00:00
net::{SocketAddr, ToSocketAddrs},
2020-09-24 17:48:13 +00:00
path::Path,
str::FromStr
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;
2020-09-24 17:48:13 +00:00
use rand::{random, seq::SliceRandom, thread_rng};
use smallvec::{smallvec, SmallVec};
2015-11-19 15:34:20 +00:00
2020-09-24 17:48:13 +00:00
use crate::{
2019-12-04 08:32:35 +00:00
beacon::BeaconSerializer,
2020-06-24 13:28:16 +00:00
config::{Config, DEFAULT_PEER_TIMEOUT, DEFAULT_PORT},
2020-09-24 17:48:13 +00:00
crypto::{is_init_message, Crypto, MessageResult, PeerCrypto},
device::{Device, Type},
error::Error,
messages::{
NodeInfo, PeerInfo, MESSAGE_TYPE_CLOSE, MESSAGE_TYPE_DATA, MESSAGE_TYPE_KEEPALIVE, MESSAGE_TYPE_NODE_INFO
},
net::{mapped_addr, Socket},
payload::Protocol,
2019-12-04 08:32:35 +00:00
poll::{WaitImpl, WaitResult},
port_forwarding::PortForwarding,
2020-09-24 17:48:13 +00:00
table::ClaimTable,
2019-12-04 08:32:35 +00:00
traffic::TrafficStats,
2020-09-24 17:48:13 +00:00
types::{Address, Mode, NodeId, Range, RangeList},
util::{addr_nice, resolve, CtrlC, Duration, MsgBuffer, StatsdMsg, 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;
2020-09-24 17:48:13 +00:00
const SPACE_BEFORE: usize = 100;
2019-01-09 16:45:12 +00:00
struct PeerData {
2020-09-24 17:48:13 +00:00
last_seen: Time,
2019-01-09 16:45:12 +00:00
timeout: Time,
peer_timeout: u16,
2019-01-09 16:45:12 +00:00
node_id: NodeId,
2020-09-24 17:48:13 +00:00
crypto: PeerCrypto<NodeInfo>
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
2020-09-24 17:48:13 +00:00
pub struct GenericCloud<D: Device, P: Protocol, S: Socket, TS: TimeSource> {
node_id: NodeId,
2020-09-24 17:48:13 +00:00
config: Config,
2015-11-22 21:00:34 +00:00
learning: bool,
2015-11-22 21:45:04 +00:00
broadcast: bool,
2020-09-24 17:48:13 +00:00
peers: HashMap<SocketAddr, PeerData, Hash>,
reconnect_peers: Vec<ReconnectEntry>,
2019-02-19 21:04:21 +00:00
own_addresses: Vec<SocketAddr>,
2020-09-24 17:48:13 +00:00
pending_inits: HashMap<SocketAddr, PeerCrypto<NodeInfo>, Hash>,
table: ClaimTable<TS>,
socket: S,
2019-02-26 00:21:15 +00:00
device: D,
2020-09-24 17:48:13 +00:00
claims: RangeList,
2015-11-23 14:40:04 +00:00
crypto: Crypto,
2020-09-24 17:48:13 +00:00
next_peers: Time,
peer_timeout_publish: u16,
update_freq: u16,
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
}
2020-09-24 17:48:13 +00:00
impl<D: Device, P: Protocol, S: Socket, TS: TimeSource> GenericCloud<D, P, S, TS> {
2019-03-01 22:25:42 +00:00
#[allow(clippy::too_many_arguments)]
2020-09-24 17:48:13 +00:00
pub fn new(config: &Config, device: D, port_forwarding: Option<PortForwarding>, stats_file: Option<File>) -> 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
};
2020-09-24 17:48:13 +00:00
let (learning, broadcast) = match config.mode {
Mode::Normal => {
match config.device_type {
Type::Tap => (true, true),
Type::Tun | Type::Dummy => (false, false)
}
}
Mode::Router => (false, false),
Mode::Switch => (true, true),
Mode::Hub => (false, true)
};
let mut claims = SmallVec::with_capacity(config.claims.len());
for s in &config.claims {
claims.push(try_fail!(Range::from_str(s), "Invalid subnet format: {} ({})", s));
}
if device.get_type() == Type::Tun && config.auto_claim {
match device.get_ip() {
Ok(ip) => {
let range = Range { base: Address::from_ipv4(ip), prefix_len: 32 };
info!("Auto-claiming {} due to interface address", range);
claims.push(range);
}
Err(e) => error!("{}", e)
}
}
2019-02-24 19:01:32 +00:00
let now = TS::now();
let update_freq = config.get_keepalive() as u16;
2020-09-24 17:48:13 +00:00
let node_id = random();
let crypto = Crypto::new(node_id, &config.crypto).unwrap();
let beacon_key = config.beacon_password.as_ref().map(|s| s.as_bytes()).unwrap_or(&[]);
2019-12-04 08:32:35 +00:00
let mut res = GenericCloud {
2020-09-24 17:48:13 +00:00
node_id,
peers: HashMap::default(),
claims,
2019-01-01 23:35:14 +00:00
learning,
broadcast,
2020-09-24 17:48:13 +00:00
pending_inits: HashMap::default(),
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,
2020-09-24 17:48:13 +00:00
table: ClaimTable::new(config.switch_timeout as Duration, config.peer_timeout as Duration),
socket,
2019-01-01 23:35:14 +00:00
device,
2020-09-24 17:48:13 +00:00
next_peers: now,
update_freq,
stats_file,
2020-05-29 09:51:04 +00:00
statsd_server: config.statsd_server.clone(),
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(),
2020-09-24 17:48:13 +00:00
beacon_serializer: BeaconSerializer::new(beacon_key),
2019-02-19 17:42:50 +00:00
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]
2020-09-24 17:48:13 +00:00
fn broadcast_msg(&mut self, type_: u8, msg: &mut MsgBuffer) -> Result<(), Error> {
debug!("Broadcasting message type {}, {:?} bytes to {} peers", type_, msg.len(), self.peers.len());
let mut msg_data = MsgBuffer::new(100);
for (addr, peer) in &mut self.peers {
msg_data.set_start(msg.get_start());
msg_data.set_length(msg.len());
msg_data.message_mut().clone_from_slice(msg.message());
peer.crypto.send_message(type_, &mut msg_data)?;
2019-01-09 16:45:12 +00:00
self.traffic.count_out_traffic(*addr, msg_data.len());
2020-09-24 17:48:13 +00:00
match self.socket.send(msg_data.message(), *addr) {
2015-12-22 21:45:52 +00:00
Ok(written) if written == msg_data.len() => Ok(()),
2020-09-24 17:48:13 +00:00
Ok(_) => Err(Error::Socket("Sent out truncated packet")),
Err(e) => Err(Error::SocketIo("IOError when sending", e))
2019-03-01 22:12:19 +00:00
}?
2015-12-22 21:45:52 +00:00
}
Ok(())
}
#[inline]
2020-09-24 17:48:13 +00:00
fn send_to(&mut self, addr: SocketAddr, msg: &mut MsgBuffer) -> Result<(), Error> {
debug!("Sending msg with {} bytes to {}", msg.len(), addr);
self.traffic.count_out_traffic(addr, msg.len());
match self.socket.send(msg.message(), addr) {
Ok(written) if written == msg.len() => Ok(()),
Ok(_) => Err(Error::Socket("Sent out truncated packet")),
Err(e) => Err(Error::SocketIo("IOError when sending", e))
2015-11-19 15:34:20 +00:00
}
}
2020-09-24 17:48:13 +00:00
#[inline]
fn send_msg(&mut self, addr: SocketAddr, type_: u8, msg: &mut MsgBuffer) -> Result<(), Error> {
debug!("Sending msg with {} bytes to {}", msg.len(), addr);
let peer = match self.peers.get_mut(&addr) {
Some(peer) => peer,
None => return Err(Error::Message("Sending to node that is not a peer"))
};
peer.crypto.send_message(type_, msg)?;
self.send_to(addr, msg)
}
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> {
2020-09-24 17:48:13 +00:00
Ok(self.socket.address().map(mapped_addr)?)
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.
2020-06-24 13:28:16 +00:00
pub fn add_reconnect_peer(&mut self, mut add: String) {
2019-02-24 19:01:32 +00:00
let now = TS::now();
if add.find(':').unwrap_or(0) <= add.find(']').unwrap_or(0) {
// : not present or only in IPv6 address
2020-06-24 13:28:16 +00:00
add = format!("{}:{}", add, DEFAULT_PORT)
}
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
})
}
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> {
2020-09-24 17:48:13 +00:00
let addrs = resolve(&addr)?.into_iter().map(mapped_addr).collect::<Vec<_>>();
for addr in &addrs {
2020-10-24 22:24:11 +00:00
if self.own_addresses.contains(addr)
|| self.peers.contains_key(addr)
|| self.pending_inits.contains_key(addr)
{
2020-09-24 17:48:13 +00:00
return Ok(())
}
}
2016-06-27 13:43:30 +00:00
// Send a message to each resolved address
2020-09-24 17:48:13 +00:00
for a in addrs {
2016-06-27 13:43:30 +00:00
// Ignore error this time
2020-09-24 17:48:13 +00:00
self.connect_sock(a).ok();
2015-12-22 21:44:25 +00:00
}
Ok(())
2015-11-19 15:34:20 +00:00
}
2020-09-24 17:48:13 +00:00
fn create_node_info(&self) -> NodeInfo {
let mut peers = smallvec![];
for (addr, peer) in &self.peers {
peers.push(PeerInfo { node_id: Some(peer.node_id), addrs: smallvec![*addr] })
}
if peers.len() > 20 {
let mut rng = rand::thread_rng();
peers.partial_shuffle(&mut rng, 20);
peers.truncate(20);
}
2020-09-28 10:50:08 +00:00
NodeInfo {
node_id: self.node_id,
peers,
claims: self.claims.clone(),
peer_timeout: Some(self.peer_timeout_publish)
}
2020-09-24 17:48:13 +00:00
}
2019-02-19 21:04:21 +00:00
fn connect_sock(&mut self, addr: SocketAddr) -> Result<(), Error> {
2020-09-24 17:48:13 +00:00
let addr = mapped_addr(addr);
if self.peers.contains_key(&addr) || self.own_addresses.contains(&addr) {
2019-02-19 21:04:21 +00:00
return Ok(())
}
debug!("Connecting to {:?}", addr);
2020-09-24 17:48:13 +00:00
let payload = self.create_node_info();
let mut peer_crypto = self.crypto.peer_instance(payload);
let mut msg = MsgBuffer::new(SPACE_BEFORE);
peer_crypto.initialize(&mut msg)?;
self.pending_inits.insert(addr, peer_crypto);
self.send_to(addr, &mut msg)
}
fn crypto_housekeep(&mut self) -> Result<(), Error> {
let mut msg = MsgBuffer::new(SPACE_BEFORE);
let mut del: SmallVec<[SocketAddr; 4]> = smallvec![];
for addr in self.pending_inits.keys().copied().collect::<Vec<_>>() {
msg.clear();
match self.pending_inits.get_mut(&addr).unwrap().every_second(&mut msg) {
Err(_) => del.push(addr),
Ok(MessageResult::None) => (),
Ok(MessageResult::Reply) => self.send_to(addr, &mut msg)?,
Ok(_) => unreachable!()
}
}
for addr in self.peers.keys().copied().collect::<Vec<_>>() {
msg.clear();
match self.peers.get_mut(&addr).unwrap().crypto.every_second(&mut msg) {
Err(_) => del.push(addr),
Ok(MessageResult::None) => (),
Ok(MessageResult::Reply) => self.send_to(addr, &mut msg)?,
Ok(_) => unreachable!()
}
}
for addr in del {
self.peers.remove(&addr);
self.pending_inits.remove(&addr);
self.connect_sock(addr)?;
}
Ok(())
2019-02-19 21:04:21 +00:00
}
2015-11-20 08:11:54 +00:00
fn housekeep(&mut self) -> Result<(), Error> {
2020-09-24 17:48:13 +00:00
let now = TS::now();
let mut buffer = MsgBuffer::new(SPACE_BEFORE);
let mut del: Vec<SocketAddr> = Vec::new();
for (&addr, ref data) in &self.peers {
if data.timeout < now {
del.push(addr);
}
}
for addr in del {
2020-10-24 20:59:14 +00:00
info!("Forgot peer {} due to timeout", addr_nice(addr));
2020-09-24 17:48:13 +00:00
self.peers.remove(&addr);
self.table.remove_claims(addr);
self.connect_sock(addr)?; // Try to reconnect
}
2015-11-21 17:09:13 +00:00
self.table.housekeep();
2020-09-24 17:48:13 +00:00
self.crypto_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();
2020-09-24 17:48:13 +00:00
if self.next_peers <= now {
2015-11-19 15:34:20 +00:00
debug!("Send peer list to all peers");
2020-09-24 17:48:13 +00:00
let info = self.create_node_info();
info.encode(&mut buffer);
self.broadcast_msg(MESSAGE_TYPE_NODE_INFO, &mut buffer)?;
2016-06-27 13:43:30 +00:00
// Reschedule for next update
2020-09-24 17:48:13 +00:00
let min_peer_timeout = self.peers.iter().map(|p| p.1.peer_timeout).min().unwrap_or(DEFAULT_PEER_TIMEOUT);
let interval = min(self.update_freq as u16, max(min_peer_timeout / 2 - 60, 1));
self.next_peers = 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
2020-09-24 17:48:13 +00:00
for addr in &entry.resolved {
if self.peers.contains_key(&addr) {
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
}
2020-09-24 17:48:13 +00:00
// Exponential back-off: 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
2020-09-24 17:48:13 +00:00
self.write_out_stats().map_err(|err| Error::FileIo("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();
2020-10-11 19:45:28 +00:00
if let Some(path) = path.strip_prefix('|') {
2019-12-04 08:32:35 +00:00
self.beacon_serializer
2020-10-11 19:45:28 +00:00
.write_to_cmd(&peers, path)
2020-09-24 17:48:13 +00:00
.map_err(|e| Error::BeaconIo("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)
2020-09-24 17:48:13 +00:00
.map_err(|e| Error::BeaconIo("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 {
2020-10-11 19:45:28 +00:00
if let Some(path) = path.strip_prefix('|') {
2019-12-04 08:32:35 +00:00
self.beacon_serializer
2020-10-11 19:45:28 +00:00
.read_from_cmd(path, Some(50))
2020-09-24 17:48:13 +00:00
.map_err(|e| Error::BeaconIo("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))
2020-09-24 17:48:13 +00:00
.map_err(|e| Error::BeaconIo("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)?;
2020-09-24 17:48:13 +00:00
writeln!(f, "peers:")?;
let now = TS::now();
for (addr, data) in &self.peers {
writeln!(f, " - \"{}\": {{ ttl_secs: {} }}", addr_nice(*addr), data.timeout - now)?;
}
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;
2020-05-30 14:12:54 +00:00
let prefix = self.config.statsd_prefix.as_ref().map(|s| s as &str).unwrap_or("vpncloud");
let msg = StatsdMsg::new()
.with_ns(prefix, |msg| {
msg.add("peer_count", self.peers.len(), "g");
2020-09-24 17:48:13 +00:00
msg.add("table_cache_entries", self.table.cache_len(), "g");
msg.add("table_claims", self.table.claim_len(), "g");
2020-05-30 14:12:54 +00:00
msg.with_ns("traffic", |msg| {
msg.with_ns("protocol", |msg| {
msg.with_ns("inbound", |msg| {
msg.add("bytes", peer_traffic.in_bytes, "c");
msg.add("packets", peer_traffic.in_packets, "c");
});
msg.with_ns("outbound", |msg| {
msg.add("bytes", peer_traffic.out_bytes, "c");
msg.add("packets", peer_traffic.out_packets, "c");
});
});
msg.with_ns("payload", |msg| {
msg.with_ns("inbound", |msg| {
msg.add("bytes", payload_traffic.in_bytes, "c");
msg.add("packets", payload_traffic.in_packets, "c");
});
msg.with_ns("outbound", |msg| {
msg.add("bytes", payload_traffic.out_bytes, "c");
msg.add("packets", payload_traffic.out_packets, "c");
});
});
});
msg.with_ns("invalid_protocol_traffic", |msg| {
msg.add("bytes", dropped.in_bytes, "c");
msg.add("packets", dropped.in_packets, "c");
});
msg.with_ns("dropped_payload", |msg| {
msg.add("bytes", dropped.out_bytes, "c");
msg.add("packets", dropped.out_packets, "c");
});
})
.build();
2020-05-29 09:51:04 +00:00
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(()),
2020-09-24 17:48:13 +00:00
Ok(_) => Err(Error::Socket("Sent out truncated packet")),
Err(e) => Err(Error::SocketIo("IOError when sending", e))
2020-05-29 09:51:04 +00:00
}?
} else {
error!("Failed to resolve statsd server {}", endpoint);
}
}
Ok(())
}
2020-09-24 17:48:13 +00:00
pub fn handle_interface_data(&mut self, data: &mut MsgBuffer) -> Result<(), Error> {
let (src, dst) = P::parse(data.message())?;
debug!("Read data from interface: src: {}, dst: {}, {} bytes", src, dst, data.len());
self.traffic.count_out_payload(dst, src, data.len());
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);
2020-09-24 17:48:13 +00:00
self.send_msg(addr, MESSAGE_TYPE_DATA, data)?;
if !self.peers.contains_key(&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));
2020-09-24 17:48:13 +00:00
self.table.remove_claims(addr);
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);
2020-09-24 17:48:13 +00:00
self.broadcast_msg(MESSAGE_TYPE_DATA, data)?;
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-09-24 17:48:13 +00:00
self.traffic.count_dropped_payload(data.len());
2015-11-22 21:45:04 +00:00
}
2015-11-19 15:34:20 +00:00
}
}
Ok(())
}
2020-09-28 10:50:08 +00:00
fn add_new_peer(&mut self, addr: SocketAddr, info: NodeInfo) -> Result<(), Error> {
2020-09-24 17:48:13 +00:00
info!("Added peer {}", addr_nice(addr));
if let Some(init) = self.pending_inits.remove(&addr) {
self.peers.insert(addr, PeerData {
crypto: init,
2020-09-28 10:50:08 +00:00
node_id: info.node_id,
2020-09-24 17:48:13 +00:00
peer_timeout: info.peer_timeout.unwrap_or(DEFAULT_PEER_TIMEOUT),
last_seen: TS::now(),
timeout: TS::now() + self.config.peer_timeout as Time
});
self.update_peer_info(addr, Some(info))?;
} else {
error!("No init for new peer {}", addr_nice(addr));
}
Ok(())
}
fn remove_peer(&mut self, addr: SocketAddr) {
if let Some(_peer) = self.peers.remove(&addr) {
2020-10-24 22:24:11 +00:00
info!("Closing connection to {}", addr_nice(addr));
2020-09-24 17:48:13 +00:00
self.table.remove_claims(addr);
}
}
fn connect_to_peers(&mut self, peers: &[PeerInfo]) -> Result<(), Error> {
'outer: for peer in peers {
for addr in &peer.addrs {
if self.peers.contains_key(addr) {
continue 'outer
2015-11-22 21:00:34 +00:00
}
2019-12-04 08:32:35 +00:00
}
2020-09-24 17:48:13 +00:00
if let Some(node_id) = peer.node_id {
if self.node_id == node_id {
continue 'outer
2019-02-15 21:42:13 +00:00
}
2020-09-24 17:48:13 +00:00
for p in self.peers.values() {
if p.node_id == node_id {
continue 'outer
}
2015-11-19 15:34:20 +00:00
}
2019-12-04 08:32:35 +00:00
}
2020-09-24 17:48:13 +00:00
self.connect(&peer.addrs as &[SocketAddr])?;
}
Ok(())
}
fn update_peer_info(&mut self, addr: SocketAddr, info: Option<NodeInfo>) -> Result<(), Error> {
if let Some(peer) = self.peers.get_mut(&addr) {
peer.last_seen = TS::now();
peer.timeout = TS::now() + self.config.peer_timeout as Time
} else {
error!("Received peer update from non peer {}", addr_nice(addr));
return Ok(())
}
if let Some(info) = info {
debug!("Adding claims of peer {}: {:?}", addr_nice(addr), info.claims);
self.table.set_claims(addr, info.claims);
debug!("Received {} peers from {}: {:?}", info.peers.len(), addr_nice(addr), info.peers);
self.connect_to_peers(&info.peers)?;
}
Ok(())
}
fn handle_payload_from(&mut self, peer: SocketAddr, data: &mut MsgBuffer) -> Result<(), Error> {
let (src, dst) = P::parse(data.message())?;
let len = data.len();
debug!("Writing data to device: {} bytes", len);
self.traffic.count_in_payload(src, dst, len);
if let Err(e) = self.device.write(data) {
error!("Failed to send via device: {}", e);
return Err(e)
}
if self.learning {
// Learn single address
self.table.cache(src, peer);
}
Ok(())
}
fn handle_message(
&mut self, src: SocketAddr, msg_result: MessageResult<NodeInfo>, data: &mut MsgBuffer
) -> Result<(), Error> {
match msg_result {
MessageResult::Message(type_) => {
match type_ {
MESSAGE_TYPE_DATA => self.handle_payload_from(src, data)?,
MESSAGE_TYPE_NODE_INFO => {
let info = match NodeInfo::decode(Cursor::new(data.message())) {
Ok(val) => val,
Err(err) => {
self.traffic.count_invalid_protocol(data.len());
return Err(err)
}
};
self.update_peer_info(src, Some(info))?
2016-05-02 07:05:34 +00:00
}
2020-09-24 17:48:13 +00:00
MESSAGE_TYPE_KEEPALIVE => self.update_peer_info(src, None)?,
MESSAGE_TYPE_CLOSE => self.remove_peer(src),
_ => return Err(Error::Message("Unknown message type"))
2015-11-22 21:00:34 +00:00
}
2019-12-04 08:32:35 +00:00
}
2020-09-28 10:50:08 +00:00
MessageResult::Initialized(info) => self.add_new_peer(src, info)?,
MessageResult::InitializedWithReply(info) => {
self.add_new_peer(src, info)?;
2020-09-24 17:48:13 +00:00
self.send_to(src, data)?
2015-11-20 09:59:01 +00:00
}
2020-09-24 17:48:13 +00:00
MessageResult::Reply => self.send_to(src, data)?,
MessageResult::None => ()
2015-11-19 15:34:20 +00:00
}
Ok(())
}
2020-09-24 17:48:13 +00:00
pub fn handle_net_message(&mut self, src: SocketAddr, data: &mut MsgBuffer) -> Result<(), Error> {
2020-10-24 22:24:11 +00:00
let src = mapped_addr(src);
2020-09-24 17:48:13 +00:00
debug!("Received {} bytes from {}", data.len(), src);
let msg_result = if let Some(init) = self.pending_inits.get_mut(&src) {
init.handle_message(data)
} else if is_init_message(data.message()) {
let mut init = self.crypto.peer_instance(self.create_node_info());
let msg_result = init.handle_message(data);
2020-10-24 20:59:14 +00:00
match msg_result {
Ok(res) => {
self.pending_inits.insert(src, init);
Ok(res)
}
Err(err) => {
warn!("Error in init message from peer {}: {}", addr_nice(src), err);
return Ok(())
}
2020-09-24 17:48:13 +00:00
}
} else if let Some(peer) = self.peers.get_mut(&src) {
peer.crypto.handle_message(data)
} else {
2020-10-24 20:59:14 +00:00
info!("Ignoring non-init message from unknown peer {}", addr_nice(src));
2020-09-24 17:48:13 +00:00
return Ok(())
};
match msg_result {
Ok(val) => self.handle_message(src, val, data),
Err(err) => {
self.traffic.count_invalid_protocol(data.len());
Err(err)
}
}
}
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
}
2020-09-24 17:48:13 +00:00
fn handle_socket_event(&mut self, buffer: &mut MsgBuffer) {
let src = try_fail!(self.socket.receive(buffer), "Failed to read from network socket: {}");
self.traffic.count_in_traffic(src, buffer.len());
if let Err(e) = self.handle_net_message(src, buffer) {
2020-10-06 20:52:14 +00:00
if let Error::CryptoInit(_) = e {
2020-10-24 20:59:14 +00:00
debug!("Crypto init error: {}", e);
info!("Closing pending connection to {} due to error in crypto init", addr_nice(src));
2020-10-06 20:52:14 +00:00
self.pending_inits.remove(&src);
2020-10-24 20:59:14 +00:00
} else {
error!("Error: {}", e);
2020-09-24 17:48:13 +00:00
}
2019-01-10 18:36:50 +00:00
}
2019-02-21 21:41:36 +00:00
}
2020-09-24 17:48:13 +00:00
fn handle_device_event(&mut self, buffer: &mut MsgBuffer) {
2020-10-06 20:52:14 +00:00
try_fail!(self.device.read(buffer), "Failed to read from device: {}");
2020-09-24 17:48:13 +00:00
if let Err(e) = self.handle_interface_data(buffer) {
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: {}");
2020-09-24 17:48:13 +00:00
let mut buffer = MsgBuffer::new(SPACE_BEFORE);
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...");
2020-09-24 17:48:13 +00:00
buffer.clear();
self.broadcast_msg(MESSAGE_TYPE_CLOSE, &mut buffer).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)]
2020-09-24 17:48:13 +00:00
impl<P: Protocol> GenericCloud<MockDevice, P, 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) {
2020-09-24 17:48:13 +00:00
let mut buffer = MsgBuffer::new(SPACE_BEFORE);
self.handle_socket_event(&mut buffer);
2019-02-26 17:36:54 +00:00
}
pub fn trigger_device_event(&mut self) {
2020-09-24 17:48:13 +00:00
let mut buffer = MsgBuffer::new(SPACE_BEFORE);
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())
}
2020-09-24 17:48:13 +00:00
pub fn is_connected(&self, addr: &SocketAddr) -> bool {
self.peers.contains_key(addr)
2019-02-26 17:36:54 +00:00
}
pub fn own_addresses(&self) -> &[SocketAddr] {
&self.own_addresses
}
2020-09-24 17:48:13 +00:00
pub fn get_num(&self) -> usize {
self.socket.address().unwrap().port() as usize
2019-02-26 17:36:54 +00:00
}
}