[INLONG-12133][Manager] Fix the security vulnerability in /api/cluster/testConnection (apache#12134)

Co-authored-by: wakefu <wakefu@tencent.com>

Author: fuweng11

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/InlongClusterServiceImpl.java

@@ -37,6 +37,7 @@
 import org.apache.inlong.manager.common.exceptions.BusinessException;
 import org.apache.inlong.manager.common.util.CommonBeanUtils;
 import org.apache.inlong.manager.common.util.Preconditions;
+import org.apache.inlong.manager.common.util.UrlVerificationUtils;
 import org.apache.inlong.manager.dao.entity.InlongClusterEntity;
 import org.apache.inlong.manager.dao.entity.InlongClusterNodeEntity;
 import org.apache.inlong.manager.dao.entity.InlongClusterTagEntity;
@@ -940,6 +941,17 @@ public String getManagerSSHPublicKey() {
 
     @Override
     public Boolean testSSHConnection(ClusterNodeRequest request) {
+        // Validate IP to prevent SSRF attacks
+        String ip = request.getIp();
+        if (StringUtils.isNotBlank(ip)) {
+            try {
+                UrlVerificationUtils.validateHostNotInternal(ip);
+            } catch (Exception e) {
+                throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER,
+                        "SSRF protection: " + e.getMessage());
+            }
+        }
+
         AgentClusterNodeRequest nodeRequest = (AgentClusterNodeRequest) request;
         try {
             CommandResult commandResult = commandExecutor.execRemote(nodeRequest, "ls");
@@ -1271,6 +1283,17 @@ public AuditConfig getAuditConfig(String clusterTag) {
     public Boolean testConnection(ClusterRequest request) {
         LOGGER.info("begin test connection for: {}", request);
 
+        // Validate URL to prevent SSRF attacks
+        String url = request.getUrl();
+        if (StringUtils.isNotBlank(url)) {
+            try {
+                UrlVerificationUtils.validateUrlNotInternal(url);
+            } catch (Exception e) {
+                throw new BusinessException(ErrorCodeEnum.INVALID_PARAMETER,
+                        "SSRF protection: " + e.getMessage());
+            }
+        }
+
         // according to the data node type, test connection
         InlongClusterOperator clusterOperator = clusterOperatorFactory.getInstance(request.getType());
         Boolean result = clusterOperator.testConnection(request);

inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InlongClusterController.java

@@ -305,12 +305,14 @@ public Response<String> getManagerSSHPublicKey() {
 
     @PostMapping("/cluster/node/testSSHConnection")
     @ApiOperation(value = "Test SSH connection for inlong cluster node")
+    @RequiresRoles(logical = Logical.OR, value = {UserRoleCode.INLONG_ADMIN, UserRoleCode.TENANT_ADMIN})
     public Response<Boolean> testSSHConnection(@RequestBody ClusterNodeRequest request) {
         return Response.success(clusterService.testSSHConnection(request));
     }
 
     @PostMapping("/cluster/testConnection")
     @ApiOperation(value = "Test connection for inlong cluster")
+    @RequiresRoles(logical = Logical.OR, value = {UserRoleCode.INLONG_ADMIN, UserRoleCode.TENANT_ADMIN})
     public Response<Boolean> testConnection(@Validated @RequestBody ClusterRequest request) {
         return Response.success(clusterService.testConnection(request));
     }