[improvement](fe) add fe_meta_auth_token for FE meta-service internal HTTP auth

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

@@ -818,6 +818,18 @@ public class Config extends ConfigBase {
     // check token when download image file.
     @ConfField public static boolean enable_token_check = true;
 
+    @ConfField(sensitive = true, description = {"Cluster token for FE meta-service internal HTTP authentication. "
+            + "When set (non-empty), FE meta-service endpoints (such as image/role/check/put/journal_id) "
+            + "additionally require the caller to present a matching token header, on top of the existing "
+            + "node-host check. Empty (default) keeps the legacy behavior of node-host check only, so "
+            + "existing clusters and rolling upgrades are unaffected. Must be identical on all FEs and "
+            + "provisioned in fe.conf before enabling, otherwise FEs will reject each other.",
+            "FE meta-service 内部 HTTP 鉴权使用的集群 token。设置(非空)后,meta-service 端点(如 "
+            + "image/role/check/put/journal_id)在原有 node-host 校验之上,额外要求调用方携带匹配的 token 头。"
+            + "为空(默认)时维持仅 node-host 校验的旧行为,存量集群与滚动升级不受影响。必须在所有 FE 上取值一致,"
+            + "并在启用前写入 fe.conf,否则 FE 之间会互相拒绝。"})
+    public static String fe_meta_auth_token = "";
+
     /**
      * Set to true if you deploy Palo using thirdparty deploy manager
      * Valid options are:

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

@@ -52,6 +52,10 @@ public class ConfigBase {
 
         boolean masterOnly() default false;
 
+        // If true, the value is a secret (e.g. a token or password) and is masked in every
+        // config dump API (Config.dump / getConfigInfo), so it is never returned in plaintext.
+        boolean sensitive() default false;
+
         String comment() default "";
 
         VariableAnnotation varType() default VariableAnnotation.NONE;
@@ -191,13 +195,26 @@ private void warnUnknownConfigKeys(String confFile, Properties props) {
         }
     }
 
+    // Placeholder returned instead of a sensitive config's real value in any dump API.
+    public static final String SENSITIVE_CONF_MASK = "********";
+
+    // Mask the value of a sensitive config (a non-empty secret) so it is never dumped in plaintext.
+    // An empty value is left as-is: it reveals nothing and keeps "unset" visible.
+    private static String maskIfSensitive(Field field, String value) {
+        ConfField anno = field.getAnnotation(ConfField.class);
+        if (anno != null && anno.sensitive() && !Strings.isNullOrEmpty(value)) {
+            return SENSITIVE_CONF_MASK;
+        }
+        return value;
+    }
+
     public static HashMap<String, String> dump() {
         HashMap<String, String> map = new HashMap<>();
         Field[] fields = confClass.getFields();
         for (Field f : fields) {
             ConfField anno = f.getAnnotation(ConfField.class);
             if (anno != null) {
-                map.put(f.getName(), getConfValue(f));
+                map.put(f.getName(), maskIfSensitive(f, getConfValue(f)));
             }
         }
         return map;
@@ -441,6 +458,7 @@ public static synchronized List<List<String>> getConfigInfo(PatternMatcher match
                 if (confKey.equals("sys_log_dir") && Strings.isNullOrEmpty(value)) {
                     value = System.getenv("DORIS_HOME") + "/log";
                 }
+                value = maskIfSensitive(f, value);
                 config.add(value);
                 config.add(f.getType().getSimpleName());
                 config.add(String.valueOf(confField.mutable()));

fe/fe-common/src/test/java/org/apache/doris/common/ConfigTest.java

@@ -23,6 +23,8 @@
 
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
 
 public class ConfigTest {
     @BeforeClass
@@ -34,6 +36,47 @@ public static void setUp() throws Exception {
         config.init(tempFile.toAbsolutePath().toString());
     }
 
+    // A sensitive config (fe_meta_auth_token) must never be dumped in plaintext by any config
+    // API: both Config.dump() and ConfigBase.getConfigInfo() return the mask instead of the value.
+    @Test
+    public void testSensitiveConfigIsMaskedWhenSet() {
+        String old = Config.fe_meta_auth_token;
+        try {
+            Config.fe_meta_auth_token = "super-secret-token";
+
+            Map<String, String> dumped = ConfigBase.dump();
+            Assert.assertEquals(ConfigBase.SENSITIVE_CONF_MASK, dumped.get("fe_meta_auth_token"));
+
+            String value = configInfoValue("fe_meta_auth_token");
+            Assert.assertEquals(ConfigBase.SENSITIVE_CONF_MASK, value);
+        } finally {
+            Config.fe_meta_auth_token = old;
+        }
+    }
+
+    // An empty sensitive config is left as-is (no secret to hide), so "unset" stays visible.
+    @Test
+    public void testEmptySensitiveConfigIsNotMasked() {
+        String old = Config.fe_meta_auth_token;
+        try {
+            Config.fe_meta_auth_token = "";
+
+            Assert.assertEquals("", ConfigBase.dump().get("fe_meta_auth_token"));
+            Assert.assertEquals("", configInfoValue("fe_meta_auth_token"));
+        } finally {
+            Config.fe_meta_auth_token = old;
+        }
+    }
+
+    private static String configInfoValue(String key) {
+        for (List<String> row : ConfigBase.getConfigInfo(null)) {
+            if (row.get(0).equals(key)) {
+                return row.get(1);
+            }
+        }
+        throw new IllegalStateException("config not found: " + key);
+    }
+
     @Test
     public void testSetEmptyArray() throws ConfigException {
         ConfigBase.setMutableConfig("s3_load_endpoint_white_list", "a,b,c");

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

@@ -20,8 +20,10 @@
 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 org.apache.http.conn.ssl.NoopHostnameVerifier;
 
@@ -50,6 +52,10 @@ public static HttpURLConnection getConnectionWithNodeIdent(String request) throw
             HostInfo selfNode = Env.getServingEnv().getSelfNode();
             conn.setRequestProperty(Env.CLIENT_NODE_HOST_KEY, selfNode.getHost());
             conn.setRequestProperty(Env.CLIENT_NODE_PORT_KEY, selfNode.getPort() + "");
+            String token = Config.fe_meta_auth_token;
+            if (!Strings.isNullOrEmpty(token)) {
+                conn.setRequestProperty(MetaBaseAction.TOKEN, token);
+            }
             return conn;
         } catch (Exception e) {
             throw new IOException(e);
@@ -65,6 +71,10 @@ public static Map<String, String> getNodeIdentHeaders() throws IOException {
         HostInfo selfNode = Env.getServingEnv().getSelfNode();
         headers.put(Env.CLIENT_NODE_HOST_KEY, selfNode.getHost());
         headers.put(Env.CLIENT_NODE_PORT_KEY, selfNode.getPort() + "");
+        String token = Config.fe_meta_auth_token;
+        if (!Strings.isNullOrEmpty(token)) {
+            headers.put(MetaBaseAction.TOKEN, token);
+        }
         return headers;
     }
 

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

@@ -20,9 +20,11 @@
 import org.apache.doris.catalog.Env;
 import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
+import org.apache.doris.common.util.HttpURLUtil;
 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,33 +57,50 @@ 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);
+        }
+
+        // If a cluster meta auth token is configured, additionally require the request to
+        // carry a matching token. An empty token keeps the legacy node-host-only behavior,
+        // so existing clusters and rolling upgrades are unaffected.
+        String clusterToken = Config.fe_meta_auth_token;
+        if (!Strings.isNullOrEmpty(clusterToken)) {
+            String requestToken = request.getHeader(MetaBaseAction.TOKEN);
+            if (!clusterToken.equals(requestToken)) {
+                LOG.warn("reject meta request with invalid token. client: {}, {}, request from: {}",
+                        clientHost, clientPort, request.getRemoteAddr());
+                throw unauthorized(clientHost, clientPort, request);
+            }
         }
     }
 
+    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)
     public Object image(HttpServletRequest request, HttpServletResponse response) {
         checkFromValidFe(request);
@@ -149,6 +168,12 @@ 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");
         }
+        // The master pushes image using HttpURLUtil.getHttpPort() (https_port when enable_https=true,
+        // otherwise http_port), so the expected port must follow the same rule to stay consistent.
+        int expectedPort = HttpURLUtil.getHttpPort();
+        if (port != expectedPort) {
+            return ResponseEntityBuilder.badRequest("port must be FE HTTP port: " + expectedPort);
+        }
 
         String versionStr = request.getParameter(VERSION);
         if (Strings.isNullOrEmpty(versionStr)) {
@@ -245,9 +270,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/httpv2/rest/manager/NodeAction.java

@@ -259,8 +259,13 @@ private static List<String> getBeList() {
      */
     @RequestMapping(path = "/config", method = RequestMethod.GET)
     public Object config(HttpServletRequest request, HttpServletResponse response) {
-        executeCheckPassword(request, response);
-        checkDbAuth(ConnectContext.get().getCurrentUserIdentity(), InfoSchemaDb.DATABASE_NAME, PrivPredicate.SELECT);
+        // This endpoint lists all FE config, matching the SQL "SHOW FRONTEND CONFIG", which
+        // requires ADMIN. Use an unconditional ADMIN check: checkAdminAuth only enforces the
+        // privilege when enable_all_http_auth is true, so it would be a no-op by default.
+        // Sensitive config values (e.g. fe_meta_auth_token) are additionally masked by ConfigBase,
+        // so they are never returned in plaintext even to an admin.
+        ActionAuthorizationInfo authInfo = executeCheckPassword(request, response);
+        checkGlobalAuth(authInfo.userIdentity, PrivPredicate.ADMIN);
 
         List<List<String>> configs = ConfigBase.getConfigInfo(null);
         // Sort all configs by config key.
@@ -320,8 +325,10 @@ public Object config(HttpServletRequest request, HttpServletResponse response) {
     public Object configurationInfo(HttpServletRequest request, HttpServletResponse response,
             @RequestParam(value = "type") String type,
             @RequestBody(required = false) ConfigInfoRequestBody requestBody) {
+        // Reads FE/BE config via fan-out to the per-node config endpoints, so it must be
+        // ADMIN-gated too. Unconditional check (see config() above for why checkAdminAuth is not).
         ActionAuthorizationInfo authInfo = executeCheckPassword(request, response);
-        checkAdminAuth(authInfo.userIdentity);
+        checkGlobalAuth(authInfo.userIdentity, PrivPredicate.ADMIN);
 
         initHttpExecutor();
 

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);
         try {
             return parseResponse(response, clazz);

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

@@ -17,11 +17,19 @@
 
 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.After;
 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 {
 
@@ -30,6 +38,57 @@ public void tearDown() {
         Config.enable_https = false;
         Config.http_port = 8030;
         Config.https_port = 8050;
+        Config.fe_meta_auth_token = "";
+    }
+
+    @Test
+    public void testNodeIdentHeadersIncludeClusterToken() throws Exception {
+        Config.fe_meta_auth_token = "cluster-token";
+        Env env = Mockito.mock(Env.class);
+        Mockito.when(env.getSelfNode()).thenReturn(new HostInfo("127.0.0.1", 9010));
+
+        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 testNodeIdentHeadersOmitTokenWhenNotConfigured() throws Exception {
+        Config.fe_meta_auth_token = "";
+        Env env = Mockito.mock(Env.class);
+        Mockito.when(env.getSelfNode()).thenReturn(new HostInfo("127.0.0.1", 9010));
+
+        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.assertFalse(headers.containsKey(MetaBaseAction.TOKEN));
+        }
+    }
+
+    @Test
+    public void testNodeIdentConnectionIncludesClusterToken() throws Exception {
+        Config.fe_meta_auth_token = "cluster-token";
+        Env env = Mockito.mock(Env.class);
+        Mockito.when(env.getSelfNode()).thenReturn(new HostInfo("127.0.0.1", 9010));
+
+        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));
+        }
     }
 
     @Test