Add LSPS5 DOS protections.

lightning-liquidity/src/lsps1/service.rs

@@ -174,6 +174,15 @@ where
 		&self.config
 	}
 
+	pub(crate) fn has_active_requests(&self, counterparty_node_id: &PublicKey) -> bool {
+		let outer_state_lock = self.per_peer_state.read().unwrap();
+		outer_state_lock.get(counterparty_node_id).map_or(false, |inner| {
+			let peer_state = inner.lock().unwrap();
+			!(peer_state.pending_requests.is_empty()
+				&& peer_state.outbound_channels_by_order_id.is_empty())
+		})
+	}
+
 	fn handle_get_info_request(
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 	) -> Result<(), LightningError> {

lightning-liquidity/src/lsps2/service.rs

@@ -107,9 +107,18 @@ struct ForwardPaymentAction(ChannelId, FeePayment);
 #[derive(Debug, PartialEq)]
 struct ForwardHTLCsAction(ChannelId, Vec<InterceptedHTLC>);
 
+#[derive(Debug, Copy, Clone, Eq, PartialEq, Ord, PartialOrd)]
+pub(crate) enum OutboundJITStage {
+	PendingInitialPayment,
+	PendingChannelOpen,
+	PendingPaymentForward,
+	PendingPayment,
+	PaymentForwarded,
+}
+
 /// The different states a requested JIT channel can be in.
-#[derive(Debug)]
-enum OutboundJITChannelState {
+#[derive(Debug, PartialEq, Eq, Clone)]
+pub(crate) enum OutboundJITChannelState {
 	/// The JIT channel SCID was created after a buy request, and we are awaiting an initial payment
 	/// of sufficient size to open the channel.
 	PendingInitialPayment { payment_queue: PaymentQueue },
@@ -134,6 +143,36 @@ enum OutboundJITChannelState {
 	PaymentForwarded { channel_id: ChannelId },
 }
 
+impl OutboundJITChannelState {
+	pub(crate) fn stage(&self) -> OutboundJITStage {
+		match self {
+			OutboundJITChannelState::PendingInitialPayment { .. } => {
+				OutboundJITStage::PendingInitialPayment
+			},
+			OutboundJITChannelState::PendingChannelOpen { .. } => {
+				OutboundJITStage::PendingChannelOpen
+			},
+			OutboundJITChannelState::PendingPaymentForward { .. } => {
+				OutboundJITStage::PendingPaymentForward
+			},
+			OutboundJITChannelState::PendingPayment { .. } => OutboundJITStage::PendingPayment,
+			OutboundJITChannelState::PaymentForwarded { .. } => OutboundJITStage::PaymentForwarded,
+		}
+	}
+}
+
+impl PartialOrd for OutboundJITChannelState {
+	fn partial_cmp(&self, other: &Self) -> Option<CmpOrdering> {
+		Some(self.cmp(other))
+	}
+}
+
+impl Ord for OutboundJITChannelState {
+	fn cmp(&self, other: &Self) -> core::cmp::Ordering {
+		self.stage().cmp(&other.stage())
+	}
+}
+
 impl OutboundJITChannelState {
 	fn new() -> Self {
 		OutboundJITChannelState::PendingInitialPayment { payment_queue: PaymentQueue::new() }
@@ -572,6 +611,16 @@ where
 		&self.config
 	}
 
+	pub(crate) fn highest_state_for_peer(
+		&self, counterparty_node_id: &PublicKey,
+	) -> Option<OutboundJITChannelState> {
+		let outer_state_lock = self.per_peer_state.read().unwrap();
+		outer_state_lock.get(counterparty_node_id).and_then(|inner| {
+			let peer_state = inner.lock().unwrap();
+			peer_state.outbound_channels_by_intercept_scid.values().map(|c| c.state.clone()).max()
+		})
+	}
+
 	/// Used by LSP to inform a client requesting a JIT Channel the token they used is invalid.
 	///
 	/// Should be called in response to receiving a [`LSPS2ServiceEvent::GetInfo`] event.
@@ -1905,4 +1954,55 @@ mod tests {
 			);
 		}
 	}
+
+	#[test]
+	fn highest_state_for_peer_orders() {
+		let opening_fee_params = LSPS2OpeningFeeParams {
+			min_fee_msat: 0,
+			proportional: 0,
+			valid_until: LSPSDateTime::from_str("1970-01-01T00:00:00Z").unwrap(),
+			min_lifetime: 0,
+			max_client_to_self_delay: 0,
+			min_payment_size_msat: 0,
+			max_payment_size_msat: 0,
+			promise: String::new(),
+		};
+
+		let mut map = new_hash_map();
+		map.insert(
+			0,
+			OutboundJITChannel {
+				state: OutboundJITChannelState::PendingInitialPayment {
+					payment_queue: PaymentQueue::new(),
+				},
+				user_channel_id: 0,
+				opening_fee_params: opening_fee_params.clone(),
+				payment_size_msat: None,
+			},
+		);
+		map.insert(
+			1,
+			OutboundJITChannel {
+				state: OutboundJITChannelState::PendingChannelOpen {
+					payment_queue: PaymentQueue::new(),
+					opening_fee_msat: 0,
+				},
+				user_channel_id: 1,
+				opening_fee_params: opening_fee_params.clone(),
+				payment_size_msat: None,
+			},
+		);
+		map.insert(
+			2,
+			OutboundJITChannel {
+				state: OutboundJITChannelState::PaymentForwarded { channel_id: ChannelId([0; 32]) },
+				user_channel_id: 2,
+				opening_fee_params,
+				payment_size_msat: None,
+			},
+		);
+
+		let max_state = map.values().map(|c| c.state.clone()).max().unwrap();
+		assert!(matches!(max_state, OutboundJITChannelState::PaymentForwarded { .. }));
+	}
 }

lightning-liquidity/src/lsps5/service.rs

@@ -22,6 +22,7 @@ use crate::prelude::*;
 use crate::sync::{Arc, Mutex};
 use crate::utils::time::TimeProvider;
 
+use crate::lsps2::service::{OutboundJITChannelState, OutboundJITStage};
 use bitcoin::secp256k1::PublicKey;
 
 use lightning::ln::channelmanager::AChannelManager;
@@ -150,6 +151,20 @@ where
 		}
 	}
 
+	/// Returns whether a request from the given client should be accepted.
+	///
+	/// Prior activity includes an existing open channel, an active LSPS1 flow,
+	/// or an LSPS2 flow that has progressed to at least
+	/// [`OutboundJITChannelState::PendingChannelOpen`].
+	pub(crate) fn can_accept_request(
+		&self, client_id: &PublicKey, lsps2_max_state: Option<OutboundJITChannelState>,
+		lsps1_has_activity: bool,
+	) -> bool {
+		self.client_has_open_channel(client_id)
+			|| lsps1_has_activity
+			|| lsps2_max_state.map_or(false, |s| s.stage() >= OutboundJITStage::PendingChannelOpen)
+	}
+
 	fn check_prune_stale_webhooks(&self) {
 		let now =
 			LSPSDateTime::new_from_duration_since_epoch(self.time_provider.duration_since_epoch());
@@ -515,7 +530,7 @@ where
 		*last_pruning = Some(now);
 	}
 
-	fn client_has_open_channel(&self, client_id: &PublicKey) -> bool {
+	pub(crate) fn client_has_open_channel(&self, client_id: &PublicKey) -> bool {
 		self.channel_manager
 			.get_cm()
 			.list_channels()

lightning-liquidity/src/manager.rs

@@ -559,6 +559,32 @@ where
 			LSPSMessage::LSPS5(msg @ LSPS5Message::Request(..)) => {
 				match &self.lsps5_service_handler {
 					Some(lsps5_service_handler) => {
+						let lsps2_max_state = self
+							.lsps2_service_handler
+							.as_ref()
+							.and_then(|h| h.highest_state_for_peer(sender_node_id));
+						#[cfg(lsps1_service)]
+						let lsps1_has_active_requests = self
+							.lsps1_service_handler
+							.as_ref()
+							.map_or(false, |h| h.has_active_requests(sender_node_id));
+						#[cfg(not(lsps1_service))]
+						let lsps1_has_active_requests = false;
+
+						if !lsps5_service_handler.can_accept_request(
+							sender_node_id,
+							lsps2_max_state,
+							lsps1_has_active_requests,
+						) {
+							return Err(LightningError {
+                                    err: format!(
+                                        "Rejecting LSPS5 request from {:?} without prior activity (requires open channel or active LSPS1 or LSPS2 flow)",
+                                        sender_node_id
+                                    ),
+                                    action: ErrorAction::IgnoreAndLog(Level::Debug),
+                                });
+						}
+
 						lsps5_service_handler.handle_message(msg, sender_node_id)?;
 					},
 					None => {