[INLONG-11608][Manager] Add permission verification for ordinary user…

…s to create streams and sinks (#11609)
apache · Dec 18, 2024 · 45b8394 · 45b8394
1 parent b66102c
commit 45b8394
Show file tree

Hide file tree

Showing 4 changed files with 22 additions and 6 deletions.
diff --git a/...r-service/src/main/java/org/apache/inlong/manager/service/sink/StreamSinkServiceImpl.java b/...r-service/src/main/java/org/apache/inlong/manager/service/sink/StreamSinkServiceImpl.java
@@ -153,7 +153,12 @@ public Integer save(SinkRequest request, String operator) {
         // Check if it can be added
         String groupId = request.getInlongGroupId();
         groupCheckService.checkGroupStatus(groupId, operator);
-
+        InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId);
+        if (groupEntity == null) {
+            throw new BusinessException(String.format("InlongGroup does not exist with InlongGroupId=%s", groupId));
+        }
+        userService.checkUser(groupEntity.getInCharges(), operator,
+                "Current user does not have permission to create sink info");
         // Make sure that there is no same sink name under the current groupId and streamId
         String streamId = request.getInlongStreamId();
         String sinkName = request.getSinkName();

diff --git a/...rvice/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java b/...rvice/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
@@ -105,6 +105,12 @@ public Integer save(SourceRequest request, String operator) {
         // Check if it can be added
         String groupId = request.getInlongGroupId();
         String streamId = request.getInlongStreamId();
+        InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId);
+        if (groupEntity == null) {
+            throw new BusinessException(String.format("InlongGroup does not exist with InlongGroupId=%s", groupId));
+        }
+        userService.checkUser(groupEntity.getInCharges(), operator,
+                "Current user does not have permission to create source info");
         InlongStreamEntity streamEntity = groupCheckService.checkStreamStatus(groupId, streamId, operator);
         String sourceName = request.getSourceName();
         List<StreamSourceEntity> existList = sourceMapper.selectByRelatedId(groupId, streamId, sourceName);

diff --git a/...rvice/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamServiceImpl.java b/...rvice/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamServiceImpl.java
@@ -166,7 +166,12 @@ public Integer save(InlongStreamRequest request, String operator) {
 
         // Check if it can be added
         checkGroupStatusIsTemp(groupId);
-
+        InlongGroupEntity groupEntity = groupMapper.selectByGroupId(groupId);
+        if (groupEntity == null) {
+            throw new BusinessException(String.format("InlongGroup does not exist with InlongGroupId=%s", groupId));
+        }
+        userService.checkUser(groupEntity.getInCharges(), operator,
+                "Current user does not have permission to create stream info");
         // The streamId under the same groupId cannot be repeated
         Integer count = streamMapper.selectExistByIdentifier(groupId, streamId);
         if (count >= 1) {

diff --git a/...ger-service/src/test/java/org/apache/inlong/manager/service/sort/SortServiceImplTest.java b/...ger-service/src/test/java/org/apache/inlong/manager/service/sort/SortServiceImplTest.java
@@ -297,7 +297,7 @@ private void prepareGroupId(String groupId) {
         request.setVersion(InlongConstants.INITIAL_VERSION);
         request.setName("test_group_name");
         request.setMqType(ClusterType.PULSAR);
-        request.setInCharges(TEST_CREATOR);
+        request.setInCharges(GLOBAL_OPERATOR);
         List<InlongGroupExtInfo> extList = new ArrayList<>();
         InlongGroupExtInfo ext1 = InlongGroupExtInfo
                 .builder()
@@ -315,7 +315,7 @@ private void prepareGroupId(String groupId) {
         extList.add(ext1);
         extList.add(ext2);
         request.setExtList(extList);
-        groupService.save(request, "test operator");
+        groupService.save(request, GLOBAL_OPERATOR);
     }
 
     private void prepareStreamId(String groupId, String streamId, String topic) {
@@ -333,7 +333,7 @@ private void prepareStreamId(String groupId, String streamId, String topic) {
         ext.setKeyName(ClusterSwitch.BACKUP_MQ_RESOURCE);
         ext.setKeyValue("backup_" + topic);
         request.setExtList(extInfos);
-        streamService.save(request, "test_operator");
+        streamService.save(request, GLOBAL_OPERATOR);
     }
 
     private void prepareCluster(String clusterName, String clusterTag) {
@@ -382,7 +382,7 @@ private void prepareTask(String taskName, String groupId, String clusterName, St
         properties.put("delimiter", "|");
         properties.put("dataType", "text");
         request.setProperties(properties);
-        streamSinkService.save(request, TEST_CREATOR);
+        streamSinkService.save(request, GLOBAL_OPERATOR);
     }
 
 }