From 14382fff6254a32a16882af5296e6546425e7959 Mon Sep 17 00:00:00 2001 From: yuluo-yx Date: Tue, 28 Jul 2026 22:53:04 +0800 Subject: [PATCH] [fix](security): stop sending AES secret over unauthenticated Netty Signed-off-by: yuluo-yx --- .../dispatch/entrance/CollectServerTest.java | 34 ++++++++ .../CollectorRuntimeHintsRegistrar.java | 2 - .../src/main/resources/application.yml | 2 + .../dispatch/entrance/CollectServer.java | 39 +++++++-- .../entrance/processor/GoOnlineProcessor.java | 14 +-- .../processor/GoOnlineProcessorTest.java | 31 +++++-- .../common/entity/dto/ServerInfo.java | 40 --------- .../scheduler/CollectorJobScheduler.java | 4 - .../process/CollectorOnlineProcessor.java | 6 +- .../scheduler/CollectorJobSchedulerTest.java | 24 ++++++ .../process/CollectorOnlineProcessorTest.java | 85 +++++++++++++++++++ 11 files changed, 200 insertions(+), 81 deletions(-) delete mode 100644 hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/dto/ServerInfo.java create mode 100644 hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/scheduler/netty/process/CollectorOnlineProcessorTest.java diff --git a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/dispatch/entrance/CollectServerTest.java b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/dispatch/entrance/CollectServerTest.java index 03a6977da63..e8c560b7ffd 100644 --- a/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/dispatch/entrance/CollectServerTest.java +++ b/hertzbeat-collector/hertzbeat-collector-basic/src/test/java/org/apache/hertzbeat/collector/dispatch/entrance/CollectServerTest.java @@ -20,6 +20,7 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +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.mock; @@ -39,7 +40,9 @@ import org.apache.hertzbeat.common.concurrent.BackgroundTaskExecutor; import org.apache.hertzbeat.common.config.VirtualThreadProperties; import org.apache.hertzbeat.common.entity.message.ClusterMsg; +import org.apache.hertzbeat.common.util.AesUtil; import org.apache.hertzbeat.remoting.RemotingClient; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -79,6 +82,11 @@ class CollectServerTest { private CollectServer.CollectNettyEventListener collectNettyEventListener; + @AfterEach + void tearDown() { + AesUtil.setDefaultSecretKey(AesUtil.DEFAULT_ENCODE_RULES); + } + @BeforeEach void setUp() { @@ -88,6 +96,7 @@ void setUp() { when(properties.getEntrance()).thenReturn(entranceProperties); collectServer = new CollectServer(collectJobService, timerDispatch, properties, threadPool, infoProperties); + ReflectionTestUtils.setField(collectServer, "commonSecret", "local-key-123456"); collectNettyEventListener = collectServer.new CollectNettyEventListener(); } @@ -100,6 +109,31 @@ void testRun() throws Exception { collectServer.run(); verify(remotingClient, times(1)).start(); + assertEquals("local-key-123456", AesUtil.getDefaultSecretKey()); + } + + @Test + void testRunRejectsMissingCommonSecret() { + RemotingClient remotingClient = mock(RemotingClient.class); + ReflectionTestUtils.setField(collectServer, "remotingClient", remotingClient); + ReflectionTestUtils.setField(collectServer, "commonSecret", " "); + + IllegalStateException exception = assertThrows(IllegalStateException.class, collectServer::run); + + assertTrue(exception.getMessage().contains("COMMON_SECRET")); + verify(remotingClient, times(0)).start(); + } + + @Test + void testRunRejectsInvalidCommonSecretLength() { + RemotingClient remotingClient = mock(RemotingClient.class); + ReflectionTestUtils.setField(collectServer, "remotingClient", remotingClient); + ReflectionTestUtils.setField(collectServer, "commonSecret", "too-short"); + + IllegalStateException exception = assertThrows(IllegalStateException.class, collectServer::run); + + assertTrue(exception.getMessage().contains("16, 24, or 32")); + verify(remotingClient, times(0)).start(); } @Test diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/nativex/CollectorRuntimeHintsRegistrar.java b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/nativex/CollectorRuntimeHintsRegistrar.java index 3464caa1b1b..7b6f75ca64b 100644 --- a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/nativex/CollectorRuntimeHintsRegistrar.java +++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/nativex/CollectorRuntimeHintsRegistrar.java @@ -32,7 +32,6 @@ import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.types.pojo.FieldType; import org.apache.arrow.vector.types.pojo.Schema; -import org.apache.hertzbeat.common.entity.dto.ServerInfo; import org.springframework.aot.hint.BindingReflectionHintsRegistrar; import org.springframework.aot.hint.MemberCategory; import org.springframework.aot.hint.RuntimeHints; @@ -55,7 +54,6 @@ public class CollectorRuntimeHintsRegistrar implements RuntimeHintsRegistrar { @Override public void registerHints(@NonNull RuntimeHints hints, ClassLoader classLoader) { BindingReflectionHintsRegistrar bindingRegistrar = new BindingReflectionHintsRegistrar(); - registerType(bindingRegistrar, hints, ServerInfo.class); scanBindingPackage(classLoader, bindingRegistrar, hints, JOB_PACKAGE); scanBindingPackage(classLoader, bindingRegistrar, hints, JOB_PROTOCOL_PACKAGE); hints.reflection().registerType(NettyAllocationManager.class, MemberCategory.DECLARED_FIELDS); diff --git a/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/application.yml b/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/application.yml index 1fb329d7d01..6e5b0d6ffec 100644 --- a/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/application.yml +++ b/hertzbeat-collector/hertzbeat-collector-collector/src/main/resources/application.yml @@ -70,6 +70,8 @@ push: common: + # Use the same key as Manager. Inject it through deployment configuration, never over Netty. + secret: ${COMMON_SECRET:} queue: type: netty diff --git a/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/CollectServer.java b/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/CollectServer.java index a5bf494750f..e775ffe06c6 100644 --- a/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/CollectServer.java +++ b/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/CollectServer.java @@ -20,7 +20,15 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder; import com.google.protobuf.ByteString; import io.netty.channel.Channel; +import java.nio.charset.StandardCharsets; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.apache.hertzbeat.collector.dispatch.CollectorInfoProperties; import org.apache.hertzbeat.collector.dispatch.DispatchProperties; import org.apache.hertzbeat.collector.dispatch.entrance.internal.CollectJobService; @@ -36,25 +44,21 @@ import org.apache.hertzbeat.common.config.VirtualThreadProperties; import org.apache.hertzbeat.common.entity.dto.CollectorInfo; import org.apache.hertzbeat.common.entity.message.ClusterMsg; +import org.apache.hertzbeat.common.util.AesUtil; import org.apache.hertzbeat.common.util.JsonUtil; import org.apache.hertzbeat.remoting.RemotingClient; import org.apache.hertzbeat.remoting.event.NettyEventListener; import org.apache.hertzbeat.remoting.netty.NettyClientConfig; import org.apache.hertzbeat.remoting.netty.NettyRemotingClient; -import org.springframework.context.ConfigurableApplicationContext; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.SpringApplication; +import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.CommandLineRunner; +import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.ConfigurableApplicationContext; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Component; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.TimeUnit; - /** * collect server */ @@ -65,6 +69,8 @@ @Slf4j public class CollectServer implements CommandLineRunner { + private static final Set AES_KEY_LENGTHS = Set.of(16, 24, 32); + private final CollectJobService collectJobService; private final TimerDispatch timerDispatch; @@ -85,6 +91,9 @@ public class CollectServer implements CommandLineRunner { private final Runnable closeApplicationAction; + @Value("${common.secret:}") + private String commonSecret; + public CollectServer(final CollectJobService collectJobService, final TimerDispatch timerDispatch, final DispatchProperties properties, @@ -181,9 +190,23 @@ void dispatchHeartbeat(String identity) { @Override public void run(String... args) throws Exception { + initializeAesSecret(); this.remotingClient.start(); } + private void initializeAesSecret() { + if (StringUtils.isBlank(commonSecret)) { + throw new IllegalStateException( + "A standalone Collector must configure the same AES key as Manager via common.secret " + + "or COMMON_SECRET"); + } + int secretLength = commonSecret.getBytes(StandardCharsets.UTF_8).length; + if (!AES_KEY_LENGTHS.contains(secretLength)) { + throw new IllegalStateException("common.secret must be 16, 24, or 32 bytes in UTF-8"); + } + AesUtil.setDefaultSecretKey(commonSecret); + } + /** * CollectNettyEventListener */ diff --git a/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/processor/GoOnlineProcessor.java b/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/processor/GoOnlineProcessor.java index e35c66a151b..af7294cd2a6 100644 --- a/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/processor/GoOnlineProcessor.java +++ b/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/dispatch/entrance/processor/GoOnlineProcessor.java @@ -22,10 +22,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.hertzbeat.collector.timer.TimerDispatch; import org.apache.hertzbeat.common.constants.CommonConstants; -import org.apache.hertzbeat.common.entity.dto.ServerInfo; import org.apache.hertzbeat.common.entity.message.ClusterMsg; -import org.apache.hertzbeat.common.util.AesUtil; -import org.apache.hertzbeat.common.util.JsonUtil; import org.apache.hertzbeat.remoting.netty.NettyRemotingProcessor; /** @@ -43,16 +40,7 @@ public GoOnlineProcessor(TimerDispatch timerDispatch) { @Override public ClusterMsg.Message handle(ChannelHandlerContext ctx, ClusterMsg.Message message) { - if (message.getMsg().isEmpty()) { - log.warn("The message that server response to collector is empty, please upgrade server"); - } else { - ServerInfo serverInfo = JsonUtil.fromJson(message.getMsg().toStringUtf8(), ServerInfo.class); - if (serverInfo == null || serverInfo.getAesSecret() == null) { - log.warn("The message that server response to collector has not secret empty, please check"); - } else { - AesUtil.setDefaultSecretKey(serverInfo.getAesSecret()); - } - } + // Ignore the response payload so an unauthenticated Manager response cannot override the local AES key. if (ClusterMsg.Direction.REQUEST.equals(message.getDirection())) { timerDispatch.goOnline(); } diff --git a/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/dispatch/entrance/processor/GoOnlineProcessorTest.java b/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/dispatch/entrance/processor/GoOnlineProcessorTest.java index b544c8c39bd..69fb3914ea6 100644 --- a/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/dispatch/entrance/processor/GoOnlineProcessorTest.java +++ b/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/dispatch/entrance/processor/GoOnlineProcessorTest.java @@ -19,17 +19,15 @@ import com.google.common.collect.Lists; import com.google.protobuf.ByteString; -import io.netty.channel.ChannelHandlerContext; import org.apache.hertzbeat.collector.timer.TimerDispatcher; import org.apache.hertzbeat.common.entity.job.Job; import org.apache.hertzbeat.common.entity.job.Metrics; import org.apache.hertzbeat.common.entity.message.ClusterMsg; +import org.apache.hertzbeat.common.util.AesUtil; import org.apache.hertzbeat.common.util.JsonUtil; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import org.mockito.Mock; -import org.mockito.MockitoAnnotations; import java.lang.reflect.Field; import java.util.Map; @@ -44,12 +42,8 @@ class GoOnlineProcessorTest { private GoOnlineProcessor goOnlineProcessor; private TimerDispatcher timerDispatcher; - @Mock - private ChannelHandlerContext channelHandlerContext; - @BeforeEach void setUp() { - MockitoAnnotations.openMocks(this); timerDispatcher = new TimerDispatcher(); goOnlineProcessor = new GoOnlineProcessor(timerDispatcher); } @@ -81,7 +75,7 @@ void verifyTaskMapPreservation() throws Exception { .setMsg(ByteString.copyFromUtf8(JsonUtil.toJson(job))) .setIdentity("test-identity") .build(); - goOnlineProcessor.handle(channelHandlerContext, responseMsg); + goOnlineProcessor.handle(null, responseMsg); assertEquals(1, currentCyclicTaskMap.size(), "Task map should still have 1 job after receiving RESPONSE"); ClusterMsg.Message requestMsg = ClusterMsg.Message.newBuilder() @@ -90,7 +84,26 @@ void verifyTaskMapPreservation() throws Exception { .setMsg(ByteString.copyFromUtf8(JsonUtil.toJson(job))) .setIdentity("test-identity") .build(); - goOnlineProcessor.handle(channelHandlerContext, requestMsg); + goOnlineProcessor.handle(null, requestMsg); assertEquals(0, currentCyclicTaskMap.size(), "Task map should be empty after receiving REQUEST"); } + + @Test + void shouldIgnoreAesSecretFromNetworkResponse() { + String localSecret = "local-key-123456"; + AesUtil.setDefaultSecretKey(localSecret); + try { + ClusterMsg.Message response = ClusterMsg.Message.newBuilder() + .setType(ClusterMsg.MessageType.GO_ONLINE) + .setDirection(ClusterMsg.Direction.RESPONSE) + .setMsg(ByteString.copyFromUtf8("{\"aesSecret\":\"network-key-1234\"}")) + .build(); + + goOnlineProcessor.handle(null, response); + + assertEquals(localSecret, AesUtil.getDefaultSecretKey()); + } finally { + AesUtil.setDefaultSecretKey(AesUtil.DEFAULT_ENCODE_RULES); + } + } } diff --git a/hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/dto/ServerInfo.java b/hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/dto/ServerInfo.java deleted file mode 100644 index 4b3fe8a4fc2..00000000000 --- a/hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/entity/dto/ServerInfo.java +++ /dev/null @@ -1,40 +0,0 @@ -/* - * 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.common.entity.dto; - -import io.swagger.v3.oas.annotations.media.Schema; -import lombok.AllArgsConstructor; -import lombok.Builder; -import lombok.Data; -import lombok.NoArgsConstructor; - -/** - * collector info - */ -@Data -@Builder -@AllArgsConstructor -@NoArgsConstructor -@Schema(description = "manager server info") -public class ServerInfo { - - /** - * the aes secret key value for aes encode decode - */ - private String aesSecret; -} diff --git a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/scheduler/CollectorJobScheduler.java b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/scheduler/CollectorJobScheduler.java index b9512f60ec6..eb10f51a3c5 100644 --- a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/scheduler/CollectorJobScheduler.java +++ b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/scheduler/CollectorJobScheduler.java @@ -36,7 +36,6 @@ import org.apache.hertzbeat.collector.dispatch.entrance.internal.CollectResponseEventListener; import org.apache.hertzbeat.common.constants.CommonConstants; import org.apache.hertzbeat.common.entity.dto.CollectorInfo; -import org.apache.hertzbeat.common.entity.dto.ServerInfo; import org.apache.hertzbeat.common.entity.job.Configmap; import org.apache.hertzbeat.common.entity.job.Job; import org.apache.hertzbeat.common.entity.job.RuntimeParamDefine; @@ -46,7 +45,6 @@ import org.apache.hertzbeat.common.entity.manager.Param; import org.apache.hertzbeat.common.entity.message.ClusterMsg; import org.apache.hertzbeat.common.entity.message.CollectRep; -import org.apache.hertzbeat.common.util.AesUtil; import org.apache.hertzbeat.common.util.JsonUtil; import org.apache.hertzbeat.common.util.SnowFlakeIdGenerator; import org.apache.hertzbeat.manager.dao.CollectorDao; @@ -265,11 +263,9 @@ public boolean onlineCollector(String identity) { if (Objects.isNull(collector)) { return false; } - ServerInfo serverInfo = ServerInfo.builder().aesSecret(AesUtil.getDefaultSecretKey()).build(); ClusterMsg.Message message = ClusterMsg.Message.newBuilder() .setType(ClusterMsg.MessageType.GO_ONLINE) .setDirection(ClusterMsg.Direction.REQUEST) - .setMsg(ByteString.copyFromUtf8(JsonUtil.toJson(serverInfo))) .setIdentity(identity) .build(); ClusterMsg.Message response = this.manageServer.sendMsgSync(identity, message); diff --git a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/scheduler/netty/process/CollectorOnlineProcessor.java b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/scheduler/netty/process/CollectorOnlineProcessor.java index 9a7e4f4e082..7f67ca55c03 100644 --- a/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/scheduler/netty/process/CollectorOnlineProcessor.java +++ b/hertzbeat-manager/src/main/java/org/apache/hertzbeat/manager/scheduler/netty/process/CollectorOnlineProcessor.java @@ -17,15 +17,12 @@ package org.apache.hertzbeat.manager.scheduler.netty.process; -import com.google.protobuf.ByteString; import io.netty.channel.ChannelHandlerContext; import java.net.InetSocketAddress; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.apache.hertzbeat.common.entity.dto.CollectorInfo; -import org.apache.hertzbeat.common.entity.dto.ServerInfo; import org.apache.hertzbeat.common.entity.message.ClusterMsg; -import org.apache.hertzbeat.common.util.AesUtil; import org.apache.hertzbeat.common.util.JsonUtil; import org.apache.hertzbeat.manager.scheduler.netty.ManageServer; import org.apache.hertzbeat.remoting.netty.NettyRemotingProcessor; @@ -55,11 +52,10 @@ public ClusterMsg.Message handle(ChannelHandlerContext ctx, ClusterMsg.Message m } this.manageServer.addChannel(collector, ctx.channel()); this.manageServer.getCollectorAndJobScheduler().collectorGoOnline(collector, collectorInfo); - ServerInfo serverInfo = ServerInfo.builder().aesSecret(AesUtil.getDefaultSecretKey()).build(); + // The AES key must be configured locally on both Manager and Collector, never sent over plaintext Netty. return ClusterMsg.Message.newBuilder() .setIdentity(message.getIdentity()) .setDirection(ClusterMsg.Direction.RESPONSE) - .setMsg(ByteString.copyFromUtf8(JsonUtil.toJson(serverInfo))) .setType(ClusterMsg.MessageType.GO_ONLINE) .build(); } diff --git a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/scheduler/CollectorJobSchedulerTest.java b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/scheduler/CollectorJobSchedulerTest.java index 10cc1761427..402ed7e578f 100644 --- a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/scheduler/CollectorJobSchedulerTest.java +++ b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/scheduler/CollectorJobSchedulerTest.java @@ -47,6 +47,7 @@ import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.eq; @@ -191,4 +192,27 @@ public void testCollectorGoOnlinePropagatesCronSchedule() { assertEquals("0 55 7 * * ?", job.getCronExpression()); } + @Test + void testOnlineCollectorDoesNotSendAesSecret() { + String identity = "collector-1"; + org.apache.hertzbeat.common.entity.manager.Collector collector = + org.apache.hertzbeat.common.entity.manager.Collector.builder() + .name(identity) + .status(CommonConstants.COLLECTOR_STATUS_ONLINE) + .build(); + when(collectorDao.findCollectorByName(identity)).thenReturn(Optional.of(collector)); + ManageServer manageServer = mock(ManageServer.class); + collectorJobScheduler.setManageServer(manageServer); + ClusterMsg.Message successResponse = ClusterMsg.Message.newBuilder() + .setMsg(com.google.protobuf.ByteString.copyFromUtf8(String.valueOf(CommonConstants.SUCCESS_CODE))) + .build(); + when(manageServer.sendMsgSync(eq(identity), any(ClusterMsg.Message.class))).thenReturn(successResponse); + + assertTrue(collectorJobScheduler.onlineCollector(identity)); + + ArgumentCaptor messageCaptor = ArgumentCaptor.forClass(ClusterMsg.Message.class); + verify(manageServer).sendMsgSync(eq(identity), messageCaptor.capture()); + assertTrue(messageCaptor.getValue().getMsg().isEmpty()); + } + } diff --git a/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/scheduler/netty/process/CollectorOnlineProcessorTest.java b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/scheduler/netty/process/CollectorOnlineProcessorTest.java new file mode 100644 index 00000000000..ac1edc52c22 --- /dev/null +++ b/hertzbeat-manager/src/test/java/org/apache/hertzbeat/manager/scheduler/netty/process/CollectorOnlineProcessorTest.java @@ -0,0 +1,85 @@ +/* + * 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.scheduler.netty.process; + +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.protobuf.ByteString; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import org.apache.hertzbeat.common.entity.dto.CollectorInfo; +import org.apache.hertzbeat.common.entity.message.ClusterMsg; +import org.apache.hertzbeat.common.util.JsonUtil; +import org.apache.hertzbeat.manager.scheduler.CollectorJobScheduler; +import org.apache.hertzbeat.manager.scheduler.netty.ManageServer; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +/** + * Security regression tests for {@link CollectorOnlineProcessor}. + */ +@ExtendWith(MockitoExtension.class) +class CollectorOnlineProcessorTest { + + @Mock + private ManageServer manageServer; + + @Mock + private CollectorJobScheduler collectorJobScheduler; + + @Mock + private ChannelHandlerContext channelHandlerContext; + + @Mock + private Channel channel; + + private CollectorOnlineProcessor processor; + + @BeforeEach + void setUp() { + when(channelHandlerContext.channel()).thenReturn(channel); + when(manageServer.getCollectorAndJobScheduler()).thenReturn(collectorJobScheduler); + processor = new CollectorOnlineProcessor(manageServer); + } + + @Test + void shouldNotReturnAesSecretWhenCollectorGoesOnline() { + CollectorInfo collectorInfo = CollectorInfo.builder() + .name("collector-1") + .ip("127.0.0.1") + .mode("public") + .version("1.0.0") + .build(); + ClusterMsg.Message request = ClusterMsg.Message.newBuilder() + .setIdentity("collector-1") + .setType(ClusterMsg.MessageType.GO_ONLINE) + .setMsg(ByteString.copyFromUtf8(JsonUtil.toJson(collectorInfo))) + .build(); + + ClusterMsg.Message response = processor.handle(channelHandlerContext, request); + + assertTrue(response.getMsg().isEmpty()); + verify(manageServer).addChannel("collector-1", channel); + verify(collectorJobScheduler).collectorGoOnline("collector-1", collectorInfo); + } +}