diff --git a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java b/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java index e3d245621e5..461846b5800 100644 --- a/hertzbeat-alerter/src/main/java/org/apache/hertzbeat/alert/config/AlertSseManager.java +++ b/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,10 +39,32 @@ @Slf4j @Component public class AlertSseManager { + + /** + * How long a subscription may stay open before the client has to reconnect. + * + *

`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 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)); @@ -47,6 +72,10 @@ public SseEmitter createEmitter(Long clientId) { return emitter; } + int subscriptionCount() { + return emitters.size(); + } + @Async public void broadcast(String data) { emitters.forEach((clientId, emitter) -> { diff --git a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java b/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java index 40e88ddef99..3fd3cdf23f6 100644 --- a/hertzbeat-alerter/src/test/java/org/apache/hertzbeat/alert/config/AlertSseManagerTest.java +++ b/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) emittersField.get(alertSseManager)).put(1L, deadEmitter); + + alertSseManager.broadcast("{\"id\":1}"); + + assertEquals(0, alertSseManager.subscriptionCount()); + assertNotNull(alertSseManager.createEmitter(2L)); + } } \ No newline at end of file diff --git a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java index 1fd681566c4..b721bd5df79 100644 --- a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/config/ManagerSseManager.java +++ b/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. + * + *

`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 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) -> { diff --git a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/ManagerSseManagerTest.java b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/config/ManagerSseManagerTest.java new file mode 100644 index 00000000000..c09461dc573 --- /dev/null +++ b/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}. + * + *

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()); + } +} diff --git a/hertzbeat-startup/src/main/resources/sureness.yml b/hertzbeat-startup/src/main/resources/sureness.yml index a3198da74b5..6e2f228ea5c 100644 --- a/hertzbeat-startup/src/main/resources/sureness.yml +++ b/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 diff --git a/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/security/SurenessSseRuleTest.java b/hertzbeat-startup/src/test/java/org/apache/hertzbeat/startup/security/SurenessSseRuleTest.java new file mode 100644 index 00000000000..041119b3a49 --- /dev/null +++ b/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. + * + *

`/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 resourceRole; + List excludedResource; + try (InputStream in = SurenessSseRuleTest.class.getResourceAsStream("/sureness.yml")) { + assertNotNull(in, "sureness.yml must be on the classpath"); + Map document = new Yaml().load(in); + resourceRole = (List) document.get("resourceRole"); + excludedResource = (List) 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")); + } +} diff --git a/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml b/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml index 28936362c9a..16d6a66b7ae 100644 --- a/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-mysql-iotdb/conf/sureness.yml @@ -86,19 +86,21 @@ resourceRole: - /api/ai/**===delete===[admin] - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] + # 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 diff --git a/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml b/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml index 28936362c9a..16d6a66b7ae 100644 --- a/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-mysql-tdengine/conf/sureness.yml @@ -86,19 +86,21 @@ resourceRole: - /api/ai/**===delete===[admin] - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] + # 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 diff --git a/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml b/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml index 28936362c9a..16d6a66b7ae 100644 --- a/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-mysql-victoria-metrics/conf/sureness.yml @@ -86,19 +86,21 @@ resourceRole: - /api/ai/**===delete===[admin] - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] + # 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 diff --git a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml index ee2e532df0b..3dd1b97dfc2 100644 --- a/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-postgresql-greptimedb/conf/sureness.yml @@ -90,19 +90,21 @@ resourceRole: - /api/ingestion/otlp/**===get===[admin,user,guest] - /api/logs/**===get===[admin,user,guest] - /api/traces/**===get===[admin,user,guest] + # 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 diff --git a/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml b/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml index 28936362c9a..16d6a66b7ae 100644 --- a/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml +++ b/script/docker-compose/hertzbeat-postgresql-victoria-metrics/conf/sureness.yml @@ -86,19 +86,21 @@ resourceRole: - /api/ai/**===delete===[admin] - /api/logs/sse/**===get===[admin,user,guest] - /api/logs/ingest/**===post===[admin,user] + # 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 diff --git a/script/sureness.yml b/script/sureness.yml index 2d6b4441eef..8acac8367d0 100644 --- a/script/sureness.yml +++ b/script/sureness.yml @@ -90,19 +90,21 @@ resourceRole: - /api/ingestion/otlp/**===get===[admin,user,guest] - /api/logs/**===get===[admin,user,guest] - /api/traces/**===get===[admin,user,guest] + # 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 diff --git a/web-app/src/app/layout/basic/widgets/notify.component.ts b/web-app/src/app/layout/basic/widgets/notify.component.ts index 022ff9a0e79..abb4b329b09 100644 --- a/web-app/src/app/layout/basic/widgets/notify.component.ts +++ b/web-app/src/app/layout/basic/widgets/notify.component.ts @@ -1,15 +1,17 @@ -import { ChangeDetectionStrategy, ChangeDetectorRef, Component, Inject, OnInit, OnDestroy } from '@angular/core'; +import { ChangeDetectionStrategy, ChangeDetectorRef, Component, Inject, NgZone, OnInit, OnDestroy } from '@angular/core'; import { Router } from '@angular/router'; import { I18NService } from '@core'; import { ALAIN_I18N_TOKEN } from '@delon/theme'; import { NzModalService } from 'ng-zorro-antd/modal'; import { NzNotificationService } from 'ng-zorro-antd/notification'; +import { Subscription } from 'rxjs'; import { finalize } from 'rxjs/operators'; import { Mute } from '../../../pojo/Mute'; import { SingleAlert } from '../../../pojo/SingleAlert'; import { AlertSoundService } from '../../../service/alert-sound.service'; import { AlertService } from '../../../service/alert.service'; +import { AuthorizedSseService } from '../../../service/authorized-sse.service'; import { GeneralConfigService } from '../../../service/general-config.service'; @Component({ @@ -125,7 +127,8 @@ export class HeaderNotifyComponent implements OnInit, OnDestroy { private previousCount = 0; // default to mute status mute: Mute = { mute: true }; - private eventSource!: EventSource; + private alertStream$!: Subscription; + private managerStream$!: Subscription; constructor( private router: Router, @Inject(ALAIN_I18N_TOKEN) private i18nSvc: I18NService, @@ -134,7 +137,9 @@ export class HeaderNotifyComponent implements OnInit, OnDestroy { private alertSvc: AlertService, private modal: NzModalService, private cdr: ChangeDetectorRef, - private alertSound: AlertSoundService + private alertSound: AlertSoundService, + private authorizedSseSvc: AuthorizedSseService, + private ngZone: NgZone ) {} ngOnInit(): void { @@ -166,9 +171,10 @@ export class HeaderNotifyComponent implements OnInit, OnDestroy { if (this.refreshInterval) { clearInterval(this.refreshInterval); } - if (this.eventSource) { - this.eventSource.close(); - } + // both streams are unsubscribed: they used to share one field, so the alert connection + // was never closed once the manager one overwrote it + this.alertStream$?.unsubscribe(); + this.managerStream$?.unsubscribe(); } onPopoverVisibleChange(visible: boolean): void { @@ -285,74 +291,70 @@ export class HeaderNotifyComponent implements OnInit, OnDestroy { } private initAlertSSEConnection(): void { - const sseUrl = '/api/alert/sse/subscribe'; - - this.eventSource = new EventSource(sseUrl); - - this.eventSource.addEventListener('ALERT_EVENT', (evt: MessageEvent) => { - let list: any[] = []; - let alert: SingleAlert = JSON.parse(evt.data); - let item = { - id: alert.id, - avatar: '/assets/img/notification.svg', - // title: `${alert.tags?.monitorName}--${this.i18nSvc.fanyi(`alert.severity.${alert.severity}`)}`, - title: alert.content, - datetime: new Date(alert.activeAt).toLocaleString(), - color: 'blue', - status: alert.status, - type: this.i18nSvc.fanyi('dashboard.alerts.title-no') - }; - console.log('alert:', alert); - list.push(item); - this.data = this.updateNoticeData(list); - if (!this.mute.mute && !this.notifiedAlert.includes(alert.id)) { - this.notifiedAlert.push(alert.id); - this.alertSound.playAlertSound(this.i18nSvc.currentLang); - const notification = new Notification(this.i18nSvc.fanyi('alert.notify.title'), { - body: this.i18nSvc.fanyi('alert.notify.body'), - icon: 'assets/logo.svg' - }); - notification.onclick = () => { - window.focus(); - this.router.navigateByUrl(`/alert/center`); - notification.close(); - }; - } - this.cdr.detectChanges(); + // read through AuthorizedSseService rather than EventSource: the stream now requires a + // credential, and EventSource cannot carry an Authorization header + this.alertStream$ = this.authorizedSseSvc.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({ + next: data => + this.ngZone.run(() => { + let list: any[] = []; + let alert: SingleAlert = JSON.parse(data); + let item = { + id: alert.id, + avatar: '/assets/img/notification.svg', + // title: `${alert.tags?.monitorName}--${this.i18nSvc.fanyi(`alert.severity.${alert.severity}`)}`, + title: alert.content, + datetime: new Date(alert.activeAt).toLocaleString(), + color: 'blue', + status: alert.status, + type: this.i18nSvc.fanyi('dashboard.alerts.title-no') + }; + console.log('alert:', alert); + list.push(item); + this.data = this.updateNoticeData(list); + if (!this.mute.mute && !this.notifiedAlert.includes(alert.id)) { + this.notifiedAlert.push(alert.id); + this.alertSound.playAlertSound(this.i18nSvc.currentLang); + const notification = new Notification(this.i18nSvc.fanyi('alert.notify.title'), { + body: this.i18nSvc.fanyi('alert.notify.body'), + icon: 'assets/logo.svg' + }); + notification.onclick = () => { + window.focus(); + this.router.navigateByUrl(`/alert/center`); + notification.close(); + }; + } + this.cdr.detectChanges(); + }), + error: error => console.error('SSE connection error:', error) }); - this.eventSource.onerror = error => { - console.error('SSE connection error:', error); - this.eventSource.close(); - }; } private initManagerSSEConnection(): void { - const sseUrl = '/api/manager/sse/subscribe'; - this.eventSource = new EventSource(sseUrl); - this.eventSource.addEventListener('IMPORT_TASK_EVENT', (evt: MessageEvent) => { - let msg = JSON.parse(evt.data); - if (msg.notifyLevel === 'SUCCESS') { - this.notifySvc.success( - this.i18nSvc.fanyi('common.notice'), - this.i18nSvc.fanyi('common.notify.import-success-detail', { taskName: msg.taskName }) - ); - } else if (msg.notifyLevel === 'ERROR') { - this.notifySvc.error( - this.i18nSvc.fanyi('common.notice'), - this.i18nSvc.fanyi('common.notify.import-fail-detail', { taskName: msg.taskName, errMsg: msg.errMsg }) - ); - } else if (msg.notifyLevel === 'INFO') { - this.notifySvc.info( - this.i18nSvc.fanyi('common.notice'), - this.i18nSvc.fanyi('common.notify.import-progress', { taskName: msg.taskName, progress: msg.progress }) - ); - } else { - console.error('Parse message error, msg:', evt.data); - } + this.managerStream$ = this.authorizedSseSvc.stream('/api/manager/sse/subscribe', 'IMPORT_TASK_EVENT').subscribe({ + next: data => + this.ngZone.run(() => { + let msg = JSON.parse(data); + if (msg.notifyLevel === 'SUCCESS') { + this.notifySvc.success( + this.i18nSvc.fanyi('common.notice'), + this.i18nSvc.fanyi('common.notify.import-success-detail', { taskName: msg.taskName }) + ); + } else if (msg.notifyLevel === 'ERROR') { + this.notifySvc.error( + this.i18nSvc.fanyi('common.notice'), + this.i18nSvc.fanyi('common.notify.import-fail-detail', { taskName: msg.taskName, errMsg: msg.errMsg }) + ); + } else if (msg.notifyLevel === 'INFO') { + this.notifySvc.info( + this.i18nSvc.fanyi('common.notice'), + this.i18nSvc.fanyi('common.notify.import-progress', { taskName: msg.taskName, progress: msg.progress }) + ); + } else { + console.error('Parse message error, msg:', data); + } + }), + error: error => console.error('Manager SSE connection error:', error) }); - this.eventSource.onerror = error => { - console.error('Manager SSE connection error:', error); - this.eventSource.close(); - }; } } diff --git a/web-app/src/app/routes/alert/alert-center/alert-center.component.ts b/web-app/src/app/routes/alert/alert-center/alert-center.component.ts index 2d734e89fba..6c94f3e1841 100644 --- a/web-app/src/app/routes/alert/alert-center/alert-center.component.ts +++ b/web-app/src/app/routes/alert/alert-center/alert-center.component.ts @@ -17,14 +17,16 @@ * under the License. */ -import { Component, Inject, OnDestroy, OnInit } from '@angular/core'; +import { Component, Inject, NgZone, OnDestroy, OnInit } from '@angular/core'; import { I18NService } from '@core'; import { ALAIN_I18N_TOKEN } from '@delon/theme'; import { NzModalService } from 'ng-zorro-antd/modal'; import { NzNotificationService } from 'ng-zorro-antd/notification'; +import { Subscription } from 'rxjs'; import { GroupAlert } from '../../../pojo/GroupAlert'; import { AlertService } from '../../../service/alert.service'; +import { AuthorizedSseService } from '../../../service/authorized-sse.service'; interface ExtendedGroupAlert extends GroupAlert { isNew?: boolean; @@ -39,6 +41,8 @@ export class AlertCenterComponent implements OnInit, OnDestroy { private notifySvc: NzNotificationService, private modal: NzModalService, private alertSvc: AlertService, + private authorizedSseSvc: AuthorizedSseService, + private ngZone: NgZone, @Inject(ALAIN_I18N_TOKEN) private i18nSvc: I18NService ) {} @@ -50,7 +54,7 @@ export class AlertCenterComponent implements OnInit, OnDestroy { checkedAlertIds = new Set(); filterStatus!: string; filterContent: string | undefined; - private eventSource!: EventSource; + private alertStream$!: Subscription; ngOnInit(): void { this.loadAlertsTable(); @@ -58,28 +62,24 @@ export class AlertCenterComponent implements OnInit, OnDestroy { } ngOnDestroy(): void { - if (this.eventSource) { - this.eventSource.close(); - } + this.alertStream$?.unsubscribe(); } // Initialize SSE subscription for real-time alerts private initSSESubscription(): void { - this.eventSource = new EventSource('/api/alert/sse/subscribe'); - this.eventSource.addEventListener('ALERT_EVENT', (evt: MessageEvent) => { - try { - const newAlert: GroupAlert = JSON.parse(evt.data); - this.updateAlertList(newAlert); - } catch (error) { - console.error('Error parsing SSE data:', error); - } + // read through AuthorizedSseService rather than EventSource: the stream now requires a + // credential, and EventSource cannot carry an Authorization header + this.alertStream$ = this.authorizedSseSvc.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({ + next: data => { + try { + const newAlert: GroupAlert = JSON.parse(data); + this.ngZone.run(() => this.updateAlertList(newAlert)); + } catch (error) { + console.error('Error parsing SSE data:', error); + } + }, + error: error => console.error('SSE connection error:', error) }); - - // Handle SSE errors - this.eventSource.onerror = error => { - console.error('SSE connection error:', error); - this.eventSource.close(); - }; } private updateAlertList(newAlert: GroupAlert): void { diff --git a/web-app/src/app/service/authorized-sse.service.spec.ts b/web-app/src/app/service/authorized-sse.service.spec.ts new file mode 100644 index 00000000000..895d67b7b3d --- /dev/null +++ b/web-app/src/app/service/authorized-sse.service.spec.ts @@ -0,0 +1,103 @@ +/* + * 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. + */ + +import { TestBed } from '@angular/core/testing'; + +import { AuthorizedSseService } from './authorized-sse.service'; +import { LocalStorageService } from './local-storage.service'; + +describe('AuthorizedSseService', () => { + let service: AuthorizedSseService; + let localStorageService: jasmine.SpyObj; + + /** Serves the given chunks as a readable body, the way a live sse response arrives. */ + function respondWith(chunks: string[], ok = true, status = 200): void { + const encoder = new TextEncoder(); + let index = 0; + const reader = { + read: () => (index < chunks.length ? Promise.resolve({ value: encoder.encode(chunks[index++]), done: false }) : new Promise(() => {})) // an open stream never completes on its own + }; + spyOn(window, 'fetch').and.returnValue(Promise.resolve({ ok, status, body: { getReader: () => reader } } as any)); + } + + beforeEach(() => { + localStorageService = jasmine.createSpyObj('LocalStorageService', ['getAuthorizationToken']); + TestBed.configureTestingModule({ + providers: [{ provide: LocalStorageService, useValue: localStorageService }] + }); + service = TestBed.inject(AuthorizedSseService); + }); + + it('sends the stored token so the stream can require a credential', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondWith(['event:ALERT_EVENT\ndata:{"id":1}\n\n']); + + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(() => { + const [, init] = (window.fetch as jasmine.Spy).calls.mostRecent().args; + expect((init.headers as Record)['Authorization']).toBe('Bearer a-token'); + subscription.unsubscribe(); + done(); + }); + }); + + it('emits the data payload of a matching event', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondWith(['event:ALERT_EVENT\ndata:{"id":1}\n\n']); + + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(data => { + expect(data).toBe('{"id":1}'); + subscription.unsubscribe(); + done(); + }); + }); + + it('ignores events of another name on the same stream', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondWith(['event:OTHER_EVENT\ndata:{"id":1}\n\n', 'event:ALERT_EVENT\ndata:{"id":2}\n\n']); + + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(data => { + expect(data).toBe('{"id":2}'); + subscription.unsubscribe(); + done(); + }); + }); + + it('reassembles an event split across chunks', done => { + localStorageService.getAuthorizationToken.and.returnValue('a-token'); + respondWith(['event:ALERT_EVENT\ndata:{"id"', ':3}\n\n']); + + const subscription = service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe(data => { + expect(data).toBe('{"id":3}'); + subscription.unsubscribe(); + done(); + }); + }); + + it('surfaces a rejected subscription as an error', done => { + localStorageService.getAuthorizationToken.and.returnValue(null as any); + respondWith([], false, 401); + + service.stream('/api/alert/sse/subscribe', 'ALERT_EVENT').subscribe({ + error: error => { + expect(String(error)).toContain('401'); + done(); + } + }); + }); +}); diff --git a/web-app/src/app/service/authorized-sse.service.ts b/web-app/src/app/service/authorized-sse.service.ts new file mode 100644 index 00000000000..2be4178ba1d --- /dev/null +++ b/web-app/src/app/service/authorized-sse.service.ts @@ -0,0 +1,109 @@ +/* + * 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. + */ + +import { Injectable, NgZone } from '@angular/core'; +import { Observable } from 'rxjs'; + +import { LocalStorageService } from './local-storage.service'; + +/** + * Reads a server sent event stream with the bearer token attached. + * + * The browser's own `EventSource` cannot carry an `Authorization` header, which is why the + * alert and manager streams used to be reachable without any credential at all. Reading the + * stream through `fetch` instead lets the token travel with the request, so the endpoints can + * be moved behind the same rbac rules as the rest of the api. + * + * The returned observable starts the request on subscribe and aborts it on unsubscribe. + * Events are emitted outside the angular zone; a caller that touches component state should + * re-enter the zone itself, as it would with any other stream. + */ +@Injectable({ providedIn: 'root' }) +export class AuthorizedSseService { + constructor(private localStorageService: LocalStorageService, private ngZone: NgZone) {} + + /** + * @param url stream endpoint, relative to the origin + * @param eventName name of the sse event to emit; other events are ignored + * @returns the `data` payload of every matching event, as raw text + */ + stream(url: string, eventName: string): Observable { + return new Observable(subscriber => { + const abortController = new AbortController(); + const token = this.localStorageService.getAuthorizationToken(); + const headers: Record = { + Accept: 'text/event-stream', + 'Cache-Control': 'no-cache' + }; + if (token) { + headers['Authorization'] = `Bearer ${token}`; + } + + this.ngZone.runOutsideAngular(() => { + fetch(url, { method: 'GET', headers, signal: abortController.signal }) + .then(async response => { + if (!response.ok) { + throw new Error(`SSE request to ${url} failed with status ${response.status}`); + } + const reader = response.body?.getReader(); + if (!reader) { + throw new Error(`SSE response from ${url} has no readable body`); + } + + const decoder = new TextDecoder(); + let buffer = ''; + while (!abortController.signal.aborted) { + const { value, done } = await reader.read(); + if (done) { + throw new Error(`SSE connection to ${url} closed`); + } + + buffer += decoder.decode(value, { stream: true }); + const frames = buffer.split(/\r?\n\r?\n/); + buffer = frames.pop() ?? ''; + + for (const frame of frames) { + let frameEvent = ''; + const dataLines: string[] = []; + for (const line of frame.split(/\r?\n/)) { + if (line.startsWith('event:')) { + frameEvent = line.substring(6).trim(); + } else if (line.startsWith('data:')) { + dataLines.push(line.substring(5)); + } + } + if (frameEvent !== eventName || dataLines.length === 0) { + continue; + } + subscriber.next(dataLines.join('\n')); + } + } + }) + .catch(error => { + if (abortController.signal.aborted) { + return; + } + subscriber.error(error); + }); + }); + + return () => abortController.abort(); + }); + } +}