+//! 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 ln::msgs::{MsgEncodable,MsgDecodable};
+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>,
}
/// 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>,
}
-macro_rules! encode_msg {
- ($msg: expr, $msg_code: expr) => {
- {
- let just_msg = $msg.encode();
- let mut encoded_msg = Vec::with_capacity(just_msg.len() + 2);
- encoded_msg.extend_from_slice(&byte_utils::be16_to_array($msg_code));
- encoded_msg.extend_from_slice(&just_msg[..]);
- encoded_msg
- }
+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 msg = VecWriter(Vec::new());
+ ($msg_code as u16).write(&mut msg).unwrap();
+ $msg.write(&mut msg).unwrap();
+ msg.0
+ }}
}
//TODO: Really should do something smarter for this
/// 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,
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 }),
- msgs::DecodeError::InvalidLength => return Err(PeerHandleError{ no_connection_possible: false }),
}
}
};
// Need an init message as first message
return Err(PeerHandleError{ no_connection_possible: false });
}
+ let mut reader = ::std::io::Cursor::new(&msg_data[2..]);
match msg_type {
// Connection control:
16 => {
- let msg = try_potential_decodeerror!(msgs::Init::decode(&msg_data[2..]));
+ 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::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::ErrorMessage::read(&mut reader));
let mut data_is_printable = true;
for b in msg.data.bytes() {
if b < 32 || b > 126 {
},
18 => {
- let msg = try_potential_decodeerror!(msgs::Ping::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::Ping::read(&mut reader));
if msg.ponglen < 65532 {
let resp = msgs::Pong { byteslen: msg.ponglen };
encode_and_send_msg!(resp, 19);
}
},
19 => {
- try_potential_decodeerror!(msgs::Pong::decode(&msg_data[2..]));
+ try_potential_decodeerror!(msgs::Pong::read(&mut reader));
},
// Channel control:
32 => {
- let msg = try_potential_decodeerror!(msgs::OpenChannel::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::OpenChannel::read(&mut reader));
let resp = try_potential_handleerror!(self.message_handler.chan_handler.handle_open_channel(&peer.their_node_id.unwrap(), &msg));
encode_and_send_msg!(resp, 33);
},
33 => {
- let msg = try_potential_decodeerror!(msgs::AcceptChannel::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::AcceptChannel::read(&mut reader));
try_potential_handleerror!(self.message_handler.chan_handler.handle_accept_channel(&peer.their_node_id.unwrap(), &msg));
},
34 => {
- let msg = try_potential_decodeerror!(msgs::FundingCreated::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::FundingCreated::read(&mut reader));
let resp = try_potential_handleerror!(self.message_handler.chan_handler.handle_funding_created(&peer.their_node_id.unwrap(), &msg));
encode_and_send_msg!(resp, 35);
},
35 => {
- let msg = try_potential_decodeerror!(msgs::FundingSigned::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::FundingSigned::read(&mut reader));
try_potential_handleerror!(self.message_handler.chan_handler.handle_funding_signed(&peer.their_node_id.unwrap(), &msg));
},
36 => {
- let msg = try_potential_decodeerror!(msgs::FundingLocked::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::FundingLocked::read(&mut reader));
let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_funding_locked(&peer.their_node_id.unwrap(), &msg));
match resp_option {
Some(resp) => encode_and_send_msg!(resp, 259),
},
38 => {
- let msg = try_potential_decodeerror!(msgs::Shutdown::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::Shutdown::read(&mut reader));
let resp_options = try_potential_handleerror!(self.message_handler.chan_handler.handle_shutdown(&peer.their_node_id.unwrap(), &msg));
if let Some(resp) = resp_options.0 {
encode_and_send_msg!(resp, 38);
}
},
39 => {
- let msg = try_potential_decodeerror!(msgs::ClosingSigned::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::ClosingSigned::read(&mut reader));
let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_closing_signed(&peer.their_node_id.unwrap(), &msg));
if let Some(resp) = resp_option {
encode_and_send_msg!(resp, 39);
},
128 => {
- let msg = try_potential_decodeerror!(msgs::UpdateAddHTLC::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::UpdateAddHTLC::read(&mut reader));
try_potential_handleerror!(self.message_handler.chan_handler.handle_update_add_htlc(&peer.their_node_id.unwrap(), &msg));
},
130 => {
- let msg = try_potential_decodeerror!(msgs::UpdateFulfillHTLC::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::UpdateFulfillHTLC::read(&mut reader));
try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fulfill_htlc(&peer.their_node_id.unwrap(), &msg));
},
131 => {
- let msg = try_potential_decodeerror!(msgs::UpdateFailHTLC::decode(&msg_data[2..]));
+ 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);
}
},
135 => {
- let msg = try_potential_decodeerror!(msgs::UpdateFailMalformedHTLC::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::UpdateFailMalformedHTLC::read(&mut reader));
try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_malformed_htlc(&peer.their_node_id.unwrap(), &msg));
},
132 => {
- let msg = try_potential_decodeerror!(msgs::CommitmentSigned::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::CommitmentSigned::read(&mut reader));
let resps = try_potential_handleerror!(self.message_handler.chan_handler.handle_commitment_signed(&peer.their_node_id.unwrap(), &msg));
encode_and_send_msg!(resps.0, 133);
if let Some(resp) = resps.1 {
}
},
133 => {
- let msg = try_potential_decodeerror!(msgs::RevokeAndACK::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::RevokeAndACK::read(&mut reader));
let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_revoke_and_ack(&peer.their_node_id.unwrap(), &msg));
match resp_option {
Some(resps) => {
}
},
134 => {
- let msg = try_potential_decodeerror!(msgs::UpdateFee::decode(&msg_data[2..]));
+ 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 => {
- let msg = try_potential_decodeerror!(msgs::AnnouncementSignatures::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::AnnouncementSignatures::read(&mut reader));
try_potential_handleerror!(self.message_handler.chan_handler.handle_announcement_signatures(&peer.their_node_id.unwrap(), &msg));
},
256 => {
- let msg = try_potential_decodeerror!(msgs::ChannelAnnouncement::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::ChannelAnnouncement::read(&mut reader));
let should_forward = try_potential_handleerror!(self.message_handler.route_handler.handle_channel_announcement(&msg));
if should_forward {
}
},
257 => {
- let msg = try_potential_decodeerror!(msgs::NodeAnnouncement::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::NodeAnnouncement::read(&mut reader));
let should_forward = try_potential_handleerror!(self.message_handler.route_handler.handle_node_announcement(&msg));
if should_forward {
}
},
258 => {
- let msg = try_potential_decodeerror!(msgs::ChannelUpdate::decode(&msg_data[2..]));
+ let msg = try_potential_decodeerror!(msgs::ChannelUpdate::read(&mut reader));
let should_forward = try_potential_handleerror!(self.message_handler.route_handler.handle_channel_update(&msg));
if should_forward {
projects / rust-lightning / blobdiff