Support onion message replies in OnionMessenger

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, OffersMessage, OffersMessageHandler, OnionMessenger};
+use lightning::onion_message::{CustomOnionMessageContents, CustomOnionMessageHandler, OffersMessage, OffersMessageHandler, OnionMessenger, ResponseError, ResponseErrorHandler};
 
 use crate::utils::test_logger;
 
@@ -56,7 +56,15 @@ pub extern "C" fn onion_message_run(data: *const u8, datalen: usize) {
 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 {}
@@ -80,14 +88,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

@@ -26,7 +26,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};
 use crate::util::atomic_counter::AtomicCounter;
 use crate::util::events::{MessageSendEvent, MessageSendEventsProvider, OnionMessageProvider};
@@ -95,18 +95,21 @@ 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) {}
+}
 
 impl CustomOnionMessageContents for Infallible {
 	fn tlv_type(&self) -> u64 { unreachable!(); }

lightning/src/onion_message/functional_tests.rs

@@ -12,7 +12,7 @@
 use crate::chain::keysinterface::{NodeSigner, Recipient};
 use crate::ln::features::InitFeatures;
 use crate::ln::msgs::{self, DecodeError, OnionMessageHandler};
-use super::{BlindedPath, CustomOnionMessageContents, CustomOnionMessageHandler, Destination, OffersMessage, OffersMessageHandler, OnionMessageContents, OnionMessenger, SendError};
+use super::{BlindedPath, CustomOnionMessageContents, CustomOnionMessageHandler, Destination, OffersMessage, OffersMessageHandler, OnionMessageContents, OnionMessenger, ResponseError, ResponseErrorHandler, SendError};
 use crate::util::ser::{Writeable, Writer};
 use crate::util::test_utils;
 
@@ -37,7 +37,13 @@ impl MessengerNode {
 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!()
 	}
 }
@@ -64,7 +70,9 @@ 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>, DecodeError> where Self: Sized {
 		if message_type == CUSTOM_MESSAGE_TYPE {
 			let mut buf = Vec::new();
@@ -76,6 +84,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

@@ -181,17 +181,34 @@ 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>;
 }
 
+///
+pub trait ResponseErrorHandler {
+	/// Called if an error occurred when sending a response to the handled message.
+	fn handle_response_error(&self, error: ResponseError);
+}
+
+///
+#[derive(Debug, PartialEq, Eq)]
+pub enum ResponseError {
+	///
+	NoReplyPath,
+	///
+	Sending(SendError),
+}
+
 impl<ES: Deref, NS: Deref, L: Deref, OMH: Deref, CMH: Deref> OnionMessenger<ES, NS, L, OMH, CMH>
 where
 	ES::Target: EntropySource,
@@ -272,6 +289,38 @@ 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
+	) {
+		match reply_path {
+			Some(reply_path) => {
+				let intermediate_nodes = vec![];  // TODO: replace with Router trait
+				let destination = Destination::BlindedPath(reply_path);
+
+				log_info!(self.logger, "Responding to onion message with path_id {:02x?}", path_id);
+				let send_result = self.send_onion_message(
+					&intermediate_nodes, destination, response, None
+				);
+
+				if let Err(e) = send_result {
+					log_info!(
+						self.logger,
+						"Failed responding to onion message with path_id {:02x?}: {:?}", path_id, e
+					);
+					error_handler.handle_response_error(ResponseError::Sending(e));
+				}
+			},
+			None => {
+				log_info!(
+					self.logger, "No reply path to respond to onion message with path_id {:02x?}",
+					path_id
+				);
+				error_handler.handle_response_error(ResponseError::NoReplyPath);
+			},
+		}
+	}
+
 	#[cfg(test)]
 	pub(super) fn release_pending_msgs(&self) -> HashMap<PublicKey, VecDeque<msgs::OnionMessage>> {
 		let mut pending_msgs = self.pending_messages.lock().unwrap();
@@ -314,7 +363,7 @@ where
 	ES::Target: EntropySource,
 	NS::Target: NodeSigner,
 	L::Target: Logger,
-	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
@@ -353,9 +402,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

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

lightning/src/onion_message/offers.rs

@@ -20,6 +20,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::util::logger::Logger;
 use crate::util::ser::{ReadableArgs, Writeable, Writer};
@@ -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!()
 	}
 }