feat: Ratelimit CSAPI endpoints

Cargo.lock

@@ -1037,6 +1037,7 @@ dependencies = [
  "either",
  "figment",
  "futures",
+ "governor",
  "hardened_malloc-rs",
  "http",
  "http-body-util",

conduwuit-example.toml

@@ -685,6 +685,13 @@
 # env: CONTINUWUITY_SENDER_SHUTDOWN_TIMEOUT
 #sender_shutdown_timeout = 5
 
+# Maximum number of requests per minute that unauthenticated clients
+# may make to ratelimited client-server API routes. This ratelimit is
+# global across all endpoints for each IP.
+#
+# env: CONTINUWUITY_UNAUTHENTICATED_REQUEST_RATELIMIT
+#unauthenticated_request_ratelimit = 30
+
 # Enables registration. If set to false, no users can register on this
 # server.
 #

src/api/router/args.rs

@@ -1,18 +1,21 @@
-use std::ops::Deref;
+use std::{
+	net::{IpAddr, Ipv4Addr},
+	ops::Deref,
+};
 
 use axum::{
 	RequestExt, RequestPartsExt,
 	body::Body,
 	extract::{FromRequest, Path, Query},
 };
-use conduwuit::{Error, Result, err};
+use conduwuit::{Error, Result, err, error};
 use ruma::{
 	CanonicalJsonObject,
 	api::{IncomingRequest, IncomingRequestExt, auth_scheme::AuthScheme},
 };
 use serde::Deserialize;
 
-use crate::{State, router::auth::CheckAuth};
+use crate::{State, client_ip::ClientIp, router::auth::CheckAuth};
 
 /// Query parameters needed to authenticate requests
 #[derive(Deserialize)]
@@ -87,6 +90,11 @@ where
 		let Path(path): Path<Vec<String>> = parts.extract().await?;
 		let borrowed_path: Vec<_> = path.iter().map(String::as_str).collect();
 		let Query(auth_query): Query<AuthQueryParams> = parts.extract().await?;
+		let ClientIp(ip) = parts
+			.extract_with_state(services)
+			.await
+			.inspect_err(|err| error!("Failed to extract client IP: {err:?}"))
+			.unwrap_or(ClientIp(IpAddr::V4(Ipv4Addr::UNSPECIFIED)));
 
 		// Assemble a new request from the read body and parts
 		let request = hyper::Request::from_parts(parts, body.as_ref());
@@ -102,7 +110,8 @@ where
 			})?;
 
 		let identity =
-			R::Authentication::check::<R>(services, &request, authentication, auth_query).await?;
+			R::Authentication::check::<R>(services, &request, ip, authentication, auth_query)
+				.await?;
 
 		// Deserialize the body
 		let body = R::try_from_http_request(request, &borrowed_path)

src/api/router/auth.rs

@@ -1,6 +1,7 @@
 use std::{
 	any::{Any, TypeId},
 	collections::BTreeSet,
+	net::IpAddr,
 };
 
 use conduwuit::{
@@ -18,7 +19,10 @@ use ruma::{
 			AuthScheme, NoAccessToken, NoAuthentication,
 		},
 		client,
-		error::{ErrorKind, InsufficientUserAuthenticationErrorData, UnknownTokenErrorData},
+		error::{
+			ErrorKind, InsufficientUserAuthenticationErrorData, LimitExceededErrorData,
+			UnknownTokenErrorData,
+		},
 		federation::authentication::ServerSignatures,
 	},
 	assign,
@@ -90,6 +94,7 @@ pub(crate) trait CheckAuth: AuthScheme {
 	fn check<R: IncomingRequest<Authentication = Self> + Any>(
 		services: &Services,
 		incoming_request: &hyper::Request<&[u8]>,
+		request_ip: IpAddr,
 		authentication: Self::Output,
 		query: AuthQueryParams,
 	) -> impl Future<Output = Result<Self::Identity>> + Send;
@@ -101,6 +106,7 @@ impl CheckAuth for ServerSignatures {
 	async fn check<R: IncomingRequest<Authentication = Self> + Any>(
 		services: &Services,
 		incoming_request: &hyper::Request<&[u8]>,
+		_request_ip: IpAddr,
 		authentication: Self::Output,
 		_query: AuthQueryParams,
 	) -> Result<Self::Identity> {
@@ -161,6 +167,7 @@ impl CheckAuth for AccessToken {
 	async fn check<R: IncomingRequest<Authentication = Self> + Any>(
 		services: &Services,
 		_incoming_request: &hyper::Request<&[u8]>,
+		_request_ip: IpAddr,
 		authentication: Self::Output,
 		query: AuthQueryParams,
 	) -> Result<Self::Identity> {
@@ -181,6 +188,7 @@ impl CheckAuth for AccessTokenOptional {
 	async fn check<R: IncomingRequest<Authentication = Self> + Any>(
 		services: &Services,
 		_incoming_request: &hyper::Request<&[u8]>,
+		request_ip: IpAddr,
 		authentication: Self::Output,
 		query: AuthQueryParams,
 	) -> Result<Self::Identity> {
@@ -195,6 +203,10 @@ impl CheckAuth for AccessTokenOptional {
 			.await
 			.map(Some)
 		} else {
+			if R::RATE_LIMITED {
+				check_unauthenticated_request_ratelimiting(services, request_ip)?;
+			}
+
 			Ok(None)
 		}
 	}
@@ -206,6 +218,7 @@ impl CheckAuth for AppserviceToken {
 	async fn check<R: IncomingRequest + Any>(
 		services: &Services,
 		_incoming_request: &hyper::Request<&[u8]>,
+		_request_ip: IpAddr,
 		authentication: Self::Output,
 		_query: AuthQueryParams,
 	) -> Result<Self::Identity> {
@@ -219,6 +232,7 @@ impl CheckAuth for AppserviceTokenOptional {
 	async fn check<R: IncomingRequest<Authentication = Self> + Any>(
 		services: &Services,
 		_incoming_request: &hyper::Request<&[u8]>,
+		_request_ip: IpAddr,
 		authentication: Self::Output,
 		_query: AuthQueryParams,
 	) -> Result<Self::Identity> {
@@ -238,6 +252,7 @@ impl CheckAuth for NoAuthentication {
 	fn check<R: IncomingRequest<Authentication = Self> + Any>(
 		_services: &Services,
 		_incoming_request: &hyper::Request<&[u8]>,
+		_request_ip: IpAddr,
 		_authentication: Self::Output,
 		_query: AuthQueryParams,
 	) -> impl Future<Output = Result<Self::Identity>> + Send {
@@ -251,6 +266,7 @@ impl CheckAuth for NoAccessToken {
 	async fn check<R: IncomingRequest<Authentication = Self> + Any>(
 		services: &Services,
 		incoming_request: &hyper::Request<&[u8]>,
+		request_ip: IpAddr,
 		_authentication: Self::Output,
 		query: AuthQueryParams,
 	) -> Result<Self::Identity> {
@@ -267,6 +283,10 @@ impl CheckAuth for NoAccessToken {
 			.await
 			.map(Some)
 		} else {
+			if R::RATE_LIMITED {
+				check_unauthenticated_request_ratelimiting(services, request_ip)?;
+			}
+
 			Ok(None)
 		}
 	}
@@ -408,3 +428,19 @@ async fn check_appservice_token(services: &Services, token: &str) -> Result<Regi
 
 	Ok(appservice_info)
 }
+
+fn check_unauthenticated_request_ratelimiting(services: &Services, ip: IpAddr) -> Result<()> {
+	if services
+		.config
+		.unauthenticated_request_ratelimit
+		.check_key(&ip)
+		.is_err()
+	{
+		return Err(Error::BadRequest(
+			ErrorKind::LimitExceeded(LimitExceededErrorData::new()),
+			"Rate limit exceeded.",
+		));
+	}
+
+	Ok(())
+}

src/core/Cargo.toml

@@ -77,6 +77,7 @@ cyborgtime.workspace = true
 either.workspace = true
 figment.workspace = true
 futures.workspace = true
+governor.workspace = true
 http-body-util.workspace = true
 http.workspace = true
 ipaddress.workspace = true

src/core/config/mod.rs

@@ -5,7 +5,10 @@ pub mod proxy;
 
 use std::{
 	collections::{BTreeMap, BTreeSet, HashMap},
+	hash::Hash,
 	net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
+	num::NonZero,
+	ops::Deref,
 	path::PathBuf,
 	time::Duration,
 };
@@ -17,6 +20,7 @@ use either::{
 };
 use figment::providers::{Env, Format, Toml};
 pub use figment::{Figment, value::Value as FigmentValue};
+use governor::{DefaultKeyedRateLimiter, Quota, RateLimiter};
 use lettre::message::Mailbox;
 use openidconnect::{Audience, ClientId, ClientSecret, Scope};
 use regex::RegexSet;
@@ -40,7 +44,7 @@ use crate::{Result, err, error::Error, utils::sys};
 /// All the config options for continuwuity.
 #[allow(clippy::struct_excessive_bools)]
 #[allow(rustdoc::broken_intra_doc_links, rustdoc::bare_urls)]
-#[derive(Clone, Debug, Deserialize)]
+#[derive(Debug, Deserialize)]
 #[config_example_generator(
 	filename = "conduwuit-example.toml",
 	section = "global",
@@ -790,6 +794,14 @@ pub struct Config {
 	#[serde(default = "default_sender_shutdown_timeout")]
 	pub sender_shutdown_timeout: u64,
 
+	/// Maximum number of requests per minute that unauthenticated clients
+	/// may make to ratelimited client-server API routes. This ratelimit is
+	/// global across all endpoints for each IP.
+	///
+	/// default: 30
+	#[serde(default = "default_unauthenticated_request_ratelimit")]
+	pub unauthenticated_request_ratelimit: PerMinuteRatelimiter<IpAddr>,
+
 	/// Enables registration. If set to false, no users can register on this
 	/// server.
 	///
@@ -2960,6 +2972,32 @@ impl TrustedServerConfig {
 	}
 }
 
+#[derive(Debug)]
+pub struct PerMinuteRatelimiter<K: Hash + Eq + Clone> {
+	ratelimiter: DefaultKeyedRateLimiter<K>,
+}
+
+impl<K: Hash + Eq + Clone> PerMinuteRatelimiter<K> {
+	fn new(quota: Quota) -> Self { Self { ratelimiter: RateLimiter::keyed(quota) } }
+}
+
+impl<K: Hash + Eq + Clone> Deref for PerMinuteRatelimiter<K> {
+	type Target = DefaultKeyedRateLimiter<K>;
+
+	fn deref(&self) -> &Self::Target { &self.ratelimiter }
+}
+
+impl<'de, K: Hash + Eq + Clone> Deserialize<'de> for PerMinuteRatelimiter<K> {
+	fn deserialize<D>(deserializer: D) -> std::prelude::v1::Result<Self, D::Error>
+	where
+		D: serde::Deserializer<'de>,
+	{
+		let max_burst = NonZero::<u32>::deserialize(deserializer)?;
+
+		Ok(Self::new(Quota::per_minute(max_burst)))
+	}
+}
+
 const DEPRECATED_KEYS: &[&str] = &[
 	"cache_capacity",
 	"conduit_cache_capacity_modifier",
@@ -3403,3 +3441,7 @@ fn default_preferred_username_claim() -> String { "preferred_username".to_owned(
 fn default_profile_key_map() -> HashMap<String, String> {
 	HashMap::from_iter([(ProfileFieldName::DisplayName.to_string(), "name".to_owned())])
 }
+
+fn default_unauthenticated_request_ratelimit<K: Hash + Eq + Clone>() -> PerMinuteRatelimiter<K> {
+	PerMinuteRatelimiter::new(Quota::per_minute(30.try_into().unwrap()))
+}