use secp256k1::key::{PublicKey,SecretKey};
use secp256k1::Secp256k1;
+use std::cell::RefCell;
use std::collections::HashMap;
+use std::hash::Hash;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize,Ordering};
fn broadcast_transaction(&self, _tx: &Transaction) {}
}
-#[derive(Clone, PartialEq, Eq, Hash)]
-struct Peer {
+#[derive(Clone)]
+struct Peer<'a> {
id: u8,
+ peers_connected: &'a RefCell<[bool; 256]>,
}
-impl SocketDescriptor for Peer {
+impl<'a> SocketDescriptor for Peer<'a> {
fn send_data(&mut self, data: &Vec<u8>, write_offset: usize, _resume_read: bool) -> usize {
assert!(write_offset < data.len());
data.len() - write_offset
}
+ fn disconnect_socket(&mut self) {
+ assert!(self.peers_connected.borrow()[self.id as usize]);
+ self.peers_connected.borrow_mut()[self.id as usize] = false;
+ }
+}
+impl<'a> PartialEq for Peer<'a> {
+ fn eq(&self, other: &Self) -> bool {
+ self.id == other.id
+ }
+}
+impl<'a> Eq for Peer<'a> {}
+impl<'a> Hash for Peer<'a> {
+ fn hash<H : std::hash::Hasher>(&self, h: &mut H) {
+ self.id.hash(h)
+ }
}
#[inline]
let channelmanager = ChannelManager::new(our_network_key, slice_to_be32(get_slice!(4)), get_slice!(1)[0] != 0, Network::Bitcoin, fee_est.clone(), monitor.clone(), watch.clone(), broadcast.clone()).unwrap();
let router = Arc::new(Router::new(PublicKey::from_secret_key(&secp_ctx, &our_network_key).unwrap()));
+ let peers = RefCell::new([false; 256]);
let handler = PeerManager::new(MessageHandler {
chan_handler: channelmanager.clone(),
route_handler: router.clone(),
}, our_network_key);
- let mut peers = [false; 256];
let mut should_forward = false;
let mut payments_received = Vec::new();
let mut payments_sent = 0;
0 => {
let mut new_id = 0;
for i in 1..256 {
- if !peers[i-1] {
+ if !peers.borrow()[i-1] {
new_id = i;
break;
}
}
if new_id == 0 { return; }
- peers[new_id - 1] = true;
- handler.new_outbound_connection(get_pubkey!(), Peer{id: (new_id - 1) as u8}).unwrap();
+ peers.borrow_mut()[new_id - 1] = true;
+ handler.new_outbound_connection(get_pubkey!(), Peer{id: (new_id - 1) as u8, peers_connected: &peers}).unwrap();
},
1 => {
let mut new_id = 0;
for i in 1..256 {
- if !peers[i-1] {
+ if !peers.borrow()[i-1] {
new_id = i;
break;
}
}
if new_id == 0 { return; }
- peers[new_id - 1] = true;
- handler.new_inbound_connection(Peer{id: (new_id - 1) as u8}).unwrap();
+ peers.borrow_mut()[new_id - 1] = true;
+ handler.new_inbound_connection(Peer{id: (new_id - 1) as u8, peers_connected: &peers}).unwrap();
},
2 => {
let peer_id = get_slice!(1)[0];
- if !peers[peer_id as usize] { return; }
- peers[peer_id as usize] = false;
- handler.disconnect_event(&Peer{id: peer_id});
+ if !peers.borrow()[peer_id as usize] { return; }
+ peers.borrow_mut()[peer_id as usize] = false;
+ handler.disconnect_event(&Peer{id: peer_id, peers_connected: &peers});
},
3 => {
let peer_id = get_slice!(1)[0];
- if !peers[peer_id as usize] { return; }
- match handler.read_event(&mut Peer{id: peer_id}, get_slice!(get_slice!(1)[0]).to_vec()) {
+ if !peers.borrow()[peer_id as usize] { return; }
+ match handler.read_event(&mut Peer{id: peer_id, peers_connected: &peers}, get_slice!(get_slice!(1)[0]).to_vec()) {
Ok(res) => assert!(!res),
- Err(_) => { peers[peer_id as usize] = false; }
+ Err(_) => { peers.borrow_mut()[peer_id as usize] = false; }
}
},
4 => {
},
5 => {
let peer_id = get_slice!(1)[0];
- if !peers[peer_id as usize] { return; }
+ if !peers.borrow()[peer_id as usize] { return; }
let their_key = get_pubkey!();
let chan_value = slice_to_be24(get_slice!(3)) as u64;
if channelmanager.create_channel(their_key, chan_value, 0).is_err() { return; }
match $res {
Ok(key) => key,
//TODO: make the error a parameter
- Err(_) => return Err(HandleError{err: $err, action: Some(msgs::ErrorAction::DisconnectPeer{})})
+ Err(_) => return Err(HandleError{err: $err, action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })})
}
};
}
fn check_remote_fee(fee_estimator: &FeeEstimator, feerate_per_kw: u32) -> Result<(), HandleError> {
if (feerate_per_kw as u64) < fee_estimator.get_est_sat_per_vbyte(ConfirmationTarget::Background) * 250 {
- return Err(HandleError{err: "Peer's feerate much too low", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "Peer's feerate much too low", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
if (feerate_per_kw as u64) > fee_estimator.get_est_sat_per_vbyte(ConfirmationTarget::HighPriority) * 375 { // 375 = 250 * 1.5x
- return Err(HandleError{err: "Peer's feerate much too high", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "Peer's feerate much too high", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
Ok(())
}
pub fn new_from_req(fee_estimator: &FeeEstimator, chan_keys: ChannelKeys, their_node_id: PublicKey, msg: &msgs::OpenChannel, user_id: u64, announce_publicly: bool) -> Result<Channel, HandleError> {
// Check sanity of message fields:
if msg.funding_satoshis >= MAX_FUNDING_SATOSHIS {
- return Err(HandleError{err: "funding value > 2^24", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "funding value > 2^24", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
if msg.channel_reserve_satoshis > msg.funding_satoshis {
- return Err(HandleError{err: "Bogus channel_reserve_satoshis", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "Bogus channel_reserve_satoshis", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
if msg.push_msat > (msg.funding_satoshis - msg.channel_reserve_satoshis) * 1000 {
- return Err(HandleError{err: "push_msat more than highest possible value", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "push_msat more than highest possible value", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
if msg.dust_limit_satoshis > msg.funding_satoshis {
- return Err(HandleError{err: "Peer never wants payout outputs?", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "Peer never wants payout outputs?", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
if msg.htlc_minimum_msat >= (msg.funding_satoshis - msg.channel_reserve_satoshis) * 1000 {
- return Err(HandleError{err: "Minimum htlc value is full channel value", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "Minimum htlc value is full channel value", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
Channel::check_remote_fee(fee_estimator, msg.feerate_per_kw)?;
if msg.to_self_delay > MAX_LOCAL_BREAKDOWN_TIMEOUT {
- return Err(HandleError{err: "They wanted our payments to be delayed by a needlessly long period", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "They wanted our payments to be delayed by a needlessly long period", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
if msg.max_accepted_htlcs < 1 {
- return Err(HandleError{err: "0 max_accpted_htlcs makes for a useless channel", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "0 max_accpted_htlcs makes for a useless channel", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
if (msg.channel_flags & 254) != 0 {
- return Err(HandleError{err: "unknown channel_flags", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "unknown channel_flags", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
// Convert things into internal flags and prep our state:
/// Call this upon creation of a funding transaction for the given channel.
/// Panics if a funding transaction has already been provided for this channel.
pub fn funding_transaction_generated(&self, temporary_channel_id: &[u8; 32], funding_txo: OutPoint) {
+
+ macro_rules! add_pending_event {
+ ($event: expr) => {
+ {
+ let mut pending_events = self.pending_events.lock().unwrap();
+ pending_events.push($event);
+ }
+ }
+ }
+
let (chan, msg, chan_monitor) = {
let mut channel_state = self.channel_state.lock().unwrap();
match channel_state.by_id.remove(temporary_channel_id) {
Ok(funding_msg) => {
(chan, funding_msg.0, funding_msg.1)
},
- Err(_e) => {
- //TODO: Push e to pendingevents
+ Err(e) => {
+ mem::drop(channel_state);
+ add_pending_event!(events::Event::DisconnectPeer {
+ node_id: chan.get_their_node_id(),
+ msg: if let Some(msgs::ErrorAction::DisconnectPeer { msg } ) = e.action { msg } else { None },
+ });
+
return;
- }
+ },
}
},
None => return
if let Err(_e) = self.monitor.add_update_monitor(chan_monitor.get_funding_txo().unwrap(), chan_monitor) {
unimplemented!(); // maybe remove from claimable_htlcs?
}
- {
- let mut pending_events = self.pending_events.lock().unwrap();
- pending_events.push(events::Event::SendFundingCreated {
- node_id: chan.get_their_node_id(),
- msg: msg,
- });
- }
+ add_pending_event!(events::Event::SendFundingCreated {
+ node_id: chan.get_their_node_id(),
+ msg: msg,
+ });
let mut channel_state = self.channel_state.lock().unwrap();
channel_state.by_id.insert(chan.channel_id(), chan);
msg: UpdateFailHTLC
},
/// The peer took some action which made us think they were useless. Disconnect them.
- DisconnectPeer,
+ DisconnectPeer {
+ msg: Option<ErrorMessage>
+ },
/// The peer did something harmless that we weren't able to process, just log and ignore
IgnoreError,
/// The peer did something incorrect. Tell them.
let mut chacha = ChaCha20Poly1305RFC::new(key, &nonce, h);
if !chacha.decrypt(&cyphertext[0..cyphertext.len() - 16], res, &cyphertext[cyphertext.len() - 16..]) {
- return Err(HandleError{err: "Bad MAC", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "Bad MAC", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
Ok(())
}
assert_eq!(act.len(), 50);
if act[0] != 0 {
- return Err(HandleError{err: "Unknown handshake version number", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "Unknown handshake version number", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
let their_pub = match PublicKey::from_slice(secp_ctx, &act[1..34]) {
- Err(_) => return Err(HandleError{err: "Invalid public key", action: Some(msgs::ErrorAction::DisconnectPeer{})}),
+ Err(_) => return Err(HandleError{err: "Invalid public key", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })}),
Ok(key) => key,
};
panic!("Requested act at wrong step");
}
if act_three[0] != 0 {
- return Err(HandleError{err: "Unknown handshake version number", action: Some(msgs::ErrorAction::DisconnectPeer{})});
+ return Err(HandleError{err: "Unknown handshake version number", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })});
}
let mut their_node_id = [0; 33];
PeerChannelEncryptor::decrypt_with_ad(&mut their_node_id, 1, &temp_k2.unwrap(), &bidirectional_state.h, &act_three[1..50])?;
self.their_node_id = Some(match PublicKey::from_slice(&self.secp_ctx, &their_node_id) {
Ok(key) => key,
- Err(_) => return Err(HandleError{err: "Bad node_id from peer", action: Some(msgs::ErrorAction::DisconnectPeer{})}),
+ Err(_) => return Err(HandleError{err: "Bad node_id from peer", action: Some(msgs::ErrorAction::DisconnectPeer{ msg: None })}),
});
let mut sha = Sha256::new();
/// indicating that read events on this descriptor should resume. A resume_read of false does
/// *not* imply that further read events should be paused.
fn send_data(&mut self, data: &Vec<u8>, write_offset: usize, resume_read: bool) -> usize;
+ /// Disconnect the socket pointed to by this SocketDescriptor. Once this function returns, no
+ /// more calls to write_event, read_event or disconnect_event may be made with this descriptor.
+ /// No disconnect_event should be generated as a result of this call, though obviously races
+ /// may occur whereby disconnect_socket is called after a call to disconnect_event but prior to
+ /// that event completing.
+ fn disconnect_socket(&mut self);
}
/// Error for PeerManager errors. If you get one of these, you must disconnect the socket and
encode_and_send_msg!(msg, 131);
continue;
},
- msgs::ErrorAction::DisconnectPeer => {
+ msgs::ErrorAction::DisconnectPeer { msg: _ } => {
return Err(PeerHandleError{ no_connection_possible: false });
},
msgs::ErrorAction::IgnoreError => {
}
continue;
},
+ Event::DisconnectPeer { ref node_id, ref msg } => {
+ if let Some(mut descriptor) = peers.node_id_to_descriptor.remove(node_id) {
+ if let Some(mut peer) = peers.peers.remove(&descriptor) {
+ if let Some(ref msg) = *msg {
+ peer.pending_outbound_buffer.push_back(peer.channel_encryptor.encrypt_message(&encode_msg!(msg, 17)));
+ // This isn't guaranteed to work, but if there is enough free
+ // room in the send buffer, put the error message there...
+ Self::do_attempt_write_data(&mut descriptor, &mut peer);
+ }
+ }
+ descriptor.disconnect_socket();
+ self.message_handler.chan_handler.peer_disconnected(&node_id, false);
+ }
+ },
}
upstream_events.push(event);
ret
}
}
+
+#[cfg(test)]
+mod tests {
+ use ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor};
+ use util::events;
+ use util::test_utils;
+
+ use secp256k1::Secp256k1;
+ use secp256k1::key::{SecretKey, PublicKey};
+
+ use rand::{thread_rng, Rng};
+
+ use std::sync::{Arc};
+
+ #[derive(PartialEq, Eq, Clone, Hash)]
+ struct FileDescriptor {
+ fd: u16,
+ }
+
+ impl SocketDescriptor for FileDescriptor {
+ fn send_data(&mut self, data: &Vec<u8>, write_offset: usize, _resume_read: bool) -> usize {
+ assert!(write_offset < data.len());
+ data.len() - write_offset
+ }
+
+ fn disconnect_socket(&mut self) {}
+ }
+
+ fn create_network(peer_count: usize) -> Vec<PeerManager<FileDescriptor>> {
+ let secp_ctx = Secp256k1::new();
+ let mut peers = Vec::new();
+ let mut rng = thread_rng();
+
+ for _ in 0..peer_count {
+ let chan_handler = test_utils::TestChannelMessageHandler::new();
+ let router = test_utils::TestRoutingMessageHandler::new();
+ let node_id = {
+ let mut key_slice = [0;32];
+ rng.fill_bytes(&mut key_slice);
+ SecretKey::from_slice(&secp_ctx, &key_slice).unwrap()
+ };
+ let msg_handler = MessageHandler { chan_handler: Arc::new(chan_handler), route_handler: Arc::new(router) };
+ let peer = PeerManager::new(msg_handler, node_id);
+ peers.push(peer);
+ }
+
+ peers
+ }
+
+ fn establish_connection(peer_a: &PeerManager<FileDescriptor>, peer_b: &PeerManager<FileDescriptor>) {
+ let secp_ctx = Secp256k1::new();
+ let their_id = PublicKey::from_secret_key(&secp_ctx, &peer_b.our_node_secret).unwrap();
+ let fd = FileDescriptor { fd: 1};
+ peer_a.new_inbound_connection(fd.clone()).unwrap();
+ peer_a.peers.lock().unwrap().node_id_to_descriptor.insert(their_id, fd.clone());
+ }
+
+ #[test]
+ fn test_disconnect_peer() {
+ // Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and
+ // push an DisconnectPeer event to remove the node flagged by id
+ let mut peers = create_network(2);
+ establish_connection(&peers[0], &peers[1]);
+ assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 1);
+
+ let secp_ctx = Secp256k1::new();
+ let their_id = PublicKey::from_secret_key(&secp_ctx, &peers[1].our_node_secret).unwrap();
+
+ let chan_handler = test_utils::TestChannelMessageHandler::new();
+ chan_handler.pending_events.lock().unwrap().push(events::Event::DisconnectPeer {
+ node_id: their_id,
+ msg: None,
+ });
+ assert_eq!(chan_handler.pending_events.lock().unwrap().len(), 1);
+ peers[0].message_handler.chan_handler = Arc::new(chan_handler);
+
+ peers[0].process_events();
+ assert_eq!(peers[0].peers.lock().unwrap().peers.len(), 0);
+ }
+}
BroadcastChannelUpdate {
msg: msgs::ChannelUpdate,
},
+
+ // Events indicating the network loop should change the state of connection with peer:
+ /// Disconnect the given peer, possibly making an attempt to send an ErrorMessage first.
+ DisconnectPeer {
+ node_id: PublicKey,
+ msg: Option<msgs::ErrorMessage>,
+ }
}
pub trait EventsProvider {
use chain::chaininterface::ConfirmationTarget;
use chain::transaction::OutPoint;
use ln::channelmonitor;
+use ln::msgs;
+use ln::msgs::{HandleError};
+use util::events;
use bitcoin::blockdata::transaction::Transaction;
+use secp256k1::PublicKey;
+
use std::sync::{Arc,Mutex};
+use std::{mem};
pub struct TestFeeEstimator {
pub sat_per_vbyte: u64,
self.txn_broadcasted.lock().unwrap().push(tx.clone());
}
}
+
+pub struct TestChannelMessageHandler {
+ pub pending_events: Mutex<Vec<events::Event>>,
+}
+
+impl TestChannelMessageHandler {
+ pub fn new() -> Self {
+ TestChannelMessageHandler {
+ pending_events: Mutex::new(Vec::new()),
+ }
+ }
+}
+
+impl msgs::ChannelMessageHandler for TestChannelMessageHandler {
+
+ fn handle_open_channel(&self, _their_node_id: &PublicKey, _msg: &msgs::OpenChannel) -> Result<msgs::AcceptChannel, HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_accept_channel(&self, _their_node_id: &PublicKey, _msg: &msgs::AcceptChannel) -> Result<(), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_funding_created(&self, _their_node_id: &PublicKey, _msg: &msgs::FundingCreated) -> Result<msgs::FundingSigned, HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_funding_signed(&self, _their_node_id: &PublicKey, _msg: &msgs::FundingSigned) -> Result<(), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_funding_locked(&self, _their_node_id: &PublicKey, _msg: &msgs::FundingLocked) -> Result<Option<msgs::AnnouncementSignatures>, HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_shutdown(&self, _their_node_id: &PublicKey, _msg: &msgs::Shutdown) -> Result<(Option<msgs::Shutdown>, Option<msgs::ClosingSigned>), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_closing_signed(&self, _their_node_id: &PublicKey, _msg: &msgs::ClosingSigned) -> Result<Option<msgs::ClosingSigned>, HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_update_add_htlc(&self, _their_node_id: &PublicKey, _msg: &msgs::UpdateAddHTLC) -> Result<(), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_update_fulfill_htlc(&self, _their_node_id: &PublicKey, _msg: &msgs::UpdateFulfillHTLC) -> Result<(), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_update_fail_htlc(&self, _their_node_id: &PublicKey, _msg: &msgs::UpdateFailHTLC) -> Result<Option<msgs::HTLCFailChannelUpdate>, HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_update_fail_malformed_htlc(&self, _their_node_id: &PublicKey, _msg: &msgs::UpdateFailMalformedHTLC) -> Result<(), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_commitment_signed(&self, _their_node_id: &PublicKey, _msg: &msgs::CommitmentSigned) -> Result<(msgs::RevokeAndACK, Option<msgs::CommitmentSigned>), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_revoke_and_ack(&self, _their_node_id: &PublicKey, _msg: &msgs::RevokeAndACK) -> Result<Option<msgs::CommitmentUpdate>, HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_update_fee(&self, _their_node_id: &PublicKey, _msg: &msgs::UpdateFee) -> Result<(), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &msgs::AnnouncementSignatures) -> Result<(), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn peer_disconnected(&self, _their_node_id: &PublicKey, _no_connection_possible: bool) {}
+}
+
+impl events::EventsProvider for TestChannelMessageHandler {
+ fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
+ let mut pending_events = self.pending_events.lock().unwrap();
+ let mut ret = Vec::new();
+ mem::swap(&mut ret, &mut *pending_events);
+ ret
+ }
+}
+
+pub struct TestRoutingMessageHandler {}
+
+impl TestRoutingMessageHandler {
+ pub fn new() -> Self {
+ TestRoutingMessageHandler {}
+ }
+}
+
+impl msgs::RoutingMessageHandler for TestRoutingMessageHandler {
+ fn handle_node_announcement(&self, _msg: &msgs::NodeAnnouncement) -> Result<(), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_channel_announcement(&self, _msg: &msgs::ChannelAnnouncement) -> Result<bool, HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_channel_update(&self, _msg: &msgs::ChannelUpdate) -> Result<(), HandleError> {
+ Err(HandleError { err: "", action: None })
+ }
+ fn handle_htlc_fail_channel_update(&self, _update: &msgs::HTLCFailChannelUpdate) {}
+}
projects / rust-lightning / commitdiff