Harden gRPC transport

ldk-server-client/src/client.rs

@@ -662,6 +662,7 @@ impl LdkServerClient {
 			body,
 			buf: Vec::new(),
 			trailers_checked: false,
+			terminated: false,
 			_marker: std::marker::PhantomData,
 		})
 	}
@@ -751,6 +752,7 @@ pub struct GrpcStream<M: Message + Default> {
 	body: Body,
 	buf: Vec<u8>,
 	trailers_checked: bool,
+	terminated: bool,
 	_marker: std::marker::PhantomData<M>,
 }
 
@@ -762,46 +764,53 @@ impl<M: Message + Default> GrpcStream<M> {
 	///
 	/// Returns `None` if the stream has ended.
 	pub async fn next_message(&mut self) -> Option<Result<M, LdkServerError>> {
+		if self.terminated {
+			return None;
+		}
+
 		loop {
 			// Try to decode a complete gRPC frame from the buffer
 			if self.buf.len() >= GRPC_FRAME_HEADER_LEN {
 				if self.buf[0] != 0 {
-					return Some(Err(LdkServerError::new(
+					return self.terminate_with_error(LdkServerError::new(
 						InternalError,
 						"gRPC stream compression is not supported",
-					)));
+					));
 				}
 				let msg_len =
 					u32::from_be_bytes([self.buf[1], self.buf[2], self.buf[3], self.buf[4]])
 						as usize;
 				if msg_len > MAX_GRPC_STREAM_MESSAGE_LEN {
-					return Some(Err(LdkServerError::new(
+					return self.terminate_with_error(LdkServerError::new(
 						InternalError,
 						format!(
 							"gRPC stream message exceeds maximum size of {} bytes",
 							MAX_GRPC_STREAM_MESSAGE_LEN
 						),
-					)));
+					));
 				}
 				let frame_len = match GRPC_FRAME_HEADER_LEN.checked_add(msg_len) {
 					Some(frame_len) => frame_len,
 					None => {
-						return Some(Err(LdkServerError::new(
+						return self.terminate_with_error(LdkServerError::new(
 							InternalError,
 							"gRPC stream frame length overflow",
-						)));
+						));
 					},
 				};
 				if self.buf.len() >= frame_len {
 					let proto_bytes = &self.buf[GRPC_FRAME_HEADER_LEN..frame_len];
-					let result = M::decode(proto_bytes).map_err(|e| {
-						LdkServerError::new(
-							InternalError,
-							format!("Failed to decode gRPC stream message: {}", e),
-						)
-					});
+					let message = match M::decode(proto_bytes) {
+						Ok(message) => message,
+						Err(e) => {
+							return self.terminate_with_error(LdkServerError::new(
+								InternalError,
+								format!("Failed to decode gRPC stream message: {}", e),
+							));
+						},
+					};
 					self.buf.drain(..frame_len);
-					return Some(result);
+					return Some(Ok(message));
 				}
 			}
 
@@ -817,10 +826,10 @@ impl<M: Message + Default> GrpcStream<M> {
 					},
 				},
 				Some(Err(e)) => {
-					return Some(Err(LdkServerError::new(
+					return self.terminate_with_error(LdkServerError::new(
 						InternalError,
 						format!("Failed to read gRPC stream: {}", e),
-					)));
+					));
 				},
 				None => {
 					if self.trailers_checked {
@@ -833,7 +842,14 @@ impl<M: Message + Default> GrpcStream<M> {
 		}
 	}
 
-	fn finish_stream(&self, trailers: Option<&HeaderMap>) -> Option<Result<M, LdkServerError>> {
+	fn terminate_with_error(&mut self, error: LdkServerError) -> Option<Result<M, LdkServerError>> {
+		self.terminated = true;
+		self.buf.clear();
+		Some(Err(error))
+	}
+
+	fn finish_stream(&mut self, trailers: Option<&HeaderMap>) -> Option<Result<M, LdkServerError>> {
+		self.terminated = true;
 		if let Some(error) = trailers.and_then(grpc_error_from_headers) {
 			return Some(Err(error));
 		}
@@ -878,6 +894,7 @@ mod tests {
 			body: Body::wrap(FramesBody(frames.into())),
 			buf: Vec::new(),
 			trailers_checked: false,
+			terminated: false,
 			_marker: std::marker::PhantomData,
 		}
 	}
@@ -952,6 +969,7 @@ mod tests {
 				MAX_GRPC_STREAM_MESSAGE_LEN
 			)
 		);
+		assert!(stream.next_message().await.is_none());
 	}
 
 	#[tokio::test]
@@ -961,6 +979,17 @@ mod tests {
 		let result = stream.next_message().await.unwrap().unwrap_err();
 		assert_eq!(result.error_code, InternalError);
 		assert_eq!(result.message, "gRPC stream compression is not supported");
+		assert!(stream.next_message().await.is_none());
+	}
+
+	#[tokio::test]
+	async fn test_event_stream_terminates_after_decode_error() {
+		let mut stream = event_stream(vec![Frame::data(&[0u8, 0, 0, 0, 1, 0xff])]);
+
+		let result = stream.next_message().await.unwrap().unwrap_err();
+		assert_eq!(result.error_code, InternalError);
+		assert!(result.message.starts_with("Failed to decode gRPC stream message:"));
+		assert!(stream.next_message().await.is_none());
 	}
 
 	#[tokio::test]

ldk-server-grpc/src/grpc.rs

@@ -25,6 +25,8 @@ pub const GRPC_STATUS_INTERNAL: u32 = 13;
 pub const GRPC_STATUS_UNAVAILABLE: u32 = 14;
 pub const GRPC_STATUS_UNAUTHENTICATED: u32 = 16;
 
+const MAX_GRPC_MESSAGE_HEADER_LEN: usize = 4 * 1024;
+
 /// A gRPC status with code and human-readable message.
 #[derive(Debug)]
 pub struct GrpcStatus {
@@ -167,16 +169,25 @@ fn ok_trailers() -> http::HeaderMap {
 	trailers
 }
 
+fn grpc_message_header_value(message: &str) -> Option<http::HeaderValue> {
+	if message.is_empty() || message.len() > MAX_GRPC_MESSAGE_HEADER_LEN {
+		return None;
+	}
+
+	let encoded = percent_encode(message);
+	if encoded.len() > MAX_GRPC_MESSAGE_HEADER_LEN {
+		return None;
+	}
+
+	http::HeaderValue::from_str(&encoded).ok()
+}
+
 /// Build trailers for a gRPC error response.
 fn error_trailers(status: &GrpcStatus) -> http::HeaderMap {
 	let mut trailers = http::HeaderMap::with_capacity(2);
 	trailers.insert("grpc-status", http::HeaderValue::from_str(&status.code.to_string()).unwrap());
-	if !status.message.is_empty() {
-		// Percent-encode the message per gRPC spec.
-		let encoded = percent_encode(&status.message);
-		if let Ok(val) = http::HeaderValue::from_str(&encoded) {
-			trailers.insert("grpc-message", val);
-		}
+	if let Some(value) = grpc_message_header_value(&status.message) {
+		trailers.insert("grpc-message", value);
 	}
 	trailers
 }
@@ -194,11 +205,8 @@ pub fn grpc_error_response(status: GrpcStatus) -> http::Response<GrpcBody> {
 		.header("grpc-accept-encoding", "identity")
 		.header("content-length", "0")
 		.header("grpc-status", status.code.to_string());
-	if !status.message.is_empty() {
-		let encoded = percent_encode(&status.message);
-		if let Ok(val) = http::HeaderValue::from_str(&encoded) {
-			builder = builder.header("grpc-message", val);
-		}
+	if let Some(value) = grpc_message_header_value(&status.message) {
+		builder = builder.header("grpc-message", value);
 	}
 	builder.body(GrpcBody::Empty).unwrap()
 }
@@ -283,7 +291,7 @@ fn hex_val(b: u8) -> Option<u8> {
 /// `H` (hours), `M` (minutes), `S` (seconds), `m` (milliseconds),
 /// `u` (microseconds), `n` (nanoseconds).
 pub fn parse_grpc_timeout(value: &str) -> Result<std::time::Duration, GrpcStatus> {
-	if !(2..=9).contains(&value.len()) {
+	if !value.is_ascii() || !(2..=9).contains(&value.len()) {
 		return Err(GrpcStatus::new(GRPC_STATUS_INVALID_ARGUMENT, "Invalid grpc-timeout header"));
 	}
 
@@ -351,6 +359,26 @@ mod tests {
 		assert_eq!(response.headers().get("content-length").unwrap(), "0");
 	}
 
+	#[test]
+	fn test_grpc_error_response_omits_oversized_message() {
+		let response = grpc_error_response(GrpcStatus::new(
+			GRPC_STATUS_INVALID_ARGUMENT,
+			"%".repeat(MAX_GRPC_MESSAGE_HEADER_LEN),
+		));
+
+		assert!(response.headers().get("grpc-message").is_none());
+	}
+
+	#[test]
+	fn test_error_trailers_omit_oversized_message() {
+		let trailers = error_trailers(&GrpcStatus::new(
+			GRPC_STATUS_INVALID_ARGUMENT,
+			"a".repeat(MAX_GRPC_MESSAGE_HEADER_LEN + 1),
+		));
+
+		assert!(trailers.get("grpc-message").is_none());
+	}
+
 	#[test]
 	fn test_decode_too_short() {
 		assert!(decode_grpc_body(&[0, 0, 0]).is_err());
@@ -517,6 +545,10 @@ mod tests {
 
 		let err = parse_grpc_timeout("18446744073709551615H").unwrap_err();
 		assert_eq!(err.code, GRPC_STATUS_INVALID_ARGUMENT);
+
+		// A multi-byte final character must not split mid-codepoint.
+		let err = parse_grpc_timeout("5\u{20ac}").unwrap_err();
+		assert_eq!(err.code, GRPC_STATUS_INVALID_ARGUMENT);
 	}
 
 	mod tonic_compat {

ldk-server/Cargo.toml

@@ -40,4 +40,5 @@ default = []
 experimental-lsps2-support = []
 
 [dev-dependencies]
+tokio = { version = "1.38.0", default-features = false, features = ["test-util"] }
 futures-util = "0.3.31"

ldk-server/src/main.rs

@@ -39,7 +39,7 @@ use prost::Message;
 use tokio::net::TcpListener;
 use tokio::select;
 use tokio::signal::unix::{Signal, SignalKind};
-use tokio::sync::broadcast;
+use tokio::sync::{broadcast, Semaphore};
 
 use crate::api::node_to_proto_custom_tlv;
 use crate::macaroons::MacaroonStore;
@@ -58,6 +58,9 @@ const KV_STORE_IN_USE_RETRY_TIMEOUT: Duration = Duration::from_secs(60);
 const KV_STORE_IN_USE_RETRY_INTERVAL: Duration = Duration::from_secs(2);
 pub(crate) const FULL_VERSION: &str =
 	concat!(env!("CARGO_PKG_VERSION"), " (", env!("GIT_HASH"), ")");
+const MAX_CONCURRENT_HTTP2_STREAMS: u32 = 32;
+const MAX_PENDING_TLS_HANDSHAKES: usize = 64;
+const TLS_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);
 
 pub fn get_default_data_dir() -> Option<PathBuf> {
 	#[cfg(target_os = "macos")]
@@ -406,6 +409,7 @@ fn main() {
 			None
 		};
 
+		let tls_handshake_semaphore = Arc::new(Semaphore::new(MAX_PENDING_TLS_HANDSHAKES));
 		info!("gRPC service listening on {}", config_file.grpc_service_addr);
 
 		systemd::notify_ready();
@@ -721,25 +725,46 @@ fn main() {
 				},
 				res = grpc_listener.accept() => {
 					match res {
-						Ok((stream, _)) => {
+						Ok((stream, peer_addr)) => {
+							let handshake_permit =
+								match Arc::clone(&tls_handshake_semaphore).try_acquire_owned() {
+									Ok(permit) => permit,
+									Err(_) => {
+										debug!("TLS handshake limit reached, rejecting connection");
+										continue;
+									},
+								};
 							let node_service = NodeService::new(
 								Arc::clone(&node),
 								Arc::clone(&macaroon_store),
 								metrics.clone(),
 								metrics_auth_header.clone(),
 								event_sender.clone(),
 								shutdown_rx.clone(),
+								peer_addr.ip(),
 							);
 							let acceptor = tls_acceptor.clone();
 							runtime.spawn(async move {
-								match acceptor.accept(stream).await {
-									Ok(tls_stream) => {
+								match tokio::time::timeout(
+									TLS_HANDSHAKE_TIMEOUT,
+									acceptor.accept(stream),
+								)
+								.await
+								{
+									Ok(Ok(tls_stream)) => {
+										// Only the handshake holds a slot. Holding it for the whole
+										// connection would let an unauthenticated peer block new
+										// connections by keeping established ones idle.
+										drop(handshake_permit);
 										let io_stream = TokioIo::new(tls_stream);
-										if let Err(err) = http2::Builder::new(TokioExecutor::new()).serve_connection(io_stream, node_service).await {
+										let mut builder = http2::Builder::new(TokioExecutor::new());
+										builder.max_concurrent_streams(MAX_CONCURRENT_HTTP2_STREAMS);
+										if let Err(err) = builder.serve_connection(io_stream, node_service).await {
 											error!("Failed to serve TLS connection: {err}");
 										}
 									},
-									Err(e) => error!("TLS handshake failed: {e}"),
+									Ok(Err(e)) => error!("TLS handshake failed: {e}"),
+									Err(_) => debug!("TLS handshake timed out"),
 								}
 							});
 						},