X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fpeer_handler.rs;h=e01d2c9b89c4494d68b8720062de63147538a1fc;hb=4fbc0b37688411a7270fce20b44fb43944669823;hp=bddc87b99fd7d6b95c9b603a7bd33062d405e0d3;hpb=e7d0d8af9f2a5d011c86edb694c315ec69f2ef21;p=rust-lightning diff --git a/src/ln/peer_handler.rs b/src/ln/peer_handler.rs index bddc87b9..e01d2c9b 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 @@ -42,11 +40,21 @@ pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone { /// generate no further read/write_events for the descriptor, only triggering a single /// 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 {} +pub struct PeerHandleError { + no_connection_possible: bool, +} +impl fmt::Debug for PeerHandleError { + fn fmt(&self, formatter: &mut fmt::Formatter) -> Result<(), fmt::Error> { + formatter.write_str("Peer Sent Invalid Data") + } +} struct Peer { channel_encryptor: PeerChannelEncryptor, + outbound: bool, their_node_id: Option, + their_global_features: Option, + their_local_features: Option, pending_outbound_buffer: LinkedList>, pending_outbound_buffer_first_msg_offset: usize, @@ -109,7 +117,10 @@ impl PeerManager { let mut peers = self.peers.lock().unwrap(); if peers.peers.insert(descriptor, Peer { channel_encryptor: peer_encryptor, + outbound: true, their_node_id: Some(their_node_id), + their_global_features: None, + their_local_features: None, pending_outbound_buffer: LinkedList::new(), pending_outbound_buffer_first_msg_offset: 0, @@ -138,7 +149,10 @@ impl PeerManager { let mut peers = self.peers.lock().unwrap(); if peers.peers.insert(descriptor, Peer { channel_encryptor: peer_encryptor, + outbound: false, their_node_id: None, + their_global_features: None, + their_local_features: None, pending_outbound_buffer: LinkedList::new(), pending_outbound_buffer_first_msg_offset: 0, @@ -206,7 +220,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) { @@ -267,6 +290,7 @@ impl PeerManager { let act_three = try_potential_handleerror!(peer.channel_encryptor.process_act_two(&peer.pending_read_buffer[..], &self.our_node_secret)).to_vec(); peer.pending_outbound_buffer.push_back(act_three); peer.pending_read_buffer = [0; 18].to_vec(); // Message length header is 18 bytes + peer.pending_read_is_header = true; insert_node_id = Some(peer.their_node_id.unwrap()); encode_and_send_msg!(msgs::Init { @@ -286,7 +310,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; @@ -295,6 +319,10 @@ impl PeerManager { assert!(msg_data.len() >= 2); let msg_type = byte_utils::slice_to_be16(&msg_data[0..2]); + if msg_type != 16 && peer.their_global_features.is_none() { + // Need an init message as first message + return Err(PeerHandleError{}); + } match msg_type { // Connection control: 16 => { @@ -305,7 +333,15 @@ impl PeerManager { if msg.local_features.requires_unknown_bits() { return Err(PeerHandleError{}); } - //TODO: Store features! + peer.their_global_features = Some(msg.global_features); + peer.their_local_features = Some(msg.local_features); + + if !peer.outbound { + encode_and_send_msg!(msgs::Init { + global_features: msgs::GlobalFeatures::new(), + local_features: msgs::LocalFeatures::new(), + }, 16); + } }, 17 => { // Error msg @@ -344,11 +380,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 +402,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 +420,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 +486,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 +529,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 +576,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 +610,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.