X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fpeer_handler.rs;h=797c55191f680a336a71fd58409d1e9c832f93d9;hb=c43e535bc03404bad16e3d30e5b2fc7215e6ca15;hp=4e710117bb805037a80457086a434de70a0bb29b;hpb=28a612f9f3c6a855c914aaf8cb10ef14772c7b90;p=rust-lightning diff --git a/src/ln/peer_handler.rs b/src/ln/peer_handler.rs index 4e710117..797c5519 100644 --- a/src/ln/peer_handler.rs +++ b/src/ln/peer_handler.rs @@ -1,7 +1,14 @@ +//! Top level peer message handling and socket handling logic lives here. +//! Instead of actually servicing sockets ourselves we require that you implement the +//! SocketDescriptor interface and use that to receive actions which you should perform on the +//! socket, and call into PeerManager with bytes read from the socket. The PeerManager will then +//! call into the provided message handlers (probably a ChannelManager and Router) with messages +//! they should handle, and encoding/sending response messages. + use secp256k1::key::{SecretKey,PublicKey}; use ln::msgs; -use util::ser::{Writer, Reader, Writeable, Readable}; +use util::ser::{Writeable, Writer, Readable}; use ln::peer_channel_encryptor::{PeerChannelEncryptor,NextNoiseStep}; use util::byte_utils; use util::events::{EventsProvider,Event}; @@ -12,8 +19,13 @@ use std::sync::{Arc, Mutex}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::{cmp,error,mem,hash,fmt}; +/// Provides references to trait impls which handle different types of messages. pub struct MessageHandler { + /// A message handler which handles messages specific to channels. Usually this is just a + /// ChannelManager object. pub chan_handler: Arc, + /// A message handler which handles messages updating our knowledge of the network channel + /// graph. Usually this is just a Router object. pub route_handler: Arc, } @@ -51,6 +63,8 @@ pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone { /// disconnect_event (unless it was provided in response to a new_*_connection event, in which case /// no such disconnect_event must be generated and the socket be silently disconencted). pub struct PeerHandleError { + /// Used to indicate that we probably can't make any future connections to this peer, implying + /// we should go ahead and force-close any channels we have with it. no_connection_possible: bool, } impl fmt::Debug for PeerHandleError { @@ -103,6 +117,8 @@ impl PeerHolder { } } +/// A PeerManager manages a set of peers, described by their SocketDescriptor and marshalls socket +/// events into messages which it passes on to its MessageHandlers. pub struct PeerManager { message_handler: MessageHandler, peers: Mutex>, @@ -112,15 +128,23 @@ pub struct PeerManager { logger: Arc, } +struct VecWriter(Vec); +impl Writer for VecWriter { + fn write_all(&mut self, buf: &[u8]) -> Result<(), ::std::io::Error> { + self.0.extend_from_slice(buf); + Ok(()) + } + fn size_hint(&mut self, size: usize) { + self.0.reserve_exact(size); + } +} + macro_rules! encode_msg { ($msg: expr, $msg_code: expr) => {{ - let mut w = Writer::new(::std::io::Cursor::new(vec![])); - 0u16.write(&mut w).unwrap(); - $msg.write(&mut w).unwrap(); - let mut msg = w.into_inner().into_inner(); - let len = msg.len(); - msg[..2].copy_from_slice(&byte_utils::be16_to_array(len as u16 - 2)); - msg + let mut msg = VecWriter(Vec::new()); + ($msg_code as u16).write(&mut msg).unwrap(); + $msg.write(&mut msg).unwrap(); + msg.0 }} } @@ -130,6 +154,7 @@ const INITIAL_SYNCS_TO_SEND: usize = 5; /// Manages and reacts to connection events. You probably want to use file descriptors as PeerIds. /// PeerIds may repeat, but only after disconnect_event() has been called. impl PeerManager { + /// Constructs a new PeerManager with the given message handlers and node_id secret key pub fn new(message_handler: MessageHandler, our_node_secret: SecretKey, logger: Arc) -> PeerManager { PeerManager { message_handler: message_handler, @@ -349,14 +374,12 @@ impl PeerManager { Ok(x) => x, Err(e) => { match e { - msgs::DecodeError::UnknownRealmByte => return Err(PeerHandleError{ no_connection_possible: false }), + msgs::DecodeError::UnknownVersion => return Err(PeerHandleError{ no_connection_possible: false }), msgs::DecodeError::UnknownRequiredFeature => { log_debug!(self, "Got a channel/node announcement with an known required feature flag, you may want to udpate!"); continue; }, - msgs::DecodeError::BadPublicKey => return Err(PeerHandleError{ no_connection_possible: false }), - msgs::DecodeError::BadSignature => return Err(PeerHandleError{ no_connection_possible: false }), - msgs::DecodeError::BadText => return Err(PeerHandleError{ no_connection_possible: false }), + msgs::DecodeError::InvalidValue => return Err(PeerHandleError{ no_connection_possible: false }), msgs::DecodeError::ShortRead => return Err(PeerHandleError{ no_connection_possible: false }), msgs::DecodeError::ExtraAddressesPerType => { log_debug!(self, "Error decoding message, ignoring due to lnd spec incompatibility. See https://github.com/lightningnetwork/lnd/issues/1407"); @@ -364,7 +387,6 @@ impl PeerManager { }, msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError{ no_connection_possible: false }), msgs::DecodeError::Io(_) => return Err(PeerHandleError{ no_connection_possible: false }), - msgs::DecodeError::InvalidValue => panic!("should not happen with message decoding"), } } }; @@ -437,20 +459,38 @@ impl PeerManager { // Need an init message as first message return Err(PeerHandleError{ no_connection_possible: false }); } - let mut reader = Reader::new(::std::io::Cursor::new(&msg_data[2..])); + let mut reader = ::std::io::Cursor::new(&msg_data[2..]); match msg_type { // Connection control: 16 => { let msg = try_potential_decodeerror!(msgs::Init::read(&mut reader)); if msg.global_features.requires_unknown_bits() { + log_info!(self, "Peer global features required unknown version bits"); return Err(PeerHandleError{ no_connection_possible: true }); } if msg.local_features.requires_unknown_bits() { + log_info!(self, "Peer local features required unknown version bits"); + return Err(PeerHandleError{ no_connection_possible: true }); + } + if msg.local_features.requires_data_loss_protect() { + log_info!(self, "Peer local features required data_loss_protect"); + return Err(PeerHandleError{ no_connection_possible: true }); + } + if msg.local_features.requires_upfront_shutdown_script() { + log_info!(self, "Peer local features required upfront_shutdown_script"); return Err(PeerHandleError{ no_connection_possible: true }); } if peer.their_global_features.is_some() { return Err(PeerHandleError{ no_connection_possible: false }); } + + log_info!(self, "Received peer Init message: data_loss_protect: {}, initial_routing_sync: {}, upfront_shutdown_script: {}, unkown local flags: {}, unknown global flags: {}", + if msg.local_features.supports_data_loss_protect() { "supported" } else { "not supported"}, + if msg.local_features.initial_routing_sync() { "requested" } else { "not requested" }, + if msg.local_features.supports_upfront_shutdown_script() { "supported" } else { "not supported"}, + if msg.local_features.supports_unknown_bits() { "present" } else { "none" }, + if msg.global_features.supports_unknown_bits() { "present" } else { "none" }); + peer.their_global_features = Some(msg.global_features); peer.their_local_features = Some(msg.local_features); @@ -465,6 +505,10 @@ impl PeerManager { local_features, }, 16); } + + for msg in self.message_handler.chan_handler.peer_connected(&peer.their_node_id.unwrap()) { + encode_and_send_msg!(msg, 136); + } }, 17 => { let msg = try_potential_decodeerror!(msgs::ErrorMessage::read(&mut reader)); @@ -596,7 +640,31 @@ impl PeerManager { let msg = try_potential_decodeerror!(msgs::UpdateFee::read(&mut reader)); try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fee(&peer.their_node_id.unwrap(), &msg)); }, - 136 => { }, // TODO: channel_reestablish + 136 => { + let msg = try_potential_decodeerror!(msgs::ChannelReestablish::read(&mut reader)); + let (funding_locked, revoke_and_ack, commitment_update) = try_potential_handleerror!(self.message_handler.chan_handler.handle_channel_reestablish(&peer.their_node_id.unwrap(), &msg)); + if let Some(lock_msg) = funding_locked { + encode_and_send_msg!(lock_msg, 36); + } + if let Some(revoke_msg) = revoke_and_ack { + encode_and_send_msg!(revoke_msg, 133); + } + match commitment_update { + Some(resps) => { + for resp in resps.update_add_htlcs { + encode_and_send_msg!(resp, 128); + } + for resp in resps.update_fulfill_htlcs { + encode_and_send_msg!(resp, 130); + } + for resp in resps.update_fail_htlcs { + encode_and_send_msg!(resp, 131); + } + encode_and_send_msg!(resps.commitment_signed, 132); + }, + None => {}, + } + }, // Routing control: 259 => {