X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fln%2Fchannelmanager.rs;h=7a5f814d8178fb88ba08a17d62d6be51847c9cb6;hb=90b545ffe3e6855e09156361ecb1d5384add1e46;hp=508997833847898147c296aab491aaf666333e87;hpb=2a93f98c86b726a3ecc8f4c73847db4318b64908;p=rust-lightning diff --git a/src/ln/channelmanager.rs b/src/ln/channelmanager.rs index 50899783..7a5f814d 100644 --- a/src/ln/channelmanager.rs +++ b/src/ln/channelmanager.rs @@ -447,6 +447,8 @@ impl ChannelManager { //TODO: We need to handle monitoring of pending offered HTLCs which just hit the chain and //may be claimed, resulting in us claiming the inbound HTLCs (and back-failing after //timeouts are hit and our claims confirm). + //TODO: In any case, we need to make sure we remove any pending htlc tracking (via + //fail_backwards or claim_funds) eventually for all HTLCs that were in the channel } /// Force closes a channel, immediately broadcasting the latest local commitment transaction to @@ -1147,7 +1149,12 @@ impl ChannelManager { if !add_htlc_msgs.is_empty() { let (commitment_msg, monitor) = match forward_chan.send_commitment() { Ok(res) => res, - Err(_e) => { + Err(e) => { + if let &Some(msgs::ErrorAction::DisconnectPeer{msg: Some(ref _err_msg)}) = &e.action { + } else if let &Some(msgs::ErrorAction::SendErrorMessage{msg: ref _err_msg}) = &e.action { + } else { + panic!("Stated return value requirements in send_commitment() were not met"); + } //TODO: Handle...this is bad! continue; }, @@ -1439,223 +1446,20 @@ impl ChannelManager { Ok(accept_msg) } - fn internal_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), MsgHandleErrInternal> { - let (chan_announcement, chan_update) = { - let mut channel_state = self.channel_state.lock().unwrap(); - match channel_state.by_id.get_mut(&msg.channel_id) { - Some(chan) => { - if chan.get_their_node_id() != *their_node_id { - return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); - } - if !chan.is_usable() { - return Err(MsgHandleErrInternal::from_no_close(HandleError{err: "Got an announcement_signatures before we were ready for it", action: Some(msgs::ErrorAction::IgnoreError)})); - } - - let our_node_id = self.get_our_node_id(); - let (announcement, our_bitcoin_sig) = chan.get_channel_announcement(our_node_id.clone(), self.genesis_hash.clone()) - .map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; - - let were_node_one = announcement.node_id_1 == our_node_id; - let msghash = Message::from_slice(&Sha256dHash::from_data(&announcement.encode()[..])[..]).unwrap(); - let bad_sig_action = MsgHandleErrInternal::send_err_msg_close_chan("Bad announcement_signatures node_signature", msg.channel_id); - secp_call!(self.secp_ctx.verify(&msghash, &msg.node_signature, if were_node_one { &announcement.node_id_2 } else { &announcement.node_id_1 }), bad_sig_action); - secp_call!(self.secp_ctx.verify(&msghash, &msg.bitcoin_signature, if were_node_one { &announcement.bitcoin_key_2 } else { &announcement.bitcoin_key_1 }), bad_sig_action); - - let our_node_sig = self.secp_ctx.sign(&msghash, &self.our_network_key); - - (msgs::ChannelAnnouncement { - node_signature_1: if were_node_one { our_node_sig } else { msg.node_signature }, - node_signature_2: if were_node_one { msg.node_signature } else { our_node_sig }, - bitcoin_signature_1: if were_node_one { our_bitcoin_sig } else { msg.bitcoin_signature }, - bitcoin_signature_2: if were_node_one { msg.bitcoin_signature } else { our_bitcoin_sig }, - contents: announcement, - }, self.get_channel_update(chan).unwrap()) // can only fail if we're not in a ready state - }, - None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) - } - }; - let mut pending_events = self.pending_events.lock().unwrap(); - pending_events.push(events::Event::BroadcastChannelAnnouncement { msg: chan_announcement, update_msg: chan_update }); - Ok(()) - } - - -} - -impl events::EventsProvider for ChannelManager { - fn get_and_clear_pending_events(&self) -> Vec { - let mut pending_events = self.pending_events.lock().unwrap(); - let mut ret = Vec::new(); - mem::swap(&mut ret, &mut *pending_events); - ret - } -} - -impl ChainListener for ChannelManager { - fn block_connected(&self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], indexes_of_txn_matched: &[u32]) { - let mut new_events = Vec::new(); - let mut failed_channels = Vec::new(); - { - let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_lock.borrow_parts(); - let short_to_id = channel_state.short_to_id; - channel_state.by_id.retain(|_, channel| { - let chan_res = channel.block_connected(header, height, txn_matched, indexes_of_txn_matched); - if let Ok(Some(funding_locked)) = chan_res { - let announcement_sigs = self.get_announcement_sigs(channel); - new_events.push(events::Event::SendFundingLocked { - node_id: channel.get_their_node_id(), - msg: funding_locked, - announcement_sigs: announcement_sigs - }); - short_to_id.insert(channel.get_short_channel_id().unwrap(), channel.channel_id()); - } else if let Err(e) = chan_res { - new_events.push(events::Event::HandleError { - node_id: channel.get_their_node_id(), - action: e.action, - }); - if channel.is_shutdown() { - return false; - } - } - if let Some(funding_txo) = channel.get_funding_txo() { - for tx in txn_matched { - for inp in tx.input.iter() { - if inp.previous_output == funding_txo.into_bitcoin_outpoint() { - if let Some(short_id) = channel.get_short_channel_id() { - short_to_id.remove(&short_id); - } - // It looks like our counterparty went on-chain. We go ahead and - // broadcast our latest local state as well here, just in case its - // some kind of SPV attack, though we expect these to be dropped. - failed_channels.push(channel.force_shutdown()); - if let Ok(update) = self.get_channel_update(&channel) { - new_events.push(events::Event::BroadcastChannelUpdate { - msg: update - }); - } - return false; - } - } - } - } - if channel.is_funding_initiated() && channel.channel_monitor().would_broadcast_at_height(height) { - if let Some(short_id) = channel.get_short_channel_id() { - short_to_id.remove(&short_id); - } - failed_channels.push(channel.force_shutdown()); - // If would_broadcast_at_height() is true, the channel_monitor will broadcast - // the latest local tx for us, so we should skip that here (it doesn't really - // hurt anything, but does make tests a bit simpler). - failed_channels.last_mut().unwrap().0 = Vec::new(); - if let Ok(update) = self.get_channel_update(&channel) { - new_events.push(events::Event::BroadcastChannelUpdate { - msg: update - }); - } - return false; - } - true - }); - } - for failure in failed_channels.drain(..) { - self.finish_force_close_channel(failure); - } - let mut pending_events = self.pending_events.lock().unwrap(); - for funding_locked in new_events.drain(..) { - pending_events.push(funding_locked); - } - self.latest_block_height.store(height as usize, Ordering::Release); - } - - /// We force-close the channel without letting our counterparty participate in the shutdown - fn block_disconnected(&self, header: &BlockHeader) { - let mut new_events = Vec::new(); - let mut failed_channels = Vec::new(); - { - let mut channel_lock = self.channel_state.lock().unwrap(); - let channel_state = channel_lock.borrow_parts(); - let short_to_id = channel_state.short_to_id; - channel_state.by_id.retain(|_, v| { - if v.block_disconnected(header) { - if let Some(short_id) = v.get_short_channel_id() { - short_to_id.remove(&short_id); - } - failed_channels.push(v.force_shutdown()); - if let Ok(update) = self.get_channel_update(&v) { - new_events.push(events::Event::BroadcastChannelUpdate { - msg: update - }); - } - false - } else { - true - } - }); - } - for failure in failed_channels.drain(..) { - self.finish_force_close_channel(failure); - } - if !new_events.is_empty() { - let mut pending_events = self.pending_events.lock().unwrap(); - for funding_locked in new_events.drain(..) { - pending_events.push(funding_locked); - } - } - self.latest_block_height.fetch_sub(1, Ordering::AcqRel); - } -} - -macro_rules! handle_error { - ($self: ident, $internal: expr, $their_node_id: expr) => { - match $internal { - Ok(msg) => Ok(msg), - Err(MsgHandleErrInternal { err, needs_channel_force_close }) => { - if needs_channel_force_close { - match &err.action { - &Some(msgs::ErrorAction::DisconnectPeer { msg: Some(ref msg) }) => { - if msg.channel_id == [0; 32] { - $self.peer_disconnected(&$their_node_id, true); - } else { - $self.force_close_channel(&msg.channel_id); - } - }, - &Some(msgs::ErrorAction::DisconnectPeer { msg: None }) => {}, - &Some(msgs::ErrorAction::IgnoreError) => {}, - &Some(msgs::ErrorAction::SendErrorMessage { ref msg }) => { - if msg.channel_id == [0; 32] { - $self.peer_disconnected(&$their_node_id, true); - } else { - $self.force_close_channel(&msg.channel_id); - } - }, - &None => {}, - } - } - Err(err) - }, - } - } -} - -impl ChannelMessageHandler for ChannelManager { - //TODO: Handle errors and close channel (or so) - fn handle_open_channel(&self, their_node_id: &PublicKey, msg: &msgs::OpenChannel) -> Result { - handle_error!(self, self.internal_open_channel(their_node_id, msg), their_node_id) - } - - fn handle_accept_channel(&self, their_node_id: &PublicKey, msg: &msgs::AcceptChannel) -> Result<(), HandleError> { + fn internal_accept_channel(&self, their_node_id: &PublicKey, msg: &msgs::AcceptChannel) -> Result<(), MsgHandleErrInternal> { let (value, output_script, user_id) = { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.get_mut(&msg.temporary_channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: see issue #153, need a consistent behavior on obnoxious behavior from random node + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.temporary_channel_id)); } - chan.accept_channel(&msg)?; + chan.accept_channel(&msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; (chan.get_value_satoshis(), chan.get_funding_redeemscript().to_v0_p2wsh(), chan.get_user_id()) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + //TODO: same as above + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.temporary_channel_id)) } }; let mut pending_events = self.pending_events.lock().unwrap(); @@ -1668,25 +1472,25 @@ impl ChannelMessageHandler for ChannelManager { Ok(()) } - fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result { + fn internal_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result { let (chan, funding_msg, monitor_update) = { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.entry(msg.temporary_channel_id.clone()) { hash_map::Entry::Occupied(mut chan) => { if chan.get().get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.temporary_channel_id)); } match chan.get_mut().funding_created(msg) { Ok((funding_msg, monitor_update)) => { (chan.remove(), funding_msg, monitor_update) }, Err(e) => { - //TODO: Possibly remove the channel depending on e.action - return Err(e); + return Err(e).map_err(|e| MsgHandleErrInternal::from_maybe_close(e)) } } }, - hash_map::Entry::Vacant(_) => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.temporary_channel_id)) } }; // Release channel lock for install_watch_outpoint call, // note that this means if the remote end is misbehaving and sends a message for the same @@ -1698,10 +1502,7 @@ impl ChannelMessageHandler for ChannelManager { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.entry(funding_msg.channel_id) { hash_map::Entry::Occupied(_) => { - return Err(HandleError { - err: "Duplicate channel_id!", - action: Some(msgs::ErrorAction::SendErrorMessage { msg: msgs::ErrorMessage { channel_id: funding_msg.channel_id, data: "Already had channel with the new channel_id".to_owned() } }) - }); + return Err(MsgHandleErrInternal::send_err_msg_no_close("Already had channel with the new channel_id", funding_msg.channel_id)) }, hash_map::Entry::Vacant(e) => { e.insert(chan); @@ -1710,18 +1511,19 @@ impl ChannelMessageHandler for ChannelManager { Ok(funding_msg) } - fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), HandleError> { + fn internal_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), MsgHandleErrInternal> { let (funding_txo, user_id, monitor) = { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } - let chan_monitor = chan.funding_signed(&msg)?; + let chan_monitor = chan.funding_signed(&msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; (chan.get_funding_txo().unwrap(), chan.get_user_id(), chan_monitor) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } }; if let Err(_e) = self.monitor.add_update_monitor(monitor.get_funding_txo().unwrap(), monitor) { @@ -1735,21 +1537,22 @@ impl ChannelMessageHandler for ChannelManager { Ok(()) } - fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result, HandleError> { + fn internal_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result, MsgHandleErrInternal> { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } - chan.funding_locked(&msg)?; + chan.funding_locked(&msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; return Ok(self.get_announcement_sigs(chan)); }, - None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) }; } - fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(Option, Option), HandleError> { + fn internal_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(Option, Option), MsgHandleErrInternal> { let (res, chan_option) = { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = channel_state_lock.borrow_parts(); @@ -1757,9 +1560,10 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.entry(msg.channel_id.clone()) { hash_map::Entry::Occupied(mut chan_entry) => { if chan_entry.get().get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } - let res = chan_entry.get_mut().shutdown(&*self.fee_estimator, &msg)?; + let res = chan_entry.get_mut().shutdown(&*self.fee_estimator, &msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; if chan_entry.get().is_shutdown() { if let Some(short_id) = chan_entry.get().get_short_channel_id() { channel_state.short_to_id.remove(&short_id); @@ -1767,7 +1571,7 @@ impl ChannelMessageHandler for ChannelManager { (res, Some(chan_entry.remove_entry().1)) } else { (res, None) } }, - hash_map::Entry::Vacant(_) => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } }; for payment_hash in res.2 { @@ -1785,16 +1589,17 @@ impl ChannelMessageHandler for ChannelManager { Ok((res.0, res.1)) } - fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result, HandleError> { + fn internal_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result, MsgHandleErrInternal> { let (res, chan_option) = { let mut channel_state_lock = self.channel_state.lock().unwrap(); let channel_state = channel_state_lock.borrow_parts(); match channel_state.by_id.entry(msg.channel_id.clone()) { hash_map::Entry::Occupied(mut chan_entry) => { if chan_entry.get().get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } - let res = chan_entry.get_mut().closing_signed(&*self.fee_estimator, &msg)?; + let res = chan_entry.get_mut().closing_signed(&*self.fee_estimator, &msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; if res.1.is_some() { // We're done with this channel, we've got a signed closing transaction and // will send the closing_signed back to the remote peer upon return. This @@ -1807,7 +1612,7 @@ impl ChannelMessageHandler for ChannelManager { (res, Some(chan_entry.remove_entry().1)) } else { (res, None) } }, - hash_map::Entry::Vacant(_) => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } }; if let Some(broadcast_tx) = res.1 { @@ -1824,7 +1629,7 @@ impl ChannelMessageHandler for ChannelManager { Ok(res.0) } - fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), msgs::HandleError> { + fn internal_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), MsgHandleErrInternal> { //TODO: BOLT 4 points out a specific attack where a peer may re-send an onion packet and //determine the state of the payment based on our response/if we forward anything/the time //we take to respond. We should take care to avoid allowing such an attack. @@ -1868,18 +1673,19 @@ impl ChannelMessageHandler for ChannelManager { let (source_short_channel_id, res) = match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } if !chan.is_usable() { - return Err(HandleError{err: "Channel not yet available for receiving HTLCs", action: None}); + return Err(MsgHandleErrInternal::from_no_close(HandleError{err: "Channel not yet available for receiving HTLCs", action: Some(msgs::ErrorAction::IgnoreError)})); } let short_channel_id = chan.get_short_channel_id().unwrap(); if let PendingHTLCStatus::Forward(ref mut forward_info) = pending_forward_info { forward_info.prev_short_channel_id = short_channel_id; } - (short_channel_id, chan.update_add_htlc(&msg, pending_forward_info)?) + (short_channel_id, chan.update_add_htlc(&msg, pending_forward_info).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}), + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) }; if will_forward { @@ -1911,7 +1717,7 @@ impl ChannelMessageHandler for ChannelManager { Ok(res) } - fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), HandleError> { + fn internal_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), MsgHandleErrInternal> { //TODO: Delay the claimed_funds relaying just like we do outbound relay! // Claim funds first, cause we don't really care if the channel we received the message on // is broken, we may have enough info to get our own money! @@ -1921,24 +1727,26 @@ impl ChannelMessageHandler for ChannelManager { match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } - chan.update_fulfill_htlc(&msg) + chan.update_fulfill_htlc(&msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e)) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } } - fn handle_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result, HandleError> { + fn internal_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result, MsgHandleErrInternal> { let mut channel_state = self.channel_state.lock().unwrap(); let payment_hash = match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } - chan.update_fail_htlc(&msg, HTLCFailReason::ErrorPacket { err: msg.reason.clone() }) + chan.update_fail_htlc(&msg, HTLCFailReason::ErrorPacket { err: msg.reason.clone() }).map_err(|e| MsgHandleErrInternal::from_maybe_close(e)) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) }?; if let Some(pending_htlc) = channel_state.claimable_htlcs.get(&payment_hash) { @@ -2004,30 +1812,32 @@ impl ChannelMessageHandler for ChannelManager { } } - fn handle_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), HandleError> { + fn internal_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), MsgHandleErrInternal> { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } - chan.update_fail_malformed_htlc(&msg, HTLCFailReason::Reason { failure_code: msg.failure_code, data: Vec::new() }) + chan.update_fail_malformed_htlc(&msg, HTLCFailReason::Reason { failure_code: msg.failure_code, data: Vec::new() }).map_err(|e| MsgHandleErrInternal::from_maybe_close(e)) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } } - fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(msgs::RevokeAndACK, Option), HandleError> { + fn internal_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(msgs::RevokeAndACK, Option), MsgHandleErrInternal> { let (revoke_and_ack, commitment_signed, chan_monitor) = { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } - chan.commitment_signed(&msg)? + chan.commitment_signed(&msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))? }, - None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } }; if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { @@ -2037,17 +1847,18 @@ impl ChannelMessageHandler for ChannelManager { Ok((revoke_and_ack, commitment_signed)) } - fn handle_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result, HandleError> { + fn internal_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result, MsgHandleErrInternal> { let (res, mut pending_forwards, mut pending_failures, chan_monitor) = { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } - chan.revoke_and_ack(&msg)? + chan.revoke_and_ack(&msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e))? }, - None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } }; if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) { @@ -2088,19 +1899,277 @@ impl ChannelMessageHandler for ChannelManager { Ok(res) } - fn handle_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), HandleError> { + fn internal_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), MsgHandleErrInternal> { let mut channel_state = self.channel_state.lock().unwrap(); match channel_state.by_id.get_mut(&msg.channel_id) { Some(chan) => { if chan.get_their_node_id() != *their_node_id { - return Err(HandleError{err: "Got a message for a channel from the wrong node!", action: None}) + //TODO: here and below MsgHandleErrInternal, #153 case + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); } - chan.update_fee(&*self.fee_estimator, &msg) + chan.update_fee(&*self.fee_estimator, &msg).map_err(|e| MsgHandleErrInternal::from_maybe_close(e)) }, - None => return Err(HandleError{err: "Failed to find corresponding channel", action: None}) + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) } } + fn internal_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), MsgHandleErrInternal> { + let (chan_announcement, chan_update) = { + let mut channel_state = self.channel_state.lock().unwrap(); + match channel_state.by_id.get_mut(&msg.channel_id) { + Some(chan) => { + if chan.get_their_node_id() != *their_node_id { + return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!", msg.channel_id)); + } + if !chan.is_usable() { + return Err(MsgHandleErrInternal::from_no_close(HandleError{err: "Got an announcement_signatures before we were ready for it", action: Some(msgs::ErrorAction::IgnoreError)})); + } + + let our_node_id = self.get_our_node_id(); + let (announcement, our_bitcoin_sig) = chan.get_channel_announcement(our_node_id.clone(), self.genesis_hash.clone()) + .map_err(|e| MsgHandleErrInternal::from_maybe_close(e))?; + + let were_node_one = announcement.node_id_1 == our_node_id; + let msghash = Message::from_slice(&Sha256dHash::from_data(&announcement.encode()[..])[..]).unwrap(); + let bad_sig_action = MsgHandleErrInternal::send_err_msg_close_chan("Bad announcement_signatures node_signature", msg.channel_id); + secp_call!(self.secp_ctx.verify(&msghash, &msg.node_signature, if were_node_one { &announcement.node_id_2 } else { &announcement.node_id_1 }), bad_sig_action); + secp_call!(self.secp_ctx.verify(&msghash, &msg.bitcoin_signature, if were_node_one { &announcement.bitcoin_key_2 } else { &announcement.bitcoin_key_1 }), bad_sig_action); + + let our_node_sig = self.secp_ctx.sign(&msghash, &self.our_network_key); + + (msgs::ChannelAnnouncement { + node_signature_1: if were_node_one { our_node_sig } else { msg.node_signature }, + node_signature_2: if were_node_one { msg.node_signature } else { our_node_sig }, + bitcoin_signature_1: if were_node_one { our_bitcoin_sig } else { msg.bitcoin_signature }, + bitcoin_signature_2: if were_node_one { msg.bitcoin_signature } else { our_bitcoin_sig }, + contents: announcement, + }, self.get_channel_update(chan).unwrap()) // can only fail if we're not in a ready state + }, + None => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", msg.channel_id)) + } + }; + let mut pending_events = self.pending_events.lock().unwrap(); + pending_events.push(events::Event::BroadcastChannelAnnouncement { msg: chan_announcement, update_msg: chan_update }); + Ok(()) + } + + +} + +impl events::EventsProvider for ChannelManager { + fn get_and_clear_pending_events(&self) -> Vec { + let mut pending_events = self.pending_events.lock().unwrap(); + let mut ret = Vec::new(); + mem::swap(&mut ret, &mut *pending_events); + ret + } +} + +impl ChainListener for ChannelManager { + fn block_connected(&self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], indexes_of_txn_matched: &[u32]) { + let mut new_events = Vec::new(); + let mut failed_channels = Vec::new(); + { + let mut channel_lock = self.channel_state.lock().unwrap(); + let channel_state = channel_lock.borrow_parts(); + let short_to_id = channel_state.short_to_id; + channel_state.by_id.retain(|_, channel| { + let chan_res = channel.block_connected(header, height, txn_matched, indexes_of_txn_matched); + if let Ok(Some(funding_locked)) = chan_res { + let announcement_sigs = self.get_announcement_sigs(channel); + new_events.push(events::Event::SendFundingLocked { + node_id: channel.get_their_node_id(), + msg: funding_locked, + announcement_sigs: announcement_sigs + }); + short_to_id.insert(channel.get_short_channel_id().unwrap(), channel.channel_id()); + } else if let Err(e) = chan_res { + new_events.push(events::Event::HandleError { + node_id: channel.get_their_node_id(), + action: e.action, + }); + if channel.is_shutdown() { + return false; + } + } + if let Some(funding_txo) = channel.get_funding_txo() { + for tx in txn_matched { + for inp in tx.input.iter() { + if inp.previous_output == funding_txo.into_bitcoin_outpoint() { + if let Some(short_id) = channel.get_short_channel_id() { + short_to_id.remove(&short_id); + } + // It looks like our counterparty went on-chain. We go ahead and + // broadcast our latest local state as well here, just in case its + // some kind of SPV attack, though we expect these to be dropped. + failed_channels.push(channel.force_shutdown()); + if let Ok(update) = self.get_channel_update(&channel) { + new_events.push(events::Event::BroadcastChannelUpdate { + msg: update + }); + } + return false; + } + } + } + } + if channel.is_funding_initiated() && channel.channel_monitor().would_broadcast_at_height(height) { + if let Some(short_id) = channel.get_short_channel_id() { + short_to_id.remove(&short_id); + } + failed_channels.push(channel.force_shutdown()); + // If would_broadcast_at_height() is true, the channel_monitor will broadcast + // the latest local tx for us, so we should skip that here (it doesn't really + // hurt anything, but does make tests a bit simpler). + failed_channels.last_mut().unwrap().0 = Vec::new(); + if let Ok(update) = self.get_channel_update(&channel) { + new_events.push(events::Event::BroadcastChannelUpdate { + msg: update + }); + } + return false; + } + true + }); + } + for failure in failed_channels.drain(..) { + self.finish_force_close_channel(failure); + } + let mut pending_events = self.pending_events.lock().unwrap(); + for funding_locked in new_events.drain(..) { + pending_events.push(funding_locked); + } + self.latest_block_height.store(height as usize, Ordering::Release); + } + + /// We force-close the channel without letting our counterparty participate in the shutdown + fn block_disconnected(&self, header: &BlockHeader) { + let mut new_events = Vec::new(); + let mut failed_channels = Vec::new(); + { + let mut channel_lock = self.channel_state.lock().unwrap(); + let channel_state = channel_lock.borrow_parts(); + let short_to_id = channel_state.short_to_id; + channel_state.by_id.retain(|_, v| { + if v.block_disconnected(header) { + if let Some(short_id) = v.get_short_channel_id() { + short_to_id.remove(&short_id); + } + failed_channels.push(v.force_shutdown()); + if let Ok(update) = self.get_channel_update(&v) { + new_events.push(events::Event::BroadcastChannelUpdate { + msg: update + }); + } + false + } else { + true + } + }); + } + for failure in failed_channels.drain(..) { + self.finish_force_close_channel(failure); + } + if !new_events.is_empty() { + let mut pending_events = self.pending_events.lock().unwrap(); + for funding_locked in new_events.drain(..) { + pending_events.push(funding_locked); + } + } + self.latest_block_height.fetch_sub(1, Ordering::AcqRel); + } +} + +macro_rules! handle_error { + ($self: ident, $internal: expr, $their_node_id: expr) => { + match $internal { + Ok(msg) => Ok(msg), + Err(MsgHandleErrInternal { err, needs_channel_force_close }) => { + if needs_channel_force_close { + match &err.action { + &Some(msgs::ErrorAction::DisconnectPeer { msg: Some(ref msg) }) => { + if msg.channel_id == [0; 32] { + $self.peer_disconnected(&$their_node_id, true); + } else { + $self.force_close_channel(&msg.channel_id); + } + }, + &Some(msgs::ErrorAction::DisconnectPeer { msg: None }) => {}, + &Some(msgs::ErrorAction::IgnoreError) => {}, + &Some(msgs::ErrorAction::SendErrorMessage { ref msg }) => { + if msg.channel_id == [0; 32] { + $self.peer_disconnected(&$their_node_id, true); + } else { + $self.force_close_channel(&msg.channel_id); + } + }, + &None => {}, + } + } + Err(err) + }, + } + } +} + +impl ChannelMessageHandler for ChannelManager { + //TODO: Handle errors and close channel (or so) + fn handle_open_channel(&self, their_node_id: &PublicKey, msg: &msgs::OpenChannel) -> Result { + handle_error!(self, self.internal_open_channel(their_node_id, msg), their_node_id) + } + + fn handle_accept_channel(&self, their_node_id: &PublicKey, msg: &msgs::AcceptChannel) -> Result<(), HandleError> { + handle_error!(self, self.internal_accept_channel(their_node_id, msg), their_node_id) + } + + fn handle_funding_created(&self, their_node_id: &PublicKey, msg: &msgs::FundingCreated) -> Result { + handle_error!(self, self.internal_funding_created(their_node_id, msg), their_node_id) + } + + fn handle_funding_signed(&self, their_node_id: &PublicKey, msg: &msgs::FundingSigned) -> Result<(), HandleError> { + handle_error!(self, self.internal_funding_signed(their_node_id, msg), their_node_id) + } + + fn handle_funding_locked(&self, their_node_id: &PublicKey, msg: &msgs::FundingLocked) -> Result, HandleError> { + handle_error!(self, self.internal_funding_locked(their_node_id, msg), their_node_id) + } + + fn handle_shutdown(&self, their_node_id: &PublicKey, msg: &msgs::Shutdown) -> Result<(Option, Option), HandleError> { + handle_error!(self, self.internal_shutdown(their_node_id, msg), their_node_id) + } + + fn handle_closing_signed(&self, their_node_id: &PublicKey, msg: &msgs::ClosingSigned) -> Result, HandleError> { + handle_error!(self, self.internal_closing_signed(their_node_id, msg), their_node_id) + } + + fn handle_update_add_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateAddHTLC) -> Result<(), msgs::HandleError> { + handle_error!(self, self.internal_update_add_htlc(their_node_id, msg), their_node_id) + } + + fn handle_update_fulfill_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFulfillHTLC) -> Result<(), HandleError> { + handle_error!(self, self.internal_update_fulfill_htlc(their_node_id, msg), their_node_id) + } + + fn handle_update_fail_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailHTLC) -> Result, HandleError> { + handle_error!(self, self.internal_update_fail_htlc(their_node_id, msg), their_node_id) + } + + fn handle_update_fail_malformed_htlc(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), HandleError> { + handle_error!(self, self.internal_update_fail_malformed_htlc(their_node_id, msg), their_node_id) + } + + fn handle_commitment_signed(&self, their_node_id: &PublicKey, msg: &msgs::CommitmentSigned) -> Result<(msgs::RevokeAndACK, Option), HandleError> { + handle_error!(self, self.internal_commitment_signed(their_node_id, msg), their_node_id) + } + + fn handle_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result, HandleError> { + handle_error!(self, self.internal_revoke_and_ack(their_node_id, msg), their_node_id) + } + + fn handle_update_fee(&self, their_node_id: &PublicKey, msg: &msgs::UpdateFee) -> Result<(), HandleError> { + handle_error!(self, self.internal_update_fee(their_node_id, msg), their_node_id) + } + fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &msgs::AnnouncementSignatures) -> Result<(), HandleError> { handle_error!(self, self.internal_announcement_signatures(their_node_id, msg), their_node_id) }