[INLONG-12211][Manager] Add tenant verification to the source/list API

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceService.java

@@ -95,9 +95,10 @@ public interface StreamSourceService {
      * Paging query source information based on conditions.
      *
      * @param request paging request.
+     * @param operator Operator's name.
      * @return source list
      */
-    PageResult<? extends StreamSource> listByCondition(SourcePageRequest request);
+    PageResult<? extends StreamSource> listByCondition(SourcePageRequest request, String operator);
 
     /**
      * Paging query source information based on conditions.

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java

@@ -223,10 +223,17 @@ public Map<String, List<StreamSource>> getSourcesMap(InlongGroupInfo groupInfo,
     }
 
     @Override
-    public PageResult<? extends StreamSource> listByCondition(SourcePageRequest request) {
+    public PageResult<? extends StreamSource> listByCondition(SourcePageRequest request, String operator) {
         PageHelper.startPage(request.getPageNum(), request.getPageSize());
         OrderFieldEnum.checkOrderField(request);
         OrderTypeEnum.checkOrderType(request);
+        InlongGroupEntity groupEntity =
+                groupMapper.selectByGroupId(request.getInlongGroupId());
+        if (groupEntity == null) {
+            throw new BusinessException(ErrorCodeEnum.GROUP_NOT_FOUND);
+        }
+        userService.checkUser(groupEntity.getInCharges(), operator,
+                "Current user does not have permission to list source info");
         Page<StreamSourceEntity> entityPage = (Page<StreamSourceEntity>) sourceMapper.selectByCondition(request);
         // Encapsulate the paging query results into the PageInfo object to obtain related paging information
         Map<String, Page<StreamSourceEntity>> sourceMap = Maps.newHashMap();
@@ -557,6 +564,12 @@ private void chkUnmodifiableParams(SourceRequest request) {
     @Override
     public List<Integer> addDataAddTask(DataAddTaskRequest request, String operator) {
         LOGGER.info("begin to add data add task info: {}", request);
+        InlongGroupEntity groupEntity = groupMapper.selectByGroupId(request.getGroupId());
+        if (groupEntity == null) {
+            throw new BusinessException(ErrorCodeEnum.GROUP_NOT_FOUND,
+                    String.format("InlongGroup does not exist with InlongGroupId=%s", request.getGroupId()));
+        }
+        userService.checkUser(request.getGroupId(), operator, "current user not allowed to add data add task");
         String auditVersion = String.valueOf(sourceMapper.selectDataAddTaskCount(request.getGroupId(), null));
         request.setAuditVersion(auditVersion);
         List<String> agentIpList = request.getAgentIpList();

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

@@ -83,7 +83,7 @@ public Response<StreamSource> get(@PathVariable Integer id) {
     @RequestMapping(value = "/source/list", method = RequestMethod.POST)
     @ApiOperation(value = "List stream sources by paginating")
     public Response<PageResult<? extends StreamSource>> listByCondition(@RequestBody SourcePageRequest request) {
-        return Response.success(sourceService.listByCondition(request));
+        return Response.success(sourceService.listByCondition(request, LoginUserUtils.getLoginUser().getName()));
     }
 
     @RequestMapping(value = "/source/update", method = RequestMethod.POST)