Trusted proxy support, unauthenticated rate limit & other fixes (#7472)

* Trusted proxies, unauthenticated rate limits and various fixes

* Fix get_groups_data

* Fix get_groups_data when not using full_access

* Fmt

* Fix org import

* deduplicate send validation

src/api/notifications.rs

@@ -33,9 +33,14 @@ pub static WS_USERS: LazyLock<Arc<WebSocketUsers>> = LazyLock::new(|| {
 pub static WS_ANONYMOUS_SUBSCRIPTIONS: LazyLock<Arc<AnonymousWebSocketSubscriptions>> = LazyLock::new(|| {
     Arc::new(AnonymousWebSocketSubscriptions {
         map: Arc::new(dashmap::DashMap::new()),
+        connections: Arc::new(dashmap::DashMap::new()),
     })
 });
 
+/// The anonymous hub needs no authentication, so bound how much a single client can hold open.
+/// One connection is needed per pending login request, several at once are only expected behind NAT.
+const MAX_ANONYMOUS_CONNECTIONS_PER_IP: u32 = 25;
+
 static NOTIFICATIONS_DISABLED: LazyLock<bool> = LazyLock::new(|| !CONFIG.enable_websocket() && !CONFIG.push_enabled());
 
 pub fn routes() -> Vec<Route> {
@@ -82,14 +87,21 @@ impl Drop for WSEntryMapGuard {
 struct WSAnonymousEntryMapGuard {
     subscriptions: Arc<AnonymousWebSocketSubscriptions>,
     token: String,
+    entry_uuid: uuid::Uuid,
     addr: IpAddr,
 }
 
 impl WSAnonymousEntryMapGuard {
-    fn new(subscriptions: Arc<AnonymousWebSocketSubscriptions>, token: String, addr: IpAddr) -> Self {
+    fn new(
+        subscriptions: Arc<AnonymousWebSocketSubscriptions>,
+        token: String,
+        entry_uuid: uuid::Uuid,
+        addr: IpAddr,
+    ) -> Self {
         Self {
             subscriptions,
             token,
+            entry_uuid,
             addr,
         }
     }
@@ -98,7 +110,11 @@ impl WSAnonymousEntryMapGuard {
 impl Drop for WSAnonymousEntryMapGuard {
     fn drop(&mut self) {
         info!("Closing WS connection from {}", self.addr);
-        self.subscriptions.map.remove(&self.token);
+        if let Some(mut entry) = self.subscriptions.map.get_mut(&self.token) {
+            entry.retain(|(uuid, _)| uuid != &self.entry_uuid);
+        }
+        self.subscriptions.map.remove_if(&self.token, |_, senders| senders.is_empty());
+        self.subscriptions.release(self.addr);
     }
 }
 
@@ -194,12 +210,19 @@ fn anonymous_websockets_hub<'r>(ws: WebSocket, token: String, ip: ClientIp) -> R
     let (mut rx, guard) = {
         let subscriptions = Arc::clone(&WS_ANONYMOUS_SUBSCRIPTIONS);
 
-        // Add a channel to send messages to this client to the map
+        if !subscriptions.try_reserve(ip.ip) {
+            err_code!("Too many connections", 429)
+        }
+
+        // Add a channel to send messages to this client to the map.
+        // Clients reconnect with the same token while a login request is still pending, so keep
+        // every subscriber instead of replacing, otherwise the older one takes the newer one down.
         let (tx, rx) = tokio::sync::mpsc::channel::<Message>(100);
-        subscriptions.map.insert(token.clone(), tx);
+        let entry_uuid = uuid::Uuid::new_v4();
+        subscriptions.map.entry(token.clone()).or_default().push((entry_uuid, tx));
 
         // Once the guard goes out of scope, the connection will have been closed and the entry will be deleted from the map
-        (rx, WSAnonymousEntryMapGuard::new(subscriptions, token, ip.ip))
+        (rx, WSAnonymousEntryMapGuard::new(subscriptions, token, entry_uuid, ip.ip))
     };
 
     Ok({
@@ -534,15 +557,42 @@ impl WebSocketUsers {
 
 #[derive(Clone)]
 pub struct AnonymousWebSocketSubscriptions {
-    map: Arc<dashmap::DashMap<String, Sender<Message>>>,
+    map: Arc<dashmap::DashMap<String, Vec<UserSenders>>>,
+    connections: Arc<dashmap::DashMap<IpAddr, u32>>,
 }
 
 impl AnonymousWebSocketSubscriptions {
+    /// Takes a connection slot for this address, returns false when it already reached the limit.
+    fn try_reserve(&self, addr: IpAddr) -> bool {
+        let mut count = self.connections.entry(addr).or_insert(0);
+        if *count >= MAX_ANONYMOUS_CONNECTIONS_PER_IP {
+            return false;
+        }
+        *count += 1;
+        true
+    }
+
+    /// Releases a slot taken by `try_reserve`.
+    fn release(&self, addr: IpAddr) {
+        let empty = if let Some(mut count) = self.connections.get_mut(&addr) {
+            *count = count.saturating_sub(1);
+            *count == 0
+        } else {
+            false
+        };
+        // Only remove once the guard above is dropped, otherwise this deadlocks.
+        if empty {
+            self.connections.remove_if(&addr, |_, count| *count == 0);
+        }
+    }
+
     async fn send_update(&self, token: &str, data: &[u8]) {
-        if let Some(sender) = self.map.get(token).map(|v| v.clone())
-            && let Err(e) = sender.send(Message::binary(data)).await
-        {
-            error!("Error sending WS update {e}");
+        // Clone the senders so the map isn't kept locked while sending.
+        let senders = self.map.get(token).map(|v| v.clone()).unwrap_or_default();
+        for (_, sender) in senders {
+            if let Err(e) = sender.send(Message::binary(data)).await {
+                error!("Error sending WS update {e}");
+            }
         }
     }