Add LSPS5 DOS protections.

When handling an incoming LSPS5 request, the manager will check
if the counterparty is 'engaged' in some way before responding.
`Engaged` meaning = active channel | LSPS2 active operation | LSPS1 active operation.

Logic: `If not engaged then reject request;`

A single test is added only checking for the active channel condition,
because it's not super easy to get LSPS1-2 on the correct state to check this (yet).
Other tangential work is happening that will make this easier and more tests will come in the near future

Author: martinsaposnic

lightning-liquidity/src/lsps1/service.rs

@@ -174,6 +174,17 @@ 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();
+		if let Some(inner_state_lock) = outer_state_lock.get(counterparty_node_id) {
+			let peer_state = inner_state_lock.lock().unwrap();
+			!(peer_state.pending_requests.is_empty()
+				&& peer_state.outbound_channels_by_order_id.is_empty())
+		} else {
+			false
+		}
+	}
+
 	fn handle_get_info_request(
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 	) -> Result<(), LightningError> {

lightning-liquidity/src/lsps2/service.rs

@@ -108,8 +108,8 @@ struct ForwardPaymentAction(ChannelId, FeePayment);
 struct ForwardHTLCsAction(ChannelId, Vec<InterceptedHTLC>);
 
 /// The different states a requested JIT channel can be in.
-#[derive(Debug)]
-enum OutboundJITChannelState {
+#[derive(Clone, Debug, PartialEq, Eq)]
+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 +134,30 @@ enum OutboundJITChannelState {
 	PaymentForwarded { channel_id: ChannelId },
 }
 
+impl OutboundJITChannelState {
+	fn ord_index(&self) -> u8 {
+		match self {
+			OutboundJITChannelState::PendingInitialPayment { .. } => 0,
+			OutboundJITChannelState::PendingChannelOpen { .. } => 1,
+			OutboundJITChannelState::PendingPaymentForward { .. } => 2,
+			OutboundJITChannelState::PendingPayment { .. } => 3,
+			OutboundJITChannelState::PaymentForwarded { .. } => 4,
+		}
+	}
+}
+
+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.ord_index().cmp(&other.ord_index())
+	}
+}
+
 impl OutboundJITChannelState {
 	fn new() -> Self {
 		OutboundJITChannelState::PendingInitialPayment { payment_queue: PaymentQueue::new() }
@@ -572,6 +596,18 @@ 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();
+		if let Some(inner_state_lock) = outer_state_lock.get(counterparty_node_id) {
+			let peer_state = inner_state_lock.lock().unwrap();
+			peer_state.outbound_channels_by_intercept_scid.values().map(|c| c.state.clone()).max()
+		} else {
+			None
+		}
+	}
+
 	/// 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 +1941,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;
 use bitcoin::secp256k1::PublicKey;
 
 use lightning::ln::channelmanager::AChannelManager;
@@ -150,6 +151,28 @@ 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| {
+				matches!(
+					s,
+					OutboundJITChannelState::PendingChannelOpen { .. }
+						| OutboundJITChannelState::PendingPaymentForward { .. }
+						| OutboundJITChannelState::PendingPayment { .. }
+						| OutboundJITChannelState::PaymentForwarded { .. }
+				)
+			})
+	}
+
 	fn check_prune_stale_webhooks(&self) {
 		let now =
 			LSPSDateTime::new_from_duration_since_epoch(self.time_provider.duration_since_epoch());
@@ -515,7 +538,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 => {