openethereum/util/network/src/host.rs

1261 lines
41 KiB
Rust
Raw Normal View History

// Copyright 2015-2017 Parity Technologies (UK) Ltd.
2016-02-05 13:40:41 +01:00
// This file is part of Parity.
// Parity is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Parity is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
use std::net::{SocketAddr, SocketAddrV4, Ipv4Addr};
use std::collections::{HashMap, HashSet};
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering as AtomicOrdering};
2016-01-21 16:48:37 +01:00
use std::ops::*;
use std::cmp::min;
2016-02-15 18:36:34 +01:00
use std::path::{Path, PathBuf};
2016-12-10 14:20:34 +01:00
use std::io::{Read, Write, ErrorKind};
2016-02-15 18:36:34 +01:00
use std::fs;
use ethkey::{KeyPair, Secret, Random, Generator};
2015-11-29 11:50:28 +01:00
use mio::*;
2016-10-30 09:56:34 +01:00
use mio::deprecated::{EventLoop};
2015-11-29 11:50:28 +01:00
use mio::tcp::*;
use util::hash::*;
use util::Hashable;
use util::version;
2016-09-01 14:49:12 +02:00
use rlp::*;
use session::{Session, SessionInfo, SessionData};
use error::*;
use io::*;
use {NetworkProtocolHandler, NonReservedPeerMode, AllowIP, PROTOCOL_VERSION};
use node_table::*;
use stats::NetworkStats;
use discovery::{Discovery, TableUpdates, NodeEntry};
use ip_utils::{map_external_address, select_public_address};
use path::restrict_permissions_owner;
use parking_lot::{Mutex, RwLock};
2016-01-15 00:50:48 +01:00
type Slab<T> = ::slab::Slab<T, usize>;
2015-11-29 11:50:28 +01:00
const MAX_SESSIONS: usize = 1024 + MAX_HANDSHAKES;
const MAX_HANDSHAKES: usize = 1024;
const DEFAULT_PORT: u16 = 30303;
// Tokens
const TCP_ACCEPT: usize = SYS_TIMER + 1;
const IDLE: usize = SYS_TIMER + 2;
const DISCOVERY: usize = SYS_TIMER + 3;
const DISCOVERY_REFRESH: usize = SYS_TIMER + 4;
const DISCOVERY_ROUND: usize = SYS_TIMER + 5;
const NODE_TABLE: usize = SYS_TIMER + 6;
const FIRST_SESSION: usize = 0;
const LAST_SESSION: usize = FIRST_SESSION + MAX_SESSIONS - 1;
const USER_TIMER: usize = LAST_SESSION + 256;
const SYS_TIMER: usize = LAST_SESSION + 1;
// Timeouts
2016-01-14 16:52:10 +01:00
const MAINTENANCE_TIMEOUT: u64 = 1000;
const DISCOVERY_REFRESH_TIMEOUT: u64 = 60_000;
const DISCOVERY_ROUND_TIMEOUT: u64 = 300;
const NODE_TABLE_TIMEOUT: u64 = 300_000;
2016-01-14 16:52:10 +01:00
cli overhaul (#1600) * cli commands * cleanup parity/signer * cleanup parity/signer * remove redundant import of signer crate from main.rs * cli cleanup in progress * cli cleanup in progress * moved few commonly used functions to separate methods with tests * cleaning up blockchain import in progress * cleaning up blockchain import in progress2 * cleaning up blockchain import in progress3 * tests for database compaction profile parsing * cleaning up blockchain import in progress4 * cleaning up blockchain import in progress5 * blockchain import * export blockchain in progress * cleanup execute_export * Configuration::to_duration cleaned up * removed unused code, tests for to_duration * cleanup Configuration::mode function * parsing some of the cli params in params.rs * rpc and signer are no longer optional * move importing extern crates to main.rs file * swipe dies from rpc module * swipe dies from dapps * finding deprecated * several tests and fixes for parity * parity cleanup in progress * cleanup price parsing * parity cleanup in progress * swiped all dies * parity cleanup in progress * replace usages of from_str with parse() in parity/params.rs * removed few more from_str * split parity/params.rs into params and helpers * removed wildcard import from configuration.rs * cleanup directories/path creation * cleaning up run cmd * moved LoggerConfig * defaults for cli params * fixed indention in raise_fd_limit * tests for rpc_apis * tests for default ipc and rpc settings * ipc socket * cleanup in progress * account service * cleanup miner config * BlockChain commands use Directiores structure now * client_config * network settings and dapps configuration * removing warnings * default logger config * fixed client_path * overhaul * fixing export && import * default export DataFormat * import and export also upgrade db * fixed export && import * polishing pr * polishing pr * fixed custom bootnodes * fixed daemonize on windows * fixed setting up enable network * finished pr * fixed compiling on windows * Fixed warning; windows build * Better cache management * Fixed tests on windows * Fixed test * Restored pruning method names * --cache alias * Fixed more tests * Ensure default options actually listed as valid [ci:skip]
2016-07-25 16:09:47 +02:00
#[derive(Debug, PartialEq, Clone)]
2016-01-23 02:36:58 +01:00
/// Network service configuration
pub struct NetworkConfiguration {
/// Directory path to store general network configuration. None means nothing will be saved
2016-02-12 09:52:32 +01:00
pub config_path: Option<String>,
/// Directory path to store network-specific configuration. None means nothing will be saved
pub net_config_path: Option<String>,
2016-02-16 02:05:36 +01:00
/// IP address to listen for incoming connections. Listen to all connections by default
pub listen_address: Option<SocketAddr>,
/// IP address to advertise. Detected automatically if none.
pub public_address: Option<SocketAddr>,
/// Port for UDP connections, same as TCP by default
pub udp_port: Option<u16>,
2016-01-23 02:36:58 +01:00
/// Enable NAT configuration
pub nat_enabled: bool,
/// Enable discovery
pub discovery_enabled: bool,
/// List of initial node addresses
pub boot_nodes: Vec<String>,
2016-01-24 18:53:54 +01:00
/// Use provided node key instead of default
pub use_secret: Option<Secret>,
2016-07-29 17:30:02 +02:00
/// Minimum number of connected peers to maintain
pub min_peers: u32,
/// Maximum allowed number of peers
2016-07-29 17:30:02 +02:00
pub max_peers: u32,
/// Maximum handshakes
pub max_handshakes: u32,
/// Reserved protocols. Peers with <key> protocol get additional <value> connection slots.
pub reserved_protocols: HashMap<ProtocolId, u32>,
/// List of reserved node addresses.
pub reserved_nodes: Vec<String>,
/// The non-reserved peer mode.
pub non_reserved_mode: NonReservedPeerMode,
/// IP filter
pub allow_ips: AllowIP,
2015-11-29 11:50:28 +01:00
}
2016-03-11 10:57:58 +01:00
impl Default for NetworkConfiguration {
fn default() -> Self {
NetworkConfiguration::new()
}
}
2015-11-29 11:50:28 +01:00
impl NetworkConfiguration {
2016-01-23 02:36:58 +01:00
/// Create a new instance of default settings.
2016-03-11 10:57:58 +01:00
pub fn new() -> Self {
2016-01-08 13:49:00 +01:00
NetworkConfiguration {
2016-02-12 09:52:32 +01:00
config_path: None,
net_config_path: None,
2016-02-16 02:05:36 +01:00
listen_address: None,
public_address: None,
udp_port: None,
2016-01-08 13:49:00 +01:00
nat_enabled: true,
discovery_enabled: true,
boot_nodes: Vec::new(),
2016-01-24 18:53:54 +01:00
use_secret: None,
2016-07-29 17:30:02 +02:00
min_peers: 25,
max_peers: 50,
max_handshakes: 64,
reserved_protocols: HashMap::new(),
allow_ips: AllowIP::All,
reserved_nodes: Vec::new(),
non_reserved_mode: NonReservedPeerMode::Accept,
2016-01-08 13:49:00 +01:00
}
}
2016-01-24 18:53:54 +01:00
/// Create new default configuration with sepcified listen port.
pub fn new_with_port(port: u16) -> NetworkConfiguration {
let mut config = NetworkConfiguration::new();
config.listen_address = Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), port)));
2016-01-24 18:53:54 +01:00
config
}
2016-03-20 11:35:46 +01:00
/// Create new default configuration for localhost-only connection with random port (usefull for testing)
pub fn new_local() -> NetworkConfiguration {
let mut config = NetworkConfiguration::new();
config.listen_address = Some(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(127, 0, 0, 1), 0)));
2016-03-20 11:35:46 +01:00
config.nat_enabled = false;
config
}
2015-11-29 11:50:28 +01:00
}
2016-01-10 22:42:27 +01:00
/// Protocol handler level packet id
2015-12-22 22:23:43 +01:00
pub type PacketId = u8;
2016-01-10 22:42:27 +01:00
/// Protocol / handler id
pub type ProtocolId = [u8; 3];
2015-11-29 11:50:28 +01:00
2016-01-10 22:42:27 +01:00
/// Messages used to communitate with the event loop from other threads.
2016-01-21 16:48:37 +01:00
#[derive(Clone)]
pub enum NetworkIoMessage {
2016-01-10 22:42:27 +01:00
/// Register a new protocol handler.
2015-12-17 11:42:30 +01:00
AddHandler {
/// Handler shared instance.
handler: Arc<NetworkProtocolHandler + Sync>,
/// Protocol Id.
2015-12-17 11:42:30 +01:00
protocol: ProtocolId,
/// Supported protocol versions.
2015-12-17 11:42:30 +01:00
versions: Vec<u8>,
/// Number of packet IDs reserved by the protocol.
packet_count: u8,
2015-12-17 11:42:30 +01:00
},
/// Register a new protocol timer
2016-01-21 16:48:37 +01:00
AddTimer {
/// Protocol Id.
2016-01-21 16:48:37 +01:00
protocol: ProtocolId,
/// Timer token.
2016-01-21 16:48:37 +01:00
token: TimerToken,
/// Timer delay in milliseconds.
2016-01-21 16:48:37 +01:00
delay: u64,
},
2016-07-09 11:21:54 +02:00
/// Initliaze public interface.
InitPublicInterface,
/// Disconnect a peer.
2016-02-02 21:10:16 +01:00
Disconnect(PeerId),
/// Disconnect and temporary disable peer.
DisablePeer(PeerId),
/// Network has been started with the host as the given enode.
NetworkStarted(String),
2015-11-29 11:50:28 +01:00
}
2016-01-10 22:42:27 +01:00
/// Local (temporary) peer session ID.
2015-12-22 22:23:43 +01:00
pub type PeerId = usize;
2015-12-17 11:42:30 +01:00
2015-12-02 20:11:13 +01:00
#[derive(Debug, PartialEq, Eq)]
2016-01-10 22:42:27 +01:00
/// Protocol info
2015-12-02 20:11:13 +01:00
pub struct CapabilityInfo {
2015-12-17 11:42:30 +01:00
pub protocol: ProtocolId,
pub version: u8,
2016-01-10 22:42:27 +01:00
/// Total number of packet IDs this protocol support.
2015-12-17 11:42:30 +01:00
pub packet_count: u8,
2015-12-02 20:11:13 +01:00
}
impl Encodable for CapabilityInfo {
2016-01-27 12:14:57 +01:00
fn rlp_append(&self, s: &mut RlpStream) {
s.begin_list(2);
s.append(&&self.protocol[..]);
2016-01-27 12:14:57 +01:00
s.append(&self.version);
2015-12-02 20:11:13 +01:00
}
}
2016-01-10 22:42:27 +01:00
/// IO access point. This is passed to all IO handlers and provides an interface to the IO subsystem.
pub struct NetworkContext<'s> {
io: &'s IoContext<NetworkIoMessage>,
2016-01-13 13:56:48 +01:00
protocol: ProtocolId,
sessions: Arc<RwLock<Slab<SharedSession>>>,
2016-03-05 23:09:51 +01:00
session: Option<SharedSession>,
session_id: Option<StreamToken>,
_reserved_peers: &'s HashSet<NodeId>,
2015-12-17 11:42:30 +01:00
}
impl<'s> NetworkContext<'s> {
/// Create a new network IO access point. Takes references to all the data that can be updated within the IO handler.
fn new(io: &'s IoContext<NetworkIoMessage>,
protocol: ProtocolId,
session: Option<SharedSession>, sessions: Arc<RwLock<Slab<SharedSession>>>,
reserved_peers: &'s HashSet<NodeId>) -> NetworkContext<'s> {
let id = session.as_ref().map(|s| s.lock().token());
NetworkContext {
io: io,
2015-12-17 11:42:30 +01:00
protocol: protocol,
2016-03-05 23:09:51 +01:00
session_id: id,
2015-12-17 11:42:30 +01:00
session: session,
sessions: sessions,
_reserved_peers: reserved_peers,
2015-12-17 11:42:30 +01:00
}
}
2016-03-05 23:09:51 +01:00
fn resolve_session(&self, peer: PeerId) -> Option<SharedSession> {
match self.session_id {
Some(id) if id == peer => self.session.clone(),
_ => self.sessions.read().get(peer).cloned(),
2016-03-05 23:09:51 +01:00
}
}
2015-12-30 12:23:36 +01:00
/// Send a packet over the network to another peer.
pub fn send(&self, peer: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), NetworkError> {
self.send_protocol(self.protocol, peer, packet_id, data)
}
/// Send a packet over the network to another peer using specified protocol.
pub fn send_protocol(&self, protocol: ProtocolId, peer: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), NetworkError> {
2016-03-05 23:09:51 +01:00
let session = self.resolve_session(peer);
if let Some(session) = session {
session.lock().send_packet(self.io, protocol, packet_id as u8, &data)?;
2016-01-21 16:48:37 +01:00
} else {
2016-02-20 01:10:27 +01:00
trace!(target: "network", "Send: Peer no longer exist")
2015-12-22 22:23:43 +01:00
}
2015-12-17 11:42:30 +01:00
Ok(())
}
2016-03-05 23:09:51 +01:00
/// Respond to a current network message. Panics if no there is no packet in the context. If the session is expired returns nothing.
pub fn respond(&self, packet_id: PacketId, data: Vec<u8>) -> Result<(), NetworkError> {
2016-03-05 23:09:51 +01:00
assert!(self.session.is_some(), "Respond called without network context");
self.session_id.map_or_else(|| Err(NetworkError::Expired), |id| self.send(id, packet_id, data))
2015-12-17 11:42:30 +01:00
}
/// Get an IoChannel.
pub fn io_channel(&self) -> IoChannel<NetworkIoMessage> {
2016-06-17 12:58:28 +02:00
self.io.channel()
}
/// Disconnect a peer and prevent it from connecting again.
2016-02-02 14:54:46 +01:00
pub fn disable_peer(&self, peer: PeerId) {
self.io.message(NetworkIoMessage::DisablePeer(peer))
.unwrap_or_else(|e| warn!("Error sending network IO message: {:?}", e));
2016-02-02 14:54:46 +01:00
}
/// Disconnect peer. Reconnect can be attempted later.
pub fn disconnect_peer(&self, peer: PeerId) {
self.io.message(NetworkIoMessage::Disconnect(peer))
.unwrap_or_else(|e| warn!("Error sending network IO message: {:?}", e));
2015-12-02 20:11:13 +01:00
}
2015-12-28 11:41:51 +01:00
/// Check if the session is still active.
pub fn is_expired(&self) -> bool {
self.session.as_ref().map_or(false, |s| s.lock().expired())
}
2016-01-21 16:48:37 +01:00
/// Register a new IO timer. 'IoHandler::timeout' will be called with the token.
pub fn register_timer(&self, token: TimerToken, ms: u64) -> Result<(), NetworkError> {
2016-01-21 16:48:37 +01:00
self.io.message(NetworkIoMessage::AddTimer {
token: token,
delay: ms,
protocol: self.protocol,
}).unwrap_or_else(|e| warn!("Error sending network IO message: {:?}", e));
2016-01-21 16:48:37 +01:00
Ok(())
}
2016-01-14 16:52:10 +01:00
/// Returns peer identification string
pub fn peer_client_version(&self, peer: PeerId) -> String {
self.resolve_session(peer).map_or("unknown".to_owned(), |s| s.lock().info.client_version.clone())
}
/// Returns information on p2p session
pub fn session_info(&self, peer: PeerId) -> Option<SessionInfo> {
self.resolve_session(peer).map(|s| s.lock().info.clone())
2016-01-14 16:52:10 +01:00
}
/// Returns max version for a given protocol.
pub fn protocol_version(&self, protocol: ProtocolId, peer: PeerId) -> Option<u8> {
let session = self.resolve_session(peer);
session.and_then(|s| s.lock().capability_version(protocol))
}
/// Returns this object's subprotocol name.
pub fn subprotocol_name(&self) -> ProtocolId { self.protocol }
2015-12-17 11:42:30 +01:00
}
2016-01-10 22:42:27 +01:00
/// Shared host information
pub struct HostInfo {
2016-01-10 22:42:27 +01:00
/// Our private and public keys.
2016-01-08 13:49:00 +01:00
keys: KeyPair,
2016-01-10 22:42:27 +01:00
/// Current network configuration
2016-01-08 13:49:00 +01:00
config: NetworkConfiguration,
2016-01-10 22:42:27 +01:00
/// Connection nonce.
2015-12-02 20:11:13 +01:00
nonce: H256,
2016-01-10 22:42:27 +01:00
/// RLPx protocol version
2015-12-02 20:11:13 +01:00
pub protocol_version: u32,
2016-01-10 22:42:27 +01:00
/// Client identifier
2015-12-02 20:11:13 +01:00
pub client_version: String,
2016-01-10 22:42:27 +01:00
/// Registered capabilities (handlers)
2016-02-23 19:38:06 +01:00
pub capabilities: Vec<CapabilityInfo>,
2016-03-20 11:35:46 +01:00
/// Local address + discovery port
pub local_endpoint: NodeEndpoint,
2016-02-23 19:38:06 +01:00
/// Public address + discovery port
2016-03-20 11:35:46 +01:00
pub public_endpoint: Option<NodeEndpoint>,
}
impl HostInfo {
2016-01-10 22:42:27 +01:00
/// Returns public key
pub fn id(&self) -> &NodeId {
self.keys.public()
}
2016-01-10 22:42:27 +01:00
/// Returns secret key
pub fn secret(&self) -> &Secret {
self.keys.secret()
}
2016-01-10 22:42:27 +01:00
/// Increments and returns connection nonce.
pub fn next_nonce(&mut self) -> H256 {
self.nonce = self.nonce.sha3();
2016-01-19 12:14:29 +01:00
self.nonce.clone()
}
}
type SharedSession = Arc<Mutex<Session>>;
2016-01-21 16:48:37 +01:00
#[derive(Copy, Clone)]
struct ProtocolTimer {
pub protocol: ProtocolId,
pub token: TimerToken, // Handler level token
}
2016-01-10 22:42:27 +01:00
/// Root IO handler. Manages protocol handlers, IO timers and network connections.
pub struct Host {
2016-01-21 16:48:37 +01:00
pub info: RwLock<HostInfo>,
2016-03-20 11:35:46 +01:00
tcp_listener: Mutex<TcpListener>,
sessions: Arc<RwLock<Slab<SharedSession>>>,
2016-02-23 19:38:06 +01:00
discovery: Mutex<Option<Discovery>>,
2016-02-12 09:52:32 +01:00
nodes: RwLock<NodeTable>,
handlers: RwLock<HashMap<ProtocolId, Arc<NetworkProtocolHandler>>>,
2016-01-21 16:48:37 +01:00
timers: RwLock<HashMap<TimerToken, ProtocolTimer>>,
timer_counter: RwLock<usize>,
2016-01-24 18:53:54 +01:00
stats: Arc<NetworkStats>,
reserved_nodes: RwLock<HashSet<NodeId>>,
num_sessions: AtomicUsize,
2016-06-17 18:26:54 +02:00
stopping: AtomicBool,
2015-11-29 11:50:28 +01:00
}
impl Host {
2016-01-23 02:36:58 +01:00
/// Create a new instance
pub fn new(mut config: NetworkConfiguration, stats: Arc<NetworkStats>) -> Result<Host, NetworkError> {
2016-03-20 11:35:46 +01:00
let mut listen_address = match config.listen_address {
None => SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), DEFAULT_PORT)),
2016-02-16 02:05:36 +01:00
Some(addr) => addr,
};
let keys = if let Some(ref secret) = config.use_secret {
KeyPair::from_secret(secret.clone())?
} else {
config.config_path.clone().and_then(|ref p| load_key(Path::new(&p)))
2016-02-15 18:36:34 +01:00
.map_or_else(|| {
let key = Random.generate().expect("Error generating random key pair");
2016-02-15 18:36:34 +01:00
if let Some(path) = config.config_path.clone() {
save_key(Path::new(&path), key.secret());
2016-02-15 18:36:34 +01:00
}
key
},
|s| KeyPair::from_secret(s).expect("Error creating node secret key"))
};
let path = config.net_config_path.clone();
2016-03-20 11:35:46 +01:00
// Setup the server socket
let tcp_listener = TcpListener::bind(&listen_address)?;
listen_address = SocketAddr::new(listen_address.ip(), tcp_listener.local_addr()?.port());
2016-12-10 14:20:34 +01:00
debug!(target: "network", "Listening at {:?}", listen_address);
2016-03-20 11:35:46 +01:00
let udp_port = config.udp_port.unwrap_or(listen_address.port());
2016-02-23 19:38:06 +01:00
let local_endpoint = NodeEndpoint { address: listen_address, udp_port: udp_port };
2016-03-20 11:35:46 +01:00
let boot_nodes = config.boot_nodes.clone();
let reserved_nodes = config.reserved_nodes.clone();
config.max_handshakes = min(config.max_handshakes, MAX_HANDSHAKES as u32);
let mut host = Host {
2016-01-21 16:48:37 +01:00
info: RwLock::new(HostInfo {
2016-02-12 09:52:32 +01:00
keys: keys,
config: config,
2015-12-02 20:11:13 +01:00
nonce: H256::random(),
2016-02-13 22:57:39 +01:00
protocol_version: PROTOCOL_VERSION,
2016-02-21 20:00:45 +01:00
client_version: version(),
2015-12-17 11:42:30 +01:00
capabilities: Vec::new(),
2016-03-20 11:35:46 +01:00
public_endpoint: None,
local_endpoint: local_endpoint,
2016-01-21 16:48:37 +01:00
}),
2016-02-23 19:38:06 +01:00
discovery: Mutex::new(None),
2016-03-20 11:35:46 +01:00
tcp_listener: Mutex::new(tcp_listener),
sessions: Arc::new(RwLock::new(Slab::new_starting_at(FIRST_SESSION, MAX_SESSIONS))),
2016-02-12 09:52:32 +01:00
nodes: RwLock::new(NodeTable::new(path)),
2016-01-21 16:48:37 +01:00
handlers: RwLock::new(HashMap::new()),
timers: RwLock::new(HashMap::new()),
timer_counter: RwLock::new(USER_TIMER),
2016-06-17 12:58:28 +02:00
stats: stats,
reserved_nodes: RwLock::new(HashSet::new()),
num_sessions: AtomicUsize::new(0),
2016-06-17 18:26:54 +02:00
stopping: AtomicBool::new(false),
2016-01-21 16:48:37 +01:00
};
for n in boot_nodes {
host.add_node(&n);
}
for n in reserved_nodes {
if let Err(e) = host.add_reserved_node(&n) {
debug!(target: "network", "Error parsing node id: {}: {:?}", n, e);
}
2016-01-23 02:36:58 +01:00
}
2016-03-20 11:35:46 +01:00
Ok(host)
2016-01-08 13:49:00 +01:00
}
2015-11-29 11:50:28 +01:00
pub fn add_node(&mut self, id: &str) {
2015-11-29 11:50:28 +01:00
match Node::from_str(id) {
Err(e) => { debug!(target: "network", "Could not add node {}: {:?}", id, e); },
2015-12-02 12:07:46 +01:00
Ok(n) => {
2016-02-13 22:57:39 +01:00
let entry = NodeEntry { endpoint: n.endpoint.clone(), id: n.id.clone() };
self.nodes.write().add_node(n);
if let Some(ref mut discovery) = *self.discovery.lock() {
2016-07-20 12:41:31 +02:00
discovery.add_node(entry);
}
2015-12-02 12:07:46 +01:00
}
2015-11-29 11:50:28 +01:00
}
}
pub fn add_reserved_node(&self, id: &str) -> Result<(), NetworkError> {
let n = Node::from_str(id)?;
let entry = NodeEntry { endpoint: n.endpoint.clone(), id: n.id.clone() };
self.reserved_nodes.write().insert(n.id.clone());
self.nodes.write().add_node(Node::new(entry.id.clone(), entry.endpoint.clone()));
if let Some(ref mut discovery) = *self.discovery.lock() {
2016-07-20 12:41:31 +02:00
discovery.add_node(entry);
}
Ok(())
}
pub fn set_non_reserved_mode(&self, mode: NonReservedPeerMode, io: &IoContext<NetworkIoMessage>) {
let mut info = self.info.write();
if info.config.non_reserved_mode != mode {
info.config.non_reserved_mode = mode.clone();
drop(info);
if let NonReservedPeerMode::Deny = mode {
// disconnect all non-reserved peers here.
let reserved: HashSet<NodeId> = self.reserved_nodes.read().clone();
let mut to_kill = Vec::new();
for e in self.sessions.write().iter_mut() {
let mut s = e.lock();
{
let id = s.id();
if id.map_or(false, |id| reserved.contains(id)) {
continue;
}
}
s.disconnect(io, DisconnectReason::ClientQuit);
to_kill.push(s.token());
}
for p in to_kill {
trace!(target: "network", "Disconnecting on reserved-only mode: {}", p);
self.kill_connection(p, io, false);
}
}
}
}
pub fn remove_reserved_node(&self, id: &str) -> Result<(), NetworkError> {
let n = Node::from_str(id)?;
self.reserved_nodes.write().remove(&n.id);
Ok(())
}
2016-06-17 12:58:28 +02:00
pub fn client_version() -> String {
version()
2016-01-21 16:48:37 +01:00
}
2016-03-20 11:35:46 +01:00
pub fn external_url(&self) -> Option<String> {
2017-01-30 17:08:21 +01:00
let info = self.info.read();
info.public_endpoint.as_ref().map(|e| format!("{}", Node::new(info.id().clone(), e.clone())))
2016-02-23 19:38:06 +01:00
}
2016-03-20 11:35:46 +01:00
pub fn local_url(&self) -> String {
2017-01-30 17:08:21 +01:00
let info = self.info.read();
format!("{}", Node::new(info.id().clone(), info.local_endpoint.clone()))
2016-03-20 11:35:46 +01:00
}
pub fn stop(&self, io: &IoContext<NetworkIoMessage>) -> Result<(), NetworkError> {
2016-06-17 18:26:54 +02:00
self.stopping.store(true, AtomicOrdering::Release);
2016-06-17 12:58:28 +02:00
let mut to_kill = Vec::new();
for e in self.sessions.write().iter_mut() {
let mut s = e.lock();
2016-06-17 18:26:54 +02:00
s.disconnect(io, DisconnectReason::ClientQuit);
to_kill.push(s.token());
2016-06-17 12:58:28 +02:00
}
for p in to_kill {
2016-06-17 18:26:54 +02:00
trace!(target: "network", "Disconnecting on shutdown: {}", p);
2016-06-17 12:58:28 +02:00
self.kill_connection(p, io, true);
}
io.unregister_handler()?;
2016-06-17 12:58:28 +02:00
Ok(())
}
/// Get all connected peers.
pub fn connected_peers(&self) -> Vec<PeerId> {
let sessions = self.sessions.read();
let sessions = &*sessions;
let mut peers = Vec::with_capacity(sessions.count());
for i in (0..MAX_SESSIONS).map(|x| x + FIRST_SESSION) {
if sessions.get(i).is_some() {
peers.push(i);
}
}
peers
}
fn init_public_interface(&self, io: &IoContext<NetworkIoMessage>) -> Result<(), NetworkError> {
if self.info.read().public_endpoint.is_some() {
2016-03-20 11:35:46 +01:00
return Ok(());
2016-02-23 19:38:06 +01:00
}
let local_endpoint = self.info.read().local_endpoint.clone();
let public_address = self.info.read().config.public_address.clone();
let allow_ips = self.info.read().config.allow_ips;
2016-03-04 22:01:36 +01:00
let public_endpoint = match public_address {
2016-02-23 19:38:06 +01:00
None => {
2016-03-20 11:35:46 +01:00
let public_address = select_public_address(local_endpoint.address.port());
let public_endpoint = NodeEndpoint { address: public_address, udp_port: local_endpoint.udp_port };
if self.info.read().config.nat_enabled {
2016-02-23 19:38:06 +01:00
match map_external_address(&local_endpoint) {
Some(endpoint) => {
2016-03-20 11:35:46 +01:00
info!("NAT mapped to external address {}", endpoint.address);
2016-02-23 19:38:06 +01:00
endpoint
},
2016-03-20 11:35:46 +01:00
None => public_endpoint
2016-02-23 19:38:06 +01:00
}
} else {
2016-03-20 11:35:46 +01:00
public_endpoint
2016-02-23 19:38:06 +01:00
}
}
2016-03-20 11:35:46 +01:00
Some(addr) => NodeEndpoint { address: addr, udp_port: local_endpoint.udp_port }
2016-02-23 19:38:06 +01:00
};
2016-02-29 23:09:51 +01:00
self.info.write().public_endpoint = Some(public_endpoint.clone());
2016-06-29 17:50:27 +02:00
if let Some(url) = self.external_url() {
io.message(NetworkIoMessage::NetworkStarted(url)).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e));
2016-06-29 17:50:27 +02:00
}
2016-02-23 19:38:06 +01:00
// Initialize discovery.
let discovery = {
let info = self.info.read();
if info.config.discovery_enabled && info.config.non_reserved_mode == NonReservedPeerMode::Accept {
2016-07-09 11:21:54 +02:00
let mut udp_addr = local_endpoint.address.clone();
udp_addr.set_port(local_endpoint.udp_port);
Some(Discovery::new(&info.keys, udp_addr, public_endpoint, DISCOVERY, allow_ips))
2016-02-23 19:38:06 +01:00
} else { None }
};
if let Some(mut discovery) = discovery {
discovery.init_node_list(self.nodes.read().unordered_entries());
2016-07-20 12:41:31 +02:00
discovery.add_node_list(self.nodes.read().unordered_entries());
*self.discovery.lock() = Some(discovery);
io.register_stream(DISCOVERY)?;
io.register_timer(DISCOVERY_REFRESH, DISCOVERY_REFRESH_TIMEOUT)?;
io.register_timer(DISCOVERY_ROUND, DISCOVERY_ROUND_TIMEOUT)?;
2016-02-23 19:38:06 +01:00
}
io.register_timer(NODE_TABLE, NODE_TABLE_TIMEOUT)?;
io.register_stream(TCP_ACCEPT)?;
2016-03-20 11:35:46 +01:00
Ok(())
2016-01-21 16:48:37 +01:00
}
fn maintain_network(&self, io: &IoContext<NetworkIoMessage>) {
2016-02-02 20:58:12 +01:00
self.keep_alive(io);
self.connect_peers(io);
2016-01-08 13:49:00 +01:00
}
2015-11-29 11:50:28 +01:00
fn have_session(&self, id: &NodeId) -> bool {
self.sessions.read().iter().any(|e| e.lock().info.id == Some(id.clone()))
2016-02-14 01:03:48 +01:00
}
fn session_count(&self) -> usize {
self.num_sessions.load(AtomicOrdering::Relaxed)
2015-11-29 11:50:28 +01:00
}
fn connecting_to(&self, id: &NodeId) -> bool {
self.sessions.read().iter().any(|e| e.lock().id() == Some(id))
2016-02-14 01:03:48 +01:00
}
fn handshake_count(&self) -> usize {
// session_count < total_count is possible because of the data race.
self.sessions.read().count().saturating_sub(self.session_count())
2015-11-29 11:50:28 +01:00
}
fn keep_alive(&self, io: &IoContext<NetworkIoMessage>) {
2016-02-02 20:58:12 +01:00
let mut to_kill = Vec::new();
for e in self.sessions.write().iter_mut() {
let mut s = e.lock();
if !s.keep_alive(io) {
s.disconnect(io, DisconnectReason::PingTimeout);
to_kill.push(s.token());
2016-02-02 20:58:12 +01:00
}
}
for p in to_kill {
trace!(target: "network", "Ping timeout: {}", p);
2016-02-14 01:03:48 +01:00
self.kill_connection(p, io, true);
2016-02-02 20:58:12 +01:00
}
}
fn connect_peers(&self, io: &IoContext<NetworkIoMessage>) {
2016-10-31 12:54:50 +01:00
let (min_peers, mut pin, max_handshakes, allow_ips, self_id) = {
let info = self.info.read();
if info.capabilities.is_empty() {
return;
}
let config = &info.config;
2016-10-31 12:54:50 +01:00
(config.min_peers, config.non_reserved_mode == NonReservedPeerMode::Deny, config.max_handshakes as usize, config.allow_ips, info.id().clone())
};
let session_count = self.session_count();
let reserved_nodes = self.reserved_nodes.read();
2016-07-29 17:30:02 +02:00
if session_count >= min_peers as usize + reserved_nodes.len() {
// check if all pinned nodes are connected.
if reserved_nodes.iter().all(|n| self.have_session(n) && self.connecting_to(n)) {
return;
}
// if not, only attempt connect to reserved peers
pin = true;
2015-11-29 11:50:28 +01:00
}
2016-02-14 01:03:48 +01:00
let handshake_count = self.handshake_count();
2016-02-16 02:05:36 +01:00
// allow 16 slots for incoming connections
if handshake_count >= max_handshakes {
2016-02-14 01:03:48 +01:00
return;
2015-11-29 11:50:28 +01:00
}
// iterate over all nodes, reserved ones coming first.
// if we are pinned to only reserved nodes, ignore all others.
let nodes = reserved_nodes.iter().cloned().chain(if !pin {
self.nodes.read().nodes(allow_ips)
} else {
Vec::new()
});
let max_handshakes_per_round = max_handshakes / 2;
let mut started: usize = 0;
2016-10-31 12:54:50 +01:00
for id in nodes.filter(|id| !self.have_session(id) && !self.connecting_to(id) && *id != self_id)
.take(min(max_handshakes_per_round, max_handshakes - handshake_count)) {
2016-02-14 01:03:48 +01:00
self.connect_peer(&id, io);
started += 1;
2015-11-29 11:50:28 +01:00
}
debug!(target: "network", "Connecting peers: {} sessions, {} pending, {} started", self.session_count(), self.handshake_count(), started);
2016-01-08 13:49:00 +01:00
}
2015-11-29 11:50:28 +01:00
2016-03-11 11:16:49 +01:00
#[cfg_attr(feature="dev", allow(single_match))]
fn connect_peer(&self, id: &NodeId, io: &IoContext<NetworkIoMessage>) {
if self.have_session(id) {
2016-02-20 01:10:27 +01:00
trace!(target: "network", "Aborted connect. Node already connected.");
2015-11-29 11:50:28 +01:00
return;
}
2016-01-21 16:48:37 +01:00
if self.connecting_to(id) {
2016-02-20 01:10:27 +01:00
trace!(target: "network", "Aborted connect. Node already connecting.");
2015-11-29 11:50:28 +01:00
return;
}
let socket = {
2016-01-21 16:48:37 +01:00
let address = {
let mut nodes = self.nodes.write();
2016-02-14 17:42:03 +01:00
if let Some(node) = nodes.get_mut(id) {
node.last_attempted = Some(::time::now());
node.endpoint.address
}
2016-02-14 17:45:00 +01:00
else {
2016-02-20 01:10:27 +01:00
debug!(target: "network", "Connection to expired node aborted");
2016-02-14 17:45:00 +01:00
return;
}
2016-01-21 16:48:37 +01:00
};
match TcpStream::connect(&address) {
2016-12-10 14:20:34 +01:00
Ok(socket) => {
trace!(target: "network", "Connecting to {:?}", address);
socket
},
2016-02-16 17:54:34 +01:00
Err(e) => {
debug!(target: "network", "Can't connect to address {:?}: {:?}", address, e);
return;
}
2015-11-29 11:50:28 +01:00
}
};
if let Err(e) = self.create_connection(socket, Some(id), io) {
debug!(target: "network", "Can't create connection: {:?}", e);
}
2016-01-24 18:53:54 +01:00
}
2015-11-29 11:50:28 +01:00
2016-03-11 11:16:49 +01:00
#[cfg_attr(feature="dev", allow(block_in_if_condition_stmt))]
fn create_connection(&self, socket: TcpStream, id: Option<&NodeId>, io: &IoContext<NetworkIoMessage>) -> Result<(), NetworkError> {
let nonce = self.info.write().next_nonce();
let mut sessions = self.sessions.write();
let token = sessions.insert_with_opt(|token| {
match Session::new(io, socket, token, id, &nonce, self.stats.clone(), &self.info.read()) {
Ok(s) => Some(Arc::new(Mutex::new(s))),
Err(e) => {
debug!(target: "network", "Session create error: {:?}", e);
None
}
}
});
match token {
Some(t) => io.register_stream(t).map(|_| ()).map_err(Into::into),
None => {
debug!(target: "network", "Max sessions reached");
Ok(())
}
}
}
2015-11-29 11:50:28 +01:00
fn accept(&self, io: &IoContext<NetworkIoMessage>) {
2016-02-20 01:10:27 +01:00
trace!(target: "network", "Accepting incoming connection");
2016-01-24 18:53:54 +01:00
loop {
let socket = match self.tcp_listener.lock().accept() {
2016-10-30 09:56:34 +01:00
Ok((sock, _addr)) => sock,
2016-01-24 18:53:54 +01:00
Err(e) => {
2016-12-10 14:20:34 +01:00
if e.kind() != ErrorKind::WouldBlock {
debug!(target: "network", "Error accepting connection: {:?}", e);
}
2016-01-24 18:53:54 +01:00
break
},
};
if let Err(e) = self.create_connection(socket, None, io) {
debug!(target: "network", "Can't accept connection: {:?}", e);
2016-01-08 13:55:44 +01:00
}
}
2015-11-29 11:50:28 +01:00
}
2015-12-17 11:42:30 +01:00
fn session_writable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage>) {
let session = { self.sessions.read().get(token).cloned() };
if let Some(session) = session {
let mut s = session.lock();
if let Err(e) = s.writable(io, &self.info.read()) {
2016-02-20 01:10:27 +01:00
trace!(target: "network", "Session write error: {}: {:?}", token, e);
}
if s.done() {
io.deregister_stream(token).unwrap_or_else(|e| debug!("Error deregistering stream: {:?}", e));
2016-06-18 15:11:10 +02:00
}
}
}
fn connection_closed(&self, token: TimerToken, io: &IoContext<NetworkIoMessage>) {
trace!(target: "network", "Connection closed: {}", token);
2016-02-14 01:03:48 +01:00
self.kill_connection(token, io, true);
2016-01-10 14:02:01 +01:00
}
2015-12-17 11:42:30 +01:00
2016-06-23 12:04:54 +02:00
#[cfg_attr(feature="dev", allow(collapsible_if))]
fn session_readable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage>) {
2015-12-22 22:23:43 +01:00
let mut ready_data: Vec<ProtocolId> = Vec::new();
let mut packet_data: Vec<(ProtocolId, PacketId, Vec<u8>)> = Vec::new();
2016-01-21 16:48:37 +01:00
let mut kill = false;
let session = { self.sessions.read().get(token).cloned() };
let mut ready_id = None;
2016-03-05 23:09:51 +01:00
if let Some(session) = session.clone() {
{
let mut s = session.lock();
loop {
let session_result = s.readable(io, &self.info.read());
match session_result {
Err(e) => {
trace!(target: "network", "Session read error: {}:{:?} ({:?}) {:?}", token, s.id(), s.remote_addr(), e);
if let NetworkError::Disconnect(DisconnectReason::IncompatibleProtocol) = e {
if let Some(id) = s.id() {
if !self.reserved_nodes.read().contains(id) {
self.nodes.write().mark_as_useless(id);
}
}
}
kill = true;
break;
},
Ok(SessionData::Ready) => {
self.num_sessions.fetch_add(1, AtomicOrdering::SeqCst);
let session_count = self.session_count();
let (min_peers, max_peers, reserved_only) = {
let info = self.info.read();
let mut max_peers = info.config.max_peers;
for cap in s.info.capabilities.iter() {
if let Some(num) = info.config.reserved_protocols.get(&cap.protocol) {
max_peers += *num;
break;
}
}
(info.config.min_peers as usize, max_peers as usize, info.config.non_reserved_mode == NonReservedPeerMode::Deny)
};
let id = s.id().expect("Ready session always has id").clone();
// Check for the session limit. session_counts accounts for the new session.
if reserved_only ||
(s.info.originated && session_count > min_peers) ||
(!s.info.originated && session_count > max_peers) {
// only proceed if the connecting peer is reserved.
if !self.reserved_nodes.read().contains(&id) {
s.disconnect(io, DisconnectReason::TooManyPeers);
kill = true;
break;
}
}
ready_id = Some(id);
// Add it to the node table
if !s.info.originated {
if let Ok(address) = s.remote_addr() {
let entry = NodeEntry { id: id, endpoint: NodeEndpoint { address: address, udp_port: address.port() } };
self.nodes.write().add_node(Node::new(entry.id.clone(), entry.endpoint.clone()));
let mut discovery = self.discovery.lock();
if let Some(ref mut discovery) = *discovery {
discovery.add_node(entry);
}
}
}
for (p, _) in self.handlers.read().iter() {
if s.have_capability(*p) {
ready_data.push(*p);
}
}
},
Ok(SessionData::Packet {
data,
protocol,
packet_id,
}) => {
match self.handlers.read().get(&protocol) {
None => { warn!(target: "network", "No handler found for protocol: {:?}", protocol) },
Some(_) => packet_data.push((protocol, packet_id, data)),
}
},
Ok(SessionData::Continue) => (),
Ok(SessionData::None) => break,
}
}
}
if kill {
self.kill_connection(token, io, true);
}
let handlers = self.handlers.read();
if !ready_data.is_empty() {
let duplicate = self.sessions.read().iter().any(|e| {
let session = e.lock();
session.token() != token && session.info.id == ready_id
});
if duplicate {
trace!(target: "network", "Rejected duplicate connection: {}", token);
session.lock().disconnect(io, DisconnectReason::DuplicatePeer);
self.kill_connection(token, io, false);
return;
}
for p in ready_data {
self.stats.inc_sessions();
let reserved = self.reserved_nodes.read();
if let Some(h) = handlers.get(&p).clone() {
h.connected(&NetworkContext::new(io, p, Some(session.clone()), self.sessions.clone(), &reserved), &token);
// accumulate pending packets.
let mut session = session.lock();
packet_data.extend(session.mark_connected(p));
}
}
}
for (p, packet_id, data) in packet_data {
let reserved = self.reserved_nodes.read();
if let Some(h) = handlers.get(&p).clone() {
h.read(&NetworkContext::new(io, p, Some(session.clone()), self.sessions.clone(), &reserved), &token, packet_id, &data[1..]);
}
}
2016-01-10 14:02:01 +01:00
}
2015-11-29 11:50:28 +01:00
}
fn connection_timeout(&self, token: StreamToken, io: &IoContext<NetworkIoMessage>) {
trace!(target: "network", "Connection timeout: {}", token);
2016-02-14 01:03:48 +01:00
self.kill_connection(token, io, true)
2015-12-02 12:07:46 +01:00
}
2016-01-08 13:55:44 +01:00
fn kill_connection(&self, token: StreamToken, io: &IoContext<NetworkIoMessage>, remote: bool) {
2016-01-10 14:02:01 +01:00
let mut to_disconnect: Vec<ProtocolId> = Vec::new();
2016-02-14 02:11:55 +01:00
let mut failure_id = None;
let mut deregister = false;
2016-03-05 23:09:51 +01:00
let mut expired_session = None;
2016-06-03 11:36:30 +02:00
if let FIRST_SESSION ... LAST_SESSION = token {
let sessions = self.sessions.write();
2016-06-03 11:36:30 +02:00
if let Some(session) = sessions.get(token).cloned() {
expired_session = Some(session.clone());
let mut s = session.lock();
2016-06-03 11:36:30 +02:00
if !s.expired() {
if s.is_ready() {
self.num_sessions.fetch_sub(1, AtomicOrdering::SeqCst);
for (p, _) in self.handlers.read().iter() {
if s.have_capability(*p) {
to_disconnect.push(*p);
2016-01-21 16:48:37 +01:00
}
}
}
2016-06-03 11:36:30 +02:00
s.set_expired();
failure_id = s.id().cloned();
2016-01-10 14:02:01 +01:00
}
2016-06-03 11:36:30 +02:00
deregister = remote || s.done();
}
2016-01-10 14:02:01 +01:00
}
2016-02-14 02:11:55 +01:00
if let Some(id) = failure_id {
if remote {
self.nodes.write().note_failure(&id);
2016-01-21 16:48:37 +01:00
}
2016-01-10 14:02:01 +01:00
}
for p in to_disconnect {
let reserved = self.reserved_nodes.read();
if let Some(h) = self.handlers.read().get(&p).clone() {
h.disconnected(&NetworkContext::new(io, p, expired_session.clone(), self.sessions.clone(), &reserved), &token);
}
2016-01-15 00:50:48 +01:00
}
if deregister {
io.deregister_stream(token).unwrap_or_else(|e| debug!("Error deregistering stream: {:?}", e));
2016-06-18 15:11:10 +02:00
}
2015-11-29 11:50:28 +01:00
}
2016-02-12 09:52:32 +01:00
fn update_nodes(&self, _io: &IoContext<NetworkIoMessage>, node_changes: TableUpdates) {
2016-02-12 09:52:32 +01:00
let mut to_remove: Vec<PeerId> = Vec::new();
2016-02-14 02:11:55 +01:00
{
let sessions = self.sessions.write();
for c in sessions.iter() {
let s = c.lock();
if let Some(id) = s.id() {
if node_changes.removed.contains(id) {
to_remove.push(s.token());
2016-02-12 09:52:32 +01:00
}
}
}
2016-01-15 00:50:48 +01:00
}
2016-02-12 09:52:32 +01:00
for i in to_remove {
trace!(target: "network", "Removed from node table: {}", i);
2016-02-12 09:52:32 +01:00
}
self.nodes.write().update(node_changes, &*self.reserved_nodes.read());
2015-11-29 11:50:28 +01:00
}
2017-02-02 17:02:46 +01:00
pub fn with_context<F>(&self, protocol: ProtocolId, io: &IoContext<NetworkIoMessage>, action: F) where F: FnOnce(&NetworkContext) {
let reserved = { self.reserved_nodes.read() };
let context = NetworkContext::new(io, protocol, None, self.sessions.clone(), &reserved);
action(&context);
}
2017-02-02 17:02:46 +01:00
pub fn with_context_eval<F, T>(&self, protocol: ProtocolId, io: &IoContext<NetworkIoMessage>, action: F) -> T where F: FnOnce(&NetworkContext) -> T {
let reserved = { self.reserved_nodes.read() };
let context = NetworkContext::new(io, protocol, None, self.sessions.clone(), &reserved);
action(&context)
}
2016-01-08 13:55:44 +01:00
}
2015-11-29 11:50:28 +01:00
impl IoHandler<NetworkIoMessage> for Host {
/// Initialize networking
fn initialize(&self, io: &IoContext<NetworkIoMessage>) {
2016-01-21 16:48:37 +01:00
io.register_timer(IDLE, MAINTENANCE_TIMEOUT).expect("Error registering Network idle timer");
2016-07-09 11:21:54 +02:00
io.message(NetworkIoMessage::InitPublicInterface).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e));
2016-02-19 16:34:31 +01:00
self.maintain_network(io)
}
fn stream_hup(&self, io: &IoContext<NetworkIoMessage>, stream: StreamToken) {
2016-02-20 01:10:27 +01:00
trace!(target: "network", "Hup: {}", stream);
match stream {
FIRST_SESSION ... LAST_SESSION => self.connection_closed(stream, io),
2016-02-20 01:10:27 +01:00
_ => warn!(target: "network", "Unexpected hup"),
};
}
fn stream_readable(&self, io: &IoContext<NetworkIoMessage>, stream: StreamToken) {
2016-06-17 18:26:54 +02:00
if self.stopping.load(AtomicOrdering::Acquire) {
return;
}
match stream {
FIRST_SESSION ... LAST_SESSION => self.session_readable(stream, io),
2016-02-12 09:52:32 +01:00
DISCOVERY => {
let node_changes = { self.discovery.lock().as_mut().map_or(None, |d| d.readable(io)) };
2016-02-21 13:58:01 +01:00
if let Some(node_changes) = node_changes {
2016-02-12 09:52:32 +01:00
self.update_nodes(io, node_changes);
}
},
TCP_ACCEPT => self.accept(io),
_ => panic!("Received unknown readable token"),
2016-01-08 13:55:44 +01:00
}
}
fn stream_writable(&self, io: &IoContext<NetworkIoMessage>, stream: StreamToken) {
2016-06-17 18:26:54 +02:00
if self.stopping.load(AtomicOrdering::Acquire) {
return;
}
match stream {
FIRST_SESSION ... LAST_SESSION => self.session_writable(stream, io),
2016-02-13 22:57:39 +01:00
DISCOVERY => {
self.discovery.lock().as_mut().map(|d| d.writable(io));
2016-02-13 22:57:39 +01:00
}
_ => panic!("Received unknown writable token"),
2015-11-29 11:50:28 +01:00
}
2016-01-08 13:55:44 +01:00
}
2016-01-08 13:49:00 +01:00
fn timeout(&self, io: &IoContext<NetworkIoMessage>, token: TimerToken) {
2016-06-17 18:26:54 +02:00
if self.stopping.load(AtomicOrdering::Acquire) {
return;
}
match token {
IDLE => self.maintain_network(io),
FIRST_SESSION ... LAST_SESSION => self.connection_timeout(token, io),
2016-02-12 09:52:32 +01:00
DISCOVERY_REFRESH => {
self.discovery.lock().as_mut().map(|d| d.refresh());
io.update_registration(DISCOVERY).unwrap_or_else(|e| debug!("Error updating discovery registration: {:?}", e));
2016-02-12 09:52:32 +01:00
},
DISCOVERY_ROUND => {
let node_changes = { self.discovery.lock().as_mut().map_or(None, |d| d.round()) };
2016-02-21 13:58:01 +01:00
if let Some(node_changes) = node_changes {
2016-02-13 22:57:39 +01:00
self.update_nodes(io, node_changes);
}
io.update_registration(DISCOVERY).unwrap_or_else(|e| debug!("Error updating discovery registration: {:?}", e));
2016-02-12 09:52:32 +01:00
},
NODE_TABLE => {
trace!(target: "network", "Refreshing node table");
self.nodes.write().clear_useless();
self.nodes.write().save();
},
_ => match self.timers.read().get(&token).cloned() {
Some(timer) => match self.handlers.read().get(&timer.protocol).cloned() {
2016-06-17 12:58:28 +02:00
None => { warn!(target: "network", "No handler found for protocol: {:?}", timer.protocol) },
Some(h) => {
let reserved = self.reserved_nodes.read();
h.timeout(&NetworkContext::new(io, timer.protocol, None, self.sessions.clone(), &reserved), timer.token);
}
2016-01-21 16:48:37 +01:00
},
None => { warn!("Unknown timer token: {}", token); } // timer is not registerd through us
2015-12-17 11:42:30 +01:00
}
}
2016-01-08 13:55:44 +01:00
}
2015-12-17 11:42:30 +01:00
fn message(&self, io: &IoContext<NetworkIoMessage>, message: &NetworkIoMessage) {
2016-06-17 18:26:54 +02:00
if self.stopping.load(AtomicOrdering::Acquire) {
return;
}
2016-01-19 12:14:29 +01:00
match *message {
NetworkIoMessage::AddHandler {
2016-01-21 16:48:37 +01:00
ref handler,
ref protocol,
ref versions,
ref packet_count,
2016-01-08 13:55:44 +01:00
} => {
2016-01-21 16:48:37 +01:00
let h = handler.clone();
let reserved = self.reserved_nodes.read();
h.initialize(&NetworkContext::new(io, *protocol, None, self.sessions.clone(), &reserved));
self.handlers.write().insert(*protocol, h);
let mut info = self.info.write();
2016-01-08 13:55:44 +01:00
for v in versions {
info.capabilities.push(CapabilityInfo { protocol: *protocol, version: *v, packet_count: *packet_count });
2016-01-08 13:55:44 +01:00
}
},
NetworkIoMessage::AddTimer {
2016-01-21 16:48:37 +01:00
ref protocol,
ref delay,
ref token,
} => {
let handler_token = {
let mut timer_counter = self.timer_counter.write();
let counter = &mut *timer_counter;
2016-01-21 16:48:37 +01:00
let handler_token = *counter;
*counter += 1;
handler_token
};
self.timers.write().insert(handler_token, ProtocolTimer { protocol: *protocol, token: *token });
io.register_timer(handler_token, *delay).unwrap_or_else(|e| debug!("Error registering timer {}: {:?}", token, e));
2016-01-21 16:48:37 +01:00
},
2016-02-02 21:10:16 +01:00
NetworkIoMessage::Disconnect(ref peer) => {
let session = { self.sessions.read().get(*peer).cloned() };
if let Some(session) = session {
session.lock().disconnect(io, DisconnectReason::DisconnectRequested);
}
trace!(target: "network", "Disconnect requested {}", peer);
self.kill_connection(*peer, io, false);
},
NetworkIoMessage::DisablePeer(ref peer) => {
let session = { self.sessions.read().get(*peer).cloned() };
if let Some(session) = session {
session.lock().disconnect(io, DisconnectReason::DisconnectRequested);
if let Some(id) = session.lock().id() {
self.nodes.write().mark_as_useless(id)
}
}
trace!(target: "network", "Disabling peer {}", peer);
2016-02-14 01:03:48 +01:00
self.kill_connection(*peer, io, false);
2016-02-02 14:54:46 +01:00
},
2016-07-09 11:21:54 +02:00
NetworkIoMessage::InitPublicInterface =>
self.init_public_interface(io).unwrap_or_else(|e| warn!("Error initializing public interface: {:?}", e)),
_ => {} // ignore others.
2015-11-29 11:50:28 +01:00
}
}
2016-01-21 16:48:37 +01:00
fn register_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage>>) {
2016-01-21 16:48:37 +01:00
match stream {
FIRST_SESSION ... LAST_SESSION => {
let session = { self.sessions.read().get(stream).cloned() };
if let Some(session) = session {
session.lock().register_socket(reg, event_loop).expect("Error registering socket");
}
2016-01-21 16:48:37 +01:00
}
DISCOVERY => self.discovery.lock().as_ref().and_then(|d| d.register_socket(event_loop).ok()).expect("Error registering discovery socket"),
2016-10-30 09:56:34 +01:00
TCP_ACCEPT => event_loop.register(&*self.tcp_listener.lock(), Token(TCP_ACCEPT), Ready::all(), PollOpt::edge()).expect("Error registering stream"),
2016-01-21 23:33:52 +01:00
_ => warn!("Unexpected stream registration")
2016-01-21 16:48:37 +01:00
}
}
fn deregister_stream(&self, stream: StreamToken, event_loop: &mut EventLoop<IoManager<NetworkIoMessage>>) {
2016-01-22 18:13:59 +01:00
match stream {
FIRST_SESSION ... LAST_SESSION => {
let mut connections = self.sessions.write();
2016-01-22 18:13:59 +01:00
if let Some(connection) = connections.get(stream).cloned() {
let c = connection.lock();
if c.expired() { // make sure it is the same connection that the event was generated for
c.deregister_socket(event_loop).expect("Error deregistering socket");
connections.remove(stream);
}
}
}
2016-02-12 09:52:32 +01:00
DISCOVERY => (),
2016-01-22 18:13:59 +01:00
_ => warn!("Unexpected stream deregistration")
}
}
fn update_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage>>) {
2016-01-21 16:48:37 +01:00
match stream {
FIRST_SESSION ... LAST_SESSION => {
let connection = { self.sessions.read().get(stream).cloned() };
if let Some(connection) = connection {
connection.lock().update_socket(reg, event_loop).expect("Error updating socket");
}
}
DISCOVERY => self.discovery.lock().as_ref().and_then(|d| d.update_registration(event_loop).ok()).expect("Error reregistering discovery socket"),
2016-10-30 09:56:34 +01:00
TCP_ACCEPT => event_loop.reregister(&*self.tcp_listener.lock(), Token(TCP_ACCEPT), Ready::all(), PollOpt::edge()).expect("Error reregistering stream"),
2016-01-21 16:48:37 +01:00
_ => warn!("Unexpected stream update")
}
}
2016-01-08 13:55:44 +01:00
}
2016-02-15 18:36:34 +01:00
fn save_key(path: &Path, key: &Secret) {
let mut path_buf = PathBuf::from(path);
if let Err(e) = fs::create_dir_all(path_buf.as_path()) {
warn!("Error creating key directory: {:?}", e);
return;
};
path_buf.push("key");
let path = path_buf.as_path();
let mut file = match fs::File::create(&path) {
2016-02-15 18:36:34 +01:00
Ok(file) => file,
Err(e) => {
warn!("Error creating key file: {:?}", e);
return;
}
};
if let Err(e) = restrict_permissions_owner(path, true, false) {
warn!(target: "network", "Failed to modify permissions of the file ({})", e);
}
2016-02-15 18:36:34 +01:00
if let Err(e) = file.write(&key.hex().into_bytes()) {
warn!("Error writing key file: {:?}", e);
}
}
fn load_key(path: &Path) -> Option<Secret> {
let mut path_buf = PathBuf::from(path);
path_buf.push("key");
let mut file = match fs::File::open(path_buf.as_path()) {
Ok(file) => file,
Err(e) => {
debug!("Error opening key file: {:?}", e);
return None;
}
};
let mut buf = String::new();
match file.read_to_string(&mut buf) {
Ok(_) => {},
Err(e) => {
2016-02-15 18:36:34 +01:00
warn!("Error reading key file: {:?}", e);
return None;
}
}
match Secret::from_str(&buf) {
Ok(key) => Some(key),
Err(e) => {
2016-02-15 18:36:34 +01:00
warn!("Error parsing key file: {:?}", e);
None
}
}
}
2016-02-16 21:25:01 +01:00
#[test]
fn key_save_load() {
use ::devtools::RandomTempPath;
2016-02-16 21:25:01 +01:00
let temp_path = RandomTempPath::create_dir();
let key = Secret::from_slice(&H256::random()).unwrap();
2016-02-16 21:25:01 +01:00
save_key(temp_path.as_path(), &key);
let r = load_key(temp_path.as_path());
assert_eq!(key, r.unwrap());
}
#[test]
fn host_client_url() {
let mut config = NetworkConfiguration::new_local();
let key = "6f7b0d801bc7b5ce7bbd930b84fd0369b3eb25d09be58d64ba811091046f3aa2".parse().unwrap();
config.use_secret = Some(key);
let host: Host = Host::new(config, Arc::new(NetworkStats::new())).unwrap();
2016-03-20 11:35:46 +01:00
assert!(host.local_url().starts_with("enode://101b3ef5a4ea7a1c7928e24c4c75fd053c235d7b80c22ae5c03d145d0ac7396e2a4ffff9adee3133a7b05044a5cee08115fd65145e5165d646bde371010d803c@"));
}