optimize: add secure authentication to interfaces in ClusterController

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

@@ -24,6 +24,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;
@@ -34,6 +35,7 @@
 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.metadata.Metadata;
@@ -46,6 +48,7 @@
 import io.seata.config.Configuration;
 import io.seata.config.ConfigurationFactory;
 import io.seata.discovery.registry.RegistryService;
+import org.apache.http.Header;
 import org.apache.http.HttpStatus;
 import org.apache.http.NameValuePair;
 import org.apache.http.StatusLine;
@@ -71,13 +74,23 @@
  * @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 String jwtToken;
+
+    private static final String NEW_TOKEN_KEY = "new_token";
+
     private static volatile RaftRegistryServiceImpl instance;
 
     private static final Configuration CONFIG = ConfigurationFactory.getInstance();
@@ -113,7 +126,8 @@ public class RaftRegistryServiceImpl implements RegistryService<ConfigChangeList
         POOLING_HTTP_CLIENT_CONNECTION_MANAGER.setDefaultMaxPerRoute(10);
     }
 
-    private RaftRegistryServiceImpl() {}
+    private RaftRegistryServiceImpl() {
+    }
 
     /**
      * Gets instance.
@@ -244,6 +258,16 @@ 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 InetSocketAddress convertInetSocketAddress(Node node) {
         Node.Endpoint endpoint = node.getTransaction();
         return new InetSocketAddress(endpoint.getHost(), endpoint.getPort());
@@ -267,16 +291,29 @@ public List<InetSocketAddress> aliveLookup(String transactionServiceGroup) {
     }
 
     private static boolean watch() {
+        Map<String, String> header = new HashMap<>();
         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);
+            String token = getToken(tcAddress);
+            if (!Objects.isNull(token)) {
+                header.put(AUTHORIZATION_HEADER, token);
+            }
             try (CloseableHttpResponse response =
-                doPost("http://" + tcAddress + "/metadata/v1/watch", param, null, 30000)) {
+                     doPost("http://" + tcAddress + "/metadata/v1/watch", param, header, 30000, "application/x-www-form-urlencoded")) {
                 if (response != null) {
+                    //refresh jwt token
+                    Header tokenHeader = response.getFirstHeader(NEW_TOKEN_KEY);
+                    if (tokenHeader != null) {
+                        jwtToken = tokenHeader.getValue();
+                    }
                     StatusLine statusLine = response.getStatusLine();
+                    if (statusLine != null && statusLine.getStatusCode() == HttpStatus.SC_UNAUTHORIZED) {
+                        LOGGER.error("Authentication failed! you should configure the correct username and password.");
+                    }
                     return statusLine != null && statusLine.getStatusCode() == HttpStatus.SC_OK;
                 }
             } catch (Exception e) {
@@ -294,7 +331,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);
@@ -315,14 +352,28 @@ 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<>();
+        String token = getToken(tcAddress);
+        if (!Objects.isNull(token)) {
+            header.put(AUTHORIZATION_HEADER, token);
+        }
         if (StringUtils.isNotBlank(tcAddress)) {
             Map<String, String> param = new HashMap<>();
             param.put("group", group);
             String response = null;
             try (CloseableHttpResponse httpResponse =
-                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);
+                     doGet("http://" + tcAddress + "/metadata/v1/cluster", param, header, 1000)) {
+                if (httpResponse != null) {
+                    //refresh jwt token
+                    Header newTokenHeader = httpResponse.getFirstHeader(NEW_TOKEN_KEY);
+                    if (!Objects.isNull(newTokenHeader)) {
+                        jwtToken = newTokenHeader.getValue();
+                    }
+                    if (httpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
+                        response = EntityUtils.toString(httpResponse.getEntity(), StandardCharsets.UTF_8);
+                    } else if (httpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_UNAUTHORIZED) {
+                        LOGGER.error("Authentication failed! you should configure the correct username and password.");
+                    }
                 }
                 MetadataResponse metadataResponse;
                 if (StringUtils.isNotBlank(response)) {
@@ -339,8 +390,42 @@ private static void acquireClusterMetaData(String clusterName, String group) {
         }
     }
 
+    public static String getToken(String tcAddress) {
+        // if the token is present ,return it
+        if (!Objects.isNull(jwtToken)) {
+            return jwtToken;
+        }
+        // if username and password is not in config , return null
+        String username = CONFIG.getConfig(getRaftUserNameKey());
+        String password = CONFIG.getConfig(getRaftPassWordKey());
+        if (Objects.isNull(username) || Objects.isNull(password)) {
+            return null;
+        }
+        // get token and set it in cache
+        if (StringUtils.isNotBlank(tcAddress)) {
+            Map<String, String> param = new HashMap<>();
+            param.put(PRO_USERNAME_KEY, username);
+            param.put(PRO_PASSWORD_KEY, password);
+            String response = null;
+            try (CloseableHttpResponse httpResponse =
+                     doPost("http://" + tcAddress + "/api/v1/auth/login", param, null, 1000, "application/json")) {
+                if (httpResponse != null && httpResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
+                    response = EntityUtils.toString(httpResponse.getEntity(), StandardCharsets.UTF_8);
+                    ObjectMapper objectMapper = new ObjectMapper();
+                    JsonNode jsonNode = objectMapper.readTree(response);
+                    jwtToken = jsonNode.get("data").asText();
+                }
+            } catch (IOException e) {
+                LOGGER.error(e.getMessage(), e);
+            }
+        }
+
+        return jwtToken;
+
+    }
+
     public static CloseableHttpResponse doGet(String url, Map<String, String> param, Map<String, String> header,
-        int timeout) {
+                                              int timeout) {
         CloseableHttpClient client;
         try {
             URIBuilder builder = new URIBuilder(url);
@@ -367,24 +452,35 @@ public static CloseableHttpResponse doGet(String url, Map<String, String> param,
     }
 
     public static CloseableHttpResponse doPost(String url, Map<String, String> params, Map<String, String> header,
-        int timeout) {
-        CloseableHttpClient client = null;
+                                               int timeout, String contentType) {
+        CloseableHttpClient client;
         try {
             URIBuilder builder = new URIBuilder(url);
             URI uri = builder.build();
             HttpPost httpPost = new HttpPost(uri);
             if (header != null) {
                 header.forEach(httpPost::addHeader);
             }
-            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");
+            if (contentType != null) {
+                httpPost.setHeader("Content-Type", contentType);
+            }
+
+            if ("application/x-www-form-urlencoded".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 ("application/json".equals(contentType)) {
+                ObjectMapper objectMapper = new ObjectMapper();
+                String requestBody = objectMapper.writeValueAsString(params);
+                StringEntity stringEntity = new StringEntity(requestBody, ContentType.APPLICATION_JSON);
+                httpPost.setEntity(stringEntity);
+            }
+
             client = HTTP_CLIENT_MAP.computeIfAbsent(timeout,
                 k -> HttpClients.custom().setConnectionManager(POOLING_HTTP_CLIENT_CONNECTION_MANAGER)
                     .setDefaultRequestConfig(RequestConfig.custom().setConnectionRequestTimeout(timeout)

script/client/conf/registry.conf

@@ -13,9 +13,16 @@
 #  limitations under the License.
 
 registry {
-  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa、custom
+  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa、custom、raft
   type = "file"
 
+   raft {
+      metadata-max-age-ms = 30000
+      serverAddr = "127.0.0.1:8848"
+      username = "seata"
+      password = "seata"
+   }
+
   nacos {
     application = "seata-server"
     serverAddr = "127.0.0.1:8848"

script/client/spring/application.properties

@@ -118,6 +118,8 @@ seata.registry.type=file
 
 seata.registry.raft.server-addr=
 seata.registry.raft.metadata-max-age-ms=30000
+seata.registry.raft.username=seata
+seata.registry.raft.password=seata
 seata.registry.consul.server-addr=127.0.0.1:8500
 
 seata.registry.etcd3.server-addr=http://localhost:2379

script/client/spring/application.yml

@@ -112,6 +112,8 @@ seata:
     raft:
       server-addr:
       metadata-max-age-ms: 30000
+      username: seata
+      password: seata
     file:
       name: file.conf
     consul:

server/src/main/resources/application.yml

@@ -50,4 +50,4 @@ seata:
     secretKey: SeataSecretKey0c382ef121d778043159209298fd40bf3850a017
     tokenValidityInMilliseconds: 1800000
     ignore:
-      urls: /,/**/*.css,/**/*.js,/**/*.html,/**/*.map,/**/*.svg,/**/*.png,/**/*.jpeg,/**/*.ico,/api/v1/auth/login,/metadata/v1/**
+      urls: /,/**/*.css,/**/*.js,/**/*.html,/**/*.map,/**/*.svg,/**/*.png,/**/*.jpeg,/**/*.ico,/api/v1/auth/login