[fix](security): stop sending AES secret over unauthenticated Netty

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

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);

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
 

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<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,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
      */

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();
         }

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);
+        }
+    }
 }

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);

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();
     }