Added per-IP rate limiting to Query Server

CHANGELOG.md

@@ -24,6 +24,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - HQX upscaling shader for people playing on low internal resolutions
 - Pets can now be traded with.
 - Crafting recipe for black lantern
+- Added Query Server to allow remote querying of server status
 
 ### Changed
 - Use fluent for translations

Cargo.lock

@@ -1591,6 +1591,17 @@ dependencies = [
  "syn 1.0.100",
 ]
 
+[[package]]
+name = "dashmap"
+version = "5.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "391b56fbd302e585b7a9494fb70e40949567b1cf9003a8e4a6041a1687c26573"
+dependencies = [
+ "cfg-if 1.0.0",
+ "hashbrown 0.12.3",
+ "lock_api 0.4.9",
+]
+
 [[package]]
 name = "deflate"
 version = "1.0.0"
@@ -2226,6 +2237,12 @@ version = "0.3.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a6508c467c73851293f390476d4491cf4d227dbabcd4170f3bb6044959b294f1"
 
+[[package]]
+name = "futures-timer"
+version = "3.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e64b03909df88034c26dc1547e8970b91f98bdb65165d6a4e9110d94263dbb2c"
+
 [[package]]
 name = "futures-util"
 version = "0.3.24"
@@ -2589,6 +2606,24 @@ dependencies = [
  "xi-unicode",
 ]
 
+[[package]]
+name = "governor"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "de1b4626e87b9eb1d603ed23067ba1e29ec1d0b35325a2b96c3fe1cf20871f56"
+dependencies = [
+ "cfg-if 1.0.0",
+ "dashmap",
+ "futures",
+ "futures-timer",
+ "no-std-compat",
+ "nonzero_ext",
+ "parking_lot 0.12.0",
+ "quanta",
+ "rand 0.8.5",
+ "smallvec",
+]
+
 [[package]]
 name = "gpu-alloc"
 version = "0.4.7"
@@ -3861,6 +3896,12 @@ dependencies = [
  "memoffset 0.6.5",
 ]
 
+[[package]]
+name = "no-std-compat"
+version = "0.4.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b93853da6d84c2e3c7d730d6473e8817692dd89be387eb01b94d7f108ecb5b8c"
+
 [[package]]
 name = "noise"
 version = "0.7.0"
@@ -3901,6 +3942,12 @@ dependencies = [
  "minimal-lexical",
 ]
 
+[[package]]
+name = "nonzero_ext"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "38bf9645c8b145698bb0b18a4637dcacbc421ea49bef2317e4fd8065a387cf21"
+
 [[package]]
 name = "notify"
 version = "5.0.0"
@@ -4717,6 +4764,22 @@ dependencies = [
  "syn 1.0.100",
 ]
 
+[[package]]
+name = "quanta"
+version = "0.9.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "20afe714292d5e879d8b12740aa223c6a88f118af41870e8b6196e39a02238a8"
+dependencies = [
+ "crossbeam-utils 0.8.11",
+ "libc",
+ "mach",
+ "once_cell",
+ "raw-cpuid",
+ "wasi 0.10.0+wasi-snapshot-preview1",
+ "web-sys",
+ "winapi 0.3.9",
+]
+
 [[package]]
 name = "quick-xml"
 version = "0.22.0"
@@ -4908,6 +4971,15 @@ name = "range-alloc"
 version = "0.1.2"
 source = "git+https://github.com/gfx-rs/gfx?rev=27a1dae3796d33d23812f2bb8c7e3b5aea18b521#27a1dae3796d33d23812f2bb8c7e3b5aea18b521"
 
+[[package]]
+name = "raw-cpuid"
+version = "10.6.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a6823ea29436221176fe662da99998ad3b4db2c7f31e7b6f5fe43adccd6320bb"
+dependencies = [
+ "bitflags",
+]
+
 [[package]]
 name = "raw-window-handle"
 version = "0.3.4"
@@ -6993,6 +7065,7 @@ dependencies = [
  "drop_guard",
  "enumset",
  "futures-util",
+ "governor",
  "hashbrown 0.12.3",
  "humantime",
  "itertools",

server/Cargo.toml

@@ -68,6 +68,7 @@ protocol = { version = "3.4", features = ["derive"] }
 
 rusqlite = { version = "0.24.2", features = ["array", "vtab", "bundled", "trace"] }
 refinery = { git = "https://gitlab.com/veloren/refinery.git", rev = "8ecf4b4772d791e6c8c0a3f9b66a7530fad1af3e", features = ["rusqlite"] }
+governor = "0.5.0"
 
 # Plugins
 plugin-api = { package = "veloren-plugin-api", path = "../plugin/api"}

server/src/query_server.rs

@@ -1,10 +1,12 @@
 use crate::{settings::ServerBattleMode, ServerInfoRequest};
 use common::resources::BattleMode;
+use governor::{clock::DefaultClock, state::keyed::DashMapStateStore, Quota, RateLimiter};
 use protocol::{wire::dgram, Protocol};
 use std::{
     io,
     io::Cursor,
-    net::{Ipv4Addr, SocketAddr, SocketAddrV4},
+    net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4},
+    num::NonZeroU32,
     time::Duration,
 };
 use tokio::{net::UdpSocket, sync::mpsc::UnboundedSender, time::timeout};
@@ -21,6 +23,7 @@ pub struct QueryServer {
     pub bind_addr: SocketAddr,
     pipeline: dgram::Pipeline<QueryServerPacketKind, protocol::wire::middleware::pipeline::Default>,
     server_info_request_sender: UnboundedSender<ServerInfoRequest>,
+    rate_limiter: RateLimiter<IpAddr, DashMapStateStore<IpAddr>, DefaultClock>,
 }
 
 impl QueryServer {
@@ -35,6 +38,10 @@ impl QueryServer {
                 protocol::Settings::default(),
             ),
             server_info_request_sender,
+            // The rate limit is currently set very low as the intended use of the query server is
+            // for the server browser to query servers periodically (no more than a few
+            // times per minute)
+            rate_limiter: RateLimiter::dashmap(Quota::per_minute(NonZeroU32::new(10).unwrap())),
         }
     }
 
@@ -42,11 +49,18 @@ impl QueryServer {
         let socket = UdpSocket::bind(self.bind_addr).await?;
 
         info!("Query Server running at {}", self.bind_addr);
-
         loop {
             let mut buf = vec![0; 1500];
             let (len, remote_addr) = socket.recv_from(&mut buf).await?;
 
+            if self.rate_limiter.check_key(&remote_addr.ip()).is_err() {
+                trace!(
+                    "Rate limit exceeded for address {:?}, dropping packet",
+                    remote_addr.ip()
+                );
+                continue;
+            }
+
             if !QueryServer::validate_datagram(len, &mut buf) {
                 continue;
             }
@@ -128,7 +142,7 @@ impl QueryServer {
     ) -> Result<(), QueryError> {
         // Ensure that the required padding is present
         if !query.padding.iter().all(|x| *x == 0xFF) {
-            return Err(QueryError::ValidationError("Missing padding".into()));
+            return Err(QueryError::Validation("Missing padding".into()));
         }
 
         // Send a request to the server_info system to retrieve the live server
@@ -141,7 +155,7 @@ impl QueryServer {
         };
         self.server_info_request_sender
             .send(req)
-            .map_err(|e| QueryError::ChannelError(format!("{}", e)))?;
+            .map_err(|e| QueryError::Channel(format!("{}", e)))?;
 
         tokio::spawn(async move {
             // Wait to receive the response from the server_info system
@@ -157,7 +171,7 @@ impl QueryServer {
                     },
                     Err(_) => {
                         //  Oneshot receive error
-                        Err(QueryError::ChannelError("Oneshot receive error".to_owned()))
+                        Err(QueryError::Channel("Oneshot receive error".to_owned()))
                     },
                 }
             })
@@ -171,33 +185,34 @@ impl QueryServer {
 
 #[derive(Debug)]
 enum QueryError {
-    NetworkError(io::Error),
+    Network(io::Error),
-    ProtocolError(protocol::Error),
+    Protocol(protocol::Error),
-    ChannelError(String),
+    Channel(String),
-    ValidationError(String),
+    Validation(String),
 }
 
 impl From<protocol::Error> for QueryError {
-    fn from(e: protocol::Error) -> Self { QueryError::ProtocolError(e) }
+    fn from(e: protocol::Error) -> Self { QueryError::Protocol(e) }
 }
 
 impl From<io::Error> for QueryError {
-    fn from(e: io::Error) -> Self { QueryError::NetworkError(e) }
+    fn from(e: io::Error) -> Self { QueryError::Network(e) }
 }
 
-#[derive(Protocol, Clone, Debug, PartialEq)]
+#[derive(Protocol, Clone, Debug, Eq, PartialEq)]
 pub struct Ping;
 
-#[derive(Protocol, Clone, Debug, PartialEq)]
+#[derive(Protocol, Clone, Debug, Eq, PartialEq)]
 pub struct Pong;
 
-#[derive(Protocol, Clone, Debug, PartialEq)]
+#[derive(Protocol, Clone, Debug, Eq, PartialEq)]
 pub struct ServerInfoQuery {
     // Used to avoid UDP amplification attacks by requiring more data to be sent than is received
     pub padding: [u8; 512],
 }
 
-#[derive(Protocol, Clone, Debug, PartialEq)]
+#[allow(clippy::large_enum_variant)]
+#[derive(Protocol, Clone, Debug, Eq, PartialEq)]
 #[protocol(discriminant = "integer")]
 #[protocol(discriminator(u8))]
 pub enum QueryServerPacketKind {
@@ -211,16 +226,16 @@ pub enum QueryServerPacketKind {
     ServerInfoResponse(ServerInfoResponse),
 }
 
-#[derive(Protocol, Debug, Clone, PartialEq)]
+#[derive(Protocol, Debug, Clone, Eq, PartialEq)]
 pub struct ServerInfoResponse {
     pub git_hash: String, /* TODO: use u8 array instead? String includes 8 bytes for capacity
                            * and length that we don't need */
     pub players_current: u16,
     pub players_max: u16,
-    pub battle_mode: QueryBattleMode, // TODO: use a custom enum to avoid accidental breakage
+    pub battle_mode: QueryBattleMode,
 }
 
-#[derive(Protocol, Debug, Clone, PartialEq)]
+#[derive(Protocol, Debug, Clone, Eq, PartialEq)]
 #[protocol(discriminant = "integer")]
 #[protocol(discriminator(u8))]
 pub enum QueryBattleMode {

server/src/settings.rs

@@ -37,7 +37,7 @@ const BANLIST_FILENAME: &str = "banlist.ron";
 const SERVER_DESCRIPTION_FILENAME: &str = "description.ron";
 const ADMINS_FILENAME: &str = "admins.ron";
 
-#[derive(Copy, Clone, Debug, PartialEq, Deserialize, Serialize)]
+#[derive(Copy, Clone, Debug, Deserialize, Serialize)]
 pub enum ServerBattleMode {
     Global(BattleMode),
     PerPlayer { default: BattleMode },

server/src/sys/server_info.rs

@@ -4,7 +4,7 @@ use specs::{Read, ReadStorage, WriteExpect};
 use tokio::sync::mpsc::UnboundedReceiver;
 use tracing::error;
 
-/// TODO: description
+/// Processes server info requests from the Query Server
 #[derive(Default)]
 pub struct Sys;
 impl<'a> System<'a> for Sys {
@@ -19,7 +19,7 @@ impl<'a> System<'a> for Sys {
     const PHASE: Phase = Phase::Create;
 
     fn run(_job: &mut Job<Self>, (clients, settings, mut receiver): Self::SystemData) {
-        let players_current = (&clients).count() as u16;
+        let players_current = clients.count() as u16;
         let server_info = ServerInfoResponse {
             players_current,
             players_max: settings.max_players,