openethereum/util/src/network/session.rs

288 lines
8.5 KiB
Rust
Raw Normal View History

2015-12-02 12:07:46 +01:00
use mio::*;
use hash::*;
2015-12-02 20:11:13 +01:00
use rlp::*;
use network::connection::{EncryptedConnection, Packet};
2015-12-02 12:07:46 +01:00
use network::handshake::Handshake;
use error::*;
2016-01-21 16:48:37 +01:00
use io::{IoContext};
use network::error::{NetworkError, DisconnectReason};
2015-12-02 12:07:46 +01:00
use network::host::*;
use network::node::NodeId;
2015-12-02 12:07:46 +01:00
2016-01-10 22:42:27 +01:00
/// Peer session over encrypted connection.
/// When created waits for Hello packet exchange and signals ready state.
/// Sends and receives protocol packets and handles basic packes such as ping/pong and disconnect.
2015-12-02 12:07:46 +01:00
pub struct Session {
2016-01-10 22:42:27 +01:00
/// Shared session information
2015-12-02 20:11:13 +01:00
pub info: SessionInfo,
2016-01-10 22:42:27 +01:00
/// Underlying connection
2015-12-02 12:07:46 +01:00
connection: EncryptedConnection,
2016-01-10 22:42:27 +01:00
/// Session ready flag. Set after successfull Hello packet exchange
2015-12-02 20:11:13 +01:00
had_hello: bool,
}
2016-01-10 22:42:27 +01:00
/// Structure used to report various session events.
2015-12-17 11:42:30 +01:00
pub enum SessionData {
None,
2016-01-10 22:42:27 +01:00
/// Session is ready to send/receive packets.
2015-12-17 11:42:30 +01:00
Ready,
2016-01-10 22:42:27 +01:00
/// A packet has been received
2015-12-17 11:42:30 +01:00
Packet {
2016-01-10 22:42:27 +01:00
/// Packet data
2015-12-17 11:42:30 +01:00
data: Vec<u8>,
2016-01-10 22:42:27 +01:00
/// Packet protocol ID
2015-12-17 11:42:30 +01:00
protocol: &'static str,
2016-01-10 22:42:27 +01:00
/// Zero based packet ID
2015-12-17 11:42:30 +01:00
packet_id: u8,
},
}
2016-01-10 22:42:27 +01:00
/// Shared session information
2015-12-02 20:11:13 +01:00
pub struct SessionInfo {
2016-01-10 22:42:27 +01:00
/// Peer public key
2015-12-02 20:11:13 +01:00
pub id: NodeId,
2016-01-10 22:42:27 +01:00
/// Peer client ID
2015-12-02 20:11:13 +01:00
pub client_version: String,
2016-01-10 22:42:27 +01:00
/// Peer RLPx protocol version
2015-12-02 20:11:13 +01:00
pub protocol_version: u32,
2016-01-10 22:42:27 +01:00
/// Peer protocol capabilities
capabilities: Vec<SessionCapabilityInfo>,
2015-12-17 11:42:30 +01:00
}
#[derive(Debug, PartialEq, Eq)]
pub struct PeerCapabilityInfo {
pub protocol: String,
pub version: u8,
}
impl Decodable for PeerCapabilityInfo {
2015-12-17 14:05:13 +01:00
fn decode<D>(decoder: &D) -> Result<Self, DecoderError> where D: Decoder {
let c = try!(decoder.as_list());
let v: u32 = try!(Decodable::decode(&c[1]));
2015-12-17 11:42:30 +01:00
Ok(PeerCapabilityInfo {
2015-12-17 14:05:13 +01:00
protocol: try!(Decodable::decode(&c[0])),
version: v as u8,
2015-12-17 11:42:30 +01:00
})
}
}
#[derive(Debug, PartialEq, Eq)]
2016-01-10 22:42:27 +01:00
struct SessionCapabilityInfo {
2015-12-17 11:42:30 +01:00
pub protocol: &'static str,
pub version: u8,
pub packet_count: u8,
pub id_offset: u8,
2015-12-02 12:07:46 +01:00
}
2015-12-02 20:11:13 +01:00
const PACKET_HELLO: u8 = 0x80;
const PACKET_DISCONNECT: u8 = 0x01;
const PACKET_PING: u8 = 0x02;
const PACKET_PONG: u8 = 0x03;
const PACKET_GET_PEERS: u8 = 0x04;
const PACKET_PEERS: u8 = 0x05;
const PACKET_USER: u8 = 0x10;
const PACKET_LAST: u8 = 0x7f;
impl Session {
2016-01-10 22:42:27 +01:00
/// Create a new session out of comepleted handshake. Consumes handshake object.
2016-01-21 16:48:37 +01:00
pub fn new<Message>(h: Handshake, _io: &IoContext<Message>, host: &HostInfo) -> Result<Session, UtilError> where Message: Send + Sync + Clone {
2015-12-02 12:07:46 +01:00
let id = h.id.clone();
let connection = try!(EncryptedConnection::new(h));
2015-12-02 20:11:13 +01:00
let mut session = Session {
2015-12-02 12:07:46 +01:00
connection: connection,
2015-12-02 20:11:13 +01:00
had_hello: false,
info: SessionInfo {
id: id,
client_version: String::new(),
protocol_version: 0,
capabilities: Vec::new(),
},
};
try!(session.write_hello(host));
try!(session.write_ping());
Ok(session)
2015-12-02 12:07:46 +01:00
}
2015-12-02 20:11:13 +01:00
2016-01-10 22:42:27 +01:00
/// Check if session is ready to send/receive data
2016-01-10 14:02:01 +01:00
pub fn is_ready(&self) -> bool {
self.had_hello
}
2016-01-10 22:42:27 +01:00
/// Readable IO handler. Returns packet data if available.
2016-01-21 16:48:37 +01:00
pub fn readable<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo) -> Result<SessionData, UtilError> where Message: Send + Sync + Clone {
match try!(self.connection.readable(io)) {
Some(data) => Ok(try!(self.read_packet(data, host))),
2015-12-17 11:42:30 +01:00
None => Ok(SessionData::None)
}
2015-12-02 12:07:46 +01:00
}
2015-12-02 20:11:13 +01:00
2016-01-10 22:42:27 +01:00
/// Writable IO handler. Sends pending packets.
2016-01-21 16:48:37 +01:00
pub fn writable<Message>(&mut self, io: &IoContext<Message>, _host: &HostInfo) -> Result<(), UtilError> where Message: Send + Sync + Clone {
self.connection.writable(io)
2015-12-02 12:07:46 +01:00
}
2015-12-02 20:11:13 +01:00
2016-01-10 22:42:27 +01:00
/// Checks if peer supports given capability
2015-12-17 11:42:30 +01:00
pub fn have_capability(&self, protocol: &str) -> bool {
self.info.capabilities.iter().any(|c| c.protocol == protocol)
}
2016-01-10 22:42:27 +01:00
/// Update registration with the event loop. Should be called at the end of the IO handler.
2016-01-21 16:48:37 +01:00
pub fn update_socket<Host:Handler>(&self, reg:Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
self.connection.update_socket(reg, event_loop)
2016-01-10 14:02:01 +01:00
}
2016-01-22 18:13:59 +01:00
/// Delete registration
pub fn deregister_socket<Host:Handler>(&self, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
self.connection.deregister_socket(event_loop)
}
2016-01-10 22:42:27 +01:00
/// Send a protocol packet to peer.
pub fn send_packet(&mut self, protocol: &str, packet_id: u8, data: &[u8]) -> Result<(), UtilError> {
2015-12-17 11:42:30 +01:00
let mut i = 0usize;
while protocol != self.info.capabilities[i].protocol {
i += 1;
if i == self.info.capabilities.len() {
debug!(target: "net", "Unkown protocol: {:?}", protocol);
return Ok(())
}
}
let pid = self.info.capabilities[i].id_offset + packet_id;
let mut rlp = RlpStream::new();
rlp.append(&(pid as u32));
rlp.append_raw(data, 1);
self.connection.send_packet(&rlp.out())
}
fn read_packet(&mut self, packet: Packet, host: &HostInfo) -> Result<SessionData, UtilError> {
2015-12-17 11:42:30 +01:00
if packet.data.len() < 2 {
return Err(From::from(NetworkError::BadProtocol));
2015-12-02 20:11:13 +01:00
}
2015-12-17 11:42:30 +01:00
let packet_id = packet.data[0];
2015-12-02 20:11:13 +01:00
if packet_id != PACKET_HELLO && packet_id != PACKET_DISCONNECT && !self.had_hello {
return Err(From::from(NetworkError::BadProtocol));
2015-12-02 20:11:13 +01:00
}
match packet_id {
2015-12-17 11:42:30 +01:00
PACKET_HELLO => {
2016-01-08 15:52:43 +01:00
let rlp = UntrustedRlp::new(&packet.data[1..]); //TODO: validate rlp expected size
2015-12-17 11:42:30 +01:00
try!(self.read_hello(&rlp, host));
Ok(SessionData::Ready)
2016-01-08 15:52:43 +01:00
},
PACKET_DISCONNECT => Err(From::from(NetworkError::Disconnect(DisconnectReason::DisconnectRequested))),
2015-12-17 11:42:30 +01:00
PACKET_PING => {
try!(self.write_pong());
Ok(SessionData::None)
2016-01-08 15:52:43 +01:00
},
2015-12-17 11:42:30 +01:00
PACKET_GET_PEERS => Ok(SessionData::None), //TODO;
PACKET_PEERS => Ok(SessionData::None),
2015-12-02 20:11:13 +01:00
PACKET_USER ... PACKET_LAST => {
2015-12-17 11:42:30 +01:00
let mut i = 0usize;
while packet_id < self.info.capabilities[i].id_offset {
i += 1;
if i == self.info.capabilities.len() {
debug!(target: "net", "Unkown packet: {:?}", packet_id);
return Ok(SessionData::None)
}
}
// map to protocol
let protocol = self.info.capabilities[i].protocol;
let pid = packet_id - self.info.capabilities[i].id_offset;
2016-01-19 12:14:29 +01:00
Ok(SessionData::Packet { data: packet.data, protocol: protocol, packet_id: pid } )
2015-12-02 20:11:13 +01:00
},
_ => {
2015-12-17 11:42:30 +01:00
debug!(target: "net", "Unkown packet: {:?}", packet_id);
Ok(SessionData::None)
2015-12-02 20:11:13 +01:00
}
}
}
fn write_hello(&mut self, host: &HostInfo) -> Result<(), UtilError> {
2015-12-02 20:11:13 +01:00
let mut rlp = RlpStream::new();
2016-01-10 14:02:01 +01:00
rlp.append_raw(&[PACKET_HELLO as u8], 0);
2016-01-27 12:14:57 +01:00
rlp.begin_list(5)
2015-12-02 20:11:13 +01:00
.append(&host.protocol_version)
.append(&host.client_version)
2016-01-27 12:14:57 +01:00
.append_list(&host.capabilities)
2015-12-02 20:11:13 +01:00
.append(&host.listen_port)
.append(host.id());
self.connection.send_packet(&rlp.out())
}
fn read_hello(&mut self, rlp: &UntrustedRlp, host: &HostInfo) -> Result<(), UtilError> {
2015-12-17 14:05:13 +01:00
let protocol = try!(rlp.val_at::<u32>(0));
let client_version = try!(rlp.val_at::<String>(1));
let peer_caps = try!(rlp.val_at::<Vec<PeerCapabilityInfo>>(2));
let id = try!(rlp.val_at::<NodeId>(4));
2015-12-02 20:11:13 +01:00
// Intersect with host capabilities
// Leave only highset mutually supported capability version
2015-12-17 11:42:30 +01:00
let mut caps: Vec<SessionCapabilityInfo> = Vec::new();
2016-01-19 12:14:29 +01:00
for hc in &host.capabilities {
2015-12-17 11:42:30 +01:00
if peer_caps.iter().any(|c| c.protocol == hc.protocol && c.version == hc.version) {
caps.push(SessionCapabilityInfo {
protocol: hc.protocol,
version: hc.version,
id_offset: 0,
packet_count: hc.packet_count,
});
}
}
caps.retain(|c| host.capabilities.iter().any(|hc| hc.protocol == c.protocol && hc.version == c.version));
2015-12-02 20:11:13 +01:00
let mut i = 0;
while i < caps.len() {
if caps.iter().any(|c| c.protocol == caps[i].protocol && c.version > caps[i].version) {
caps.remove(i);
}
else {
i += 1;
}
}
2015-12-17 11:42:30 +01:00
i = 0;
let mut offset: u8 = PACKET_USER;
while i < caps.len() {
caps[i].id_offset = offset;
offset += caps[i].packet_count;
i += 1;
}
2015-12-02 20:11:13 +01:00
trace!(target: "net", "Hello: {} v{} {} {:?}", client_version, protocol, id, caps);
2016-01-15 00:50:48 +01:00
self.info.client_version = client_version;
2015-12-17 11:42:30 +01:00
self.info.capabilities = caps;
2015-12-02 20:11:13 +01:00
if protocol != host.protocol_version {
return Err(From::from(self.disconnect(DisconnectReason::UselessPeer)));
2015-12-02 20:11:13 +01:00
}
self.had_hello = true;
Ok(())
}
fn write_ping(&mut self) -> Result<(), UtilError> {
2016-01-10 14:02:01 +01:00
self.send(try!(Session::prepare(PACKET_PING)))
2015-12-02 20:11:13 +01:00
}
fn write_pong(&mut self) -> Result<(), UtilError> {
2016-01-10 14:02:01 +01:00
self.send(try!(Session::prepare(PACKET_PONG)))
2015-12-02 20:11:13 +01:00
}
fn disconnect(&mut self, reason: DisconnectReason) -> NetworkError {
2015-12-02 20:11:13 +01:00
let mut rlp = RlpStream::new();
rlp.append(&(PACKET_DISCONNECT as u32));
2016-01-27 12:14:57 +01:00
rlp.begin_list(1);
2015-12-02 20:11:13 +01:00
rlp.append(&(reason.clone() as u32));
self.connection.send_packet(&rlp.out()).ok();
NetworkError::Disconnect(reason)
2015-12-02 20:11:13 +01:00
}
fn prepare(packet_id: u8) -> Result<RlpStream, UtilError> {
2016-01-10 14:02:01 +01:00
let mut rlp = RlpStream::new();
2015-12-02 20:11:13 +01:00
rlp.append(&(packet_id as u32));
2016-01-27 12:14:57 +01:00
rlp.begin_list(0);
2015-12-02 20:11:13 +01:00
Ok(rlp)
}
fn send(&mut self, rlp: RlpStream) -> Result<(), UtilError> {
2015-12-02 20:11:13 +01:00
self.connection.send_packet(&rlp.out())
}
2015-12-02 12:07:46 +01:00
}