Refuse to send and forward OMs to disconnected peers
authorValentine Wallace <vwallace@protonmail.com>
Thu, 4 Aug 2022 15:05:07 +0000 (11:05 -0400)
committerValentine Wallace <vwallace@protonmail.com>
Fri, 2 Sep 2022 20:27:30 +0000 (16:27 -0400)
We also refuse to connect to peers that don't advertise onion message
forwarding support.

lightning/src/ln/features.rs
lightning/src/ln/msgs.rs
lightning/src/ln/peer_handler.rs
lightning/src/onion_message/functional_tests.rs
lightning/src/onion_message/messenger.rs

index dbc17a34138b04b59ca9207b574494c9a8485009..c978c61a3a563428e38f3368e557e544a6c623b9 100644 (file)
@@ -433,6 +433,11 @@ mod sealed {
        define_feature!(27, ShutdownAnySegwit, [InitContext, NodeContext],
                "Feature flags for `opt_shutdown_anysegwit`.", set_shutdown_any_segwit_optional,
                set_shutdown_any_segwit_required, supports_shutdown_anysegwit, requires_shutdown_anysegwit);
+       // We do not yet advertise the onion messages feature bit, but we need to detect when peers
+       // support it.
+       define_feature!(39, OnionMessages, [InitContext, NodeContext],
+               "Feature flags for `option_onion_messages`.", set_onion_messages_optional,
+               set_onion_messages_required, supports_onion_messages, requires_onion_messages);
        define_feature!(45, ChannelType, [InitContext, NodeContext],
                "Feature flags for `option_channel_type`.", set_channel_type_optional,
                set_channel_type_required, supports_channel_type, requires_channel_type);
@@ -767,7 +772,7 @@ impl<T: sealed::GossipQueries> Features<T> {
 
 impl<T: sealed::InitialRoutingSync> Features<T> {
        // We are no longer setting initial_routing_sync now that gossip_queries
-       // is enabled. This feature is ignored by a peer when gossip_queries has 
+       // is enabled. This feature is ignored by a peer when gossip_queries has
        // been negotiated.
        #[cfg(test)]
        pub(crate) fn clear_initial_routing_sync(&mut self) {
index 190907ce26e3a9f01edcc613d1112086692c6567..a32b17b9f0d331c50780bec4963ff4e17c3d4306 100644 (file)
@@ -949,6 +949,12 @@ pub trait RoutingMessageHandler : MessageSendEventsProvider {
 pub trait OnionMessageHandler : OnionMessageProvider {
        /// Handle an incoming onion_message message from the given peer.
        fn handle_onion_message(&self, peer_node_id: &PublicKey, msg: &OnionMessage);
+       /// Called when a connection is established with a peer. Can be used to track which peers
+       /// advertise onion message support and are online.
+       fn peer_connected(&self, their_node_id: &PublicKey, init: &Init);
+       /// Indicates a connection to the peer failed/an existing connection was lost. Allows handlers to
+       /// drop and refuse to forward onion messages to this peer.
+       fn peer_disconnected(&self, their_node_id: &PublicKey, no_connection_possible: bool);
 }
 
 mod fuzzy_internal_msgs {
index 1258026e17ec123f9e28b9047dbf65b225a8f3db..f80c8984c1cb0437831e70510c6d9fa0e4e7e0fe 100644 (file)
@@ -81,6 +81,8 @@ impl OnionMessageProvider for IgnoringMessageHandler {
 }
 impl OnionMessageHandler for IgnoringMessageHandler {
        fn handle_onion_message(&self, _their_node_id: &PublicKey, _msg: &msgs::OnionMessage) {}
+       fn peer_connected(&self, _their_node_id: &PublicKey, _init: &msgs::Init) {}
+       fn peer_disconnected(&self, _their_node_id: &PublicKey, _no_connection_possible: bool) {}
 }
 impl Deref for IgnoringMessageHandler {
        type Target = IgnoringMessageHandler;
@@ -1173,8 +1175,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
                        }
 
                        self.message_handler.route_handler.peer_connected(&their_node_id, &msg);
-
                        self.message_handler.chan_handler.peer_connected(&their_node_id, &msg);
+                       self.message_handler.onion_message_handler.peer_connected(&their_node_id, &msg);
+
                        peer_lock.their_features = Some(msg.features);
                        return Ok(None);
                } else if peer_lock.their_features.is_none() {
@@ -1729,6 +1732,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
                                        }
                                        descriptor.disconnect_socket();
                                        self.message_handler.chan_handler.peer_disconnected(&node_id, false);
+                                       self.message_handler.onion_message_handler.peer_disconnected(&node_id, false);
                                }
                        }
                }
@@ -1756,6 +1760,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
                                                log_pubkey!(node_id), if no_connection_possible { "no " } else { "" });
                                        self.node_id_to_descriptor.lock().unwrap().remove(&node_id);
                                        self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible);
+                                       self.message_handler.onion_message_handler.peer_disconnected(&node_id, no_connection_possible);
                                }
                        }
                };
@@ -1776,6 +1781,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
                        log_trace!(self.logger, "Disconnecting peer with id {} due to client request", node_id);
                        peers_lock.remove(&descriptor);
                        self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible);
+                       self.message_handler.onion_message_handler.peer_disconnected(&node_id, no_connection_possible);
                        descriptor.disconnect_socket();
                }
        }
@@ -1791,6 +1797,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
                        if let Some(node_id) = peer.lock().unwrap().their_node_id {
                                log_trace!(self.logger, "Disconnecting peer with id {} due to client request to disconnect all peers", node_id);
                                self.message_handler.chan_handler.peer_disconnected(&node_id, false);
+                               self.message_handler.onion_message_handler.peer_disconnected(&node_id, false);
                        }
                        descriptor.disconnect_socket();
                }
@@ -1881,6 +1888,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
                                                        log_trace!(self.logger, "Disconnecting peer with id {} due to ping timeout", node_id);
                                                        self.node_id_to_descriptor.lock().unwrap().remove(&node_id);
                                                        self.message_handler.chan_handler.peer_disconnected(&node_id, false);
+                                                       self.message_handler.onion_message_handler.peer_disconnected(&node_id, false);
                                                }
                                        }
                                }
index 001d96e834264c68ae8e467b8a024e471d923dcc..b18431a66dd7755d6e7c1aabf1b3e98f0318f9a7 100644 (file)
 //! Onion message testing and test utilities live here.
 
 use chain::keysinterface::{KeysInterface, Recipient};
-use ln::msgs::OnionMessageHandler;
+use ln::features::InitFeatures;
+use ln::msgs::{self, OnionMessageHandler};
 use super::{BlindedRoute, Destination, OnionMessenger, SendError};
 use util::enforcing_trait_impls::EnforcingSigner;
 use util::test_utils;
 
 use bitcoin::network::constants::Network;
-use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey};
+use bitcoin::secp256k1::{PublicKey, Secp256k1};
 
 use sync::Arc;
 
@@ -34,18 +35,26 @@ impl MessengerNode {
 }
 
 fn create_nodes(num_messengers: u8) -> Vec<MessengerNode> {
-       let mut res = Vec::new();
+       let mut nodes = Vec::new();
        for i in 0..num_messengers {
                let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i)));
                let seed = [i as u8; 32];
                let keys_manager = Arc::new(test_utils::TestKeysInterface::new(&seed, Network::Testnet));
-               res.push(MessengerNode {
+               nodes.push(MessengerNode {
                        keys_manager: keys_manager.clone(),
                        messenger: OnionMessenger::new(keys_manager, logger.clone()),
                        logger,
                });
        }
-       res
+       for idx in 0..num_messengers - 1 {
+               let i = idx as usize;
+               let mut features = InitFeatures::known();
+               features.set_onion_messages_optional();
+               let init_msg = msgs::Init { features, remote_network_address: None };
+               nodes[i].messenger.peer_connected(&nodes[i + 1].get_node_pk(), &init_msg.clone());
+               nodes[i + 1].messenger.peer_connected(&nodes[i].get_node_pk(), &init_msg.clone());
+       }
+       nodes
 }
 
 fn pass_along_path(path: &Vec<MessengerNode>, expected_path_id: Option<[u8; 32]>) {
@@ -53,7 +62,6 @@ fn pass_along_path(path: &Vec<MessengerNode>, expected_path_id: Option<[u8; 32]>
        let num_nodes = path.len();
        for (idx, node) in path.into_iter().skip(1).enumerate() {
                let events = prev_node.messenger.release_pending_msgs();
-               assert_eq!(events.len(), 1);
                let onion_msg =  {
                        let msgs = events.get(&node.get_node_pk()).unwrap();
                        assert_eq!(msgs.len(), 1);
@@ -110,12 +118,9 @@ fn three_blinded_hops() {
 #[test]
 fn too_big_packet_error() {
        // Make sure we error as expected if a packet is too big to send.
-       let nodes = create_nodes(1);
-
-       let hop_secret = SecretKey::from_slice(&hex::decode("0101010101010101010101010101010101010101010101010101010101010101").unwrap()[..]).unwrap();
-       let secp_ctx = Secp256k1::new();
-       let hop_node_id = PublicKey::from_secret_key(&secp_ctx, &hop_secret);
+       let nodes = create_nodes(2);
 
+       let hop_node_id = nodes[1].get_node_pk();
        let hops = [hop_node_id; 400];
        let err = nodes[0].messenger.send_onion_message(&hops, Destination::Node(hop_node_id), None).unwrap_err();
        assert_eq!(err, SendError::TooBigPacket);
index 3a14c78d0fba724a6517776352d37fdefc6244bf..b9da9246cb103bcf44eb82050bd5870f45521ae5 100644 (file)
@@ -122,6 +122,8 @@ pub enum SendError {
        /// The provided [`Destination`] was an invalid [`BlindedRoute`], due to having fewer than two
        /// blinded hops.
        TooFewBlindedHops,
+       /// Our next-hop peer was offline or does not support onion message forwarding.
+       InvalidFirstHop,
 }
 
 impl<Signer: Sign, K: Deref, L: Deref> OnionMessenger<Signer, K, L>
@@ -165,25 +167,28 @@ impl<Signer: Sign, K: Deref, L: Deref> OnionMessenger<Signer, K, L>
                        .map_err(|e| SendError::Secp256k1(e))?;
 
                let prng_seed = self.keys_manager.get_secure_random_bytes();
-               let onion_packet = construct_onion_message_packet(
+               let onion_routing_packet = construct_onion_message_packet(
                        packet_payloads, packet_keys, prng_seed).map_err(|()| SendError::TooBigPacket)?;
 
                let mut pending_per_peer_msgs = self.pending_messages.lock().unwrap();
-               let pending_msgs = pending_per_peer_msgs.entry(introduction_node_id).or_insert_with(VecDeque::new);
-               pending_msgs.push_back(
-                       msgs::OnionMessage {
-                               blinding_point,
-                               onion_routing_packet: onion_packet,
+               match pending_per_peer_msgs.entry(introduction_node_id) {
+                       hash_map::Entry::Vacant(_) => Err(SendError::InvalidFirstHop),
+                       hash_map::Entry::Occupied(mut e) => {
+                               e.get_mut().push_back(msgs::OnionMessage { blinding_point, onion_routing_packet });
+                               Ok(())
                        }
-               );
-               Ok(())
+               }
        }
 
        #[cfg(test)]
        pub(super) fn release_pending_msgs(&self) -> HashMap<PublicKey, VecDeque<msgs::OnionMessage>> {
                let mut pending_msgs = self.pending_messages.lock().unwrap();
                let mut msgs = HashMap::new();
-               core::mem::swap(&mut *pending_msgs, &mut msgs);
+               // We don't want to disconnect the peers by removing them entirely from the original map, so we
+               // swap the pending message buffers individually.
+               for (peer_node_id, pending_messages) in &mut *pending_msgs {
+                       msgs.insert(*peer_node_id, core::mem::take(pending_messages));
+               }
                msgs
        }
 }
@@ -252,32 +257,43 @@ impl<Signer: Sign, K: Deref, L: Deref> OnionMessageHandler for OnionMessenger<Si
                                };
 
                                let mut pending_per_peer_msgs = self.pending_messages.lock().unwrap();
-                               let pending_msgs = pending_per_peer_msgs.entry(next_node_id).or_insert_with(VecDeque::new);
-                               pending_msgs.push_back(
-                                       msgs::OnionMessage {
-                                               blinding_point: match next_blinding_override {
-                                                       Some(blinding_point) => blinding_point,
-                                                       None => {
-                                                               let blinding_factor = {
-                                                                       let mut sha = Sha256::engine();
-                                                                       sha.input(&msg.blinding_point.serialize()[..]);
-                                                                       sha.input(control_tlvs_ss.as_ref());
-                                                                       Sha256::from_engine(sha).into_inner()
-                                                               };
-                                                               let next_blinding_point = msg.blinding_point;
-                                                               match next_blinding_point.mul_tweak(&self.secp_ctx, &Scalar::from_be_bytes(blinding_factor).unwrap()) {
-                                                                       Ok(bp) => bp,
-                                                                       Err(e) => {
-                                                                               log_trace!(self.logger, "Failed to compute next blinding point: {}", e);
-                                                                               return
-                                                                       }
-                                                               }
-                                                       },
-                                               },
-                                               onion_routing_packet: outgoing_packet,
+
+                               #[cfg(fuzzing)]
+                               pending_per_peer_msgs.entry(next_node_id).or_insert_with(VecDeque::new);
+
+                               match pending_per_peer_msgs.entry(next_node_id) {
+                                       hash_map::Entry::Vacant(_) => {
+                                               log_trace!(self.logger, "Dropping forwarded onion message to disconnected peer {:?}", next_node_id);
+                                               return
                                        },
-                               );
-                               log_trace!(self.logger, "Forwarding an onion message to peer {}", next_node_id);
+                                       hash_map::Entry::Occupied(mut e) => {
+                                               e.get_mut().push_back(
+                                                       msgs::OnionMessage {
+                                                               blinding_point: match next_blinding_override {
+                                                                       Some(blinding_point) => blinding_point,
+                                                                       None => {
+                                                                               let blinding_factor = {
+                                                                                       let mut sha = Sha256::engine();
+                                                                                       sha.input(&msg.blinding_point.serialize()[..]);
+                                                                                       sha.input(control_tlvs_ss.as_ref());
+                                                                                       Sha256::from_engine(sha).into_inner()
+                                                                               };
+                                                                               let next_blinding_point = msg.blinding_point;
+                                                                               match next_blinding_point.mul_tweak(&self.secp_ctx, &Scalar::from_be_bytes(blinding_factor).unwrap()) {
+                                                                                       Ok(bp) => bp,
+                                                                                       Err(e) => {
+                                                                                               log_trace!(self.logger, "Failed to compute next blinding point: {}", e);
+                                                                                               return
+                                                                                       }
+                                                                               }
+                                                                       },
+                                                               },
+                                                               onion_routing_packet: outgoing_packet,
+                                                       },
+                                               );
+                                               log_trace!(self.logger, "Forwarding an onion message to peer {}", next_node_id);
+                                       }
+                               };
                        },
                        Err(e) => {
                                log_trace!(self.logger, "Errored decoding onion message packet: {:?}", e);
@@ -287,6 +303,18 @@ impl<Signer: Sign, K: Deref, L: Deref> OnionMessageHandler for OnionMessenger<Si
                        },
                };
        }
+
+       fn peer_connected(&self, their_node_id: &PublicKey, init: &msgs::Init) {
+               if init.features.supports_onion_messages() {
+                       let mut peers = self.pending_messages.lock().unwrap();
+                       peers.insert(their_node_id.clone(), VecDeque::new());
+               }
+       }
+
+       fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) {
+               let mut pending_msgs = self.pending_messages.lock().unwrap();
+               pending_msgs.remove(their_node_id);
+       }
 }
 
 impl<Signer: Sign, K: Deref, L: Deref> OnionMessageProvider for OnionMessenger<Signer, K, L>