optimize: add secure authentication to interfaces in ClusterController

common/pom.xml

@@ -45,5 +45,9 @@
             <artifactId>httpclient</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
     </dependencies>
 </project>

common/src/main/java/io/seata/common/exception/AuthenticationFailedException.java

@@ -0,0 +1,65 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed 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 io.seata.common.exception;
+
+
+/**
+ * Exception indicating authentication failure. This exception is typically thrown
+ * when the authentication process fails, and it extends SecurityException to
+ * signal that it is a security-related issue.
+ */
+public class AuthenticationFailedException extends SecurityException {
+
+    /**
+     * Constructs a new AuthenticationFailedException with no detailed message.
+     */
+    public AuthenticationFailedException() {
+        super();
+    }
+
+    /**
+     * Constructs a new AuthenticationFailedException with the specified detail message.
+     *
+     * @param message the detail message (which is saved for later retrieval
+     *                by the getMessage() method).
+     */
+    public AuthenticationFailedException(String message) {
+        super(message);
+    }
+
+    /**
+     * Constructs a new AuthenticationFailedException with the specified cause.
+     *
+     * @param cause the cause (which is saved for later retrieval by the
+     *              getCause() method).
+     */
+    public AuthenticationFailedException(Throwable cause) {
+        super(cause);
+    }
+
+    /**
+     * Constructs a new AuthenticationFailedException with the specified detail
+     * message and cause.
+     *
+     * @param message the detail message (which is saved for later retrieval
+     *                by the getMessage() method).
+     * @param cause   the cause (which is saved for later retrieval by the
+     *                getCause() method).
+     */
+    public AuthenticationFailedException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

common/src/main/java/io/seata/common/util/HttpClientUtil.java

@@ -16,6 +16,7 @@
 package io.seata.common.util;
 
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.http.NameValuePair;
 import org.apache.http.client.ClientProtocolException;
 import org.apache.http.client.config.RequestConfig;
@@ -52,7 +53,7 @@ public class HttpClientUtil {
     private static final Map<Integer/*timeout*/, CloseableHttpClient> HTTP_CLIENT_MAP = new ConcurrentHashMap<>();
 
     private static final PoolingHttpClientConnectionManager POOLING_HTTP_CLIENT_CONNECTION_MANAGER =
-            new PoolingHttpClientConnectionManager();
+        new PoolingHttpClientConnectionManager();
 
     static {
         POOLING_HTTP_CLIENT_CONNECTION_MANAGER.setMaxTotal(10);
@@ -66,25 +67,35 @@ public class HttpClientUtil {
         })));
     }
 
+
     // post request
     public static CloseableHttpResponse doPost(String url, Map<String, String> params, Map<String, String> header,
-        int timeout) throws IOException {
+                                               int timeout) throws IOException {
         try {
             URIBuilder builder = new URIBuilder(url);
             URI uri = builder.build();
             HttpPost httpPost = new HttpPost(uri);
+            String contentType = "";
             if (header != null) {
                 header.forEach(httpPost::addHeader);
+                contentType = header.get("Content-Type");
+            }
+            if (StringUtils.isNotBlank(contentType)) {
+                if (ContentType.APPLICATION_FORM_URLENCODED.getMimeType().equals(contentType)) {
+                    List<NameValuePair> nameValuePairs = new ArrayList<>();
+                    params.forEach((k, v) -> {
+                        nameValuePairs.add(new BasicNameValuePair(k, v));
+                    });
+                    String requestBody = URLEncodedUtils.format(nameValuePairs, StandardCharsets.UTF_8);
+                    StringEntity stringEntity = new StringEntity(requestBody, ContentType.APPLICATION_FORM_URLENCODED);
+                    httpPost.setEntity(stringEntity);
+                } else if (ContentType.APPLICATION_JSON.getMimeType().equals(contentType)) {
+                    ObjectMapper objectMapper = new ObjectMapper();
+                    String requestBody = objectMapper.writeValueAsString(params);
+                    StringEntity stringEntity = new StringEntity(requestBody, ContentType.APPLICATION_JSON);
+                    httpPost.setEntity(stringEntity);
+                }
             }
-            List<NameValuePair> nameValuePairs = new ArrayList<>();
-            params.forEach((k, v) -> {
-                nameValuePairs.add(new BasicNameValuePair(k, v));
-            });
-            String requestBody = URLEncodedUtils.format(nameValuePairs, StandardCharsets.UTF_8);
-
-            StringEntity stringEntity = new StringEntity(requestBody, ContentType.APPLICATION_FORM_URLENCODED);
-            httpPost.setEntity(stringEntity);
-            httpPost.setHeader("Content-Type", "application/x-www-form-urlencoded");
             CloseableHttpClient client = HTTP_CLIENT_MAP.computeIfAbsent(timeout,
                 k -> HttpClients.custom().setConnectionManager(POOLING_HTTP_CLIENT_CONNECTION_MANAGER)
                     .setDefaultRequestConfig(RequestConfig.custom().setConnectionRequestTimeout(timeout)
@@ -99,7 +110,7 @@ public static CloseableHttpResponse doPost(String url, Map<String, String> param
 
     // get request
     public static CloseableHttpResponse doGet(String url, Map<String, String> param, Map<String, String> header,
-        int timeout) throws IOException {
+                                              int timeout) throws IOException {
         try {
             URIBuilder builder = new URIBuilder(url);
             if (param != null) {

discovery/seata-discovery-raft/src/main/java/io/seata/discovery/registry/raft/RaftRegistryServiceImpl.java

@@ -23,6 +23,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ThreadLocalRandom;
@@ -33,8 +34,10 @@
 import java.util.stream.Stream;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.seata.common.ConfigurationKeys;
+import io.seata.common.exception.AuthenticationFailedException;
 import io.seata.common.metadata.Metadata;
 import io.seata.common.metadata.MetadataResponse;
 import io.seata.common.metadata.Node;
@@ -49,7 +52,9 @@
 import org.apache.http.HttpStatus;
 import org.apache.http.StatusLine;
 import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.entity.ContentType;
 import org.apache.http.util.EntityUtils;
+import org.apache.http.protocol.HTTP;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -59,13 +64,31 @@
  * @author funkye
  */
 public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeListener> {
-   
+
     private static final Logger LOGGER = LoggerFactory.getLogger(RaftRegistryServiceImpl.class);
 
     private static final String REGISTRY_TYPE = "raft";
 
     private static final String PRO_SERVER_ADDR_KEY = "serverAddr";
 
+    private static final String PRO_USERNAME_KEY = "username";
+
+    private static final String PRO_PASSWORD_KEY = "password";
+
+    private static final String AUTHORIZATION_HEADER = "Authorization";
+
+    private static final String TOKEN_VALID_TIME_MS_KEY = "tokenValidityInMilliseconds";
+
+    private static final long TOKEN_EXPIRE_TIME_IN_MILLISECONDS;
+
+    private static final String USERNAME;
+
+    private static final String PASSWORD;
+
+    public static String jwtToken;
+
+    private static long tokenTimeStamp = -1;
+
     private static volatile RaftRegistryServiceImpl instance;
 
     private static final Configuration CONFIG = ConfigurationFactory.getInstance();
@@ -91,7 +114,20 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
      */
     private static final Map<String, List<InetSocketAddress>> ALIVE_NODES = new ConcurrentHashMap<>();
 
-    private RaftRegistryServiceImpl() {}
+    static {
+        TOKEN_EXPIRE_TIME_IN_MILLISECONDS = CONFIG.getLong(getTokenExpireTimeInMillisecondsKey(), 29 * 60 * 1000L);
+        USERNAME = CONFIG.getConfig(getRaftUserNameKey());
+        PASSWORD = CONFIG.getConfig(getRaftPassWordKey());
+    }
+
+    private RaftRegistryServiceImpl() {
+
+        try {
+            refreshToken();
+        } catch (IOException e) {
+            throw new RuntimeException("Init fetch token failed!", e);
+        }
+    }
 
     /**
      * Gets instance.
@@ -215,6 +251,29 @@ private static String getRaftAddrFileKey() {
             REGISTRY_TYPE, PRO_SERVER_ADDR_KEY);
     }
 
+    private static String getRaftUserNameKey() {
+        return String.join(ConfigurationKeys.FILE_CONFIG_SPLIT_CHAR, ConfigurationKeys.FILE_ROOT_REGISTRY,
+            REGISTRY_TYPE, PRO_USERNAME_KEY);
+    }
+
+    private static String getRaftPassWordKey() {
+        return String.join(ConfigurationKeys.FILE_CONFIG_SPLIT_CHAR, ConfigurationKeys.FILE_ROOT_REGISTRY,
+            REGISTRY_TYPE, PRO_PASSWORD_KEY);
+    }
+
+    private static String getTokenExpireTimeInMillisecondsKey() {
+        return String.join(ConfigurationKeys.FILE_CONFIG_SPLIT_CHAR, ConfigurationKeys.FILE_ROOT_REGISTRY,
+            REGISTRY_TYPE, TOKEN_VALID_TIME_MS_KEY);
+    }
+
+    private static boolean isTokenExpired() {
+        if (tokenTimeStamp == -1) {
+            return true;
+        }
+        long tokenExpiredTime = tokenTimeStamp + TOKEN_EXPIRE_TIME_IN_MILLISECONDS;
+        return System.currentTimeMillis() >= tokenExpiredTime;
+    }
+
     private InetSocketAddress convertInetSocketAddress(Node node) {
         Node.Endpoint endpoint = node.getTransaction();
         return new InetSocketAddress(endpoint.getHost(), endpoint.getPort());
@@ -238,17 +297,35 @@ public List<InetSocketAddress> aliveLookup(String transactionServiceGroup) {
     }
 
     private static boolean watch() {
+        Map<String, String> header = new HashMap<>();
+        header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
         Map<String, String> param = new HashMap<>();
         String clusterName = CURRENT_TRANSACTION_CLUSTER_NAME;
         Map<String, Long> groupTerms = METADATA.getClusterTerm(clusterName);
         groupTerms.forEach((k, v) -> param.put(k, String.valueOf(v)));
         for (String group : groupTerms.keySet()) {
             String tcAddress = queryHttpAddress(clusterName, group);
-            try (CloseableHttpResponse response =
-                HttpClientUtil.doPost("http://" + tcAddress + "/metadata/v1/watch", param, null, 30000)) {
-                if (response != null) {
-                    StatusLine statusLine = response.getStatusLine();
-                    return statusLine != null && statusLine.getStatusCode() == HttpStatus.SC_OK;
+            try {
+                if (isTokenExpired()) {
+                    refreshToken();
+                }
+                if (!Objects.isNull(jwtToken)) {
+                    header.put(AUTHORIZATION_HEADER, jwtToken);
+                }
+                try (CloseableHttpResponse response =
+                         HttpClientUtil.doPost("http://" + tcAddress + "/metadata/v1/watch", param, header, 30000)) {
+                    if (response != null) {
+                        StatusLine statusLine = response.getStatusLine();
+                        if (statusLine != null && statusLine.getStatusCode() == HttpStatus.SC_UNAUTHORIZED) {
+                            if (StringUtils.isNotBlank(USERNAME) && StringUtils.isNotBlank(PASSWORD)) {
+                                LOGGER.warn("Authentication failed!");
+                                throw new IOException();
+                            } else {
+                                throw new AuthenticationFailedException("Authentication failed! you should configure the correct username and password.");
+                            }
+                        }
+                        return statusLine != null && statusLine.getStatusCode() == HttpStatus.SC_OK;
+                    }
                 }
             } catch (IOException e) {
                 LOGGER.error("watch cluster node: {}, fail: {}", tcAddress, e.getMessage());
@@ -265,7 +342,7 @@ private static boolean watch() {
 
     @Override
     public List<InetSocketAddress> refreshAliveLookup(String transactionServiceGroup,
-        List<InetSocketAddress> aliveAddress) {
+                                                      List<InetSocketAddress> aliveAddress) {
         if (METADATA.isRaftMode()) {
             Node leader = METADATA.getLeader(getServiceGroup(transactionServiceGroup));
             InetSocketAddress leaderAddress = convertInetSocketAddress(leader);
@@ -286,14 +363,35 @@ private static void acquireClusterMetaDataByClusterName(String clusterName) {
 
     private static void acquireClusterMetaData(String clusterName, String group) {
         String tcAddress = queryHttpAddress(clusterName, group);
+        Map<String, String> header = new HashMap<>();
+        header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_FORM_URLENCODED.getMimeType());
+        if (isTokenExpired()) {
+            try {
+                refreshToken();
+            } catch (IOException e) {
+                LOGGER.error(e.getMessage(), e);
+            }
+        }
+        if (StringUtils.isNotBlank(jwtToken)) {
+            header.put(AUTHORIZATION_HEADER, jwtToken);
+        }
         if (StringUtils.isNotBlank(tcAddress)) {
             Map<String, String> param = new HashMap<>();
             param.put("group", group);
             String response = null;
             try (CloseableHttpResponse httpResponse =
-                HttpClientUtil.doGet("http://" + tcAddress + "/metadata/v1/cluster", param, null, 1000)) {
-                if (httpResponse != null && httpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
-                    response = EntityUtils.toString(httpResponse.getEntity(), StandardCharsets.UTF_8);
+                     HttpClientUtil.doGet("http://" + tcAddress + "/metadata/v1/cluster", param, header, 1000)) {
+                if (httpResponse != null) {
+                    if (httpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
+                        response = EntityUtils.toString(httpResponse.getEntity(), StandardCharsets.UTF_8);
+                    } else if (httpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_UNAUTHORIZED) {
+                        if (StringUtils.isNotBlank(USERNAME) && StringUtils.isNotBlank(PASSWORD)) {
+                            LOGGER.warn("Authentication failed !");
+                            throw new IOException();
+                        } else {
+                            throw new AuthenticationFailedException("Authentication failed! you should configure the correct username and password.");
+                        }
+                    }
                 }
                 MetadataResponse metadataResponse;
                 if (StringUtils.isNotBlank(response)) {
@@ -310,6 +408,45 @@ private static void acquireClusterMetaData(String clusterName, String group) {
         }
     }
 
+    public static void refreshToken() throws IOException {
+        // if username and password is not in config , return
+        if (Objects.isNull(USERNAME) || Objects.isNull(PASSWORD)) {
+            return;
+        }
+        String raftClusterAddress = CONFIG.getConfig(getRaftAddrFileKey());
+        // get token and set it in cache
+        if (StringUtils.isNotBlank(raftClusterAddress)) {
+            String[] tcAddressList = raftClusterAddress.split(",");
+            String tcAddress = tcAddressList[0];
+            Map<String, String> param = new HashMap<>();
+            param.put(PRO_USERNAME_KEY, USERNAME);
+            param.put(PRO_PASSWORD_KEY, PASSWORD);
+            Map<String, String> header = new HashMap<>();
+            header.put(HTTP.CONTENT_TYPE, ContentType.APPLICATION_JSON.getMimeType());
+            String response = null;
+            tokenTimeStamp = System.currentTimeMillis();
+            try (CloseableHttpResponse httpResponse =
+                     HttpClientUtil.doPost("http://" + tcAddress + "/api/v1/auth/login", param, header, 1000)) {
+                if (httpResponse != null) {
+                    if (httpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
+                        response = EntityUtils.toString(httpResponse.getEntity(), StandardCharsets.UTF_8);
+                        JsonNode jsonNode = OBJECT_MAPPER.readTree(response);
+                        String codeStatus = jsonNode.get("code").asText();
+                        if (!StringUtils.equals(codeStatus, "200")) {
+                            //authorized failed,throw exception to kill process
+                            throw new AuthenticationFailedException("Authentication failed! you should configure the correct username and password.");
+                        }
+                        jwtToken = jsonNode.get("data").asText();
+                    } else {
+                        //authorized failed,throw exception to kill process
+                        throw new AuthenticationFailedException("Authentication failed! you should configure the correct username and password.");
+                    }
+                }
+            }
+        }
+    }
+
+
     @Override
     public List<InetSocketAddress> lookup(String key) throws Exception {
         String clusterName = getServiceGroup(key);