use bitcoin::hashes::sha256::Hash as Sha256;
use bitcoin::secp256k1::{self, PublicKey, Scalar, Secp256k1, SecretKey};
-use crate::blinded_path::{BlindedPath, IntroductionNode, NodeIdLookUp};
-use crate::blinded_path::message::{advance_path_by_one, ForwardTlvs, NextHop, ReceiveTlvs};
+use crate::blinded_path::{BlindedPath, IntroductionNode, NextMessageHop, NodeIdLookUp};
+use crate::blinded_path::message::{advance_path_by_one, ForwardNode, ForwardTlvs, ReceiveTlvs};
use crate::blinded_path::utils;
use crate::events::{Event, EventHandler, EventsProvider};
use crate::sign::{EntropySource, NodeSigner, Recipient};
/// # use bitcoin::hashes::hex::FromHex;
/// # use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey, self};
/// # use lightning::blinded_path::{BlindedPath, EmptyNodeIdLookUp};
+/// # use lightning::blinded_path::message::ForwardNode;
/// # use lightning::sign::{EntropySource, KeysManager};
/// # use lightning::ln::peer_handler::IgnoringMessageHandler;
/// # use lightning::onion_message::messenger::{Destination, MessageRouter, OnionMessagePath, OnionMessenger};
/// # })
/// # }
/// # fn create_blinded_paths<T: secp256k1::Signing + secp256k1::Verification>(
-/// # &self, _recipient: PublicKey, _peers: Vec<PublicKey>, _secp_ctx: &Secp256k1<T>
+/// # &self, _recipient: PublicKey, _peers: Vec<ForwardNode>, _secp_ctx: &Secp256k1<T>
/// # ) -> Result<Vec<BlindedPath>, ()> {
/// # unreachable!()
/// # }
/// # let your_custom_message_type = 42;
/// your_custom_message_type
/// }
+/// fn msg_type(&self) -> &'static str { "YourCustomMessageType" }
/// }
/// // Send a custom onion message to a node id.
/// let destination = Destination::Node(destination_node_id);
///
/// // Create a blinded path to yourself, for someone to send an onion message to.
/// # let your_node_id = hop_node_id1;
-/// let hops = [hop_node_id3, hop_node_id4, your_node_id];
-/// let blinded_path = BlindedPath::new_for_message(&hops, &keys_manager, &secp_ctx).unwrap();
+/// let hops = [
+/// ForwardNode { node_id: hop_node_id3, short_channel_id: None },
+/// ForwardNode { node_id: hop_node_id4, short_channel_id: None },
+/// ];
+/// let blinded_path = BlindedPath::new_for_message(&hops, your_node_id, &keys_manager, &secp_ctx).unwrap();
///
/// // Send a custom onion message to a blinded path.
/// let destination = Destination::BlindedPath(blinded_path);
}
}
+
+/// The `Responder` struct creates an appropriate [`ResponseInstruction`]
+/// for responding to a message.
+pub struct Responder {
+ /// The path along which a response can be sent.
+ reply_path: BlindedPath,
+ path_id: Option<[u8; 32]>
+}
+
+impl Responder {
+ /// Creates a new [`Responder`] instance with the provided reply path.
+ pub(super) fn new(reply_path: BlindedPath, path_id: Option<[u8; 32]>) -> Self {
+ Responder {
+ reply_path,
+ path_id,
+ }
+ }
+
+ /// Creates a [`ResponseInstruction::WithoutReplyPath`] for a given response.
+ ///
+ /// Use when the recipient doesn't need to send back a reply to us.
+ pub fn respond<T: OnionMessageContents>(self, response: T) -> ResponseInstruction<T> {
+ ResponseInstruction::WithoutReplyPath(OnionMessageResponse {
+ message: response,
+ reply_path: self.reply_path,
+ path_id: self.path_id,
+ })
+ }
+
+ /// Creates a [`ResponseInstruction::WithReplyPath`] for a given response.
+ ///
+ /// Use when the recipient needs to send back a reply to us.
+ pub fn respond_with_reply_path<T: OnionMessageContents>(self, response: T) -> ResponseInstruction<T> {
+ ResponseInstruction::WithReplyPath(OnionMessageResponse {
+ message: response,
+ reply_path: self.reply_path,
+ path_id: self.path_id,
+ })
+ }
+}
+
+/// This struct contains the information needed to reply to a received message.
+pub struct OnionMessageResponse<T: OnionMessageContents> {
+ message: T,
+ reply_path: BlindedPath,
+ path_id: Option<[u8; 32]>,
+}
+
+/// `ResponseInstruction` represents instructions for responding to received messages.
+pub enum ResponseInstruction<T: OnionMessageContents> {
+ /// Indicates that a response should be sent including a reply path for
+ /// the recipient to respond back.
+ WithReplyPath(OnionMessageResponse<T>),
+ /// Indicates that a response should be sent without including a reply path
+ /// for the recipient to respond back.
+ WithoutReplyPath(OnionMessageResponse<T>),
+ /// Indicates that there's no response to send back.
+ NoResponse,
+}
+
/// An [`OnionMessage`] for [`OnionMessenger`] to send.
///
/// These are obtained when released from [`OnionMessenger`]'s handlers after which they are
fn create_blinded_paths<
T: secp256k1::Signing + secp256k1::Verification
>(
- &self, recipient: PublicKey, peers: Vec<PublicKey>, secp_ctx: &Secp256k1<T>,
+ &self, recipient: PublicKey, peers: Vec<ForwardNode>, secp_ctx: &Secp256k1<T>,
) -> Result<Vec<BlindedPath>, ()>;
}
fn create_blinded_paths<
T: secp256k1::Signing + secp256k1::Verification
>(
- &self, recipient: PublicKey, peers: Vec<PublicKey>, secp_ctx: &Secp256k1<T>,
+ &self, recipient: PublicKey, peers: Vec<ForwardNode>, secp_ctx: &Secp256k1<T>,
) -> Result<Vec<BlindedPath>, ()> {
// Limit the number of blinded paths that are computed.
const MAX_PATHS: usize = 3;
let is_recipient_announced =
network_graph.nodes().contains_key(&NodeId::from_pubkey(&recipient));
- let mut peer_info = peers.iter()
+ let mut peer_info = peers.into_iter()
// Limit to peers with announced channels
- .filter_map(|pubkey|
+ .filter_map(|peer|
network_graph
- .node(&NodeId::from_pubkey(pubkey))
+ .node(&NodeId::from_pubkey(&peer.node_id))
.filter(|info| info.channels.len() >= MIN_PEER_CHANNELS)
- .map(|info| (*pubkey, info.is_tor_only(), info.channels.len()))
+ .map(|info| (peer, info.is_tor_only(), info.channels.len()))
)
// Exclude Tor-only nodes when the recipient is announced.
.filter(|(_, is_tor_only, _)| !(*is_tor_only && is_recipient_announced))
});
let paths = peer_info.into_iter()
- .map(|(pubkey, _, _)| vec![pubkey, recipient])
- .map(|node_pks| BlindedPath::new_for_message(&node_pks, &*self.entropy_source, secp_ctx))
+ .map(|(peer, _, _)| {
+ BlindedPath::new_for_message(&[peer], recipient, &*self.entropy_source, secp_ctx)
+ })
.take(MAX_PATHS)
.collect::<Result<Vec<_>, _>>();
- match paths {
+ let mut paths = match paths {
Ok(paths) if !paths.is_empty() => Ok(paths),
_ => {
if is_recipient_announced {
Err(())
}
},
+ }?;
+ for path in &mut paths {
+ path.use_compact_introduction_node(&network_graph);
}
+
+ Ok(paths)
}
}
TooFewBlindedHops,
/// The first hop is not a peer and doesn't have a known [`SocketAddress`].
InvalidFirstHop(PublicKey),
- /// A path from the sender to the destination could not be found by the [`MessageRouter`].
+ /// Indicates that a path could not be found by the [`MessageRouter`].
+ ///
+ /// This occurs when either:
+ /// - No path from the sender to the destination was found to send the onion message
+ /// - No reply path to the sender could be created when responding to an onion message
PathNotFound,
/// Onion message contents must have a TLV type >= 64.
InvalidMessage,
/// Called with the custom message that was received, returning a response to send, if any.
///
/// The returned [`Self::CustomMessage`], if any, is enqueued to be sent by [`OnionMessenger`].
- fn handle_custom_message(&self, msg: Self::CustomMessage) -> Option<Self::CustomMessage>;
+ fn handle_custom_message(&self, message: Self::CustomMessage, responder: Option<Responder>) -> ResponseInstruction<Self::CustomMessage>;
/// Read a custom message of type `message_type` from `buffer`, returning `Ok(None)` if the
/// message type is unknown.
/// A processed incoming onion message, containing either a Forward (another onion message)
/// or a Receive payload with decrypted contents.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
pub enum PeeledOnion<T: OnionMessageContents> {
/// Forwarded onion, with the next node id and a new onion
- Forward(NextHop, OnionMessage),
+ Forward(NextMessageHop, OnionMessage),
/// Received onion message, with decrypted contents, path_id, and reply path
Receive(ParsedOnionMessageContents<T>, Option<[u8; 32]>, Option<BlindedPath>)
}
)
}
+ /// Similar to [`Self::new`], but rather than dropping onion messages that are
+ /// intended to be forwarded to offline peers, we will intercept them for
+ /// later forwarding.
+ ///
+ /// Interception flow:
+ /// 1. If an onion message for an offline peer is received, `OnionMessenger` will
+ /// generate an [`Event::OnionMessageIntercepted`]. Event handlers can
+ /// then choose to persist this onion message for later forwarding, or drop
+ /// it.
+ /// 2. When the offline peer later comes back online, `OnionMessenger` will
+ /// generate an [`Event::OnionMessagePeerConnected`]. Event handlers will
+ /// then fetch all previously intercepted onion messages for this peer.
+ /// 3. Once the stored onion messages are fetched, they can finally be
+ /// forwarded to the now-online peer via [`Self::forward_onion_message`].
///
+ /// # Note
+ ///
+ /// LDK will not rate limit how many [`Event::OnionMessageIntercepted`]s
+ /// are generated, so it is the caller's responsibility to limit how many
+ /// onion messages are persisted and only persist onion messages for relevant
+ /// peers.
pub fn new_with_offline_peer_interception(
entropy_source: ES, node_signer: NS, logger: L, node_id_lookup: NL,
message_router: MR, offers_handler: OMH, custom_handler: CMH
&self, contents: T, destination: Destination, reply_path: Option<BlindedPath>,
log_suffix: fmt::Arguments
) -> Result<SendSuccess, SendError> {
- let mut logger = WithContext::from(&self.logger, None, None);
- let result = self.find_path(destination)
- .and_then(|path| {
- let first_hop = path.intermediate_nodes.get(0).map(|p| *p);
- logger = WithContext::from(&self.logger, first_hop, None);
- self.enqueue_onion_message(path, contents, reply_path, log_suffix)
- });
+ let mut logger = WithContext::from(&self.logger, None, None, None);
+ let result = self.find_path(destination).and_then(|path| {
+ let first_hop = path.intermediate_nodes.get(0).map(|p| *p);
+ logger = WithContext::from(&self.logger, first_hop, None, None);
+ self.enqueue_onion_message(path, contents, reply_path, log_suffix)
+ });
match result.as_ref() {
Err(SendError::GetNodeIdFailed) => {
.map_err(|_| SendError::PathNotFound)
}
+ fn create_blinded_path(&self) -> Result<BlindedPath, SendError> {
+ let recipient = self.node_signer
+ .get_node_id(Recipient::Node)
+ .map_err(|_| SendError::GetNodeIdFailed)?;
+ let secp_ctx = &self.secp_ctx;
+
+ let peers = self.message_recipients.lock().unwrap()
+ .iter()
+ .filter(|(_, peer)| matches!(peer, OnionMessageRecipient::ConnectedPeer(_)))
+ .map(|(node_id, _ )| ForwardNode {
+ node_id: *node_id,
+ short_channel_id: None,
+ })
+ .collect::<Vec<_>>();
+
+ self.message_router
+ .create_blinded_paths(recipient, peers, secp_ctx)
+ .and_then(|paths| paths.into_iter().next().ok_or(()))
+ .map_err(|_| SendError::PathNotFound)
+ }
+
fn enqueue_onion_message<T: OnionMessageContents>(
&self, path: OnionMessagePath, contents: T, reply_path: Option<BlindedPath>,
log_suffix: fmt::Arguments
}
}
+ /// Forwards an [`OnionMessage`] to `peer_node_id`. Useful if we initialized
+ /// the [`OnionMessenger`] with [`Self::new_with_offline_peer_interception`]
+ /// and want to forward a previously intercepted onion message to a peer that
+ /// has just come online.
+ pub fn forward_onion_message(
+ &self, message: OnionMessage, peer_node_id: &PublicKey
+ ) -> Result<(), SendError> {
+ let mut message_recipients = self.message_recipients.lock().unwrap();
+ if outbound_buffer_full(&peer_node_id, &message_recipients) {
+ return Err(SendError::BufferFull);
+ }
+
+ match message_recipients.entry(*peer_node_id) {
+ hash_map::Entry::Occupied(mut e) if e.get().is_connected() => {
+ e.get_mut().enqueue_message(message);
+ Ok(())
+ },
+ _ => Err(SendError::InvalidFirstHop(*peer_node_id))
+ }
+ }
+
#[cfg(any(test, feature = "_test_utils"))]
pub fn send_onion_message_using_path<T: OnionMessageContents>(
&self, path: OnionMessagePath, contents: T, reply_path: Option<BlindedPath>
)
}
- fn handle_onion_message_response<T: OnionMessageContents>(
- &self, response: Option<T>, reply_path: Option<BlindedPath>, log_suffix: fmt::Arguments
- ) {
- if let Some(response) = response {
- match reply_path {
- Some(reply_path) => {
- let _ = self.find_path_and_enqueue_onion_message(
- response, Destination::BlindedPath(reply_path), None, log_suffix
+ /// Handles the response to an [`OnionMessage`] based on its [`ResponseInstruction`],
+ /// enqueueing any response for sending.
+ ///
+ /// This function is useful for asynchronous handling of [`OnionMessage`]s.
+ /// Handlers have the option to return [`ResponseInstruction::NoResponse`], indicating that
+ /// no immediate response should be sent. Then, they can transfer the associated [`Responder`]
+ /// to another task responsible for generating the response asynchronously. Subsequently, when
+ /// the response is prepared and ready for sending, that task can invoke this method to enqueue
+ /// the response for delivery.
+ pub fn handle_onion_message_response<T: OnionMessageContents>(
+ &self, response: ResponseInstruction<T>
+ ) -> Result<Option<SendSuccess>, SendError> {
+ let (response, create_reply_path) = match response {
+ ResponseInstruction::WithReplyPath(response) => (response, true),
+ ResponseInstruction::WithoutReplyPath(response) => (response, false),
+ ResponseInstruction::NoResponse => return Ok(None),
+ };
+
+ let message_type = response.message.msg_type();
+ let reply_path = if create_reply_path {
+ match self.create_blinded_path() {
+ Ok(reply_path) => Some(reply_path),
+ Err(err) => {
+ log_trace!(
+ self.logger,
+ "Failed to create reply path when responding with {} to an onion message \
+ with path_id {:02x?}: {:?}",
+ message_type, response.path_id, err
);
- },
- None => {
- log_trace!(self.logger, "Missing reply path {}", log_suffix);
- },
+ return Err(err);
+ }
}
- }
+ } else { None };
+
+ self.find_path_and_enqueue_onion_message(
+ response.message, Destination::BlindedPath(response.reply_path), reply_path,
+ format_args!(
+ "when responding with {} to an onion message with path_id {:02x?}",
+ message_type,
+ response.path_id
+ )
+ ).map(|result| Some(result))
}
#[cfg(test)]
}
msgs
}
+
+ fn enqueue_event(&self, event: Event) {
+ const MAX_EVENTS_BUFFER_SIZE: usize = (1 << 10) * 256;
+ let mut pending_events = self.pending_events.lock().unwrap();
+ let total_buffered_bytes: usize = pending_events
+ .iter()
+ .map(|ev| ev.serialized_length())
+ .sum();
+ if total_buffered_bytes >= MAX_EVENTS_BUFFER_SIZE {
+ log_trace!(self.logger, "Dropping event {:?}: buffer full", event);
+ return
+ }
+ pending_events.push(event);
+ }
}
fn outbound_buffer_full(peer_node_id: &PublicKey, buffer: &HashMap<PublicKey, OnionMessageRecipient>) -> bool {
CMH::Target: CustomOnionMessageHandler,
{
fn handle_onion_message(&self, peer_node_id: &PublicKey, msg: &OnionMessage) {
- let logger = WithContext::from(&self.logger, Some(*peer_node_id), None);
+ let logger = WithContext::from(&self.logger, Some(*peer_node_id), None, None);
match self.peel_onion_message(msg) {
Ok(PeeledOnion::Receive(message, path_id, reply_path)) => {
log_trace!(
match message {
ParsedOnionMessageContents::Offers(msg) => {
- let response = self.offers_handler.handle_message(msg);
- self.handle_onion_message_response(
- response, reply_path, format_args!(
- "when responding to Offers onion message with path_id {:02x?}",
- path_id
- )
+ let responder = reply_path.map(
+ |reply_path| Responder::new(reply_path, path_id)
);
+ let response_instructions = self.offers_handler.handle_message(msg, responder);
+ let _ = self.handle_onion_message_response(response_instructions);
},
ParsedOnionMessageContents::Custom(msg) => {
- let response = self.custom_handler.handle_custom_message(msg);
- self.handle_onion_message_response(
- response, reply_path, format_args!(
- "when responding to Custom onion message with path_id {:02x?}",
- path_id
- )
+ let responder = reply_path.map(
+ |reply_path| Responder::new(reply_path, path_id)
);
+ let response_instructions = self.custom_handler.handle_custom_message(msg, responder);
+ let _ = self.handle_onion_message_response(response_instructions);
},
}
},
Ok(PeeledOnion::Forward(next_hop, onion_message)) => {
let next_node_id = match next_hop {
- NextHop::NodeId(pubkey) => pubkey,
- NextHop::ShortChannelId(scid) => match self.node_id_lookup.next_node_id(scid) {
+ NextMessageHop::NodeId(pubkey) => pubkey,
+ NextMessageHop::ShortChannelId(scid) => match self.node_id_lookup.next_node_id(scid) {
Some(pubkey) => pubkey,
None => {
log_trace!(self.logger, "Dropping forwarded onion messager: unable to resolve next hop using SCID {}", scid);
log_trace!(logger, "Forwarding an onion message to peer {}", next_node_id);
},
_ if self.intercept_messages_for_offline_peers => {
- self.pending_events.lock().unwrap().push(
+ self.enqueue_event(
Event::OnionMessageIntercepted {
peer_node_id: next_node_id, message: onion_message
}
.or_insert_with(|| OnionMessageRecipient::ConnectedPeer(VecDeque::new()))
.mark_connected();
if self.intercept_messages_for_offline_peers {
- self.pending_events.lock().unwrap().push(
+ self.enqueue_event(
Event::OnionMessagePeerConnected { peer_node_id: *their_node_id }
);
}
if let Some(ss) = prev_control_tlvs_ss.take() {
payloads.push((Payload::Forward(ForwardControlTlvs::Unblinded(
ForwardTlvs {
- next_hop: NextHop::NodeId(unblinded_pk_opt.unwrap()),
+ next_hop: NextMessageHop::NodeId(unblinded_pk_opt.unwrap()),
next_blinding_override: None,
}
)), ss));
} else if let Some((intro_node_id, blinding_pt)) = intro_node_id_blinding_pt.take() {
if let Some(control_tlvs_ss) = prev_control_tlvs_ss.take() {
payloads.push((Payload::Forward(ForwardControlTlvs::Unblinded(ForwardTlvs {
- next_hop: NextHop::NodeId(intro_node_id),
+ next_hop: NextMessageHop::NodeId(intro_node_id),
next_blinding_override: Some(blinding_pt),
})), control_tlvs_ss));
}
projects / rust-lightning / blobdiff