Support onion message replies in OnionMessenger

Modify onion message handlers to return an optional response message for
OnionMessenger to reply with. Also, add a ResponseErrorHandler for
handling any errors that may occur when responding.

Author: jkczyz

fuzz/src/onion_message.rs

@@ -11,7 +11,7 @@ use lightning::ln::script::ShutdownScript;
 use lightning::util::enforcing_trait_impls::EnforcingSigner;
 use lightning::util::logger::Logger;
 use lightning::util::ser::{Readable, Writeable, Writer};
-use lightning::onion_message::{CustomOnionMessageContents, CustomOnionMessageHandler, Destination, MessageRouter, OffersMessage, OffersMessageHandler, OnionMessenger};
+use lightning::onion_message::{CustomOnionMessageContents, CustomOnionMessageHandler, Destination, MessageRouter, OffersMessage, OffersMessageHandler, OnionMessenger, ResponseError, ResponseErrorHandler};
 
 use crate::utils::test_logger;
 
@@ -66,7 +66,15 @@ impl MessageRouter for TestMessageRouter {
 struct TestOffersMessageHandler {}
 
 impl OffersMessageHandler for TestOffersMessageHandler {
-	fn handle_message(&self, _message: OffersMessage) {}
+	fn handle_message(&self, _message: OffersMessage) -> Option<OffersMessage> {
+		None
+	}
+}
+
+impl ResponseErrorHandler for TestOffersMessageHandler {
+	fn handle_response_error(&self, _error: ResponseError) {
+		unreachable!()
+	}
 }
 
 struct TestCustomMessage {}
@@ -90,14 +98,22 @@ struct TestCustomMessageHandler {}
 
 impl CustomOnionMessageHandler for TestCustomMessageHandler {
 	type CustomMessage = TestCustomMessage;
-	fn handle_custom_message(&self, _msg: Self::CustomMessage) {}
+	fn handle_custom_message(&self, _msg: Self::CustomMessage) -> Option<Self::CustomMessage> {
+		None
+	}
 	fn read_custom_message<R: io::Read>(&self, _message_type: u64, buffer: &mut R) -> Result<Option<Self::CustomMessage>, msgs::DecodeError> {
 		let mut buf = Vec::new();
 		buffer.read_to_end(&mut buf)?;
 		return Ok(Some(TestCustomMessage {}))
 	}
 }
 
+impl ResponseErrorHandler for TestCustomMessageHandler {
+	fn handle_response_error(&self, _error: ResponseError) {
+		unreachable!()
+	}
+}
+
 pub struct VecWriter(pub Vec<u8>);
 impl Writer for VecWriter {
 	fn write_all(&mut self, buf: &[u8]) -> Result<(), ::std::io::Error> {

lightning/src/ln/peer_handler.rs

@@ -27,7 +27,7 @@ use crate::util::ser::{VecWriter, Writeable, Writer};
 use crate::ln::peer_channel_encryptor::{PeerChannelEncryptor,NextNoiseStep};
 use crate::ln::wire;
 use crate::ln::wire::Encode;
-use crate::onion_message::{CustomOnionMessageContents, CustomOnionMessageHandler, OffersMessage, OffersMessageHandler, SimpleArcOnionMessenger, SimpleRefOnionMessenger};
+use crate::onion_message::{CustomOnionMessageContents, CustomOnionMessageHandler, OffersMessage, OffersMessageHandler, ResponseError, ResponseErrorHandler, SimpleArcOnionMessenger, SimpleRefOnionMessenger};
 use crate::routing::gossip::{NetworkGraph, P2PGossipSync, NodeId, NodeAlias};
 use crate::util::atomic_counter::AtomicCounter;
 use crate::util::logger::Logger;
@@ -103,18 +103,24 @@ impl OnionMessageHandler for IgnoringMessageHandler {
 	}
 }
 impl OffersMessageHandler for IgnoringMessageHandler {
-	fn handle_message(&self, _msg: OffersMessage) {}
+	fn handle_message(&self, _msg: OffersMessage) -> Option<OffersMessage> { None }
 }
 impl CustomOnionMessageHandler for IgnoringMessageHandler {
 	type CustomMessage = Infallible;
-	fn handle_custom_message(&self, _msg: Infallible) {
+	fn handle_custom_message(&self, _msg: Infallible) -> Option<Infallible> {
 		// Since we always return `None` in the read the handle method should never be called.
 		unreachable!();
 	}
 	fn read_custom_message<R: io::Read>(&self, _msg_type: u64, _buffer: &mut R) -> Result<Option<Infallible>, msgs::DecodeError> where Self: Sized {
 		Ok(None)
 	}
 }
+impl ResponseErrorHandler for IgnoringMessageHandler {
+	fn handle_response_error(&self, _error: ResponseError) {
+		// Since handler implementations return None for a response.
+		unreachable!();
+	}
+}
 
 impl CustomOnionMessageContents for Infallible {
 	fn tlv_type(&self) -> u64 { unreachable!(); }

lightning/src/onion_message/functional_tests.rs

@@ -13,7 +13,7 @@ use crate::blinded_path::BlindedPath;
 use crate::sign::{NodeSigner, Recipient};
 use crate::ln::features::InitFeatures;
 use crate::ln::msgs::{self, DecodeError, OnionMessageHandler};
-use super::{CustomOnionMessageContents, CustomOnionMessageHandler, Destination, MessageRouter, OffersMessage, OffersMessageHandler, OnionMessageContents, OnionMessenger, SendError};
+use super::{CustomOnionMessageContents, CustomOnionMessageHandler, Destination, MessageRouter, OffersMessage, OffersMessageHandler, OnionMessageContents, OnionMessenger, ResponseError, ResponseErrorHandler, SendError};
 use crate::util::ser::{Writeable, Writer};
 use crate::util::test_utils;
 
@@ -56,7 +56,13 @@ impl MessageRouter for TestMessageRouter {
 struct TestOffersMessageHandler {}
 
 impl OffersMessageHandler for TestOffersMessageHandler {
-	fn handle_message(&self, _message: OffersMessage) {
+	fn handle_message(&self, _message: OffersMessage) -> Option<OffersMessage> {
+		todo!()
+	}
+}
+
+impl ResponseErrorHandler for TestOffersMessageHandler {
+	fn handle_response_error(&self, _error: ResponseError) {
 		todo!()
 	}
 }
@@ -102,8 +108,9 @@ impl Drop for TestCustomMessageHandler {
 
 impl CustomOnionMessageHandler for TestCustomMessageHandler {
 	type CustomMessage = TestCustomMessage;
-	fn handle_custom_message(&self, _msg: Self::CustomMessage) {
+	fn handle_custom_message(&self, _msg: Self::CustomMessage) -> Option<Self::CustomMessage> {
 		self.num_messages_expected.fetch_sub(1, Ordering::SeqCst);
+		None
 	}
 	fn read_custom_message<R: io::Read>(&self, message_type: u64, buffer: &mut R) -> Result<Option<Self::CustomMessage>, DecodeError> where Self: Sized {
 		if message_type == CUSTOM_MESSAGE_TYPE {
@@ -115,6 +122,12 @@ impl CustomOnionMessageHandler for TestCustomMessageHandler {
 	}
 }
 
+impl ResponseErrorHandler for TestCustomMessageHandler {
+	fn handle_response_error(&self, _error: ResponseError) {
+		todo!()
+	}
+}
+
 fn create_nodes(num_messengers: u8) -> Vec<MessengerNode> {
 	let mut nodes = Vec::new();
 	for i in 0..num_messengers {

lightning/src/onion_message/messenger.rs

@@ -202,17 +202,39 @@ pub enum SendError {
 ///
 /// [`IgnoringMessageHandler`]: crate::ln::peer_handler::IgnoringMessageHandler
 /// [`CustomMessage`]: Self::CustomMessage
-pub trait CustomOnionMessageHandler {
+pub trait CustomOnionMessageHandler: ResponseErrorHandler {
 	/// The message known to the handler. To support multiple message types, you may want to make this
 	/// an enum with a variant for each supported message.
 	type CustomMessage: CustomOnionMessageContents;
-	/// Called with the custom message that was received.
-	fn handle_custom_message(&self, msg: Self::CustomMessage);
+
+	/// Called with the custom message that was received, returning a response to send, if any.
+	fn handle_custom_message(&self, msg: Self::CustomMessage) -> Option<Self::CustomMessage>;
+
 	/// Read a custom message of type `message_type` from `buffer`, returning `Ok(None)` if the
 	/// message type is unknown.
 	fn read_custom_message<R: io::Read>(&self, message_type: u64, buffer: &mut R) -> Result<Option<Self::CustomMessage>, msgs::DecodeError>;
 }
 
+/// Handler for errors occurring when responding to an onion message.
+///
+/// Each message handler may choose to reply to an onion message with a response. This handler is
+/// used when an error occurs when responding.
+pub trait ResponseErrorHandler {
+	/// Called if an error occurred when sending a response to the handled message.
+	fn handle_response_error(&self, error: ResponseError);
+}
+
+/// An error when replying to an onion message.
+#[derive(Debug, PartialEq, Eq)]
+pub enum ResponseError {
+	/// Failed to get the node id from the [`NodeSigner`].
+	GetNodeIdFailed,
+	/// No reply path in the onion message.
+	NoReplyPath,
+	/// Failed to send the reply.
+	Sending(SendError),
+}
+
 impl<ES: Deref, NS: Deref, L: Deref, MR: Deref, OMH: Deref, CMH: Deref>
 OnionMessenger<ES, NS, L, MR, OMH, CMH>
 where
@@ -297,6 +319,45 @@ where
 		}
 	}
 
+	fn respond_with_onion_message<T: CustomOnionMessageContents, EH: ResponseErrorHandler>(
+		&self, response: OnionMessageContents<T>, path_id: Option<[u8; 32]>,
+		reply_path: Option<BlindedPath>, error_handler: &EH
+	) {
+		let sender = match self.node_signer.get_node_id(Recipient::Node) {
+			Ok(node_id) => node_id,
+			Err(_) => {
+				log_info!(
+					self.logger, "No sender to respond to onion message with path_id {:02x?}",
+					path_id
+				);
+				return error_handler.handle_response_error(ResponseError::GetNodeIdFailed);
+			}
+		};
+
+		let destination = match reply_path {
+			Some(reply_path) => Destination::BlindedPath(reply_path),
+			None => {
+				log_info!(
+					self.logger, "No reply path to respond to onion message with path_id {:02x?}",
+					path_id
+				);
+				return error_handler.handle_response_error(ResponseError::NoReplyPath);
+			},
+		};
+
+		let intermediate_nodes = self.message_router.find_route(&sender, &destination);
+
+		log_info!(self.logger, "Responding to onion message with path_id {:02x?}", path_id);
+
+		if let Err(e) = self.send_onion_message(&intermediate_nodes, destination, response, None) {
+			log_info!(
+				self.logger, "Failed responding to onion message with path_id {:02x?}: {:?}",
+				path_id, e
+			);
+			return error_handler.handle_response_error(ResponseError::Sending(e));
+		}
+	}
+
 	#[cfg(test)]
 	pub(super) fn release_pending_msgs(&self) -> HashMap<PublicKey, VecDeque<msgs::OnionMessage>> {
 		let mut pending_msgs = self.pending_messages.lock().unwrap();
@@ -340,7 +401,7 @@ where
 	NS::Target: NodeSigner,
 	L::Target: Logger,
 	MR::Target: MessageRouter,
-	OMH::Target: OffersMessageHandler,
+	OMH::Target: OffersMessageHandler + Sized,
 	CMH::Target: CustomOnionMessageHandler + Sized,
 {
 	/// Handle an incoming onion message. Currently, if a message was destined for us we will log, but
@@ -379,9 +440,27 @@ where
 				log_info!(self.logger,
 					"Received an onion message with path_id {:02x?} and {} reply_path",
 						path_id, if reply_path.is_some() { "a" } else { "no" });
-				match message {
-					OnionMessageContents::Offers(msg) => self.offers_handler.handle_message(msg),
-					OnionMessageContents::Custom(msg) => self.custom_handler.handle_custom_message(msg),
+
+				let response = match message {
+					OnionMessageContents::Offers(msg) => {
+						self.offers_handler.handle_message(msg)
+							.map(|msg| OnionMessageContents::Offers(msg))
+					},
+					OnionMessageContents::Custom(msg) => {
+						self.custom_handler.handle_custom_message(msg)
+							.map(|msg| OnionMessageContents::Custom(msg))
+					},
+				};
+
+				if let Some(response) = response {
+					match response {
+						OnionMessageContents::Offers(_) => self.respond_with_onion_message(
+							response, path_id, reply_path, &*self.offers_handler
+						),
+						OnionMessageContents::Custom(_) => self.respond_with_onion_message(
+							response, path_id, reply_path, &*self.custom_handler
+						),
+					}
 				}
 			},
 			Ok((Payload::Forward(ForwardControlTlvs::Unblinded(ForwardTlvs {

lightning/src/onion_message/mod.rs

@@ -27,6 +27,6 @@ mod packet;
 mod functional_tests;
 
 // Re-export structs so they can be imported with just the `onion_message::` module prefix.
-pub use self::messenger::{CustomOnionMessageContents, CustomOnionMessageHandler, Destination, MessageRouter, OnionMessageContents, OnionMessenger, SendError, SimpleArcOnionMessenger, SimpleRefOnionMessenger};
+pub use self::messenger::{CustomOnionMessageContents, CustomOnionMessageHandler, Destination, MessageRouter, OnionMessageContents, OnionMessenger, ResponseError, ResponseErrorHandler, SendError, SimpleArcOnionMessenger, SimpleRefOnionMessenger};
 pub use self::offers::{OffersMessage, OffersMessageHandler};
 pub(crate) use self::packet::{ControlTlvs, Packet};

lightning/src/onion_message/offers.rs

@@ -19,6 +19,7 @@ use crate::ln::msgs::DecodeError;
 use crate::offers::invoice_request::InvoiceRequest;
 use crate::offers::invoice::Invoice;
 use crate::offers::parse::ParseError;
+use crate::onion_message::messenger::{ResponseError, ResponseErrorHandler};
 use crate::routing::router::Router;
 use crate::sign::{EntropySource, NodeSigner, SignerProvider};
 use crate::util::logger::Logger;
@@ -34,10 +35,10 @@ const INVOICE_ERROR_TLV_TYPE: u64 = 68;
 /// A handler for an [`OnionMessage`] containing a BOLT 12 Offers message as its payload.
 ///
 /// [`OnionMessage`]: crate::ln::msgs::OnionMessage
-pub trait OffersMessageHandler {
+pub trait OffersMessageHandler: ResponseErrorHandler {
 	/// Handles the given message by either responding with an [`Invoice`], sending a payment, or
 	/// replying with an error.
-	fn handle_message(&self, message: OffersMessage);
+	fn handle_message(&self, message: OffersMessage) -> Option<OffersMessage>;
 }
 
 /// Possible BOLT 12 Offers messages sent and received via an [`OnionMessage`].
@@ -122,7 +123,24 @@ where
 	R::Target: Router,
 	L::Target: Logger,
 {
-	fn handle_message(&self, _message: OffersMessage) {
+	fn handle_message(&self, _message: OffersMessage) -> Option<OffersMessage> {
+		todo!()
+	}
+}
+
+impl<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, L: Deref>
+ResponseErrorHandler for ChannelManager<M, T, ES, NS, SP, F, R, L>
+where
+	M::Target: chain::Watch<<SP::Target as SignerProvider>::Signer>,
+	T::Target: BroadcasterInterface,
+	ES::Target: EntropySource,
+	NS::Target: NodeSigner,
+	SP::Target: SignerProvider,
+	F::Target: FeeEstimator,
+	R::Target: Router,
+	L::Target: Logger,
+{
+	fn handle_response_error(&self, _error: ResponseError) {
 		todo!()
 	}
 }