Add methods to fetch an `OfferBuilder` for "phantom" node configs

diff --git a/ext-functional-test-demo/src/main.rs b/ext-functional-test-demo/src/main.rs
index 654cf91..67eb8c7 100644
--- a/ext-functional-test-demo/src/main.rs
+++ b/ext-functional-test-demo/src/main.rs
@@ -17,6 +17,7 @@ mod tests {
 	impl TestSignerFactory for BrokenSignerFactory {
 		fn make_signer(
 			&self, _seed: &[u8; 32], _now: Duration, _v2_remote_key_derivation: bool,
+			_phantom_seed: Option<&[u8; 32]>,
 		) -> Box<dyn DynKeysInterfaceTrait<EcdsaSigner = DynSigner>> {
 			panic!()
 		}
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
index bbede95..64cbc92 100644
--- a/lightning/src/ln/channelmanager.rs
+++ b/lightning/src/ln/channelmanager.rs
@@ -13402,6 +13402,47 @@ macro_rules! create_offer_builder { ($self: ident, $builder: ty) => {
 
 		Ok(builder.into())
 	}
+
+	/// Creates an [`OfferBuilder`] such that the [`Offer`] it builds is recognized by any
+	/// [`ChannelManager`] (or [`OffersMessageFlow`]) using the same [`ExpandedKey`] (as returned
+	/// from [`NodeSigner::get_expanded_key`]). This allows any nodes participating in a BOLT 11
+	/// "phantom node" cluster to also receive BOLT 12 payments.
+	///
+	/// Note that, unlike with BOLT 11 invoices, BOLT 12 "phantom" offers do not in fact have any
+	/// "phantom node" appended to receiving paths. Instead, multiple blinded paths are simply
+	/// included which terminate at different final nodes.
+	///
+	/// `other_nodes_channels` must be set to a list of each participating node's `node_id` (from
+	/// [`NodeSigner::get_node_id`] with a [`Recipient::Node`]) and its channels.
+	///
+	/// `path_count_limit` is used to limit the number of blinded paths included in the resulting
+	/// [`Offer`]. Note that if this is less than the number of participating nodes (i.e.
+	/// `other_nodes_channels.len() + 1`) not all nodes will participate in receiving funds.
+	/// Because the parameterized [`MessageRouter`] will only get a chance to limit the number of
+	/// paths *per-node*, it is important to set this for offers that will be included in a QR
+	/// code.
+	///
+	/// See [`Self::create_offer_builder`] for more details on the blinded path construction.
+	///
+	/// [`ExpandedKey`]: inbound_payment::ExpandedKey
+	pub fn create_phantom_offer_builder(
+		&$self, other_nodes_channels: Vec<(PublicKey, Vec<ChannelDetails>)>,
+		path_count_limit: usize,
+	) -> Result<$builder, Bolt12SemanticError> {
+		let mut peers = Vec::with_capacity(other_nodes_channels.len() + 1);
+		if !other_nodes_channels.iter().any(|(node_id, _)| *node_id == $self.get_our_node_id()) {
+			peers.push(($self.get_our_node_id(), $self.get_peers_for_blinded_path()));
+		}
+		for (node_id, peer_chans) in other_nodes_channels {
+			peers.push((node_id, Self::channel_details_to_forward_nodes(peer_chans)));
+		}
+
+		let builder = $self.flow.create_phantom_offer_builder(
+			&$self.entropy_source, peers, path_count_limit
+		)?;
+
+		Ok(builder.into())
+	}
 } }
 
 macro_rules! create_refund_builder { ($self: ident, $builder: ty) => {
@@ -14018,6 +14059,41 @@ impl<
 		now
 	}
 
+	/// Converts a list of channels to a list of peers which may be suitable to receive onion
+	/// messages through.
+	fn channel_details_to_forward_nodes(
+		mut channel_list: Vec<ChannelDetails>,
+	) -> Vec<MessageForwardNode> {
+		channel_list.sort_unstable_by_key(|chan| chan.counterparty.node_id);
+		let mut res = Vec::new();
+		// TODO: When MSRV reaches 1.77 use chunk_by
+		let mut start = 0;
+		while start < channel_list.len() {
+			let counterparty_node_id = channel_list[start].counterparty.node_id;
+			let end = channel_list[start..]
+				.iter()
+				.position(|chan| chan.counterparty.node_id != counterparty_node_id)
+				.map(|pos| start + pos)
+				.unwrap_or(channel_list.len());
+
+			let peer_chans = &channel_list[start..end];
+			if peer_chans.iter().any(|chan| chan.is_usable)
+				&& peer_chans.iter().any(|c| c.counterparty.features.supports_onion_messages())
+			{
+				res.push(MessageForwardNode {
+					node_id: peer_chans[0].counterparty.node_id,
+					short_channel_id: peer_chans
+						.iter()
+						.filter(|chan| chan.is_usable)
+						.min_by_key(|chan| chan.short_channel_id)
+						.and_then(|chan| chan.get_inbound_payment_scid()),
+				})
+			}
+			start = end;
+		}
+		res
+	}
+
 	fn get_peers_for_blinded_path(&self) -> Vec<MessageForwardNode> {
 		let per_peer_state = self.per_peer_state.read().unwrap();
 		per_peer_state
diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs
index e896575..01de988 100644
--- a/lightning/src/ln/functional_test_utils.rs
+++ b/lightning/src/ln/functional_test_utils.rs
@@ -4405,21 +4405,41 @@ pub fn create_chanmon_cfgs(node_count: usize) -> Vec<TestChanMonCfg> {
 
 pub fn create_chanmon_cfgs_with_legacy_keys(
 	node_count: usize, predefined_keys_ids: Option<Vec<[u8; 32]>>,
+) -> Vec<TestChanMonCfg> {
+	create_chanmon_cfgs_internal(node_count, predefined_keys_ids, false)
+}
+
+pub fn create_phantom_chanmon_cfgs(node_count: usize) -> Vec<TestChanMonCfg> {
+	create_chanmon_cfgs_internal(node_count, None, true)
+}
+
+pub fn create_chanmon_cfgs_internal(
+	node_count: usize, predefined_keys_ids: Option<Vec<[u8; 32]>>, phantom: bool,
 ) -> Vec<TestChanMonCfg> {
 	let mut chan_mon_cfgs = Vec::new();
+	let phantom_seed = if phantom { Some(&[42; 32]) } else { None };
 	for i in 0..node_count {
 		let tx_broadcaster = test_utils::TestBroadcaster::new(Network::Testnet);
 		let fee_estimator = test_utils::TestFeeEstimator::new(253);
 		let chain_source = test_utils::TestChainSource::new(Network::Testnet);
 		let logger = test_utils::TestLogger::with_id(format!("node {}", i));
 		let persister = test_utils::TestPersister::new();
-		let seed = [i as u8; 32];
-		let keys_manager = if predefined_keys_ids.is_some() {
+		let mut seed = [i as u8; 32];
+		if phantom {
+			// We would ideally randomize keys on every test run, but some tests fail in that case.
+			// Instead, we only randomize in the phantom case.
+			use core::hash::{BuildHasher, Hasher};
+			// Get a random value using the only std API to do so - the DefaultHasher
+			let rand_val = std::collections::hash_map::RandomState::new().build_hasher().finish();
+			seed[..8].copy_from_slice(&rand_val.to_ne_bytes());
+		}
+		let keys_manager = test_utils::TestKeysInterface::with_settings(
+			&seed,
+			Network::Testnet,
 			// Use legacy (V1) remote_key derivation for tests using legacy key sets.
-			test_utils::TestKeysInterface::with_v1_remote_key_derivation(&seed, Network::Testnet)
-		} else {
-			test_utils::TestKeysInterface::new(&seed, Network::Testnet)
-		};
+			predefined_keys_ids.is_some(),
+			phantom_seed,
+		);
 		let scorer = RwLock::new(test_utils::TestScorer::new());
 
 		// Set predefined keys_id if provided
diff --git a/lightning/src/ln/offers_tests.rs b/lightning/src/ln/offers_tests.rs
index 12e631b..a4a09dd 100644
--- a/lightning/src/ln/offers_tests.rs
+++ b/lightning/src/ln/offers_tests.rs
@@ -75,15 +75,21 @@ const MAX_SHORT_LIVED_RELATIVE_EXPIRY: Duration = Duration::from_secs(60 * 60 *
 use crate::prelude::*;
 
 macro_rules! expect_recent_payment {
-	($node: expr, $payment_state: path, $payment_id: expr) => {
-		match $node.node.list_recent_payments().first() {
-			Some(&$payment_state { payment_id: actual_payment_id, .. }) => {
-				assert_eq!($payment_id, actual_payment_id);
-			},
-			Some(_) => panic!("Unexpected recent payment state"),
-			None => panic!("No recent payments"),
+	($node: expr, $payment_state: path, $payment_id: expr) => {{
+		let mut found_payment = false;
+		for payment in $node.node.list_recent_payments().iter() {
+			match payment {
+				$payment_state { payment_id: actual_payment_id, .. } => {
+					if $payment_id == *actual_payment_id {
+						found_payment = true;
+						break;
+					}
+				},
+				_ => {},
+			}
 		}
-	}
+		assert!(found_payment);
+	}}
 }
 
 fn connect_peers<'a, 'b, 'c>(node_a: &Node<'a, 'b, 'c>, node_b: &Node<'a, 'b, 'c>) {
@@ -2572,3 +2578,92 @@ fn no_double_pay_with_stale_channelmanager() {
 	// generated in response to the duplicate invoice.
 	assert!(nodes[0].node.get_and_clear_pending_events().is_empty());
 }
+
+#[test]
+fn creates_and_pays_for_phantom_offer() {
+	// Tests that we can pay a "phantom offer" to any participating node.
+	let mut chanmon_cfgs = create_chanmon_cfgs(1);
+	chanmon_cfgs.append(&mut create_phantom_chanmon_cfgs(2));
+	let node_cfgs = create_node_cfgs(3, &chanmon_cfgs);
+	let node_chanmgrs = create_node_chanmgrs(3, &node_cfgs, &[None, None, None]);
+	let nodes = create_network(3, &node_cfgs, &node_chanmgrs);
+
+	create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 10_000_000, 1_000_000_000);
+	create_announced_chan_between_nodes_with_value(&nodes, 0, 2, 10_000_000, 1_000_000_000);
+
+	let node_a_id = nodes[0].node.get_our_node_id();
+	let node_b_id = nodes[1].node.get_our_node_id();
+	let node_c_id = nodes[2].node.get_our_node_id();
+
+	let offer = nodes[1].node
+		.create_phantom_offer_builder(vec![(node_c_id, nodes[2].node.list_channels())], 2)
+		.unwrap()
+		.amount_msats(10_000_000)
+		.build().unwrap();
+
+	// The offer should be resolvable by either of node B or C but signed by a derived key
+	assert!(offer.issuer_signing_pubkey().is_some());
+	assert_ne!(offer.issuer_signing_pubkey(), Some(node_b_id));
+	assert_ne!(offer.issuer_signing_pubkey(), Some(node_c_id));
+	assert_eq!(offer.paths().len(), 2);
+	let mut b_path_count = 0;
+	let mut c_path_count = 0;
+	for path in offer.paths() {
+		if check_compact_path_introduction_node(&path, &nodes[0], node_b_id) {
+			b_path_count += 1;
+		}
+		if check_compact_path_introduction_node(&path, &nodes[0], node_c_id) {
+			c_path_count += 1;
+		}
+	}
+	assert_eq!(b_path_count, 1);
+	assert_eq!(c_path_count, 1);
+
+	// Pay twice, first via node B (the node that actually built the offer) then pay via node C
+	// (which won't have seen the offer until it receives the invoice_request).
+	for (payment_id, recipient) in [([1; 32], &nodes[1]), ([2; 32], &nodes[2])] {
+		let payment_id = PaymentId(payment_id);
+		nodes[0].node.pay_for_offer(&offer, None, payment_id, Default::default()).unwrap();
+		expect_recent_payment!(nodes[0], RecentPaymentDetails::AwaitingInvoice, payment_id);
+
+		let recipient_id = recipient.node.get_our_node_id();
+		let non_recipient_id = if node_b_id == recipient_id {
+			node_c_id
+		} else {
+			node_b_id
+		};
+
+		let onion_message =
+			nodes[0].onion_messenger.next_onion_message_for_peer(recipient_id).unwrap();
+		let _discard =
+			nodes[0].onion_messenger.next_onion_message_for_peer(non_recipient_id).unwrap();
+		recipient.onion_messenger.handle_onion_message(node_a_id, &onion_message);
+
+		let (invoice_request, _) = extract_invoice_request(&recipient, &onion_message);
+		let payment_context = PaymentContext::Bolt12Offer(Bolt12OfferContext {
+			offer_id: offer.id(),
+			invoice_request: InvoiceRequestFields {
+				payer_signing_pubkey: invoice_request.payer_signing_pubkey(),
+				quantity: None,
+				payer_note_truncated: None,
+				human_readable_name: None,
+			},
+		});
+
+		let onion_message =
+			recipient.onion_messenger.next_onion_message_for_peer(node_a_id).unwrap();
+		nodes[0].onion_messenger.handle_onion_message(recipient_id, &onion_message);
+
+		let (invoice, _) = extract_invoice(&nodes[0], &onion_message);
+		assert_eq!(invoice.amount_msats(), 10_000_000);
+
+		route_bolt12_payment(&nodes[0], &[recipient], &invoice);
+		expect_recent_payment!(&nodes[0], RecentPaymentDetails::Pending, payment_id);
+
+		claim_bolt12_payment(&nodes[0], &[recipient], payment_context, &invoice);
+		expect_recent_payment!(&nodes[0], RecentPaymentDetails::Fulfilled, payment_id);
+
+		assert!(nodes[0].onion_messenger.next_onion_message_for_peer(node_b_id).is_none());
+		assert!(nodes[0].onion_messenger.next_onion_message_for_peer(node_c_id).is_none());
+	}
+}
diff --git a/lightning/src/offers/flow.rs b/lightning/src/offers/flow.rs
index 0bb9877..efd5303 100644
--- a/lightning/src/offers/flow.rs
+++ b/lightning/src/offers/flow.rs
@@ -286,6 +286,39 @@ impl<MR: MessageRouter, L: Logger> OffersMessageFlow<MR, L> {
 		self.create_blinded_paths(peers, context)
 	}
 
+	fn blinded_paths_for_phantom_offer(
+		&self, per_node_peers: Vec<(PublicKey, Vec<MessageForwardNode>)>, path_count_limit: usize,
+		context: MessageContext,
+	) -> Result<Vec<BlindedMessagePath>, ()> {
+		let receive_key = ReceiveAuthKey(self.inbound_payment_key.phantom_node_blinded_path_key);
+		let secp_ctx = &self.secp_ctx;
+
+		let mut per_node_paths: Vec<_> = per_node_peers
+			.into_iter()
+			.filter_map(|(recipient, peers)| {
+				self.message_router
+					.create_blinded_paths(recipient, receive_key, context.clone(), peers, secp_ctx)
+					.ok()
+			})
+			.collect();
+
+		let mut res = Vec::new();
+		while res.len() < path_count_limit && !per_node_paths.is_empty() {
+			for node_paths in per_node_paths.iter_mut() {
+				if let Some(path) = node_paths.pop() {
+					res.push(path);
+				}
+			}
+			per_node_paths.retain(|node_paths| !node_paths.is_empty());
+		}
+
+		if res.is_empty() {
+			Err(())
+		} else {
+			Ok(res)
+		}
+	}
+
 	/// Creates a collection of blinded paths by delegating to
 	/// [`MessageRouter::create_blinded_paths`].
 	///
@@ -559,8 +592,7 @@ impl<MR: MessageRouter, L: Logger> OffersMessageFlow<MR, L> {
 
 	/// Creates an [`OfferBuilder`] such that the [`Offer`] it builds is recognized by the
 	/// [`OffersMessageFlow`], and any corresponding [`InvoiceRequest`] can be verified using
-	/// [`Self::verify_invoice_request`]. The offer will expire at `absolute_expiry` if `Some`,
-	/// or will not expire if `None`.
+	/// [`Self::verify_invoice_request`].
 	///
 	/// # Privacy
 	///
@@ -634,6 +666,25 @@ impl<MR: MessageRouter, L: Logger> OffersMessageFlow<MR, L> {
 		})
 	}
 
+	/// Creates an [`OfferBuilder`] such that the [`Offer`] it builds is recognized by any
+	/// [`OffersMessageFlow`] using the same [`ExpandedKey`] (provided in the constructor as
+	/// `inbound_payment_key`), and any corresponding [`InvoiceRequest`] can be verified using
+	/// [`Self::verify_invoice_request`].
+	///
+	/// See [`Self::create_offer_builder`] for more details on privacy and limitations.
+	///
+	/// [`ExpandedKey`]: inbound_payment::ExpandedKey
+	pub fn create_phantom_offer_builder<ES: EntropySource>(
+		&self, entropy_source: ES, per_node_peers: Vec<(PublicKey, Vec<MessageForwardNode>)>,
+		path_count_limit: usize,
+	) -> Result<OfferBuilder<'_, DerivedMetadata, secp256k1::All>, Bolt12SemanticError> {
+		self.create_offer_builder_intern(entropy_source, |_, context, _| {
+			self.blinded_paths_for_phantom_offer(per_node_peers, path_count_limit, context)
+				.map_err(|_| Bolt12SemanticError::MissingPaths)
+		})
+		.map(|(builder, _)| builder)
+	}
+
 	fn create_refund_builder_intern<ES: EntropySource, PF, I>(
 		&self, entropy_source: ES, make_paths: PF, amount_msats: u64, absolute_expiry: Duration,
 		payment_id: PaymentId,
diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs
index a12b113..f9115e4 100644
--- a/lightning/src/util/test_utils.rs
+++ b/lightning/src/util/test_utils.rs
@@ -1954,6 +1954,7 @@ pub trait TestSignerFactory: Send + Sync {
 	/// Make a dynamic signer
 	fn make_signer(
 		&self, seed: &[u8; 32], now: Duration, v2_remote_key_derivation: bool,
+		phantom_seed: Option<&[u8; 32]>,
 	) -> Box<dyn DynKeysInterfaceTrait<EcdsaSigner = DynSigner>>;
 }
 
@@ -1963,12 +1964,13 @@ struct DefaultSignerFactory();
 impl TestSignerFactory for DefaultSignerFactory {
 	fn make_signer(
 		&self, seed: &[u8; 32], now: Duration, v2_remote_key_derivation: bool,
+		phantom_seed: Option<&[u8; 32]>,
 	) -> Box<dyn DynKeysInterfaceTrait<EcdsaSigner = DynSigner>> {
 		let phantom = sign::PhantomKeysManager::new(
 			seed,
 			now.as_secs(),
 			now.subsec_nanos(),
-			seed,
+			if let Some(provided_seed) = phantom_seed { provided_seed } else { seed },
 			v2_remote_key_derivation,
 		);
 		let dphantom = DynPhantomKeysInterface::new(phantom);
@@ -2000,7 +2002,7 @@ impl TestKeysInterface {
 		let factory = DefaultSignerFactory();
 
 		let now = Duration::from_secs(genesis_block(network).header.time as u64);
-		let backing = factory.make_signer(seed, now, true);
+		let backing = factory.make_signer(seed, now, true, None);
 		Self::build(backing)
 	}
 
@@ -2012,7 +2014,21 @@ impl TestKeysInterface {
 		let factory = DefaultSignerFactory();
 
 		let now = Duration::from_secs(genesis_block(network).header.time as u64);
-		let backing = factory.make_signer(seed, now, false);
+		let backing = factory.make_signer(seed, now, false, None);
+		Self::build(backing)
+	}
+
+	pub fn with_settings(
+		seed: &[u8; 32], network: Network, v1_derivation: bool, phantom_seed: Option<&[u8; 32]>,
+	) -> Self {
+		#[cfg(feature = "std")]
+		let factory = SIGNER_FACTORY.get();
+
+		#[cfg(not(feature = "std"))]
+		let factory = DefaultSignerFactory();
+
+		let now = Duration::from_secs(genesis_block(network).header.time as u64);
+		let backing = factory.make_signer(seed, now, !v1_derivation, phantom_seed);
 		Self::build(backing)
 	}