X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fpeer_handler.rs;h=a6992eae4ff1fcf62dab20875407e2741dcebec5;hb=f7dd69240afcd3a0087f7796824a9e2697ed85cf;hp=bddc87b99fd7d6b95c9b603a7bd33062d405e0d3;hpb=f9fc2dfbe20f86ab8bc3e637f3ca33a5a868cb98;p=rust-lightning diff --git a/src/ln/peer_handler.rs b/src/ln/peer_handler.rs index bddc87b9..a6992eae 100644 --- a/src/ln/peer_handler.rs +++ b/src/ln/peer_handler.rs @@ -8,13 +8,11 @@ use util::events::{EventsProvider,Event}; use std::collections::{HashMap,LinkedList}; use std::sync::{Arc, Mutex}; -use std::cmp; -use std::mem; -use std::hash; +use std::{cmp,mem,hash,fmt}; pub struct MessageHandler { - pub chan_handler: Arc, - pub route_handler: Arc, + pub chan_handler: Arc, + pub route_handler: Arc, } /// Provides an object which can be used to send data to and which uniquely identifies a connection @@ -43,6 +41,11 @@ 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 {} +impl fmt::Debug for PeerHandleError { + fn fmt(&self, formatter: &mut fmt::Formatter) -> Result<(), fmt::Error> { + formatter.write_str("Peer Send Invalid Data") + } +} struct Peer { channel_encryptor: PeerChannelEncryptor, @@ -206,7 +209,16 @@ impl PeerManager { /// 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) -> Result { - let mut upstream_events = Vec::new(); + match self.do_read_event(peer_descriptor, data) { + Ok(res) => Ok(res), + Err(e) => { + self.disconnect_event(peer_descriptor); + Err(e) + } + } + } + + fn do_read_event(&self, peer_descriptor: &mut Descriptor, data: Vec) -> Result { let pause_read = { let mut peers = self.peers.lock().unwrap(); let (should_insert_node_id, pause_read) = match peers.peers.get_mut(peer_descriptor) { @@ -286,7 +298,7 @@ impl PeerManager { let msg_len = try_potential_handleerror!(peer.channel_encryptor.decrypt_length_header(&peer.pending_read_buffer[..])); peer.pending_read_buffer = Vec::with_capacity(msg_len as usize + 16); peer.pending_read_buffer.resize(msg_len as usize + 16, 0); - if msg_len < 2 + 16 { // Need at least the message type tag + if msg_len < 2 { // Need at least the message type tag return Err(PeerHandleError{}); } peer.pending_read_is_header = false; @@ -344,11 +356,20 @@ impl PeerManager { 38 => { let msg = try_potential_decodeerror!(msgs::Shutdown::decode(&msg_data[2..])); - try_potential_handleerror!(self.message_handler.chan_handler.handle_shutdown(&peer.their_node_id.unwrap(), &msg)); + 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); + } + if let Some(resp) = resp_options.1 { + encode_and_send_msg!(resp, 39); + } }, 39 => { let msg = try_potential_decodeerror!(msgs::ClosingSigned::decode(&msg_data[2..])); - try_potential_handleerror!(self.message_handler.chan_handler.handle_closing_signed(&peer.their_node_id.unwrap(), &msg)); + 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 => { @@ -357,42 +378,15 @@ impl PeerManager { }, 130 => { let msg = try_potential_decodeerror!(msgs::UpdateFulfillHTLC::decode(&msg_data[2..])); - let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fulfill_htlc(&peer.their_node_id.unwrap(), &msg)); - match resp_option { - Some(resps) => { - for resp in resps.0 { - encode_and_send_msg!(resp, 128); - } - encode_and_send_msg!(resps.1, 132); - }, - None => {}, - } + 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 resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_htlc(&peer.their_node_id.unwrap(), &msg)); - match resp_option { - Some(resps) => { - for resp in resps.0 { - encode_and_send_msg!(resp, 128); - } - encode_and_send_msg!(resps.1, 132); - }, - None => {}, - } + 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::decode(&msg_data[2..])); - let resp_option = try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_malformed_htlc(&peer.their_node_id.unwrap(), &msg)); - match resp_option { - Some(resps) => { - for resp in resps.0 { - encode_and_send_msg!(resp, 128); - } - encode_and_send_msg!(resps.1, 132); - }, - None => {}, - } + try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fail_malformed_htlc(&peer.their_node_id.unwrap(), &msg)); }, 132 => { @@ -402,9 +396,17 @@ impl PeerManager { }, 133 => { let msg = try_potential_decodeerror!(msgs::RevokeAndACK::decode(&msg_data[2..])); - try_potential_handleerror!(self.message_handler.chan_handler.handle_revoke_and_ack(&peer.their_node_id.unwrap(), &msg)); + 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) => { + for resp in resps.0 { + encode_and_send_msg!(resp, 128); + } + encode_and_send_msg!(resps.1, 132); + }, + None => {}, + } }, - 134 => { let msg = try_potential_decodeerror!(msgs::UpdateFee::decode(&msg_data[2..])); try_potential_handleerror!(self.message_handler.chan_handler.handle_update_fee(&peer.their_node_id.unwrap(), &msg)); @@ -460,11 +462,25 @@ impl PeerManager { None => {} }; + pause_read + }; + + self.process_events(); + + Ok(pause_read) + } + + /// Checks for any events generated by our handlers and processes them. May be needed after eg + /// calls to ChannelManager::process_pending_htlc_forward. + pub fn process_events(&self) { + let mut upstream_events = Vec::new(); + { // TODO: There are some DoS attacks here where you can flood someone's outbound send // buffer by doing things like announcing channels on another node. We should be willing to // drop optional-ish messages when send buffers get full! let mut events_generated = self.message_handler.chan_handler.get_and_clear_pending_events(); + let mut peers = self.peers.lock().unwrap(); for event in events_generated.drain(..) { macro_rules! get_peer_for_forwarding { ($node_id: expr, $handle_no_such_peer: block) => { @@ -489,6 +505,8 @@ impl PeerManager { Event::FundingGenerationReady {..} => { /* Hand upstream */ }, Event::FundingBroadcastSafe {..} => { /* Hand upstream */ }, Event::PaymentReceived {..} => { /* Hand upstream */ }, + Event::PaymentSent {..} => { /* Hand upstream */ }, + Event::PaymentFailed {..} => { /* Hand upstream */ }, Event::PendingHTLCsForwardable {..} => { //TODO: Handle upstream in some confused form so that upstream just knows @@ -534,6 +552,14 @@ impl PeerManager { Self::do_attempt_write_data(&mut descriptor, peer); continue; }, + Event::SendFailHTLC { ref node_id, ref msg } => { + 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, 131))); + Self::do_attempt_write_data(&mut descriptor, peer); + continue; + }, Event::BroadcastChannelAnnouncement { ref msg, ref update_msg } => { let encoded_msg = encode_msg!(msg, 256); let encoded_update_msg = encode_msg!(update_msg, 258); @@ -560,16 +586,12 @@ impl PeerManager { upstream_events.push(event); } - - pause_read - }; + } let mut pending_events = self.pending_events.lock().unwrap(); for event in upstream_events.drain(..) { pending_events.push(event); } - - Ok(pause_read) } /// Indicates that the given socket descriptor's connection is now closed.