This is an automated email from the ASF dual-hosted git repository. yuluo-yx pushed a commit to branch 0728-yuluo/feat-1 in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
commit 3f851e11a5f83bb871a430835004210854f13315 Author: yuluo-yx <[email protected]> AuthorDate: Tue Jul 28 22:53:04 2026 +0800 [fix](security): stop sending AES secret over unauthenticated Netty Signed-off-by: yuluo-yx <[email protected]> --- .../dispatch/entrance/CollectServerTest.java | 34 +++++++++ .../nativex/CollectorRuntimeHintsRegistrar.java | 2 - .../src/main/resources/application.yml | 2 + .../collector/dispatch/entrance/CollectServer.java | 38 ++++++++-- .../entrance/processor/GoOnlineProcessor.java | 14 +--- .../entrance/processor/GoOnlineProcessorTest.java | 31 +++++--- .../hertzbeat/common/entity/dto/ServerInfo.java | 40 ---------- .../manager/scheduler/CollectorJobScheduler.java | 4 - .../netty/process/CollectorOnlineProcessor.java | 6 +- .../scheduler/CollectorJobSchedulerTest.java | 24 ++++++ .../process/CollectorOnlineProcessorTest.java | 85 ++++++++++++++++++++++ 11 files changed, 199 insertions(+), 81 deletions(-) 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 03a6977da6..1df89fb215 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 @@ package org.apache.hertzbeat.collector.dispatch.entrance; 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.collector.timer.TimerDispatch; 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 @@ class CollectServerTest { 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 @@ class CollectServerTest { 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 或 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 3464caa1b1..7b6f75ca64 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.DictionaryEncoding; 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 1fb329d7d0..f1ac042dd1 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: + # 与 Manager 使用同一密钥；密钥只通过部署配置注入，禁止由 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 a5bf494750..420fca0a00 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 @@ package org.apache.hertzbeat.collector.dispatch.entrance; 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.concurrent.BackgroundTaskExecutor; 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 @@ import java.util.concurrent.TimeUnit; @Slf4j public class CollectServer implements CommandLineRunner { + private static final Set<Integer> 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,22 @@ public class CollectServer implements CommandLineRunner { @Override public void run(String... args) throws Exception { + initializeAesSecret(); this.remotingClient.start(); } + private void initializeAesSecret() { + if (StringUtils.isBlank(commonSecret)) { + throw new IllegalStateException( + "独立 Collector 必须通过 common.secret 或 COMMON_SECRET 配置与 Manager 相同的 AES 密钥"); + } + int secretLength = commonSecret.getBytes(StandardCharsets.UTF_8).length; + if (!AES_KEY_LENGTHS.contains(secretLength)) { + throw new IllegalStateException("common.secret 的 UTF-8 字节长度必须为 16、24 或 32"); + } + 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 e35c66a151..8e88647335 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 io.netty.channel.ChannelHandlerContext; 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 class GoOnlineProcessor implements NettyRemotingProcessor { @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()); - } - } + // 不处理响应载荷，避免未认证的 Manager 响应覆盖 Collector 本地配置的 AES 密钥。 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 b544c8c39b..69fb3914ea 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 @@ package org.apache.hertzbeat.collector.dispatch.entrance.processor; 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 @@ class GoOnlineProcessorTest { .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 @@ class GoOnlineProcessorTest { .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 4b3fe8a4fc..0000000000 --- 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 b9512f60ec..eb10f51a3c 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.CollectJobServi 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.Monitor; 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 class CollectorJobScheduler implements CollectorScheduling, CollectJobSch 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 9a7e4f4e08..afefb49a99 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 class CollectorOnlineProcessor implements NettyRemotingProcessor { } this.manageServer.addChannel(collector, ctx.channel()); this.manageServer.getCollectorAndJobScheduler().collectorGoOnline(collector, collectorInfo); - ServerInfo serverInfo = ServerInfo.builder().aesSecret(AesUtil.getDefaultSecretKey()).build(); + // AES 密钥必须由 Manager 与 Collector 在本地预配置，禁止通过未加密的 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 10cc176142..402ed7e578 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 java.util.Optional; 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 class CollectorJobSchedulerTest { 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<ClusterMsg.Message> 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 0000000000..04c2896773 --- /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; + +/** + * {@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); + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
