compose.yaml

@@ -8,7 +8,7 @@ services:
       - "${HOST_PORT:-8080}:8080"
     environment:
       # Comma-separated client versions served at GET /versions.
-      SUPPORTED_VERSIONS: ${SUPPORTED_VERSIONS:-0.1.0-beta3}
+      SUPPORTED_VERSIONS: ${SUPPORTED_VERSIONS:-0.1.1-beta}
       DEPRECATED_VERSIONS: ${DEPRECATED_VERSIONS:-}
     volumes:
       - server-data:/data

src/cosmetics.rs

@@ -44,8 +44,7 @@ pub async fn buy(
     item_id: String,
     pool: Arc<SqlitePool>,
 ) -> Result<String, String> {
-    // Lock once
+    let uuid = crate::methods::auth::require(&uuid).await?;
-    let uuid = uuid.lock().await.clone();
 
     let mut user = user::get(&uuid, &pool).await?;
 
@@ -88,8 +87,7 @@ pub async fn equip(
     item_id: String,
     pool: Arc<SqlitePool>,
 ) -> Result<String, String> {
-    // Lock once
+    let uuid = crate::methods::auth::require(&uuid).await?;
-    let uuid = uuid.lock().await.clone();
 
     let mut user = user::get(&uuid, &pool).await?;
 

src/limits.rs

@@ -0,0 +1,79 @@
+use std::{
+    collections::HashSet,
+    time::{Duration, Instant},
+};
+
+use tokio::sync::Mutex;
+
+/// Most players a single `emote` or `send_player` request may target.
+pub const MAX_TARGETS: usize = 256;
+
+/// Validates and de-duplicates a request's target UUIDs.
+pub fn targets(targets: Vec<String>) -> Result<HashSet<String>, String> {
+    if targets.len() > MAX_TARGETS {
+        return Err(format!("Too many targets (max {MAX_TARGETS})"));
+    }
+
+    Ok(targets.into_iter().collect())
+}
+
+/// Token bucket: allows bursts of up to `burst` requests, refilled at one
+/// token per `refill`.
+pub struct RateLimiter {
+    state: Mutex<(f64, Instant)>,
+    burst: f64,
+    per_sec: f64,
+}
+
+impl RateLimiter {
+    pub fn new(burst: u32, refill: Duration) -> Self {
+        Self {
+            state: Mutex::new((burst as f64, Instant::now())),
+            burst: burst as f64,
+            per_sec: 1.0 / refill.as_secs_f64(),
+        }
+    }
+
+    pub async fn check(&self) -> Result<(), String> {
+        let mut state = self.state.lock().await;
+        let (tokens, last) = &mut *state;
+
+        let now = Instant::now();
+        *tokens = (*tokens + now.duration_since(*last).as_secs_f64() * self.per_sec).min(self.burst);
+        *last = now;
+
+        if *tokens < 1.0 {
+            return Err("Rate limited, try again shortly".to_string());
+        }
+
+        *tokens -= 1.0;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn allows_a_burst_then_refills() {
+        let limiter = RateLimiter::new(2, Duration::from_millis(50));
+
+        assert!(limiter.check().await.is_ok());
+        assert!(limiter.check().await.is_ok());
+        assert!(limiter.check().await.is_err());
+
+        tokio::time::sleep(Duration::from_millis(60)).await;
+        assert!(limiter.check().await.is_ok());
+        assert!(limiter.check().await.is_err());
+    }
+
+    #[test]
+    fn caps_and_dedupes_targets() {
+        let many = (0..=MAX_TARGETS).map(|i| i.to_string()).collect();
+        assert!(targets(many).is_err());
+
+        let dupes = vec!["a".to_string(), "a".to_string(), "b".to_string()];
+        assert_eq!(targets(dupes).unwrap().len(), 2);
+    }
+}

src/main.rs

@@ -1,4 +1,5 @@
 mod cosmetics;
+mod limits;
 mod methods;
 mod types;
 mod user;
@@ -51,7 +52,7 @@ impl VersionManifest {
         };
 
         Self {
-            supported: list("SUPPORTED_VERSIONS", "0.1.0-beta3"),
+            supported: list("SUPPORTED_VERSIONS", "0.1.1-beta"),
             deprecated: list("DEPRECATED_VERSIONS", ""),
         }
     }
@@ -121,6 +122,10 @@ async fn register_handlers(session: &Session, state: AppState) {
     let name = Arc::new(Mutex::new(String::new()));
     let pending: methods::auth::PendingChallenge = Arc::new(Mutex::new(None));
 
+    // Per-connection limits on the methods that fan out to other players.
+    let emote_limit = Arc::new(limits::RateLimiter::new(5, Duration::from_secs(1)));
+    let send_player_limit = Arc::new(limits::RateLimiter::new(10, Duration::from_secs(1)));
+
     session
         .on_close({
             let session = session.clone();
@@ -245,7 +250,14 @@ async fn register_handlers(session: &Session, state: AppState) {
             let sessions = Arc::clone(&sessions);
             let uuid = Arc::clone(&uuid);
 
-            move |_, emote| methods::emote::send_emote(Arc::clone(&sessions), Arc::clone(&uuid), emote)
+            move |_, emote| {
+                methods::emote::send_emote(
+                    Arc::clone(&sessions),
+                    Arc::clone(&uuid),
+                    Arc::clone(&emote_limit),
+                    emote,
+                )
+            }
         })
         .await;
 
@@ -254,7 +266,9 @@ async fn register_handlers(session: &Session, state: AppState) {
             let sessions = Arc::clone(&sessions);
             let pool = Arc::clone(&pool);
 
-            move |_, uuid| methods::user::get_user(sessions.clone(), uuid, pool.clone())
+            let uuid = Arc::clone(&uuid);
+
+            move |_, target| methods::user::get_user(sessions.clone(), uuid.clone(), target, pool.clone())
         })
         .await;
 
@@ -270,6 +284,7 @@ async fn register_handlers(session: &Session, state: AppState) {
                     sessions.clone(),
                     name.clone(),
                     uuid.clone(),
+                    Arc::clone(&send_player_limit),
                     targets.targets,
                     pool.clone(),
                 )

src/methods/auth.rs

@@ -20,6 +20,17 @@ pub struct Challenge {
 
 pub type PendingChallenge = Arc<Mutex<Option<Challenge>>>;
 
+/// Returns the connection's authenticated UUID, or an error before `auth_verify`.
+pub async fn require(uuid: &UUID) -> Result<String, String> {
+    let uuid = uuid.lock().await;
+
+    if uuid.is_empty() {
+        return Err("Not authenticated".to_string());
+    }
+
+    Ok(uuid.clone())
+}
+
 /// Minecraft usernames: 1–16 characters of letters, digits and underscores.
 fn is_valid_username(name: &str) -> bool {
     (1..=16).contains(&name.len()) && name.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'_')

src/methods/emote.rs

@@ -1,16 +1,26 @@
-use crate::types::{SessionMap, UUID};
+use std::sync::Arc;
+
+use crate::{
+    limits::{self, RateLimiter},
+    types::{SessionMap, UUID},
+};
 
 pub async fn send_emote(
     sessions: SessionMap,
     uuid: UUID,
+    limit: Arc<RateLimiter>,
     emote: crate::types::EmoteRequest,
 ) -> Result<(), String> {
+    let from = crate::methods::auth::require(&uuid).await?;
+    let targets = limits::targets(emote.targets)?;
+    limit.check().await?;
+
     let emote_event = crate::types::EventEmote {
-        from: uuid.lock().await.clone(),
+        from,
         emote: emote.emote,
     };
 
-    for target in emote.targets {
+    for target in targets {
         if let Some(sessions) = sessions.lock().await.get_mut(&target) {
             let emote_event = emote_event.clone();
             let mut bad_sessions = Vec::new();

src/methods/user.rs

@@ -3,6 +3,7 @@ use std::sync::Arc;
 use sqlx::SqlitePool;
 
 use crate::{
+    limits::{self, RateLimiter},
     methods,
     types::{PlayerStream, SessionMap, UUID},
     user::User,
@@ -10,9 +11,12 @@ use crate::{
 
 pub async fn get_user(
     sessions: SessionMap,
+    caller: UUID,
     uuid: String,
     pool: Arc<SqlitePool>,
 ) -> Result<Option<User>, String> {
+    crate::methods::auth::require(&caller).await?;
+
     if !sessions.lock().await.contains_key(&uuid) {
         return Ok(None);
     }
@@ -24,11 +28,15 @@ pub async fn send_user(
     sessions: SessionMap,
     name: UUID,
     uuid: UUID,
+    limit: Arc<RateLimiter>,
     targets: Vec<String>,
     pool: Arc<SqlitePool>,
 ) -> Result<(), String> {
+    let uuid = crate::methods::auth::require(&uuid).await?;
+    let targets = limits::targets(targets)?;
+    limit.check().await?;
+
     println!("send player {targets:?}");
-    let uuid = uuid.lock().await.to_string();
 
     let user = PlayerStream {
         player: crate::user::get(&uuid, pool.as_ref()).await?,