[fix] require an account to subscribe to the alert and manager streams

hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java

@@ -19,9 +19,12 @@
 
 package org.apache.hertzbeat.alert.config;
 
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpStatus;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
+import org.springframework.web.server.ResponseStatusException;
 import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
@@ -36,17 +39,43 @@
 @Slf4j
 @Component
 public class AlertSseManager {
+
+    /**
+     * How long a subscription may stay open before the client has to reconnect.
+     *
+     * <p>`Long.MAX_VALUE` meant a subscription never expired on its own, so a client that
+     * went away without closing cleanly held its request thread until the container noticed.
+     * A finite timeout bounds that; browsers reconnect on timeout, and the ui re-subscribes.
+     */
+    private static final long EMITTER_TIMEOUT_MILLIS = 30 * 60 * 1000L;
+
+    /**
+     * Cap on concurrently held subscriptions. Each one occupies a request thread, so without
+     * a ceiling enough parallel subscriptions exhaust the container's thread pool and take
+     * the whole application down with them.
+     */
+    @Setter
+    private int maxEmitters = 1000;
+
     private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();
 
     public SseEmitter createEmitter(Long clientId) {
-        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
+        if (emitters.size() >= maxEmitters) {
+            log.warn("Refused alert subscription, already holding {} of at most {}", emitters.size(), maxEmitters);
+            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "Too many alert subscriptions");
+        }
+        SseEmitter emitter = new SseEmitter(EMITTER_TIMEOUT_MILLIS);
         emitter.onCompletion(() -> removeEmitter(clientId));
         emitter.onTimeout(() -> removeEmitter(clientId));
         emitter.onError((ex) -> removeEmitter(clientId));
         emitters.put(clientId, emitter);
         return emitter;
     }
 
+    int subscriptionCount() {
+        return emitters.size();
+    }
+
     @Async
     public void broadcast(String data) {
         emitters.forEach((clientId, emitter) -> {

hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java

@@ -19,15 +19,19 @@
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.server.ResponseStatusException;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 import java.lang.reflect.Field;
 import java.util.HashMap;
 import java.util.Map;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
@@ -66,4 +70,56 @@ void testCompleteThrowsException() throws Exception {
         assertFalse(currentEmitters.containsKey(1L), "Emitter should still exist because complete() threw exception");
     }
 
+    /**
+     * An unbounded emitter never expires on its own, so a client that goes away without
+     * closing cleanly keeps holding its request thread until the container notices.
+     */
+    @Test
+    void testSubscriptionsAreGivenFiniteTimeout() {
+        SseEmitter emitter = alertSseManager.createEmitter(1L);
+
+        assertNotNull(emitter.getTimeout());
+        assertTrue(emitter.getTimeout() > 0 && emitter.getTimeout() < Long.MAX_VALUE,
+                "timeout must be finite, was " + emitter.getTimeout());
+    }
+
+    /**
+     * Each open subscription occupies a request thread, so enough of them in parallel
+     * exhaust the container's pool and take the whole application down.
+     */
+    @Test
+    void testSubscriptionsBeyondLimitAreRefused() {
+        alertSseManager.setMaxEmitters(2);
+
+        alertSseManager.createEmitter(1L);
+        alertSseManager.createEmitter(2L);
+        ResponseStatusException thrown =
+                assertThrows(ResponseStatusException.class, () -> alertSseManager.createEmitter(3L));
+
+        assertEquals(HttpStatus.SERVICE_UNAVAILABLE, thrown.getStatusCode());
+        assertEquals(2, alertSseManager.subscriptionCount());
+    }
+
+    /**
+     * The cap must not become a permanent lockout: once a dead subscription is cleaned up,
+     * its slot has to be available again.
+     */
+    @Test
+    void testDroppedSubscriptionFreesItsSlot() throws Exception {
+        alertSseManager.setMaxEmitters(1);
+        alertSseManager.createEmitter(1L);
+        assertThrows(ResponseStatusException.class, () -> alertSseManager.createEmitter(2L));
+
+        // a client that went away makes the next send fail, which is how the manager notices
+        SseEmitter deadEmitter = mock(SseEmitter.class);
+        doThrow(new IllegalStateException("client gone")).when(deadEmitter).send(any(SseEmitter.SseEventBuilder.class));
+        Field emittersField = AlertSseManager.class.getDeclaredField("emitters");
+        emittersField.setAccessible(true);
+        ((Map<Long, SseEmitter>) emittersField.get(alertSseManager)).put(1L, deadEmitter);
+
+        alertSseManager.broadcast("{\"id\":1}");
+
+        assertEquals(0, alertSseManager.subscriptionCount());
+        assertNotNull(alertSseManager.createEmitter(2L));
+    }
 }

hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java

@@ -19,13 +19,16 @@
 
 package org.apache.hertzbeat.manager.config;
 
+import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.common.constants.ManagerEventTypeEnum;
 import org.apache.hertzbeat.common.entity.dto.ImportTaskMessage;
 import org.apache.hertzbeat.common.entity.dto.ManagerMessage;
 import org.apache.hertzbeat.common.util.JsonUtil;
+import org.springframework.http.HttpStatus;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
+import org.springframework.web.server.ResponseStatusException;
 import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
@@ -40,16 +43,42 @@
 @Slf4j
 @Component
 public class ManagerSseManager {
+
+    /**
+     * How long a subscription may stay open before the client has to reconnect.
+     *
+     * <p>`Long.MAX_VALUE` meant a subscription never expired on its own, so a client that
+     * went away without closing cleanly held its request thread until the container noticed.
+     * A finite timeout bounds that; browsers reconnect on timeout, and the ui re-subscribes.
+     */
+    private static final long EMITTER_TIMEOUT_MILLIS = 30 * 60 * 1000L;
+
+    /**
+     * Cap on concurrently held subscriptions. Each one occupies a request thread, so without
+     * a ceiling enough parallel subscriptions exhaust the container's thread pool and take
+     * the whole application down with them.
+     */
+    @Setter
+    private int maxEmitters = 1000;
+
     private final Map<Long, SseEmitter> emitters = new ConcurrentHashMap<>();
 
     public SseEmitter createEmitter(Long clientId) {
-        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
+        if (emitters.size() >= maxEmitters) {
+            log.warn("Refused manager subscription, already holding {} of at most {}", emitters.size(), maxEmitters);
+            throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, "Too many manager subscriptions");
+        }
+        SseEmitter emitter = new SseEmitter(EMITTER_TIMEOUT_MILLIS);
         emitter.onCompletion(() -> removeEmitter(clientId));
         emitter.onTimeout(() -> removeEmitter(clientId));
         emitters.put(clientId, emitter);
         return emitter;
     }
 
+    int subscriptionCount() {
+        return emitters.size();
+    }
+
     @Async
     public void broadcast(String eventName, String data) {
         emitters.forEach((clientId, emitter) -> {

hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/ManagerSseManagerTest.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.manager.config;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.server.ResponseStatusException;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+/**
+ * Test case for {@link ManagerSseManager}.
+ *
+ * <p>Every open subscription holds a request thread for as long as it lives, so both how
+ * long one may live and how many may exist at once have to be bounded.
+ */
+class ManagerSseManagerTest {
+
+    private ManagerSseManager managerSseManager;
+
+    @BeforeEach
+    void setUp() {
+        managerSseManager = new ManagerSseManager();
+    }
+
+    @Test
+    void testSubscriptionsAreGivenFiniteTimeout() {
+        SseEmitter emitter = managerSseManager.createEmitter(1L);
+
+        assertNotNull(emitter.getTimeout());
+        assertTrue(emitter.getTimeout() > 0 && emitter.getTimeout() < Long.MAX_VALUE,
+                "timeout must be finite, was " + emitter.getTimeout());
+    }
+
+    @Test
+    void testSubscriptionsBeyondLimitAreRefused() {
+        managerSseManager.setMaxEmitters(1);
+
+        managerSseManager.createEmitter(1L);
+        ResponseStatusException thrown =
+                assertThrows(ResponseStatusException.class, () -> managerSseManager.createEmitter(2L));
+
+        assertEquals(HttpStatus.SERVICE_UNAVAILABLE, thrown.getStatusCode());
+        assertEquals(1, managerSseManager.subscriptionCount());
+    }
+}

hertzbeat-startup/src/main/resources/sureness.yml

@@ -93,19 +93,21 @@ resourceRole:
   - /api/account/token===get===[admin]
   - /api/account/token/**===post===[admin]
   - /api/account/token/**===delete===[admin]
+  # the alert stream carries full alert payloads and the manager stream carries import
+  # progress; both are scoped like the log stream above rather than left anonymous
+  - /api/alert/sse/**===get===[admin,user,guest]
+  - /api/manager/sse/**===get===[admin,user,guest]
 
 # config the resource restful api that need bypass auth protection
 # rule: api===method 
 # eg: /api/v1/source3===get means /api/v1/source3===get can be access by anyone, no need auth.
 excludedResource:
-  - /api/alert/sse/**===*
   - /api/account/auth/**===*
   - /api/i18n/**===get
   - /api/apps/hierarchy===get
   - /api/observability/capability===get
   - /api/push/**===*
   - /api/status/page/public/**===*
-  - /api/manager/sse/**===*
   # web ui resource
   - /===get
   - /assets/**===get

hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/security/SurenessSseRuleTest.java

@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.startup.security;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import com.usthe.sureness.matcher.util.TirePathTree;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.yaml.snakeyaml.Yaml;
+
+/**
+ * Guards the rbac rules covering the server sent event streams.
+ *
+ * <p>`/api/alert/sse/**` and `/api/manager/sse/**` used to sit in `excludedResource`.
+ * Sureness evaluates the exclusion tree before any credential check, so an anonymous
+ * `curl -N` stayed subscribed and received every alert the deployment raised - internal
+ * hostnames, addresses, metric values and alert content - because
+ * `AlertNoticeDispatch` broadcasts each alert to every subscriber with no per subscriber
+ * filtering. `/api/logs/sse/**` was already scoped this way; these now match it.
+ */
+class SurenessSseRuleTest {
+
+    private static final String SEPARATOR = "===";
+
+    private static TirePathTree roleTree;
+
+    private static TirePathTree excludeTree;
+
+    @BeforeAll
+    @SuppressWarnings("unchecked")
+    static void loadSurenessConfig() throws IOException {
+        List<String> resourceRole;
+        List<String> excludedResource;
+        try (InputStream in = SurenessSseRuleTest.class.getResourceAsStream("/sureness.yml")) {
+            assertNotNull(in, "sureness.yml must be on the classpath");
+            Map<String, Object> document = new Yaml().load(in);
+            resourceRole = (List<String>) document.get("resourceRole");
+            excludedResource = (List<String>) document.get("excludedResource");
+        }
+        assertNotNull(resourceRole, "resourceRole must be present");
+        assertNotNull(excludedResource, "excludedResource must be present");
+        roleTree = new TirePathTree();
+        roleTree.buildTree(new LinkedHashSet<>(resourceRole));
+        excludeTree = new TirePathTree();
+        excludeTree.buildTree(new LinkedHashSet<>(excludedResource));
+    }
+
+    @Test
+    void subscribingToAlertsRequiresAnAccount() {
+        assertEquals("[admin,user,guest]",
+                roleTree.searchPathFilterRoles("/api/alert/sse/subscribe" + SEPARATOR + "get"));
+    }
+
+    @Test
+    void subscribingToManagerEventsRequiresAnAccount() {
+        assertEquals("[admin,user,guest]",
+                roleTree.searchPathFilterRoles("/api/manager/sse/subscribe" + SEPARATOR + "get"));
+    }
+
+    /**
+     * The rule above only takes effect if the path stops matching an exclusion: sureness
+     * returns from `checkIn` as soon as `isExcludedResource` matches, before any credential
+     * is looked at.
+     */
+    @Test
+    void theStreamsAreNoLongerAnonymous() {
+        assertNull(excludeTree.searchPathFilterRoles("/api/alert/sse/subscribe" + SEPARATOR + "get"));
+        assertNull(excludeTree.searchPathFilterRoles("/api/manager/sse/subscribe" + SEPARATOR + "get"));
+    }
+
+    @Test
+    void theLogStreamScopingIsUnchanged() {
+        assertEquals("[admin,user,guest]",
+                roleTree.searchPathFilterRoles("/api/logs/sse/subscribe" + SEPARATOR + "get"));
+    }
+}