]> git.bitcoin.ninja Git - rust-lightning/commitdiff
Add unknown message handler to peer manager 2021-07-custom-msg-suggestion
authorTibo-lg <thibaut@cryptogarage.co.jp>
Wed, 14 Jul 2021 02:44:36 +0000 (11:44 +0900)
committerMatt Corallo <git@bluematt.me>
Tue, 3 Aug 2021 15:33:11 +0000 (15:33 +0000)
fuzz/src/full_stack.rs
lightning-background-processor/src/lib.rs
lightning-net-tokio/src/lib.rs
lightning/src/ln/peer_handler.rs
lightning/src/ln/wire.rs
lightning/src/util/events.rs

index f68cc8f3df7a16bf43d81993c2cff9ad1f963793..b57869871320d2d36a59b42d89570b8341ab7d47 100644 (file)
@@ -34,7 +34,7 @@ use lightning::chain::transaction::OutPoint;
 use lightning::chain::keysinterface::{InMemorySigner, KeysInterface};
 use lightning::ln::{PaymentHash, PaymentPreimage, PaymentSecret};
 use lightning::ln::channelmanager::{ChainParameters, ChannelManager};
-use lightning::ln::peer_handler::{MessageHandler,PeerManager,SocketDescriptor};
+use lightning::ln::peer_handler::{MessageHandler,PeerManager,SocketDescriptor,IgnoringUnknownMessageHandler};
 use lightning::ln::msgs::DecodeError;
 use lightning::routing::router::get_route;
 use lightning::routing::network_graph::NetGraphMsgHandler;
@@ -159,7 +159,7 @@ type ChannelMan = ChannelManager<
        EnforcingSigner,
        Arc<chainmonitor::ChainMonitor<EnforcingSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
        Arc<TestBroadcaster>, Arc<KeyProvider>, Arc<FuzzEstimator>, Arc<dyn Logger>>;
-type PeerMan<'a> = PeerManager<Peer<'a>, Arc<ChannelMan>, Arc<NetGraphMsgHandler<Arc<dyn chain::Access>, Arc<dyn Logger>>>, Arc<dyn Logger>>;
+type PeerMan<'a> = PeerManager<Peer<'a>, Arc<ChannelMan>, Arc<NetGraphMsgHandler<Arc<dyn chain::Access>, Arc<dyn Logger>>>, Arc<dyn Logger>, Arc<IgnoringUnknownMessageHandler>>;
 
 struct MoneyLossDetector<'a> {
        manager: Arc<ChannelMan>,
@@ -374,7 +374,7 @@ pub fn do_test(data: &[u8], logger: &Arc<dyn Logger>) {
        let mut loss_detector = MoneyLossDetector::new(&peers, channelmanager.clone(), monitor.clone(), PeerManager::new(MessageHandler {
                chan_handler: channelmanager.clone(),
                route_handler: net_graph_msg_handler.clone(),
-       }, our_network_key, &[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 15, 0], Arc::clone(&logger)));
+       }, our_network_key, &[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 15, 0], Arc::clone(&logger), Arc::new(IgnoringUnknownMessageHandler{})));
 
        let mut should_forward = false;
        let mut payments_received: Vec<PaymentHash> = Vec::new();
index 3a36bb5625eab033b05c5aec3cfa183eb2d67a7d..93d259a51ff3f90f2de92a36930df443eb876fcb 100644 (file)
@@ -15,7 +15,7 @@ use lightning::chain::channelmonitor;
 use lightning::chain::keysinterface::{Sign, KeysInterface};
 use lightning::ln::channelmanager::ChannelManager;
 use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler};
-use lightning::ln::peer_handler::{PeerManager, SocketDescriptor};
+use lightning::ln::peer_handler::{PeerManager, SocketDescriptor, UnknownMessageHandler};
 use lightning::util::events::{EventHandler, EventsProvider};
 use lightning::util::logger::Logger;
 use std::sync::Arc;
@@ -118,7 +118,8 @@ impl BackgroundProcessor {
                CMP: 'static + Send + ChannelManagerPersister<Signer, CW, T, K, F, L>,
                M: 'static + Deref<Target = ChainMonitor<Signer, CF, T, F, L, P>> + Send + Sync,
                CM: 'static + Deref<Target = ChannelManager<Signer, CW, T, K, F, L>> + Send + Sync,
-               PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, L>> + Send + Sync,
+               UMH: 'static + Deref + Send + Sync,
+               PM: 'static + Deref<Target = PeerManager<Descriptor, CMH, RMH, L, UMH>> + Send + Sync,
        >
        (persister: CMP, event_handler: EH, chain_monitor: M, channel_manager: CM, peer_manager: PM, logger: L) -> Self
        where
@@ -131,6 +132,7 @@ impl BackgroundProcessor {
                P::Target: 'static + channelmonitor::Persist<Signer>,
                CMH::Target: 'static + ChannelMessageHandler,
                RMH::Target: 'static + RoutingMessageHandler,
+               UMH::Target: 'static + UnknownMessageHandler,
        {
                let stop_thread = Arc::new(AtomicBool::new(false));
                let stop_thread_clone = stop_thread.clone();
@@ -182,7 +184,7 @@ mod tests {
        use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, ChannelManager, SimpleArcChannelManager};
        use lightning::ln::features::InitFeatures;
        use lightning::ln::msgs::ChannelMessageHandler;
-       use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor};
+       use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringUnknownMessageHandler};
        use lightning::util::config::UserConfig;
        use lightning::util::events::{Event, MessageSendEventsProvider, MessageSendEvent};
        use lightning::util::ser::Writeable;
@@ -210,7 +212,7 @@ mod tests {
 
        struct Node {
                node: Arc<SimpleArcChannelManager<ChainMonitor, test_utils::TestBroadcaster, test_utils::TestFeeEstimator, test_utils::TestLogger>>,
-               peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, Arc<test_utils::TestLogger>>>,
+               peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, Arc<test_utils::TestLogger>, Arc<IgnoringUnknownMessageHandler>>>,
                chain_monitor: Arc<ChainMonitor>,
                persister: Arc<FilesystemPersister>,
                tx_broadcaster: Arc<test_utils::TestBroadcaster>,
@@ -251,7 +253,7 @@ mod tests {
                        let params = ChainParameters { network, best_block };
                        let manager = Arc::new(ChannelManager::new(fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster.clone(), logger.clone(), keys_manager.clone(), UserConfig::default(), params));
                        let msg_handler = MessageHandler { chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new()), route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new() )};
-                       let peer_manager = Arc::new(PeerManager::new(msg_handler, keys_manager.get_node_secret(), &seed, logger.clone()));
+                       let peer_manager = Arc::new(PeerManager::new(msg_handler, keys_manager.get_node_secret(), &seed, logger.clone(), Arc::new(IgnoringUnknownMessageHandler {})));
                        let node = Node { node: manager, peer_manager, chain_monitor, persister, tx_broadcaster, logger, best_block };
                        nodes.push(node);
                }
index 5f5fece0d2ed34e447e2905f9f3bd6d3f42ce102..68a25a75ef39fa33813f55343aab8407b20813fd 100644 (file)
@@ -37,7 +37,7 @@
 //! type DataPersister = dyn lightning::chain::channelmonitor::Persist<lightning::chain::keysinterface::InMemorySigner> + Send + Sync;
 //! type ChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::chain::keysinterface::InMemorySigner, Arc<ChainFilter>, Arc<TxBroadcaster>, Arc<FeeEstimator>, Arc<Logger>, Arc<DataPersister>>;
 //! type ChannelManager = Arc<lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor, TxBroadcaster, FeeEstimator, Logger>>;
-//! type PeerManager = Arc<lightning::ln::peer_handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChainMonitor, TxBroadcaster, FeeEstimator, ChainAccess, Logger>>;
+//! type PeerManager = Arc<lightning::ln::peer_handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChainMonitor, TxBroadcaster, FeeEstimator, ChainAccess, Logger, Arc<lightning::ln::peer_handler::IgnoringUnknownMessageHandler>>>;
 //!
 //! // Connect to node with pubkey their_node_id at addr:
 //! async fn connect_to_node(peer_manager: PeerManager, chain_monitor: Arc<ChainMonitor>, channel_manager: ChannelManager, their_node_id: PublicKey, addr: SocketAddr) {
@@ -80,6 +80,7 @@ use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
 
 use lightning::ln::peer_handler;
 use lightning::ln::peer_handler::SocketDescriptor as LnSocketTrait;
+use lightning::ln::peer_handler::UnknownMessageHandler;
 use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler};
 use lightning::util::logger::Logger;
 
@@ -119,10 +120,11 @@ struct Connection {
        id: u64,
 }
 impl Connection {
-       async fn schedule_read<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
+       async fn schedule_read<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
                        CMH: ChannelMessageHandler + 'static,
                        RMH: RoutingMessageHandler + 'static,
-                       L: Logger + 'static + ?Sized {
+                       L: Logger + 'static + ?Sized,
+                       UMH: UnknownMessageHandler + 'static {
                // 8KB is nice and big but also should never cause any issues with stack overflowing.
                let mut buf = [0; 8192];
 
@@ -222,10 +224,11 @@ impl Connection {
 /// The returned future will complete when the peer is disconnected and associated handling
 /// futures are freed, though, because all processing futures are spawned with tokio::spawn, you do
 /// not need to poll the provided future in order to make progress.
-pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
+pub fn setup_inbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
                CMH: ChannelMessageHandler + 'static + Send + Sync,
                RMH: RoutingMessageHandler + 'static + Send + Sync,
-               L: Logger + 'static + ?Sized + Send + Sync {
+               L: Logger + 'static + ?Sized + Send + Sync,
+               UMH: UnknownMessageHandler + 'static + Send + Sync {
        let (reader, write_receiver, read_receiver, us) = Connection::new(stream);
        #[cfg(debug_assertions)]
        let last_us = Arc::clone(&us);
@@ -262,10 +265,11 @@ pub fn setup_inbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<So
 /// The returned future will complete when the peer is disconnected and associated handling
 /// futures are freed, though, because all processing futures are spawned with tokio::spawn, you do
 /// not need to poll the provided future in order to make progress.
-pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
+pub fn setup_outbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, their_node_id: PublicKey, stream: StdTcpStream) -> impl std::future::Future<Output=()> where
                CMH: ChannelMessageHandler + 'static + Send + Sync,
                RMH: RoutingMessageHandler + 'static + Send + Sync,
-               L: Logger + 'static + ?Sized + Send + Sync {
+               L: Logger + 'static + ?Sized + Send + Sync,
+               UMH: UnknownMessageHandler + 'static + Send + Sync {
        let (reader, mut write_receiver, read_receiver, us) = Connection::new(stream);
        #[cfg(debug_assertions)]
        let last_us = Arc::clone(&us);
@@ -332,10 +336,11 @@ pub fn setup_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<S
 /// disconnected and associated handling futures are freed, though, because all processing in said
 /// futures are spawned with tokio::spawn, you do not need to poll the second future in order to
 /// make progress.
-pub async fn connect_outbound<CMH, RMH, L>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>>>, their_node_id: PublicKey, addr: SocketAddr) -> Option<impl std::future::Future<Output=()>> where
+pub async fn connect_outbound<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, their_node_id: PublicKey, addr: SocketAddr) -> Option<impl std::future::Future<Output=()>> where
                CMH: ChannelMessageHandler + 'static + Send + Sync,
                RMH: RoutingMessageHandler + 'static + Send + Sync,
-               L: Logger + 'static + ?Sized + Send + Sync {
+               L: Logger + 'static + ?Sized + Send + Sync,
+               UMH: UnknownMessageHandler + 'static + Send + Sync {
        if let Ok(Ok(stream)) = time::timeout(Duration::from_secs(10), async { TcpStream::connect(&addr).await.map(|s| s.into_std().unwrap()) }).await {
                Some(setup_outbound(peer_manager, their_node_id, stream))
        } else { None }
@@ -467,7 +472,7 @@ impl Hash for SocketDescriptor {
 mod tests {
        use lightning::ln::features::*;
        use lightning::ln::msgs::*;
-       use lightning::ln::peer_handler::{MessageHandler, PeerManager};
+       use lightning::ln::peer_handler::{MessageHandler, PeerManager, IgnoringUnknownMessageHandler};
        use lightning::util::events::*;
        use bitcoin::secp256k1::{Secp256k1, SecretKey, PublicKey};
 
@@ -563,7 +568,7 @@ mod tests {
                let a_manager = Arc::new(PeerManager::new(MessageHandler {
                        chan_handler: Arc::clone(&a_handler),
                        route_handler: Arc::clone(&a_handler),
-               }, a_key.clone(), &[1; 32], Arc::new(TestLogger())));
+               }, a_key.clone(), &[1; 32], Arc::new(TestLogger()), Arc::new(IgnoringUnknownMessageHandler {})));
 
                let (b_connected_sender, mut b_connected) = mpsc::channel(1);
                let (b_disconnected_sender, mut b_disconnected) = mpsc::channel(1);
@@ -577,7 +582,7 @@ mod tests {
                let b_manager = Arc::new(PeerManager::new(MessageHandler {
                        chan_handler: Arc::clone(&b_handler),
                        route_handler: Arc::clone(&b_handler),
-               }, b_key.clone(), &[2; 32], Arc::new(TestLogger())));
+               }, b_key.clone(), &[2; 32], Arc::new(TestLogger()), Arc::new(IgnoringUnknownMessageHandler {})));
 
                // We bind on localhost, hoping the environment is properly configured with a local
                // address. This may not always be the case in containers and the like, so if this test is
index 96ec31c98ee1b4aa289f2ba78ee32c76a4e5f22d..5a47f39b591bf6b53f838b034d8a53457de5a274 100644 (file)
@@ -68,6 +68,47 @@ impl Deref for IgnoringMessageHandler {
        fn deref(&self) -> &Self { self }
 }
 
+/// A dummy implementation of `UnknownMessageHandler` that does nothing.
+pub struct IgnoringUnknownMessageHandler{}
+impl MessageSendEventsProvider for IgnoringUnknownMessageHandler {
+       fn get_and_clear_pending_msg_events(&self) -> Vec<MessageSendEvent> {
+               Vec::new()
+       }
+}
+
+/// Define a dummy type to satisfy the constraint of UnknownMessageHandle `Message`
+/// associated type for implementing it for IgnoringUnknownMessageHandler.
+type DummyType = ();
+impl Encode for DummyType {
+       const TYPE: u16 = 0;
+}
+impl Writeable for DummyType {
+       fn write<W: ::util::ser::Writer>(&self, _writer: &mut W) -> Result<(), ::std::io::Error> {
+               Ok(())
+       }
+}
+
+impl UnknownMessageHandler for IgnoringUnknownMessageHandler {
+       type MessageEnum = ();
+       type Message = DummyType;
+       fn read<R: ::std::io::Read>(&self, _message_type: u16, _buffer: &mut R) -> Result<Option<Self::MessageEnum>, msgs::DecodeError> {
+               Ok(None)
+       }
+
+       fn handle_unknown_message(&self, _msg: Self::MessageEnum) -> Result<(), MessageHandlingError> {
+               // Since we always return `None` in the read the handle method should never be called.
+               unreachable!();
+       }
+
+       fn get_and_clear_pending_msgs(&self) -> Vec<(&PublicKey, Self::Message)> {
+               Vec::new()
+       }
+}
+impl Deref for IgnoringUnknownMessageHandler {
+       type Target = IgnoringUnknownMessageHandler;
+       fn deref(&self) -> &Self { self }
+}
+
 /// A dummy struct which implements `ChannelMessageHandler` without having any channels.
 /// You can provide one of these as the route_handler in a MessageHandler.
 pub struct ErroringMessageHandler {
@@ -173,6 +214,23 @@ pub struct MessageHandler<CM: Deref, RM: Deref> where
        pub route_handler: RM,
 }
 
+/// Handler for messages external to the LN protocol.
+pub trait UnknownMessageHandler where Self::Message : Encode + Writeable + Debug {
+       /// A type that represents a message that can be sent over the wire
+       type Message;
+       /// A type that represents an enumeration of messages that can be handled by the handler.
+       type MessageEnum;
+       ///
+       fn read<R: ::std::io::Read>(&self, msg_type: u16, buffer: &mut R) -> Result<Option<Self::MessageEnum>, msgs::DecodeError>;
+       /// Called with the message type that was received and the buffer to be read. If the handler
+       /// could handle the message, should return `Ok(Some(wire::Message::HandledUnknownMessage(msg_type)))`,
+       /// otherwise Ok(None). Can also return a `DecodingError` if the buffer contained unexpected data
+       /// for the given message type.
+       fn handle_unknown_message(&self, msg: Self::MessageEnum) -> Result<(), MessageHandlingError>;
+       /// Get messages to be sent to specified peers.
+       fn get_and_clear_pending_msgs(&self) -> Vec<(&PublicKey, Self::Message)>;
+}
+
 /// Provides an object which can be used to send data to and which uniquely identifies a connection
 /// to a remote host. You will need to be able to generate multiple of these which meet Eq and
 /// implement Hash to meet the PeerManager API.
@@ -311,7 +369,7 @@ fn _check_usize_is_32_or_64() {
 /// lifetimes). Other times you can afford a reference, which is more efficient, in which case
 /// SimpleRefPeerManager is the more appropriate type. Defining these type aliases prevents
 /// issues such as overly long function definitions.
-pub type SimpleArcPeerManager<SD, M, T, F, C, L> = PeerManager<SD, Arc<SimpleArcChannelManager<M, T, F, L>>, Arc<NetGraphMsgHandler<Arc<C>, Arc<L>>>, Arc<L>>;
+pub type SimpleArcPeerManager<SD, M, T, F, C, L, UMH> = PeerManager<SD, Arc<SimpleArcChannelManager<M, T, F, L>>, Arc<NetGraphMsgHandler<Arc<C>, Arc<L>>>, Arc<L>, UMH>;
 
 /// SimpleRefPeerManager is a type alias for a PeerManager reference, and is the reference
 /// counterpart to the SimpleArcPeerManager type alias. Use this type by default when you don't
@@ -319,7 +377,7 @@ pub type SimpleArcPeerManager<SD, M, T, F, C, L> = PeerManager<SD, Arc<SimpleArc
 /// usage of lightning-net-tokio (since tokio::spawn requires parameters with static lifetimes).
 /// But if this is not necessary, using a reference is more efficient. Defining these type aliases
 /// helps with issues such as long function definitions.
-pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, SD, M, T, F, C, L> = PeerManager<SD, SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L>, &'e NetGraphMsgHandler<&'g C, &'f L>, &'f L>;
+pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, SD, M, T, F, C, L, UMH> = PeerManager<SD, SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, M, T, F, L>, &'e NetGraphMsgHandler<&'g C, &'f L>, &'f L, UMH>;
 
 /// A PeerManager manages a set of peers, described by their [`SocketDescriptor`] and marshalls
 /// socket events into messages which it passes on to its [`MessageHandler`].
@@ -340,14 +398,16 @@ pub type SimpleRefPeerManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, SD, M, T, F, C, L> = P
 /// you're using lightning-net-tokio.
 ///
 /// [`read_event`]: PeerManager::read_event
-pub struct PeerManager<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> where
+pub struct PeerManager<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, UMH: Deref> where
                CM::Target: ChannelMessageHandler,
                RM::Target: RoutingMessageHandler,
-               L::Target: Logger {
+               L::Target: Logger,
+               UMH::Target: UnknownMessageHandler {
        message_handler: MessageHandler<CM, RM>,
        peers: Mutex<PeerHolder<Descriptor>>,
        our_node_secret: SecretKey,
        ephemeral_key_midstate: Sha256Engine,
+       unknown_message_handler: UMH,
 
        // Usize needs to be at least 32 bits to avoid overflowing both low and high. If usize is 64
        // bits we will never realistically count into high:
@@ -357,8 +417,11 @@ pub struct PeerManager<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: De
        logger: L,
 }
 
-enum MessageHandlingError {
+/// An error indicating a failure to handle a received message.
+pub enum MessageHandlingError {
+       /// An error related to communication with a peer.
        PeerHandleError(PeerHandleError),
+       /// An error related to the LN protocol.
        LightningError(LightningError),
 }
 
@@ -382,7 +445,7 @@ macro_rules! encode_msg {
        }}
 }
 
-impl<Descriptor: SocketDescriptor, CM: Deref, L: Deref> PeerManager<Descriptor, CM, IgnoringMessageHandler, L> where
+impl<Descriptor: SocketDescriptor, CM: Deref, L: Deref> PeerManager<Descriptor, CM, IgnoringMessageHandler, L, IgnoringUnknownMessageHandler> where
                CM::Target: ChannelMessageHandler,
                L::Target: Logger {
        /// Constructs a new PeerManager with the given ChannelMessageHandler. No routing message
@@ -396,11 +459,11 @@ impl<Descriptor: SocketDescriptor, CM: Deref, L: Deref> PeerManager<Descriptor,
                Self::new(MessageHandler {
                        chan_handler: channel_message_handler,
                        route_handler: IgnoringMessageHandler{},
-               }, our_node_secret, ephemeral_random_data, logger)
+               }, our_node_secret, ephemeral_random_data, logger, IgnoringUnknownMessageHandler{})
        }
 }
 
-impl<Descriptor: SocketDescriptor, RM: Deref, L: Deref> PeerManager<Descriptor, ErroringMessageHandler, RM, L> where
+impl<Descriptor: SocketDescriptor, RM: Deref, L: Deref> PeerManager<Descriptor, ErroringMessageHandler, RM, L, IgnoringUnknownMessageHandler> where
                RM::Target: RoutingMessageHandler,
                L::Target: Logger {
        /// Constructs a new PeerManager with the given RoutingMessageHandler. No channel message
@@ -416,18 +479,19 @@ impl<Descriptor: SocketDescriptor, RM: Deref, L: Deref> PeerManager<Descriptor,
                Self::new(MessageHandler {
                        chan_handler: ErroringMessageHandler::new(),
                        route_handler: routing_message_handler,
-               }, our_node_secret, ephemeral_random_data, logger)
+               }, our_node_secret, ephemeral_random_data, logger, IgnoringUnknownMessageHandler{})
        }
 }
 
-impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<Descriptor, CM, RM, L> where
+impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, UMH: Deref> PeerManager<Descriptor, CM, RM, L, UMH> where
                CM::Target: ChannelMessageHandler,
                RM::Target: RoutingMessageHandler,
-               L::Target: Logger {
+               L::Target: Logger,
+               UMH::Target: UnknownMessageHandler {
        /// Constructs a new PeerManager with the given message handlers and node_id secret key
        /// ephemeral_random_data is used to derive per-connection ephemeral keys and must be
        /// cryptographically secure random bytes.
-       pub fn new(message_handler: MessageHandler<CM, RM>, our_node_secret: SecretKey, ephemeral_random_data: &[u8; 32], logger: L) -> Self {
+       pub fn new(message_handler: MessageHandler<CM, RM>, our_node_secret: SecretKey, ephemeral_random_data: &[u8; 32], logger: L, unknown_message_handler: UMH) -> Self {
                let mut ephemeral_key_midstate = Sha256::engine();
                ephemeral_key_midstate.input(ephemeral_random_data);
 
@@ -442,6 +506,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
                        peer_counter_low: AtomicUsize::new(0),
                        peer_counter_high: AtomicUsize::new(0),
                        logger,
+                       unknown_message_handler,
                }
        }
 
@@ -801,47 +866,81 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
                                                                                peer.pending_read_buffer = [0; 18].to_vec();
                                                                                peer.pending_read_is_header = true;
 
-                                                                               let mut reader = ::std::io::Cursor::new(&msg_data[..]);
-                                                                               let message_result = wire::read(&mut reader);
-                                                                               let message = match message_result {
-                                                                                       Ok(x) => x,
-                                                                                       Err(e) => {
-                                                                                               match e {
-                                                                                                       msgs::DecodeError::UnknownVersion => return Err(PeerHandleError { no_connection_possible: false }),
-                                                                                                       msgs::DecodeError::UnknownRequiredFeature => {
-                                                                                                               log_trace!(self.logger, "Got a channel/node announcement with an known required feature flag, you may want to update!");
-                                                                                                               continue;
-                                                                                                       }
-                                                                                                       msgs::DecodeError::InvalidValue => {
-                                                                                                               log_debug!(self.logger, "Got an invalid value while deserializing message");
-                                                                                                               return Err(PeerHandleError { no_connection_possible: false });
-                                                                                                       }
-                                                                                                       msgs::DecodeError::ShortRead => {
-                                                                                                               log_debug!(self.logger, "Deserialization failed due to shortness of message");
-                                                                                                               return Err(PeerHandleError { no_connection_possible: false });
-                                                                                                       }
-                                                                                                       msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError { no_connection_possible: false }),
-                                                                                                       msgs::DecodeError::Io(_) => return Err(PeerHandleError { no_connection_possible: false }),
-                                                                                                       msgs::DecodeError::UnsupportedCompression => {
-                                                                                                               log_trace!(self.logger, "We don't support zlib-compressed message fields, ignoring message");
-                                                                                                               continue;
+                                                                               let mut message_result = wire::read(&mut ::std::io::Cursor::new(&msg_data[..]));
+
+                                                                               // Need an Init as first message
+                                                                               if let Ok(wire::Message::Init(_)) = message_result {
+                                                                               } else if peer.their_features.is_none() {
+                                                                                       log_debug!(self.logger, "Peer {} sent non-Init first message", log_pubkey!(peer.their_node_id.unwrap()));
+                                                                                       return Err(PeerHandleError{ no_connection_possible: false }.into());
+                                                                               }
+
+                                                                               let mut message_err = Ok(());
+                                                                               let mut handle_err = Ok(());
+                                                                               match message_result {
+                                                                                       Ok(wire::Message::Unknown(msg_type)) => {
+                                                                                               let mut type_bytes = [0; 2];
+                                                                                               let mut reader = ::std::io::Cursor::new(&msg_data[..]);
+                                                                                               reader.read_exact(&mut type_bytes).expect("How did we read these to begin with?");
+                                                                                               match self.unknown_message_handler.read(*msg_type, &mut reader) {
+                                                                                                       Ok(Some(msg)) => {
+                                                                                                               handle_err = self.unknown_message_handler.handle_unknown_message(msg);
+                                                                                                       },
+                                                                                                       Ok(None) => {
+                                                                                                               if *msg_type % 2 == 0 {
+                                                                                                                       return Err(PeerHandleError { no_connection_possible: true });
+                                                                                                               }
                                                                                                        }
+                                                                                                       Err(e) => { message_err = Err(e); },
                                                                                                }
-                                                                                       }
-                                                                               };
-
-                                                                               match self.handle_message(peer, message) {
-                                                                                       Err(handling_error) => match handling_error {
-                                                                                               MessageHandlingError::PeerHandleError(e) => { return Err(e) },
-                                                                                               MessageHandlingError::LightningError(e) => {
-                                                                                                       try_potential_handleerror!(Err(e));
-                                                                                               },
                                                                                        },
-                                                                                       Ok(Some(msg)) => {
-                                                                                               peer_node_id = Some(peer.their_node_id.expect("After noise is complete, their_node_id is always set"));
-                                                                                               msgs_to_forward.push(msg);
+                                                                                       Ok(msg) => {
+                                                                                               match self.handle_message(peer, msg) {
+                                                                                                       Ok(Some(forward_msg)) => {
+                                                                                                               peer_node_id = Some(peer.their_node_id.expect("After noise is complete, their_node_id is always set"));
+                                                                                                               msgs_to_forward.push(forward_msg);
+                                                                                                       },
+                                                                                                       Ok(None) => {},
+                                                                                                       Err(e) => { handle_err = Err(e); },
+                                                                                               }
                                                                                        },
-                                                                                       Ok(None) => {},
+                                                                                       Err(e) => { message_err = Err(e); },
+                                                                               }
+
+                                                                               match message_err {
+                                                                                       Ok(()) => {},
+                                                                                       Err(e) => match e {
+                                                                                               msgs::DecodeError::UnknownVersion => return Err(PeerHandleError { no_connection_possible: false }),
+                                                                                               msgs::DecodeError::UnknownRequiredFeature => {
+                                                                                                       log_trace!(self.logger, "Got a channel/node announcement with an known required feature flag, you may want to update!");
+                                                                                                       continue;
+                                                                                               }
+                                                                                               msgs::DecodeError::InvalidValue => {
+                                                                                                       log_debug!(self.logger, "Got an invalid value while deserializing message");
+                                                                                                       return Err(PeerHandleError { no_connection_possible: false });
+                                                                                               }
+                                                                                               msgs::DecodeError::ShortRead => {
+                                                                                                       log_debug!(self.logger, "Deserialization failed due to shortness of message");
+                                                                                                       return Err(PeerHandleError { no_connection_possible: false });
+                                                                                               }
+                                                                                               msgs::DecodeError::BadLengthDescriptor => return Err(PeerHandleError { no_connection_possible: false }),
+                                                                                               msgs::DecodeError::Io(_) => return Err(PeerHandleError { no_connection_possible: false }),
+                                                                                               msgs::DecodeError::UnsupportedCompression => {
+                                                                                                       log_trace!(self.logger, "We don't support zlib-compressed message fields, ignoring message");
+                                                                                                       continue;
+                                                                                               }
+                                                                                       },
+                                                                               }
+
+                                                                               match handle_err {
+                                                                                       Ok(()) => {},
+                                                                                       Err(e) =>
+                                                                                               match e {
+                                                                                                       MessageHandlingError::PeerHandleError(e) => { return Err(e) },
+                                                                                                       MessageHandlingError::LightningError(e) => {
+                                                                                                               try_potential_handleerror!(Err(e));
+                                                                                                       }
+                                                                                               }
                                                                                }
                                                                        }
                                                                }
@@ -868,13 +967,6 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
        fn handle_message(&self, peer: &mut Peer, message: wire::Message) -> Result<Option<wire::Message>, MessageHandlingError> {
                log_trace!(self.logger, "Received message {:?} from {}", message, log_pubkey!(peer.their_node_id.unwrap()));
 
-               // Need an Init as first message
-               if let wire::Message::Init(_) = message {
-               } else if peer.their_features.is_none() {
-                       log_debug!(self.logger, "Peer {} sent non-Init first message", log_pubkey!(peer.their_node_id.unwrap()));
-                       return Err(PeerHandleError{ no_connection_possible: false }.into());
-               }
-
                let mut should_forward = None;
 
                match message {
@@ -1033,14 +1125,9 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
                        },
 
                        // Unknown messages:
-                       wire::Message::Unknown(msg_type) if msg_type.is_even() => {
-                               log_debug!(self.logger, "Received unknown even message of type {}, disconnecting peer!", msg_type);
-                               // Fail the channel if message is an even, unknown type as per BOLT #1.
-                               return Err(PeerHandleError{ no_connection_possible: true }.into());
+                       wire::Message::Unknown(_msg) => {
+                               // Handled in `do_read_event`.
                        },
-                       wire::Message::Unknown(msg_type) => {
-                               log_trace!(self.logger, "Received unknown odd message of type {}, ignoring", msg_type);
-                       }
                };
                Ok(should_forward)
        }
@@ -1135,27 +1222,29 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
                        let mut events_generated = self.message_handler.chan_handler.get_and_clear_pending_msg_events();
                        events_generated.append(&mut self.message_handler.route_handler.get_and_clear_pending_msg_events());
                        let peers = &mut *peers_lock;
-                       for event in events_generated.drain(..) {
-                               macro_rules! get_peer_for_forwarding {
-                                       ($node_id: expr) => {
-                                               {
-                                                       match peers.node_id_to_descriptor.get($node_id) {
-                                                               Some(descriptor) => match peers.peers.get_mut(&descriptor) {
-                                                                       Some(peer) => {
-                                                                               if peer.their_features.is_none() {
-                                                                                       continue;
-                                                                               }
-                                                                               peer
-                                                                       },
-                                                                       None => panic!("Inconsistent peers set state!"),
-                                                               },
-                                                               None => {
-                                                                       continue;
+
+                       macro_rules! get_peer_for_forwarding {
+                               ($node_id: expr) => {
+                                       {
+                                               match peers.node_id_to_descriptor.get($node_id) {
+                                                       Some(descriptor) => match peers.peers.get_mut(&descriptor) {
+                                                               Some(peer) => {
+                                                                       if peer.their_features.is_none() {
+                                                                               continue;
+                                                                       }
+                                                                       peer
                                                                },
-                                                       }
+                                                               None => panic!("Inconsistent peers set state!"),
+                                                       },
+                                                       None => {
+                                                               continue;
+                                                       },
                                                }
                                        }
                                }
+                       }
+
+                       for event in events_generated.drain(..) {
                                match event {
                                        MessageSendEvent::SendAcceptChannel { ref node_id, ref msg } => {
                                                log_debug!(self.logger, "Handling SendAcceptChannel event in peer_handler for node {} for channel {}",
@@ -1322,10 +1411,14 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
                                                        msg.number_of_blocks,
                                                        msg.sync_complete);
                                                self.enqueue_message(get_peer_for_forwarding!(node_id), msg);
-                                       }
+                                       },
                                }
                        }
 
+                       for (node_id, message) in self.unknown_message_handler.get_and_clear_pending_msgs() {
+                               self.enqueue_message(get_peer_for_forwarding!(node_id), &message);
+                       }
+
                        for (descriptor, peer) in peers.peers.iter_mut() {
                                self.do_attempt_write_data(&mut (*descriptor).clone(), peer);
                        }
@@ -1438,7 +1531,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref> PeerManager<D
 
 #[cfg(test)]
 mod tests {
-       use ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor};
+       use ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringUnknownMessageHandler};
        use ln::msgs;
        use util::events;
        use util::test_utils;
@@ -1497,20 +1590,20 @@ mod tests {
                cfgs
        }
 
-       fn create_network<'a>(peer_count: usize, cfgs: &'a Vec<PeerManagerCfg>) -> Vec<PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger>> {
+       fn create_network<'a>(peer_count: usize, cfgs: &'a Vec<PeerManagerCfg>) -> Vec<PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger, IgnoringUnknownMessageHandler>> {
                let mut peers = Vec::new();
                for i in 0..peer_count {
                        let node_secret = SecretKey::from_slice(&[42 + i as u8; 32]).unwrap();
                        let ephemeral_bytes = [i as u8; 32];
                        let msg_handler = MessageHandler { chan_handler: &cfgs[i].chan_handler, route_handler: &cfgs[i].routing_handler };
-                       let peer = PeerManager::new(msg_handler, node_secret, &ephemeral_bytes, &cfgs[i].logger);
+                       let peer = PeerManager::new(msg_handler, node_secret, &ephemeral_bytes, &cfgs[i].logger, IgnoringUnknownMessageHandler {});
                        peers.push(peer);
                }
 
                peers
        }
 
-       fn establish_connection<'a>(peer_a: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger>, peer_b: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger>) -> (FileDescriptor, FileDescriptor) {
+       fn establish_connection<'a>(peer_a: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger, IgnoringUnknownMessageHandler>, peer_b: &PeerManager<FileDescriptor, &'a test_utils::TestChannelMessageHandler, &'a test_utils::TestRoutingMessageHandler, &'a test_utils::TestLogger, IgnoringUnknownMessageHandler>) -> (FileDescriptor, FileDescriptor) {
                let secp_ctx = Secp256k1::new();
                let a_id = PublicKey::from_secret_key(&secp_ctx, &peer_a.our_node_secret);
                let mut fd_a = FileDescriptor { fd: 1, outbound_data: Arc::new(Mutex::new(Vec::new())) };
index 0ee280b50e4e72ffc4290333a960a9a09853ae93..7a495d98f518f6e83aee31a9b993b8d2407fa80e 100644 (file)
@@ -62,6 +62,13 @@ pub enum Message {
 #[derive(Clone, Copy, Debug)]
 pub struct MessageType(u16);
 
+impl ::std::ops::Deref for MessageType {
+       type Target = u16;
+       fn deref(&self) -> &Self::Target {
+               &self.0
+       }
+}
+
 impl Message {
        #[allow(dead_code)] // This method is only used in tests
        /// Returns the type that was used to decode the message payload.
index 0fc7c6b3a3d166096019407784b929588ddbddd8..122b2efc1e6f4f8d33926898e6cf48c936163844 100644 (file)
@@ -433,7 +433,7 @@ pub enum MessageSendEvent {
                node_id: PublicKey,
                /// The reply_channel_range which should be sent.
                msg: msgs::ReplyChannelRange,
-       }
+       },
 }
 
 /// A trait indicating an object may generate message send events