rpc: add option to whitelist ips in rate limiting

cumulus/test/service/src/lib.rs

@@ -803,6 +803,7 @@ pub fn node_config(
 		rpc_message_buffer_capacity: Default::default(),
 		rpc_batch_config: RpcBatchRequestConfig::Unlimited,
 		rpc_rate_limit: None,
+		rpc_rate_limit_whitelisted_hosts: Default::default(),
 		prometheus_config: None,
 		telemetry_endpoints: None,
 		default_heap_pages: None,

polkadot/node/test/service/src/lib.rs

@@ -188,6 +188,7 @@ pub fn node_config(
 		rpc_message_buffer_capacity: Default::default(),
 		rpc_batch_config: RpcBatchRequestConfig::Unlimited,
 		rpc_rate_limit: None,
+		rpc_rate_limit_whitelisted_hosts: Default::default(),
 		prometheus_config: None,
 		telemetry_endpoints: None,
 		default_heap_pages: None,

substrate/bin/node/cli/benches/block_production.rs

@@ -86,6 +86,7 @@ fn new_node(tokio_handle: Handle) -> node_cli::service::NewFullBase {
 		rpc_message_buffer_capacity: Default::default(),
 		rpc_batch_config: RpcBatchRequestConfig::Unlimited,
 		rpc_rate_limit: None,
+		rpc_rate_limit_whitelisted_hosts: Default::default(),
 		prometheus_config: None,
 		telemetry_endpoints: None,
 		default_heap_pages: None,

substrate/bin/node/cli/benches/transaction_pool.rs

@@ -82,6 +82,7 @@ fn new_node(tokio_handle: Handle) -> node_cli::service::NewFullBase {
 		rpc_message_buffer_capacity: Default::default(),
 		rpc_batch_config: RpcBatchRequestConfig::Unlimited,
 		rpc_rate_limit: None,
+		rpc_rate_limit_whitelisted_hosts: Default::default(),
 		prometheus_config: None,
 		telemetry_endpoints: None,
 		default_heap_pages: None,

substrate/client/cli/src/commands/run_cmd.rs

@@ -94,6 +94,10 @@ pub struct RunCmd {
 	#[arg(long)]
 	pub rpc_rate_limit: Option<NonZeroU32>,
 
+	/// Disable RPC rate limiting for certain hosts.
+	#[arg(long, num_args = 1..)]
+	pub rpc_rate_limit_whitelisted_hosts: Vec<String>,
+
 	/// Set the maximum RPC request payload size for both HTTP and WS in megabytes.
 	#[arg(long, default_value_t = RPC_DEFAULT_MAX_REQUEST_SIZE_MB)]
 	pub rpc_max_request_size: u32,
@@ -439,6 +443,10 @@ impl CliConfiguration for RunCmd {
 		Ok(self.rpc_rate_limit)
 	}
 
+	fn rpc_rate_limit_whitelisted_hosts(&self) -> Result<Vec<String>> {
+		Ok(self.rpc_rate_limit_whitelisted_hosts.clone())
+	}
+
 	fn transaction_pool(&self, is_dev: bool) -> Result<TransactionPoolOptions> {
 		Ok(self.pool_config.transaction_pool(is_dev))
 	}

substrate/client/cli/src/config.rs

@@ -349,6 +349,11 @@ pub trait CliConfiguration<DCV: DefaultConfigurationValues = ()>: Sized {
 		Ok(None)
 	}
 
+	/// RPC rate limit whitelisted hosts.
+	fn rpc_rate_limit_whitelisted_hosts(&self) -> Result<Vec<String>> {
+		Ok(vec![])
+	}
+
 	/// Get the prometheus configuration (`None` if disabled)
 	///
 	/// By default this is `None`.
@@ -523,6 +528,7 @@ pub trait CliConfiguration<DCV: DefaultConfigurationValues = ()>: Sized {
 			rpc_message_buffer_capacity: self.rpc_buffer_capacity_per_connection()?,
 			rpc_batch_config: self.rpc_batch_config()?,
 			rpc_rate_limit: self.rpc_rate_limit()?,
+			rpc_rate_limit_whitelisted_hosts: self.rpc_rate_limit_whitelisted_hosts()?,
 			prometheus_config: self
 				.prometheus_config(DCV::prometheus_listen_port(), &chain_spec)?,
 			telemetry_endpoints,

substrate/client/cli/src/runner.rs

@@ -273,6 +273,7 @@ mod tests {
 				rpc_port: 9944,
 				rpc_batch_config: sc_service::config::RpcBatchRequestConfig::Unlimited,
 				rpc_rate_limit: None,
+				rpc_rate_limit_whitelisted_hosts: Default::default(),
 				prometheus_config: None,
 				telemetry_endpoints: None,
 				default_heap_pages: None,

substrate/client/rpc-servers/src/lib.rs

@@ -21,26 +21,29 @@
 #![warn(missing_docs)]
 
 pub mod middleware;
+pub mod utils;
 
 use std::{
 	convert::Infallible, error::Error as StdError, net::SocketAddr, num::NonZeroU32, time::Duration,
 };
 
-use http::header::HeaderValue;
 use hyper::{
 	server::conn::AddrStream,
 	service::{make_service_fn, service_fn},
 };
 use jsonrpsee::{
 	server::{
-		middleware::http::{HostFilterLayer, ProxyGetRequestLayer},
-		stop_channel, ws, PingConfig, StopHandle, TowerServiceBuilder,
+		middleware::http::ProxyGetRequestLayer, stop_channel, ws, PingConfig, StopHandle,
+		TowerServiceBuilder,
 	},
 	Methods, RpcModule,
 };
 use tokio::net::TcpListener;
 use tower::Service;
-use tower_http::cors::{AllowOrigin, CorsLayer};
+use utils::{
+	build_rpc_api, format_cors, host_filtering, hosts_to_ip_addrs, read_ip_from_proxy,
+	try_into_cors,
+};
 
 pub use jsonrpsee::{
 	core::{
@@ -85,6 +88,8 @@ pub struct Config<'a, M: Send + Sync + 'static> {
 	pub batch_config: BatchRequestConfig,
 	/// Rate limit calls per minute.
 	pub rate_limit: Option<NonZeroU32>,
+	/// Disable rate limit for hosts.
+	pub rate_limit_whitelisted_hosts: &'a [String],
 }
 
 #[derive(Debug, Clone)]
@@ -117,11 +122,13 @@ where
 		tokio_handle,
 		rpc_api,
 		rate_limit,
+		rate_limit_whitelisted_hosts,
 	} = config;
 
 	let std_listener = TcpListener::bind(addrs.as_slice()).await?.into_std()?;
 	let local_addr = std_listener.local_addr().ok();
-	let host_filter = hosts_filtering(cors.is_some(), local_addr);
+	let host_filter = host_filtering(cors.is_some(), local_addr);
+	let rate_limit_whitelisted_ip_addrs = hosts_to_ip_addrs(rate_limit_whitelisted_hosts)?;
 
 	let http_middleware = tower::ServiceBuilder::new()
 		.option_layer(host_filter)
@@ -160,20 +167,33 @@ where
 		stop_handle: stop_handle.clone(),
 	};
 
-	let make_service = make_service_fn(move |_conn: &AddrStream| {
+	let make_service = make_service_fn(move |conn: &AddrStream| {
 		let cfg = cfg.clone();
+		let peer_ip = conn.remote_addr().ip();
+		let rate_limit_whitelisted_ip_addrs = rate_limit_whitelisted_ip_addrs.clone();
 
 		async move {
 			let cfg = cfg.clone();
+			let rate_limit_whitelisted_ip_addrs = rate_limit_whitelisted_ip_addrs.clone();
 
 			Ok::<_, Infallible>(service_fn(move |req| {
+				let ip =
+					if let Some(proxy_ip) = read_ip_from_proxy(&req) { proxy_ip } else { peer_ip };
+
+				let rate_limit_cfg = if rate_limit_whitelisted_ip_addrs.iter().any(|ip2| ip2 == &ip)
+				{
+					None
+				} else {
+					rate_limit
+				};
+
 				let PerConnection { service_builder, metrics, tokio_handle, stop_handle, methods } =
 					cfg.clone();
 
 				let is_websocket = ws::is_upgrade_request(&req);
 				let transport_label = if is_websocket { "ws" } else { "http" };
 
-				let middleware_layer = match (metrics, rate_limit) {
+				let middleware_layer = match (metrics, rate_limit_cfg) {
 					(None, None) => None,
 					(Some(metrics), None) => Some(
 						MiddlewareLayer::new().with_metrics(Metrics::new(metrics, transport_label)),
@@ -227,57 +247,3 @@ where
 
 	Ok(server_handle)
 }
-
-fn hosts_filtering(enabled: bool, addr: Option<SocketAddr>) -> Option<HostFilterLayer> {
-	// If the local_addr failed, fallback to wildcard.
-	let port = addr.map_or("*".to_string(), |p| p.port().to_string());
-
-	if enabled {
-		// NOTE: The listening addresses are whitelisted by default.
-		let hosts =
-			[format!("localhost:{port}"), format!("127.0.0.1:{port}"), format!("[::1]:{port}")];
-		Some(HostFilterLayer::new(hosts).expect("Valid hosts; qed"))
-	} else {
-		None
-	}
-}
-
-fn build_rpc_api<M: Send + Sync + 'static>(mut rpc_api: RpcModule<M>) -> RpcModule<M> {
-	let mut available_methods = rpc_api.method_names().collect::<Vec<_>>();
-	// The "rpc_methods" is defined below and we want it to be part of the reported methods.
-	available_methods.push("rpc_methods");
-	available_methods.sort();
-
-	rpc_api
-		.register_method("rpc_methods", move |_, _| {
-			serde_json::json!({
-				"methods": available_methods,
-			})
-		})
-		.expect("infallible all other methods have their own address space; qed");
-
-	rpc_api
-}
-
-fn try_into_cors(
-	maybe_cors: Option<&Vec<String>>,
-) -> Result<CorsLayer, Box<dyn StdError + Send + Sync>> {
-	if let Some(cors) = maybe_cors {
-		let mut list = Vec::new();
-		for origin in cors {
-			list.push(HeaderValue::from_str(origin)?);
-		}
-		Ok(CorsLayer::new().allow_origin(AllowOrigin::list(list)))
-	} else {
-		// allow all cors
-		Ok(CorsLayer::permissive())
-	}
-}
-
-fn format_cors(maybe_cors: Option<&Vec<String>>) -> String {
-	if let Some(cors) = maybe_cors {
-		format!("{:?}", cors)
-	} else {
-		format!("{:?}", ["*"])
-	}
-}

substrate/client/rpc-servers/src/utils.rs

@@ -0,0 +1,161 @@
+// This file is part of Substrate.
+
+// Copyright (C) Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! Substrate RPC server utils.
+
+use std::{
+	error::Error as StdError,
+	net::{IpAddr, SocketAddr, ToSocketAddrs},
+	str::FromStr,
+};
+
+use hyper::{
+	header::{HeaderName, HeaderValue},
+	Request,
+};
+use jsonrpsee::{server::middleware::http::HostFilterLayer, RpcModule};
+use tower_http::cors::{AllowOrigin, CorsLayer};
+
+static X_FORWARDED_FOR: HeaderName = HeaderName::from_static("x-forwarded-for");
+
+/// Helper to read the ip address of the client `X_FORWARDED_FOR` header.
+pub(crate) fn read_ip_from_proxy<B>(req: &Request<B>) -> Option<IpAddr> {
+	// https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/X-Forwarded-For
+	//
+	// "X-Forwarded-For" returns a list of ip addresses
+	//
+	// X-Forwarded-For: 203.0.113.195,198.51.100.178
+	if let Some(ips) = req.headers().get(&X_FORWARDED_FOR).and_then(|v| v.to_str().ok()) {
+		if let Some(proxy_ip) = ips.split_once(',').and_then(|(v, _)| IpAddr::from_str(v).ok()) {
+			// NOTE: we assume that ip addr is global
+			// and it may not work with local proxies.
+			return Some(proxy_ip);
+		}
+	}
+
+	None
+}
+
+pub(crate) fn host_filtering(enabled: bool, addr: Option<SocketAddr>) -> Option<HostFilterLayer> {
+	// If the local_addr failed, fallback to wildcard.
+	let port = addr.map_or("*".to_string(), |p| p.port().to_string());
+
+	if enabled {
+		// NOTE: The listening addresses are whitelisted by default.
+		let hosts =
+			[format!("localhost:{port}"), format!("127.0.0.1:{port}"), format!("[::1]:{port}")];
+		Some(HostFilterLayer::new(hosts).expect("Valid hosts; qed"))
+	} else {
+		None
+	}
+}
+
+pub(crate) fn build_rpc_api<M: Send + Sync + 'static>(mut rpc_api: RpcModule<M>) -> RpcModule<M> {
+	let mut available_methods = rpc_api.method_names().collect::<Vec<_>>();
+	// The "rpc_methods" is defined below and we want it to be part of the reported methods.
+	available_methods.push("rpc_methods");
+	available_methods.sort();
+
+	rpc_api
+		.register_method("rpc_methods", move |_, _| {
+			serde_json::json!({
+				"methods": available_methods,
+			})
+		})
+		.expect("infallible all other methods have their own address space; qed");
+
+	rpc_api
+}
+
+pub(crate) fn try_into_cors(
+	maybe_cors: Option<&Vec<String>>,
+) -> Result<CorsLayer, Box<dyn StdError + Send + Sync>> {
+	if let Some(cors) = maybe_cors {
+		let mut list = Vec::new();
+		for origin in cors {
+			list.push(HeaderValue::from_str(origin)?);
+		}
+		Ok(CorsLayer::new().allow_origin(AllowOrigin::list(list)))
+	} else {
+		// allow all cors
+		Ok(CorsLayer::permissive())
+	}
+}
+
+pub(crate) fn format_cors(maybe_cors: Option<&Vec<String>>) -> String {
+	if let Some(cors) = maybe_cors {
+		format!("{:?}", cors)
+	} else {
+		format!("{:?}", ["*"])
+	}
+}
+
+pub(crate) fn hosts_to_ip_addrs(
+	hosts: &[String],
+) -> Result<Vec<IpAddr>, Box<dyn StdError + Send + Sync>> {
+	let mut ip_list = Vec::new();
+
+	for host in hosts {
+		// The host may contain a port such as `hostname:8080`
+		// and we don't care about the port to lookup the IP addr.
+		//
+		// to_socket_addr without the port will fail though
+		let host_no_port = if let Some((h, _port)) = host.split_once(":") { h } else { host };
+
+		let sockaddrs = (host_no_port, 0).to_socket_addrs()?;
+
+		for sockaddr in sockaddrs {
+			ip_list.push(sockaddr.ip());
+		}
+	}
+
+	Ok(ip_list)
+}
+
+#[cfg(test)]
+mod tests {
+	use hyper::header::HeaderValue;
+
+	use super::*;
+
+	#[test]
+	fn socket_ip_works() {
+		let req = hyper::Request::new(());
+		let ip = read_ip_from_proxy(&req);
+		assert!(ip.is_none())
+	}
+
+	#[test]
+	fn ip_from_proxy() {
+		let mut req = hyper::Request::new(());
+
+		req.headers_mut()
+			.insert(&X_FORWARDED_FOR, HeaderValue::from_static("203.0.113.195,198.51.100.178"));
+		let ip = read_ip_from_proxy(&req);
+		assert_eq!(Some(IpAddr::from_str("203.0.113.195").unwrap()), ip);
+	}
+
+	#[test]
+	fn ip_from_proxy_faulty() {
+		let mut req = hyper::Request::new(());
+
+		req.headers_mut().insert(&X_FORWARDED_FOR, HeaderValue::from_static("    "));
+		let ip = read_ip_from_proxy(&req);
+		assert!(ip.is_none())
+	}
+}