+ /// Process an incoming message and return a decision (ok, lightning error, peer handling error) regarding the next action with the peer
+ /// Returns the message back if it needs to be broadcasted to all other peers.
+ fn handle_message(
+ &self,
+ peer: &mut Peer,
+ message: wire::Message<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>
+ ) -> Result<Option<wire::Message<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>>, MessageHandlingError> {
+ log_trace!(self.logger, "Received message {:?} from {}", message, log_pubkey!(peer.their_node_id.unwrap()));
+
+ // Need an Init as first message
+ if let wire::Message::Init(_) = message {
+ } else if peer.their_features.is_none() {
+ log_debug!(self.logger, "Peer {} sent non-Init first message", log_pubkey!(peer.their_node_id.unwrap()));
+ return Err(PeerHandleError{ no_connection_possible: false }.into());
+ }
+
+ let mut should_forward = None;
+
+ match message {
+ // Setup and Control messages:
+ wire::Message::Init(msg) => {
+ if msg.features.requires_unknown_bits() {
+ log_debug!(self.logger, "Peer features required unknown version bits");
+ return Err(PeerHandleError{ no_connection_possible: true }.into());
+ }
+ if peer.their_features.is_some() {
+ return Err(PeerHandleError{ no_connection_possible: false }.into());
+ }
+
+ log_info!(self.logger, "Received peer Init message: {}", msg.features);
+
+ if msg.features.initial_routing_sync() {
+ peer.sync_status = InitSyncTracker::ChannelsSyncing(0);
+ }
+ if !msg.features.supports_static_remote_key() {
+ log_debug!(self.logger, "Peer {} does not support static remote key, disconnecting with no_connection_possible", log_pubkey!(peer.their_node_id.unwrap()));
+ return Err(PeerHandleError{ no_connection_possible: true }.into());
+ }
+
+ self.message_handler.route_handler.sync_routing_table(&peer.their_node_id.unwrap(), &msg);
+
+ self.message_handler.chan_handler.peer_connected(&peer.their_node_id.unwrap(), &msg);
+ peer.their_features = Some(msg.features);
+ },
+ wire::Message::Error(msg) => {
+ let mut data_is_printable = true;
+ for b in msg.data.bytes() {
+ if b < 32 || b > 126 {
+ data_is_printable = false;
+ break;
+ }
+ }
+
+ if data_is_printable {
+ log_debug!(self.logger, "Got Err message from {}: {}", log_pubkey!(peer.their_node_id.unwrap()), msg.data);
+ } else {
+ log_debug!(self.logger, "Got Err message from {} with non-ASCII error message", log_pubkey!(peer.their_node_id.unwrap()));
+ }
+ self.message_handler.chan_handler.handle_error(&peer.their_node_id.unwrap(), &msg);
+ if msg.channel_id == [0; 32] {
+ return Err(PeerHandleError{ no_connection_possible: true }.into());
+ }
+ },
+
+ wire::Message::Ping(msg) => {
+ if msg.ponglen < 65532 {
+ let resp = msgs::Pong { byteslen: msg.ponglen };
+ self.enqueue_message(peer, &resp);
+ }
+ },
+ wire::Message::Pong(_msg) => {
+ peer.awaiting_pong = false;
+ },
+
+ // Channel messages:
+ wire::Message::OpenChannel(msg) => {
+ self.message_handler.chan_handler.handle_open_channel(&peer.their_node_id.unwrap(), peer.their_features.clone().unwrap(), &msg);
+ },
+ wire::Message::AcceptChannel(msg) => {
+ self.message_handler.chan_handler.handle_accept_channel(&peer.their_node_id.unwrap(), peer.their_features.clone().unwrap(), &msg);
+ },
+
+ wire::Message::FundingCreated(msg) => {
+ self.message_handler.chan_handler.handle_funding_created(&peer.their_node_id.unwrap(), &msg);
+ },
+ wire::Message::FundingSigned(msg) => {
+ self.message_handler.chan_handler.handle_funding_signed(&peer.their_node_id.unwrap(), &msg);
+ },
+ wire::Message::FundingLocked(msg) => {
+ self.message_handler.chan_handler.handle_funding_locked(&peer.their_node_id.unwrap(), &msg);
+ },
+
+ wire::Message::Shutdown(msg) => {
+ self.message_handler.chan_handler.handle_shutdown(&peer.their_node_id.unwrap(), peer.their_features.as_ref().unwrap(), &msg);
+ },
+ wire::Message::ClosingSigned(msg) => {
+ self.message_handler.chan_handler.handle_closing_signed(&peer.their_node_id.unwrap(), &msg);
+ },
+
+ // Commitment messages:
+ wire::Message::UpdateAddHTLC(msg) => {
+ self.message_handler.chan_handler.handle_update_add_htlc(&peer.their_node_id.unwrap(), &msg);
+ },
+ wire::Message::UpdateFulfillHTLC(msg) => {
+ self.message_handler.chan_handler.handle_update_fulfill_htlc(&peer.their_node_id.unwrap(), &msg);
+ },
+ wire::Message::UpdateFailHTLC(msg) => {
+ self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg);
+ },
+ wire::Message::UpdateFailMalformedHTLC(msg) => {
+ self.message_handler.chan_handler.handle_update_fail_malformed_htlc(&peer.their_node_id.unwrap(), &msg);
+ },
+
+ wire::Message::CommitmentSigned(msg) => {
+ self.message_handler.chan_handler.handle_commitment_signed(&peer.their_node_id.unwrap(), &msg);
+ },
+ wire::Message::RevokeAndACK(msg) => {
+ self.message_handler.chan_handler.handle_revoke_and_ack(&peer.their_node_id.unwrap(), &msg);
+ },
+ wire::Message::UpdateFee(msg) => {
+ self.message_handler.chan_handler.handle_update_fee(&peer.their_node_id.unwrap(), &msg);
+ },
+ wire::Message::ChannelReestablish(msg) => {
+ self.message_handler.chan_handler.handle_channel_reestablish(&peer.their_node_id.unwrap(), &msg);
+ },
+
+ // Routing messages:
+ wire::Message::AnnouncementSignatures(msg) => {
+ self.message_handler.chan_handler.handle_announcement_signatures(&peer.their_node_id.unwrap(), &msg);
+ },
+ wire::Message::ChannelAnnouncement(msg) => {
+ if self.message_handler.route_handler.handle_channel_announcement(&msg)
+ .map_err(|e| -> MessageHandlingError { e.into() })? {
+ should_forward = Some(wire::Message::ChannelAnnouncement(msg));
+ }
+ },
+ wire::Message::NodeAnnouncement(msg) => {
+ if self.message_handler.route_handler.handle_node_announcement(&msg)
+ .map_err(|e| -> MessageHandlingError { e.into() })? {
+ should_forward = Some(wire::Message::NodeAnnouncement(msg));
+ }
+ },
+ wire::Message::ChannelUpdate(msg) => {
+ self.message_handler.chan_handler.handle_channel_update(&peer.their_node_id.unwrap(), &msg);
+ if self.message_handler.route_handler.handle_channel_update(&msg)
+ .map_err(|e| -> MessageHandlingError { e.into() })? {
+ should_forward = Some(wire::Message::ChannelUpdate(msg));
+ }
+ },
+ wire::Message::QueryShortChannelIds(msg) => {
+ self.message_handler.route_handler.handle_query_short_channel_ids(&peer.their_node_id.unwrap(), msg)?;
+ },
+ wire::Message::ReplyShortChannelIdsEnd(msg) => {
+ self.message_handler.route_handler.handle_reply_short_channel_ids_end(&peer.their_node_id.unwrap(), msg)?;
+ },
+ wire::Message::QueryChannelRange(msg) => {
+ self.message_handler.route_handler.handle_query_channel_range(&peer.their_node_id.unwrap(), msg)?;
+ },
+ wire::Message::ReplyChannelRange(msg) => {
+ self.message_handler.route_handler.handle_reply_channel_range(&peer.their_node_id.unwrap(), msg)?;
+ },
+ wire::Message::GossipTimestampFilter(_msg) => {
+ // TODO: handle message
+ },
+
+ // Unknown messages:
+ wire::Message::Unknown(type_id) if message.is_even() => {
+ log_debug!(self.logger, "Received unknown even message of type {}, disconnecting peer!", type_id);
+ // Fail the channel if message is an even, unknown type as per BOLT #1.
+ return Err(PeerHandleError{ no_connection_possible: true }.into());
+ },
+ wire::Message::Unknown(type_id) => {
+ log_trace!(self.logger, "Received unknown odd message of type {}, ignoring", type_id);
+ },
+ wire::Message::Custom(custom) => {
+ self.custom_message_handler.handle_custom_message(custom)?;
+ },
+ };
+ Ok(should_forward)
+ }
+
+ fn forward_broadcast_msg(&self, peers: &mut PeerHolder<Descriptor>, msg: &wire::Message<<<CMH as core::ops::Deref>::Target as wire::CustomMessageReader>::CustomMessage>, except_node: Option<&PublicKey>) {
+ match msg {
+ wire::Message::ChannelAnnouncement(ref msg) => {
+ log_trace!(self.logger, "Sending message to all peers except {:?} or the announced channel's counterparties: {:?}", except_node, msg);
+ let encoded_msg = encode_msg!(msg);
+
+ for (_, peer) in peers.peers.iter_mut() {
+ if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
+ !peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
+ continue
+ }
+ if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+ log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
+ continue;
+ }
+ if peer.their_node_id.as_ref() == Some(&msg.contents.node_id_1) ||
+ peer.their_node_id.as_ref() == Some(&msg.contents.node_id_2) {
+ continue;
+ }
+ if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
+ continue;
+ }
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+ }
+ },
+ wire::Message::NodeAnnouncement(ref msg) => {
+ log_trace!(self.logger, "Sending message to all peers except {:?} or the announced node: {:?}", except_node, msg);
+ let encoded_msg = encode_msg!(msg);
+
+ for (_, peer) in peers.peers.iter_mut() {
+ if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
+ !peer.should_forward_node_announcement(msg.contents.node_id) {
+ continue
+ }
+ if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+ log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
+ continue;
+ }
+ if peer.their_node_id.as_ref() == Some(&msg.contents.node_id) {
+ continue;
+ }
+ if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
+ continue;
+ }
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+ }
+ },
+ wire::Message::ChannelUpdate(ref msg) => {
+ log_trace!(self.logger, "Sending message to all peers except {:?}: {:?}", except_node, msg);
+ let encoded_msg = encode_msg!(msg);
+
+ for (_, peer) in peers.peers.iter_mut() {
+ if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() ||
+ !peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
+ continue
+ }
+ if peer.pending_outbound_buffer.len() > OUTBOUND_BUFFER_LIMIT_DROP_GOSSIP {
+ log_trace!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
+ continue;
+ }
+ if except_node.is_some() && peer.their_node_id.as_ref() == except_node {
+ continue;
+ }
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encoded_msg[..]));
+ }
+ },
+ _ => debug_assert!(false, "We shouldn't attempt to forward anything but gossip messages"),
+ }
+ }
+