+//! 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};
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<msgs::ChannelMessageHandler>,
+ /// 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<msgs::RoutingMessageHandler>,
}
/// Provides an object which can be used to send data to and which uniquely identifies a connection
/// to a remote host. You will need to be able to generate multiple of these which meet Eq and
/// implement Hash to meet the PeerManager API.
+///
/// For efficiency, Clone should be relatively cheap for this type.
+///
/// You probably want to just extend an int and put a file descriptor in a struct and implement
/// send_data. Note that if you are using a higher-level net library that may close() itself, be
/// careful to ensure you don't have races whereby you might register a new connection with an fd
/// Returns the amount of data which was sent, possibly 0 if the socket has since disconnected.
/// Note that in the disconnected case, a disconnect_event must still fire and further write
/// attempts may occur until that time.
+ ///
/// If the returned size is smaller than data.len() - write_offset, a write_available event must
/// trigger the next time more data can be written. Additionally, until the a send_data event
/// completes fully, no further read_events should trigger on the same peer!
+ ///
/// If a read_event on this descriptor had previously returned true (indicating that read
/// events should be paused to prevent DoS in the send buffer), resume_read may be set
/// indicating that read events on this descriptor should resume. A resume_read of false does
/// 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 {
}
}
+/// 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<Descriptor: SocketDescriptor> {
message_handler: MessageHandler,
peers: Mutex<PeerHolder<Descriptor>>,
logger: Arc<Logger>,
}
+struct VecWriter(Vec<u8>);
+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
}}
}
/// 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<Descriptor: SocketDescriptor> PeerManager<Descriptor> {
+ /// 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<Logger>) -> PeerManager<Descriptor> {
PeerManager {
message_handler: message_handler,
}
/// Get the list of node ids for peers which have completed the initial handshake.
+ ///
/// For outbound connections, this will be the same as the their_node_id parameter passed in to
/// new_outbound_connection, however entries will only appear once the initial handshake has
/// completed and we are sure the remote peer has the private key for the given node_id.
/// Indicates a new outbound connection has been established to a node with the given node_id.
/// Note that if an Err is returned here you MUST NOT call disconnect_event for the new
/// descriptor but must disconnect the connection immediately.
+ ///
/// Returns some bytes to send to the remote node.
+ ///
/// Panics if descriptor is duplicative with some other descriptor which has not yet has a
/// disconnect_event.
pub fn new_outbound_connection(&self, their_node_id: PublicKey, descriptor: Descriptor) -> Result<Vec<u8>, PeerHandleError> {
}
/// Indicates a new inbound connection has been established.
+ ///
/// May refuse the connection by returning an Err, but will never write bytes to the remote end
/// (outbound connector always speaks first). Note that if an Err is returned here you MUST NOT
/// call disconnect_event for the new descriptor but must disconnect the connection
/// immediately.
+ ///
/// Panics if descriptor is duplicative with some other descriptor which has not yet has a
/// disconnect_event.
pub fn new_inbound_connection(&self, descriptor: Descriptor) -> Result<(), PeerHandleError> {
}
/// Indicates that there is room to write data to the given socket descriptor.
+ ///
/// May return an Err to indicate that the connection should be closed.
+ ///
/// Will most likely call send_data on the descriptor passed in (or the descriptor handed into
- /// new_*_connection) before returning. Thus, be very careful with reentrancy issues! The
+ /// new_*\_connection) before returning. Thus, be very careful with reentrancy issues! The
/// invariants around calling write_event in case a write did not fully complete must still
/// hold - be ready to call write_event again if a write call generated here isn't sufficient!
- /// Panics if the descriptor was not previously registered in a new_*_connection event.
+ /// Panics if the descriptor was not previously registered in a new_\*_connection event.
pub fn write_event(&self, descriptor: &mut Descriptor) -> Result<(), PeerHandleError> {
let mut peers = self.peers.lock().unwrap();
match peers.peers.get_mut(descriptor) {
}
/// Indicates that data was read from the given socket descriptor.
+ ///
/// May return an Err to indicate that the connection should be closed.
+ ///
/// Will very likely call send_data on the descriptor passed in (or a descriptor handed into
/// new_*_connection) before returning. Thus, be very careful with reentrancy issues! The
/// invariants around calling write_event in case a write did not fully complete must still
/// hold. Note that this function will often call send_data on many peers before returning, not
/// just this peer!
+ ///
/// If Ok(true) is returned, further read_events should not be triggered until a write_event on
/// this file descriptor has resume_read set (preventing DoS issues in the send buffer). Note
/// that this must be true even if a send_data call with resume_read=true was made during the
/// course of this function!
+ ///
/// Panics if the descriptor was not previously registered in a new_*_connection event.
pub fn read_event(&self, peer_descriptor: &mut Descriptor, data: Vec<u8>) -> Result<bool, PeerHandleError> {
match self.do_read_event(peer_descriptor, data) {
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");
},
msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError{ no_connection_possible: false }),
msgs::DecodeError::Io(_) => return Err(PeerHandleError{ no_connection_possible: false }),
- msgs::DecodeError::InvalidValue => return Err(PeerHandleError{ no_connection_possible: false }),
}
}
};
// 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);
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));
},
131 => {
let msg = try_potential_decodeerror!(msgs::UpdateFailHTLC::read(&mut reader));
- let chan_update = try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg));
- if let Some(update) = chan_update {
- self.message_handler.route_handler.handle_htlc_fail_channel_update(&update);
- }
+ try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg));
},
135 => {
let msg = try_potential_decodeerror!(msgs::UpdateFailMalformedHTLC::read(&mut reader));
for resp in resps.update_fail_htlcs {
encode_and_send_msg!(resp, 131);
}
+ if let Some(resp) = resps.update_fee {
+ encode_and_send_msg!(resp, 134);
+ }
encode_and_send_msg!(resps.commitment_signed, 132);
},
None => {},
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, order) = 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);
+ }
+ macro_rules! handle_raa { () => {
+ if let Some(revoke_msg) = revoke_and_ack {
+ encode_and_send_msg!(revoke_msg, 133);
+ }
+ } }
+ macro_rules! handle_cu { () => {
+ 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);
+ }
+ if let Some(resp) = resps.update_fee {
+ encode_and_send_msg!(resp, 134);
+ }
+ encode_and_send_msg!(resps.commitment_signed, 132);
+ },
+ None => {},
+ }
+ } }
+ match order {
+ msgs::RAACommitmentOrder::RevokeAndACKFirst => {
+ handle_raa!();
+ handle_cu!();
+ },
+ msgs::RAACommitmentOrder::CommitmentFirst => {
+ handle_cu!();
+ handle_raa!();
+ },
+ }
+ },
// Routing control:
259 => {
Event::PaymentSent {..} => { /* Hand upstream */ },
Event::PaymentFailed {..} => { /* Hand upstream */ },
Event::PendingHTLCsForwardable {..} => { /* Hand upstream */ },
+ Event::SpendableOutputs { .. } => { /* Hand upstream */ },
Event::SendOpenChannel { ref node_id, ref msg } => {
log_trace!(self, "Handling SendOpenChannel event in peer_handler for node {} for channel {}",
Self::do_attempt_write_data(&mut descriptor, peer);
continue;
},
- Event::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref commitment_signed } } => {
+ Event::UpdateHTLCs { ref node_id, updates: msgs::CommitmentUpdate { ref update_add_htlcs, ref update_fulfill_htlcs, ref update_fail_htlcs, ref update_fail_malformed_htlcs, ref update_fee, ref commitment_signed } } => {
log_trace!(self, "Handling UpdateHTLCs event in peer_handler for node {} with {} adds, {} fulfills, {} fails for channel {}",
log_pubkey!(node_id),
update_add_htlcs.len(),
for msg in update_fail_malformed_htlcs {
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 135)));
}
+ if let &Some(ref msg) = update_fee {
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 134)));
+ }
peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(commitment_signed, 132)));
Self::do_attempt_write_data(&mut descriptor, peer);
continue;
},
+ Event::SendRevokeAndACK { ref node_id, ref msg } => {
+ log_trace!(self, "Handling SendRevokeAndACK event in peer_handler for node {} for channel {}",
+ log_pubkey!(node_id),
+ log_bytes!(msg.channel_id));
+ let (mut descriptor, peer) = get_peer_for_forwarding!(node_id, {
+ //TODO: Do whatever we're gonna do for handling dropped messages
+ });
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 133)));
+ Self::do_attempt_write_data(&mut descriptor, peer);
+ continue;
+ },
Event::SendShutdown { ref node_id, ref msg } => {
log_trace!(self, "Handling Shutdown event in peer_handler for node {} for channel {}",
log_pubkey!(node_id),
}
continue;
},
+ Event::PaymentFailureNetworkUpdate { ref update } => {
+ self.message_handler.route_handler.handle_htlc_fail_channel_update(update);
+ continue;
+ },
Event::HandleError { ref node_id, ref action } => {
if let Some(ref action) = *action {
match *action {
}
/// Indicates that the given socket descriptor's connection is now closed.
+ ///
/// This must be called even if a PeerHandleError was given for a read_event or write_event,
- /// but must NOT be called if a PeerHandleError was provided out of a new_*_connection event!
+ /// but must NOT be called if a PeerHandleError was provided out of a new_\*\_connection event!
+ ///
/// Panics if the descriptor was not previously registered in a successful new_*_connection event.
pub fn disconnect_event(&self, descriptor: &Descriptor) {
self.disconnect_event_internal(descriptor, false);