[improvement](fe) Enhance FE meta service request validation

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -813,6 +813,12 @@ public class Config extends ConfigBase {
     // check token when download image file.
     @ConfField public static boolean enable_token_check = true;
 
+    @ConfField(mutable = true, description = {
+            "Whether FE meta service accepts legacy FE node identity headers without a cluster token. "
+                    + "This is only for rolling upgrade compatibility and should be disabled after all FEs "
+                    + "are upgraded."})
+    public static boolean enable_meta_service_legacy_node_ident_auth = false;
+
     /**
      * Set to true if you deploy Palo using thirdparty deploy manager
      * Valid options are:

fe/fe-core/src/main/java/org/apache/doris/common/util/HttpURLUtil.java

@@ -19,8 +19,11 @@
 
 import org.apache.doris.catalog.Env;
 import org.apache.doris.cloud.security.SecurityChecker;
+import org.apache.doris.common.Config;
+import org.apache.doris.httpv2.meta.MetaBaseAction;
 import org.apache.doris.system.SystemInfoService.HostInfo;
 
+import com.google.common.base.Strings;
 import com.google.common.collect.Maps;
 
 import java.io.IOException;
@@ -37,9 +40,11 @@ public static HttpURLConnection getConnectionWithNodeIdent(String request) throw
             HttpURLConnection conn = (HttpURLConnection) url.openConnection();
             // Must use Env.getServingEnv() instead of getCurrentEnv(),
             // because here we need to obtain selfNode through the official service catalog.
-            HostInfo selfNode = Env.getServingEnv().getSelfNode();
+            Env env = Env.getServingEnv();
+            HostInfo selfNode = env.getSelfNode();
             conn.setRequestProperty(Env.CLIENT_NODE_HOST_KEY, selfNode.getHost());
             conn.setRequestProperty(Env.CLIENT_NODE_PORT_KEY, selfNode.getPort() + "");
+            setClusterToken(conn, getClusterToken(env));
             return conn;
         } catch (Exception e) {
             throw new IOException(e);
@@ -52,10 +57,25 @@ public static Map<String, String> getNodeIdentHeaders() throws IOException {
         Map<String, String> headers = Maps.newHashMap();
         // Must use Env.getServingEnv() instead of getCurrentEnv(),
         // because here we need to obtain selfNode through the official service catalog.
-        HostInfo selfNode = Env.getServingEnv().getSelfNode();
+        Env env = Env.getServingEnv();
+        HostInfo selfNode = env.getSelfNode();
         headers.put(Env.CLIENT_NODE_HOST_KEY, selfNode.getHost());
         headers.put(Env.CLIENT_NODE_PORT_KEY, selfNode.getPort() + "");
+        String token = getClusterToken(env);
+        if (!Strings.isNullOrEmpty(token)) {
+            headers.put(MetaBaseAction.TOKEN, token);
+        }
         return headers;
     }
 
+    private static String getClusterToken(Env env) {
+        String token = env.getToken();
+        return Strings.isNullOrEmpty(token) ? Config.auth_token : token;
+    }
+
+    private static void setClusterToken(HttpURLConnection conn, String token) {
+        if (!Strings.isNullOrEmpty(token)) {
+            conn.setRequestProperty(MetaBaseAction.TOKEN, token);
+        }
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/httpv2/meta/MetaService.java

@@ -23,6 +23,7 @@
 import org.apache.doris.common.util.NetUtils;
 import org.apache.doris.ha.FrontendNodeType;
 import org.apache.doris.httpv2.entity.ResponseEntityBuilder;
+import org.apache.doris.httpv2.exception.UnauthorizedException;
 import org.apache.doris.httpv2.rest.RestBaseController;
 import org.apache.doris.master.MetaHelper;
 import org.apache.doris.persist.MetaCleaner;
@@ -55,31 +56,56 @@ public class MetaService extends RestBaseController {
 
     private File imageDir = MetaHelper.getMasterImageDir();
 
-    private boolean isFromValidFe(String clientHost, String clientPortStr) {
+    private Frontend getValidFe(String clientHost, String clientPortStr) {
         Integer clientPort;
         try {
             clientPort = Integer.valueOf(clientPortStr);
         } catch (Exception e) {
             LOG.warn("get clientPort error. clientPortStr: {}", clientPortStr, e.getMessage());
-            return false;
+            return null;
         }
 
         Frontend fe = Env.getCurrentEnv().checkFeExist(clientHost, clientPort);
         if (fe == null) {
             LOG.warn("request is not from valid FE. client: {}, {}", clientHost, clientPortStr);
-            return false;
         }
-        return true;
+        return fe;
     }
 
     private void checkFromValidFe(HttpServletRequest request)
-            throws InvalidClientException {
+            throws UnauthorizedException {
         String clientHost = request.getHeader(Env.CLIENT_NODE_HOST_KEY);
         String clientPort = request.getHeader(Env.CLIENT_NODE_PORT_KEY);
-        if (!isFromValidFe(clientHost, clientPort)) {
-            throw new InvalidClientException("invalid client host: " + clientHost + ":" + clientPort
-                + ", request from " + request.getRemoteHost());
+        Frontend fe = getValidFe(clientHost, clientPort);
+        if (fe == null) {
+            throw unauthorized(clientHost, clientPort, request);
+        }
+
+        String requestToken = request.getHeader(MetaBaseAction.TOKEN);
+        if (!Strings.isNullOrEmpty(requestToken)) {
+            if (requestToken.equals(getClusterToken())) {
+                return;
+            }
+            LOG.warn("reject meta request with invalid token. client: {}, {}, request from: {}",
+                    clientHost, clientPort, request.getRemoteAddr());
+            throw unauthorized(clientHost, clientPort, request);
+        }
+
+        if (Config.enable_meta_service_legacy_node_ident_auth) {
+            return;
         }
+
+        throw unauthorized(clientHost, clientPort, request);
+    }
+
+    private String getClusterToken() {
+        String token = Env.getCurrentEnv().getToken();
+        return Strings.isNullOrEmpty(token) ? Config.auth_token : token;
+    }
+
+    private UnauthorizedException unauthorized(String clientHost, String clientPort, HttpServletRequest request) {
+        return new UnauthorizedException("invalid client host: " + clientHost + ":" + clientPort
+                + ", request from " + request.getRemoteAddr());
     }
 
     @RequestMapping(path = "/image", method = RequestMethod.GET)
@@ -149,6 +175,9 @@ public Object put(HttpServletRequest request, HttpServletResponse response) thro
         if (port < 0 || port > 65535) {
             return ResponseEntityBuilder.badRequest("port is invalid. The port number is between 0-65535");
         }
+        if (port != Config.http_port) {
+            return ResponseEntityBuilder.badRequest("port must be FE HTTP port: " + Config.http_port);
+        }
 
         String versionStr = request.getParameter(VERSION);
         if (Strings.isNullOrEmpty(versionStr)) {
@@ -242,9 +271,7 @@ public Object check(HttpServletRequest request, HttpServletResponse response) th
 
     @RequestMapping(value = "/dump", method = RequestMethod.GET)
     public Object dump(HttpServletRequest request, HttpServletResponse response) throws DdlException {
-        if (Config.enable_all_http_auth) {
-            executeCheckPassword(request, response);
-        }
+        executeCheckPassword(request, response);
 
         /*
          * Before dump, we acquired the catalog read lock and all databases' read lock and all

fe/fe-core/src/main/java/org/apache/doris/master/MetaHelper.java

@@ -149,7 +149,7 @@ private static void checkFile(File file) throws IOException {
 
     public static <T> ResponseBody doGet(String url, int timeout, Class<T> clazz) throws IOException {
         Map<String, String> headers = HttpURLUtil.getNodeIdentHeaders();
-        LOG.info("meta helper, url: {}, timeout{}, headers: {}", url, timeout, headers);
+        LOG.info("meta helper, url: {}, timeout: {}, header names: {}", url, timeout, headers.keySet());
         String response = HttpUtils.doGet(url, headers, timeout);
         return parseResponse(response, clazz);
     }

fe/fe-core/src/test/java/org/apache/doris/common/util/HttpURLUtilTest.java

@@ -0,0 +1,87 @@
+// 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.doris.common.util;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.Config;
+import org.apache.doris.httpv2.meta.MetaBaseAction;
+import org.apache.doris.system.SystemInfoService.HostInfo;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+
+import java.net.HttpURLConnection;
+import java.util.Map;
+
+public class HttpURLUtilTest {
+
+    @Test
+    public void testNodeIdentHeadersIncludeClusterToken() throws Exception {
+        Env env = Mockito.mock(Env.class);
+        Mockito.when(env.getSelfNode()).thenReturn(new HostInfo("127.0.0.1", 9010));
+        Mockito.when(env.getToken()).thenReturn("cluster-token");
+
+        try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+            envStatic.when(Env::getServingEnv).thenReturn(env);
+
+            Map<String, String> headers = HttpURLUtil.getNodeIdentHeaders();
+
+            Assert.assertEquals("127.0.0.1", headers.get(Env.CLIENT_NODE_HOST_KEY));
+            Assert.assertEquals("9010", headers.get(Env.CLIENT_NODE_PORT_KEY));
+            Assert.assertEquals("cluster-token", headers.get(MetaBaseAction.TOKEN));
+        }
+    }
+
+    @Test
+    public void testNodeIdentHeadersUseConfigAuthTokenWhenEnvTokenIsEmpty() throws Exception {
+        String oldAuthToken = Config.auth_token;
+        Config.auth_token = "config-token";
+        Env env = Mockito.mock(Env.class);
+        Mockito.when(env.getSelfNode()).thenReturn(new HostInfo("127.0.0.1", 9010));
+        Mockito.when(env.getToken()).thenReturn("");
+
+        try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+            envStatic.when(Env::getServingEnv).thenReturn(env);
+
+            Map<String, String> headers = HttpURLUtil.getNodeIdentHeaders();
+
+            Assert.assertEquals("config-token", headers.get(MetaBaseAction.TOKEN));
+        } finally {
+            Config.auth_token = oldAuthToken;
+        }
+    }
+
+    @Test
+    public void testNodeIdentConnectionIncludesClusterToken() throws Exception {
+        Env env = Mockito.mock(Env.class);
+        Mockito.when(env.getSelfNode()).thenReturn(new HostInfo("127.0.0.1", 9010));
+        Mockito.when(env.getToken()).thenReturn("cluster-token");
+
+        try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+            envStatic.when(Env::getServingEnv).thenReturn(env);
+
+            HttpURLConnection connection = HttpURLUtil.getConnectionWithNodeIdent("http://127.0.0.1:8030/info");
+
+            Assert.assertEquals("127.0.0.1", connection.getRequestProperty(Env.CLIENT_NODE_HOST_KEY));
+            Assert.assertEquals("9010", connection.getRequestProperty(Env.CLIENT_NODE_PORT_KEY));
+            Assert.assertEquals("cluster-token", connection.getRequestProperty(MetaBaseAction.TOKEN));
+        }
+    }
+}