use bitcoin::hashes::{Hash, HashEngine};
use bitcoin::hashes::hmac::{Hmac, HmacEngine};
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::utils;
use crate::events::{Event, EventHandler, EventsProvider};
use crate::sign::{EntropySource, NodeSigner, Recipient};
use crate::ln::features::{InitFeatures, NodeFeatures};
use crate::ln::msgs::{self, OnionMessage, OnionMessageHandler, SocketAddress};
use crate::ln::onion_utils;
use crate::routing::gossip::{NetworkGraph, NodeId, ReadOnlyNetworkGraph};
use super::packet::OnionMessageContents;
use super::packet::ParsedOnionMessageContents;
use super::offers::OffersMessageHandler;
use super::packet::{BIG_PACKET_HOP_DATA_LEN, ForwardControlTlvs, Packet, Payload, ReceiveControlTlvs, SMALL_PACKET_HOP_DATA_LEN};
use crate::util::logger::{Logger, WithContext};
use crate::util::ser::Writeable;
use core::fmt;
use core::ops::Deref;
use crate::io;
use crate::sync::Mutex;
use crate::prelude::*;
#[cfg(not(c_bindings))]
use {
crate::sign::KeysManager,
crate::ln::channelmanager::{SimpleArcChannelManager, SimpleRefChannelManager},
crate::ln::peer_handler::IgnoringMessageHandler,
crate::sync::Arc,
};
pub(super) const MAX_TIMER_TICKS: usize = 2;
pub struct OnionMessenger<ES: Deref, NS: Deref, L: Deref, NL: Deref, MR: Deref, OMH: Deref, CMH: Deref>
where
ES::Target: EntropySource,
NS::Target: NodeSigner,
L::Target: Logger,
NL::Target: NodeIdLookUp,
MR::Target: MessageRouter,
OMH::Target: OffersMessageHandler,
CMH::Target: CustomOnionMessageHandler,
{
entropy_source: ES,
node_signer: NS,
logger: L,
message_recipients: Mutex<HashMap<PublicKey, OnionMessageRecipient>>,
secp_ctx: Secp256k1<secp256k1::All>,
node_id_lookup: NL,
message_router: MR,
offers_handler: OMH,
custom_handler: CMH,
}
enum OnionMessageRecipient {
ConnectedPeer(VecDeque<OnionMessage>),
PendingConnection(VecDeque<OnionMessage>, Option<Vec<SocketAddress>>, usize),
}
impl OnionMessageRecipient {
fn pending_connection(addresses: Vec<SocketAddress>) -> Self {
Self::PendingConnection(VecDeque::new(), Some(addresses), 0)
}
fn pending_messages(&self) -> &VecDeque<OnionMessage> {
match self {
OnionMessageRecipient::ConnectedPeer(pending_messages) => pending_messages,
OnionMessageRecipient::PendingConnection(pending_messages, _, _) => pending_messages,
}
}
fn enqueue_message(&mut self, message: OnionMessage) {
let pending_messages = match self {
OnionMessageRecipient::ConnectedPeer(pending_messages) => pending_messages,
OnionMessageRecipient::PendingConnection(pending_messages, _, _) => pending_messages,
};
pending_messages.push_back(message);
}
fn dequeue_message(&mut self) -> Option<OnionMessage> {
let pending_messages = match self {
OnionMessageRecipient::ConnectedPeer(pending_messages) => pending_messages,
OnionMessageRecipient::PendingConnection(pending_messages, _, _) => {
debug_assert!(false);
pending_messages
},
};
pending_messages.pop_front()
}
#[cfg(test)]
fn release_pending_messages(&mut self) -> VecDeque<OnionMessage> {
let pending_messages = match self {
OnionMessageRecipient::ConnectedPeer(pending_messages) => pending_messages,
OnionMessageRecipient::PendingConnection(pending_messages, _, _) => pending_messages,
};
core::mem::take(pending_messages)
}
fn mark_connected(&mut self) {
if let OnionMessageRecipient::PendingConnection(pending_messages, _, _) = self {
let mut new_pending_messages = VecDeque::new();
core::mem::swap(pending_messages, &mut new_pending_messages);
*self = OnionMessageRecipient::ConnectedPeer(new_pending_messages);
}
}
fn is_connected(&self) -> bool {
match self {
OnionMessageRecipient::ConnectedPeer(..) => true,
OnionMessageRecipient::PendingConnection(..) => false,
}
}
}
#[cfg(not(c_bindings))]
pub struct PendingOnionMessage<T: OnionMessageContents> {
pub contents: T,
pub destination: Destination,
pub reply_path: Option<BlindedPath>,
}
#[cfg(c_bindings)]
pub type PendingOnionMessage<T> = (T, Destination, Option<BlindedPath>);
pub(crate) fn new_pending_onion_message<T: OnionMessageContents>(
contents: T, destination: Destination, reply_path: Option<BlindedPath>
) -> PendingOnionMessage<T> {
#[cfg(not(c_bindings))]
return PendingOnionMessage { contents, destination, reply_path };
#[cfg(c_bindings)]
return (contents, destination, reply_path);
}
pub trait MessageRouter {
fn find_path(
&self, sender: PublicKey, peers: Vec<PublicKey>, destination: Destination
) -> Result<OnionMessagePath, ()>;
fn create_blinded_paths<
T: secp256k1::Signing + secp256k1::Verification
>(
&self, recipient: PublicKey, peers: Vec<PublicKey>, secp_ctx: &Secp256k1<T>,
) -> Result<Vec<BlindedPath>, ()>;
}
pub struct DefaultMessageRouter<G: Deref<Target=NetworkGraph<L>>, L: Deref, ES: Deref>
where
L::Target: Logger,
ES::Target: EntropySource,
{
network_graph: G,
entropy_source: ES,
}
impl<G: Deref<Target=NetworkGraph<L>>, L: Deref, ES: Deref> DefaultMessageRouter<G, L, ES>
where
L::Target: Logger,
ES::Target: EntropySource,
{
pub fn new(network_graph: G, entropy_source: ES) -> Self {
Self { network_graph, entropy_source }
}
}
impl<G: Deref<Target=NetworkGraph<L>>, L: Deref, ES: Deref> MessageRouter for DefaultMessageRouter<G, L, ES>
where
L::Target: Logger,
ES::Target: EntropySource,
{
fn find_path(
&self, sender: PublicKey, peers: Vec<PublicKey>, mut destination: Destination
) -> Result<OnionMessagePath, ()> {
let network_graph = self.network_graph.deref().read_only();
destination.resolve(&network_graph);
let first_node = match destination.first_node() {
Some(first_node) => first_node,
None => return Err(()),
};
if peers.contains(&first_node) || sender == first_node {
Ok(OnionMessagePath {
intermediate_nodes: vec![], destination, first_node_addresses: None
})
} else {
let node_announcement = network_graph
.node(&NodeId::from_pubkey(&first_node))
.and_then(|node_info| node_info.announcement_info.as_ref())
.and_then(|announcement_info| announcement_info.announcement_message.as_ref())
.map(|node_announcement| &node_announcement.contents);
match node_announcement {
Some(node_announcement) if node_announcement.features.supports_onion_messages() => {
let first_node_addresses = Some(node_announcement.addresses.clone());
Ok(OnionMessagePath {
intermediate_nodes: vec![], destination, first_node_addresses
})
},
_ => Err(()),
}
}
}
fn create_blinded_paths<
T: secp256k1::Signing + secp256k1::Verification
>(
&self, recipient: PublicKey, peers: Vec<PublicKey>, secp_ctx: &Secp256k1<T>,
) -> Result<Vec<BlindedPath>, ()> {
const MAX_PATHS: usize = 3;
const MIN_PEER_CHANNELS: usize = 3;
let network_graph = self.network_graph.deref().read_only();
let is_recipient_announced =
network_graph.nodes().contains_key(&NodeId::from_pubkey(&recipient));
let mut peer_info = peers.iter()
.filter_map(|pubkey|
network_graph
.node(&NodeId::from_pubkey(pubkey))
.filter(|info| info.channels.len() >= MIN_PEER_CHANNELS)
.map(|info| (*pubkey, info.is_tor_only(), info.channels.len()))
)
.filter(|(_, is_tor_only, _)| !(*is_tor_only && is_recipient_announced))
.collect::<Vec<_>>();
peer_info.sort_unstable_by(|(_, a_tor_only, a_channels), (_, b_tor_only, b_channels)| {
a_tor_only.cmp(b_tor_only).then(a_channels.cmp(b_channels).reverse())
});
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))
.take(MAX_PATHS)
.collect::<Result<Vec<_>, _>>();
match paths {
Ok(paths) if !paths.is_empty() => Ok(paths),
_ => {
if is_recipient_announced {
BlindedPath::one_hop_for_message(recipient, &*self.entropy_source, secp_ctx)
.map(|path| vec![path])
} else {
Err(())
}
},
}
}
}
#[derive(Clone)]
pub struct OnionMessagePath {
pub intermediate_nodes: Vec<PublicKey>,
pub destination: Destination,
pub first_node_addresses: Option<Vec<SocketAddress>>,
}
impl OnionMessagePath {
pub fn first_node(&self) -> Option<PublicKey> {
self.intermediate_nodes
.first()
.copied()
.or_else(|| self.destination.first_node())
}
}
#[derive(Clone, Hash, Debug, PartialEq, Eq)]
pub enum Destination {
Node(PublicKey),
BlindedPath(BlindedPath),
}
impl Destination {
pub fn resolve(&mut self, network_graph: &ReadOnlyNetworkGraph) {
if let Destination::BlindedPath(path) = self {
if let IntroductionNode::DirectedShortChannelId(..) = path.introduction_node {
if let Some(pubkey) = path
.public_introduction_node_id(network_graph)
.and_then(|node_id| node_id.as_pubkey().ok())
{
path.introduction_node = IntroductionNode::NodeId(pubkey);
}
}
}
}
pub(super) fn num_hops(&self) -> usize {
match self {
Destination::Node(_) => 1,
Destination::BlindedPath(BlindedPath { blinded_hops, .. }) => blinded_hops.len(),
}
}
fn first_node(&self) -> Option<PublicKey> {
match self {
Destination::Node(node_id) => Some(*node_id),
Destination::BlindedPath(BlindedPath { introduction_node, .. }) => {
match introduction_node {
IntroductionNode::NodeId(pubkey) => Some(*pubkey),
IntroductionNode::DirectedShortChannelId(..) => None,
}
},
}
}
}
#[derive(Clone, Hash, Debug, PartialEq, Eq)]
pub enum SendSuccess {
Buffered,
BufferedAwaitingConnection(PublicKey),
}
#[derive(Clone, Hash, Debug, PartialEq, Eq)]
pub enum SendError {
Secp256k1(secp256k1::Error),
TooBigPacket,
TooFewBlindedHops,
InvalidFirstHop(PublicKey),
PathNotFound,
InvalidMessage,
BufferFull,
GetNodeIdFailed,
UnresolvedIntroductionNode,
BlindedPathAdvanceFailed,
}
pub trait CustomOnionMessageHandler {
type CustomMessage: OnionMessageContents;
fn handle_custom_message(&self, msg: Self::CustomMessage) -> Option<Self::CustomMessage>;
fn read_custom_message<R: io::Read>(&self, message_type: u64, buffer: &mut R) -> Result<Option<Self::CustomMessage>, msgs::DecodeError>;
#[cfg(not(c_bindings))]
fn release_pending_custom_messages(&self) -> Vec<PendingOnionMessage<Self::CustomMessage>>;
#[cfg(c_bindings)]
fn release_pending_custom_messages(&self) -> Vec<(Self::CustomMessage, Destination, Option<BlindedPath>)>;
}
#[derive(Debug)]
pub enum PeeledOnion<T: OnionMessageContents> {
Forward(NextHop, OnionMessage),
Receive(ParsedOnionMessageContents<T>, Option<[u8; 32]>, Option<BlindedPath>)
}
pub fn create_onion_message_resolving_destination<
ES: Deref, NS: Deref, NL: Deref, T: OnionMessageContents
>(
entropy_source: &ES, node_signer: &NS, node_id_lookup: &NL,
network_graph: &ReadOnlyNetworkGraph, secp_ctx: &Secp256k1<secp256k1::All>,
mut path: OnionMessagePath, contents: T, reply_path: Option<BlindedPath>,
) -> Result<(PublicKey, OnionMessage, Option<Vec<SocketAddress>>), SendError>
where
ES::Target: EntropySource,
NS::Target: NodeSigner,
NL::Target: NodeIdLookUp,
{
path.destination.resolve(network_graph);
create_onion_message(
entropy_source, node_signer, node_id_lookup, secp_ctx, path, contents, reply_path,
)
}
pub fn create_onion_message<ES: Deref, NS: Deref, NL: Deref, T: OnionMessageContents>(
entropy_source: &ES, node_signer: &NS, node_id_lookup: &NL,
secp_ctx: &Secp256k1<secp256k1::All>, path: OnionMessagePath, contents: T,
reply_path: Option<BlindedPath>,
) -> Result<(PublicKey, OnionMessage, Option<Vec<SocketAddress>>), SendError>
where
ES::Target: EntropySource,
NS::Target: NodeSigner,
NL::Target: NodeIdLookUp,
{
let OnionMessagePath { intermediate_nodes, mut destination, first_node_addresses } = path;
if let Destination::BlindedPath(BlindedPath { ref blinded_hops, .. }) = destination {
if blinded_hops.is_empty() {
return Err(SendError::TooFewBlindedHops);
}
}
if contents.tlv_type() < 64 { return Err(SendError::InvalidMessage) }
if intermediate_nodes.len() == 0 {
if let Destination::BlindedPath(ref mut blinded_path) = destination {
let our_node_id = node_signer.get_node_id(Recipient::Node)
.map_err(|()| SendError::GetNodeIdFailed)?;
let introduction_node_id = match blinded_path.introduction_node {
IntroductionNode::NodeId(pubkey) => pubkey,
IntroductionNode::DirectedShortChannelId(direction, scid) => {
match node_id_lookup.next_node_id(scid) {
Some(next_node_id) => *direction.select_pubkey(&our_node_id, &next_node_id),
None => return Err(SendError::UnresolvedIntroductionNode),
}
},
};
if introduction_node_id == our_node_id {
advance_path_by_one(blinded_path, node_signer, node_id_lookup, &secp_ctx)
.map_err(|()| SendError::BlindedPathAdvanceFailed)?;
}
}
}
let blinding_secret_bytes = entropy_source.get_secure_random_bytes();
let blinding_secret = SecretKey::from_slice(&blinding_secret_bytes[..]).expect("RNG is busted");
let (first_node_id, blinding_point) = if let Some(first_node_id) = intermediate_nodes.first() {
(*first_node_id, PublicKey::from_secret_key(&secp_ctx, &blinding_secret))
} else {
match &destination {
Destination::Node(pk) => (*pk, PublicKey::from_secret_key(&secp_ctx, &blinding_secret)),
Destination::BlindedPath(BlindedPath { introduction_node, blinding_point, .. }) => {
match introduction_node {
IntroductionNode::NodeId(pubkey) => (*pubkey, *blinding_point),
IntroductionNode::DirectedShortChannelId(..) => {
return Err(SendError::UnresolvedIntroductionNode);
},
}
}
}
};
let (packet_payloads, packet_keys) = packet_payloads_and_keys(
&secp_ctx, &intermediate_nodes, destination, contents, reply_path, &blinding_secret
)?;
let prng_seed = entropy_source.get_secure_random_bytes();
let onion_routing_packet = construct_onion_message_packet(
packet_payloads, packet_keys, prng_seed).map_err(|()| SendError::TooBigPacket)?;
let message = OnionMessage { blinding_point, onion_routing_packet };
Ok((first_node_id, message, first_node_addresses))
}
pub fn peel_onion_message<NS: Deref, L: Deref, CMH: Deref>(
msg: &OnionMessage, secp_ctx: &Secp256k1<secp256k1::All>, node_signer: NS, logger: L,
custom_handler: CMH,
) -> Result<PeeledOnion<<<CMH>::Target as CustomOnionMessageHandler>::CustomMessage>, ()>
where
NS::Target: NodeSigner,
L::Target: Logger,
CMH::Target: CustomOnionMessageHandler,
{
let control_tlvs_ss = match node_signer.ecdh(Recipient::Node, &msg.blinding_point, None) {
Ok(ss) => ss,
Err(e) => {
log_error!(logger, "Failed to retrieve node secret: {:?}", e);
return Err(());
}
};
let onion_decode_ss = {
let blinding_factor = {
let mut hmac = HmacEngine::<Sha256>::new(b"blinded_node_id");
hmac.input(control_tlvs_ss.as_ref());
Hmac::from_engine(hmac).to_byte_array()
};
match node_signer.ecdh(Recipient::Node, &msg.onion_routing_packet.public_key,
Some(&Scalar::from_be_bytes(blinding_factor).unwrap()))
{
Ok(ss) => ss.secret_bytes(),
Err(()) => {
log_trace!(logger, "Failed to compute onion packet shared secret");
return Err(());
}
}
};
match onion_utils::decode_next_untagged_hop(
onion_decode_ss, &msg.onion_routing_packet.hop_data[..], msg.onion_routing_packet.hmac,
(control_tlvs_ss, custom_handler.deref(), logger.deref())
) {
Ok((Payload::Receive::<ParsedOnionMessageContents<<<CMH as Deref>::Target as CustomOnionMessageHandler>::CustomMessage>> {
message, control_tlvs: ReceiveControlTlvs::Unblinded(ReceiveTlvs { path_id }), reply_path,
}, None)) => {
Ok(PeeledOnion::Receive(message, path_id, reply_path))
},
Ok((Payload::Forward(ForwardControlTlvs::Unblinded(ForwardTlvs {
next_hop, next_blinding_override
})), Some((next_hop_hmac, new_packet_bytes)))) => {
let new_pubkey = match onion_utils::next_hop_pubkey(&secp_ctx, msg.onion_routing_packet.public_key, &onion_decode_ss) {
Ok(pk) => pk,
Err(e) => {
log_trace!(logger, "Failed to compute next hop packet pubkey: {}", e);
return Err(())
}
};
let outgoing_packet = Packet {
version: 0,
public_key: new_pubkey,
hop_data: new_packet_bytes,
hmac: next_hop_hmac,
};
let onion_message = OnionMessage {
blinding_point: match next_blinding_override {
Some(blinding_point) => blinding_point,
None => {
match onion_utils::next_hop_pubkey(
&secp_ctx, msg.blinding_point, control_tlvs_ss.as_ref()
) {
Ok(bp) => bp,
Err(e) => {
log_trace!(logger, "Failed to compute next blinding point: {}", e);
return Err(())
}
}
}
},
onion_routing_packet: outgoing_packet,
};
Ok(PeeledOnion::Forward(next_hop, onion_message))
},
Err(e) => {
log_trace!(logger, "Errored decoding onion message packet: {:?}", e);
Err(())
},
_ => {
log_trace!(logger, "Received bogus onion message packet, either the sender encoded a final hop as a forwarding hop or vice versa");
Err(())
},
}
}
impl<ES: Deref, NS: Deref, L: Deref, NL: Deref, MR: Deref, OMH: Deref, CMH: Deref>
OnionMessenger<ES, NS, L, NL, MR, OMH, CMH>
where
ES::Target: EntropySource,
NS::Target: NodeSigner,
L::Target: Logger,
NL::Target: NodeIdLookUp,
MR::Target: MessageRouter,
OMH::Target: OffersMessageHandler,
CMH::Target: CustomOnionMessageHandler,
{
pub fn new(
entropy_source: ES, node_signer: NS, logger: L, node_id_lookup: NL, message_router: MR,
offers_handler: OMH, custom_handler: CMH
) -> Self {
let mut secp_ctx = Secp256k1::new();
secp_ctx.seeded_randomize(&entropy_source.get_secure_random_bytes());
OnionMessenger {
entropy_source,
node_signer,
message_recipients: Mutex::new(new_hash_map()),
secp_ctx,
logger,
node_id_lookup,
message_router,
offers_handler,
custom_handler,
}
}
#[cfg(test)]
pub(crate) fn set_offers_handler(&mut self, offers_handler: OMH) {
self.offers_handler = offers_handler;
}
pub fn send_onion_message<T: OnionMessageContents>(
&self, contents: T, destination: Destination, reply_path: Option<BlindedPath>
) -> Result<SendSuccess, SendError> {
self.find_path_and_enqueue_onion_message(
contents, destination, reply_path, format_args!("")
)
}
fn find_path_and_enqueue_onion_message<T: OnionMessageContents>(
&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)
});
match result.as_ref() {
Err(SendError::GetNodeIdFailed) => {
log_warn!(logger, "Unable to retrieve node id {}", log_suffix);
},
Err(SendError::PathNotFound) => {
log_trace!(logger, "Failed to find path {}", log_suffix);
},
Err(e) => {
log_trace!(logger, "Failed sending onion message {}: {:?}", log_suffix, e);
},
Ok(SendSuccess::Buffered) => {
log_trace!(logger, "Buffered onion message {}", log_suffix);
},
Ok(SendSuccess::BufferedAwaitingConnection(node_id)) => {
log_trace!(
logger,
"Buffered onion message waiting on peer connection {}: {}",
log_suffix, node_id
);
},
}
result
}
fn find_path(&self, destination: Destination) -> Result<OnionMessagePath, SendError> {
let sender = self.node_signer
.get_node_id(Recipient::Node)
.map_err(|_| SendError::GetNodeIdFailed)?;
let peers = self.message_recipients.lock().unwrap()
.iter()
.filter(|(_, recipient)| matches!(recipient, OnionMessageRecipient::ConnectedPeer(_)))
.map(|(node_id, _)| *node_id)
.collect();
self.message_router
.find_path(sender, peers, destination)
.map_err(|_| SendError::PathNotFound)
}
fn enqueue_onion_message<T: OnionMessageContents>(
&self, path: OnionMessagePath, contents: T, reply_path: Option<BlindedPath>,
log_suffix: fmt::Arguments
) -> Result<SendSuccess, SendError> {
log_trace!(self.logger, "Constructing onion message {}: {:?}", log_suffix, contents);
let (first_node_id, onion_message, addresses) = create_onion_message(
&self.entropy_source, &self.node_signer, &self.node_id_lookup, &self.secp_ctx, path,
contents, reply_path,
)?;
let mut message_recipients = self.message_recipients.lock().unwrap();
if outbound_buffer_full(&first_node_id, &message_recipients) {
return Err(SendError::BufferFull);
}
match message_recipients.entry(first_node_id) {
hash_map::Entry::Vacant(e) => match addresses {
None => Err(SendError::InvalidFirstHop(first_node_id)),
Some(addresses) => {
e.insert(OnionMessageRecipient::pending_connection(addresses))
.enqueue_message(onion_message);
Ok(SendSuccess::BufferedAwaitingConnection(first_node_id))
},
},
hash_map::Entry::Occupied(mut e) => {
e.get_mut().enqueue_message(onion_message);
if e.get().is_connected() {
Ok(SendSuccess::Buffered)
} else {
Ok(SendSuccess::BufferedAwaitingConnection(first_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>
) -> Result<SendSuccess, SendError> {
self.enqueue_onion_message(path, contents, reply_path, format_args!(""))
}
pub(crate) fn peel_onion_message(
&self, msg: &OnionMessage
) -> Result<PeeledOnion<<<CMH>::Target as CustomOnionMessageHandler>::CustomMessage>, ()> {
peel_onion_message(
msg, &self.secp_ctx, &*self.node_signer, &*self.logger, &*self.custom_handler
)
}
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
);
},
None => {
log_trace!(self.logger, "Missing reply path {}", log_suffix);
},
}
}
}
#[cfg(test)]
pub(super) fn release_pending_msgs(&self) -> HashMap<PublicKey, VecDeque<OnionMessage>> {
let mut message_recipients = self.message_recipients.lock().unwrap();
let mut msgs = new_hash_map();
for (node_id, recipient) in &mut *message_recipients {
msgs.insert(*node_id, recipient.release_pending_messages());
}
msgs
}
}
fn outbound_buffer_full(peer_node_id: &PublicKey, buffer: &HashMap<PublicKey, OnionMessageRecipient>) -> bool {
const MAX_TOTAL_BUFFER_SIZE: usize = (1 << 20) * 128;
const MAX_PER_PEER_BUFFER_SIZE: usize = (1 << 10) * 256;
let mut total_buffered_bytes = 0;
let mut peer_buffered_bytes = 0;
for (pk, peer_buf) in buffer {
for om in peer_buf.pending_messages() {
let om_len = om.serialized_length();
if pk == peer_node_id {
peer_buffered_bytes += om_len;
}
total_buffered_bytes += om_len;
if total_buffered_bytes >= MAX_TOTAL_BUFFER_SIZE ||
peer_buffered_bytes >= MAX_PER_PEER_BUFFER_SIZE
{
return true
}
}
}
false
}
impl<ES: Deref, NS: Deref, L: Deref, NL: Deref, MR: Deref, OMH: Deref, CMH: Deref> EventsProvider
for OnionMessenger<ES, NS, L, NL, MR, OMH, CMH>
where
ES::Target: EntropySource,
NS::Target: NodeSigner,
L::Target: Logger,
NL::Target: NodeIdLookUp,
MR::Target: MessageRouter,
OMH::Target: OffersMessageHandler,
CMH::Target: CustomOnionMessageHandler,
{
fn process_pending_events<H: Deref>(&self, handler: H) where H::Target: EventHandler {
for (node_id, recipient) in self.message_recipients.lock().unwrap().iter_mut() {
if let OnionMessageRecipient::PendingConnection(_, addresses, _) = recipient {
if let Some(addresses) = addresses.take() {
handler.handle_event(Event::ConnectionNeeded { node_id: *node_id, addresses });
}
}
}
}
}
impl<ES: Deref, NS: Deref, L: Deref, NL: Deref, MR: Deref, OMH: Deref, CMH: Deref> OnionMessageHandler
for OnionMessenger<ES, NS, L, NL, MR, OMH, CMH>
where
ES::Target: EntropySource,
NS::Target: NodeSigner,
L::Target: Logger,
NL::Target: NodeIdLookUp,
MR::Target: MessageRouter,
OMH::Target: OffersMessageHandler,
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);
match self.peel_onion_message(msg) {
Ok(PeeledOnion::Receive(message, path_id, reply_path)) => {
log_trace!(
logger,
"Received an onion message with path_id {:02x?} and {} reply_path: {:?}",
path_id, if reply_path.is_some() { "a" } else { "no" }, message);
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
)
);
},
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
)
);
},
}
},
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) {
Some(pubkey) => pubkey,
None => {
log_trace!(self.logger, "Dropping forwarded onion messager: unable to resolve next hop using SCID {}", scid);
return
},
},
};
let mut message_recipients = self.message_recipients.lock().unwrap();
if outbound_buffer_full(&next_node_id, &message_recipients) {
log_trace!(
logger,
"Dropping forwarded onion message to peer {}: outbound buffer full",
next_node_id);
return
}
#[cfg(fuzzing)]
message_recipients
.entry(next_node_id)
.or_insert_with(|| OnionMessageRecipient::ConnectedPeer(VecDeque::new()));
match message_recipients.entry(next_node_id) {
hash_map::Entry::Occupied(mut e) if matches!(
e.get(), OnionMessageRecipient::ConnectedPeer(..)
) => {
e.get_mut().enqueue_message(onion_message);
log_trace!(logger, "Forwarding an onion message to peer {}", next_node_id);
},
_ => {
log_trace!(
logger,
"Dropping forwarded onion message to disconnected peer {}",
next_node_id);
return
},
}
},
Err(e) => {
log_error!(logger, "Failed to process onion message {:?}", e);
}
}
}
fn peer_connected(&self, their_node_id: &PublicKey, init: &msgs::Init, _inbound: bool) -> Result<(), ()> {
if init.features.supports_onion_messages() {
self.message_recipients.lock().unwrap()
.entry(*their_node_id)
.or_insert_with(|| OnionMessageRecipient::ConnectedPeer(VecDeque::new()))
.mark_connected();
} else {
self.message_recipients.lock().unwrap().remove(their_node_id);
}
Ok(())
}
fn peer_disconnected(&self, their_node_id: &PublicKey) {
match self.message_recipients.lock().unwrap().remove(their_node_id) {
Some(OnionMessageRecipient::ConnectedPeer(..)) => {},
Some(_) => debug_assert!(false),
None => {},
}
}
fn timer_tick_occurred(&self) {
let mut message_recipients = self.message_recipients.lock().unwrap();
message_recipients.retain(|_, recipient| match recipient {
OnionMessageRecipient::PendingConnection(_, None, ticks) => *ticks < MAX_TIMER_TICKS,
OnionMessageRecipient::PendingConnection(_, Some(_), _) => true,
_ => true,
});
for recipient in message_recipients.values_mut() {
if let OnionMessageRecipient::PendingConnection(_, None, ticks) = recipient {
*ticks += 1;
}
}
}
fn provided_node_features(&self) -> NodeFeatures {
let mut features = NodeFeatures::empty();
features.set_onion_messages_optional();
features
}
fn provided_init_features(&self, _their_node_id: &PublicKey) -> InitFeatures {
let mut features = InitFeatures::empty();
features.set_onion_messages_optional();
features
}
fn next_onion_message_for_peer(&self, peer_node_id: PublicKey) -> Option<OnionMessage> {
for message in self.offers_handler.release_pending_messages() {
#[cfg(not(c_bindings))]
let PendingOnionMessage { contents, destination, reply_path } = message;
#[cfg(c_bindings)]
let (contents, destination, reply_path) = message;
let _ = self.find_path_and_enqueue_onion_message(
contents, destination, reply_path, format_args!("when sending OffersMessage")
);
}
for message in self.custom_handler.release_pending_custom_messages() {
#[cfg(not(c_bindings))]
let PendingOnionMessage { contents, destination, reply_path } = message;
#[cfg(c_bindings)]
let (contents, destination, reply_path) = message;
let _ = self.find_path_and_enqueue_onion_message(
contents, destination, reply_path, format_args!("when sending CustomMessage")
);
}
self.message_recipients.lock().unwrap()
.get_mut(&peer_node_id)
.and_then(|buffer| buffer.dequeue_message())
}
}
#[cfg(not(c_bindings))]
pub type SimpleArcOnionMessenger<M, T, F, L> = OnionMessenger<
Arc<KeysManager>,
Arc<KeysManager>,
Arc<L>,
Arc<SimpleArcChannelManager<M, T, F, L>>,
Arc<DefaultMessageRouter<Arc<NetworkGraph<Arc<L>>>, Arc<L>, Arc<KeysManager>>>,
Arc<SimpleArcChannelManager<M, T, F, L>>,
IgnoringMessageHandler
>;
#[cfg(not(c_bindings))]
pub type SimpleRefOnionMessenger<
'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, 'j, M, T, F, L
> = OnionMessenger<
&'a KeysManager,
&'a KeysManager,
&'b L,
&'i SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, M, T, F, L>,
&'j DefaultMessageRouter<&'g NetworkGraph<&'b L>, &'b L, &'a KeysManager>,
&'i SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, M, T, F, L>,
IgnoringMessageHandler
>;
fn packet_payloads_and_keys<T: OnionMessageContents, S: secp256k1::Signing + secp256k1::Verification>(
secp_ctx: &Secp256k1<S>, unblinded_path: &[PublicKey], destination: Destination, message: T,
mut reply_path: Option<BlindedPath>, session_priv: &SecretKey
) -> Result<(Vec<(Payload<T>, [u8; 32])>, Vec<onion_utils::OnionKeys>), SendError> {
let num_hops = unblinded_path.len() + destination.num_hops();
let mut payloads = Vec::with_capacity(num_hops);
let mut onion_packet_keys = Vec::with_capacity(num_hops);
let (mut intro_node_id_blinding_pt, num_blinded_hops) = match &destination {
Destination::Node(_) => (None, 0),
Destination::BlindedPath(BlindedPath { introduction_node, blinding_point, blinded_hops }) => {
let introduction_node_id = match introduction_node {
IntroductionNode::NodeId(pubkey) => pubkey,
IntroductionNode::DirectedShortChannelId(..) => {
return Err(SendError::UnresolvedIntroductionNode);
},
};
(Some((*introduction_node_id, *blinding_point)), blinded_hops.len())
},
};
let num_unblinded_hops = num_hops - num_blinded_hops;
let mut unblinded_path_idx = 0;
let mut blinded_path_idx = 0;
let mut prev_control_tlvs_ss = None;
let mut final_control_tlvs = None;
utils::construct_keys_callback(secp_ctx, unblinded_path.iter(), Some(destination), session_priv,
|_, onion_packet_ss, ephemeral_pubkey, control_tlvs_ss, unblinded_pk_opt, enc_payload_opt| {
if num_unblinded_hops != 0 && unblinded_path_idx < num_unblinded_hops {
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_blinding_override: None,
}
)), ss));
}
prev_control_tlvs_ss = Some(control_tlvs_ss);
unblinded_path_idx += 1;
} 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_blinding_override: Some(blinding_pt),
})), control_tlvs_ss));
}
}
if blinded_path_idx < num_blinded_hops.saturating_sub(1) && enc_payload_opt.is_some() {
payloads.push((Payload::Forward(ForwardControlTlvs::Blinded(enc_payload_opt.unwrap())),
control_tlvs_ss));
blinded_path_idx += 1;
} else if let Some(encrypted_payload) = enc_payload_opt {
final_control_tlvs = Some(ReceiveControlTlvs::Blinded(encrypted_payload));
prev_control_tlvs_ss = Some(control_tlvs_ss);
}
let (rho, mu) = onion_utils::gen_rho_mu_from_shared_secret(onion_packet_ss.as_ref());
onion_packet_keys.push(onion_utils::OnionKeys {
#[cfg(test)]
shared_secret: onion_packet_ss,
#[cfg(test)]
blinding_factor: [0; 32],
ephemeral_pubkey,
rho,
mu,
});
}
).map_err(|e| SendError::Secp256k1(e))?;
if let Some(control_tlvs) = final_control_tlvs {
payloads.push((Payload::Receive {
control_tlvs,
reply_path: reply_path.take(),
message,
}, prev_control_tlvs_ss.unwrap()));
} else {
payloads.push((Payload::Receive {
control_tlvs: ReceiveControlTlvs::Unblinded(ReceiveTlvs { path_id: None, }),
reply_path: reply_path.take(),
message,
}, prev_control_tlvs_ss.unwrap()));
}
Ok((payloads, onion_packet_keys))
}
fn construct_onion_message_packet<T: OnionMessageContents>(payloads: Vec<(Payload<T>, [u8; 32])>, onion_keys: Vec<onion_utils::OnionKeys>, prng_seed: [u8; 32]) -> Result<Packet, ()> {
let payloads_ser_len = onion_utils::payloads_serialized_length(&payloads);
let hop_data_len = if payloads_ser_len <= SMALL_PACKET_HOP_DATA_LEN {
SMALL_PACKET_HOP_DATA_LEN
} else if payloads_ser_len <= BIG_PACKET_HOP_DATA_LEN {
BIG_PACKET_HOP_DATA_LEN
} else { return Err(()) };
onion_utils::construct_onion_message_packet::<_, _>(
payloads, onion_keys, prng_seed, hop_data_len)
}