From aa1f33a9ec6df06c9d49a1daa7d7203898d33fd4 Mon Sep 17 00:00:00 2001 From: fuwenkai <834260992@qq.com> Date: Tue, 17 Dec 2024 14:58:36 +0800 Subject: [PATCH] [INLONG-11608][Manager] Add permission verification for ordinary users to create streams and sinks --- .../service/sink/StreamSinkServiceImpl.java | 7 +++- .../source/StreamSourceServiceImpl.java | 6 ++++ .../stream/InlongStreamServiceImpl.java | 34 +++++++++++-------- 3 files changed, 31 insertions(+), 16 deletions(-) diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/StreamSinkServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/StreamSinkServiceImpl.java index dab2cdc93ae..a278bc51c0c 100644 --- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/StreamSinkServiceImpl.java +++ b/inlong-manager/manager-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/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java index 324c9969103..04e39415000 100644 --- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java +++ b/inlong-manager/manager-service/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 existList = sourceMapper.selectByRelatedId(groupId, streamId, sourceName); diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamServiceImpl.java index 101c39d24fb..e8aa01aa2c8 100644 --- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamServiceImpl.java +++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/stream/InlongStreamServiceImpl.java @@ -17,6 +17,19 @@ package org.apache.inlong.manager.service.stream; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.github.pagehelper.Page; +import com.github.pagehelper.PageHelper; +import net.sf.jsqlparser.JSQLParserException; +import net.sf.jsqlparser.parser.CCJSqlParserManager; +import net.sf.jsqlparser.statement.Statement; +import net.sf.jsqlparser.statement.create.table.ColDataType; +import net.sf.jsqlparser.statement.create.table.ColumnDefinition; +import net.sf.jsqlparser.statement.create.table.CreateTable; +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; import org.apache.inlong.manager.common.consts.InlongConstants; import org.apache.inlong.manager.common.enums.ErrorCodeEnum; import org.apache.inlong.manager.common.enums.GroupStatus; @@ -68,20 +81,6 @@ import org.apache.inlong.manager.service.sink.StreamSinkService; import org.apache.inlong.manager.service.source.StreamSourceService; import org.apache.inlong.manager.service.user.UserService; - -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.core.type.TypeReference; -import com.fasterxml.jackson.databind.ObjectMapper; -import com.github.pagehelper.Page; -import com.github.pagehelper.PageHelper; -import net.sf.jsqlparser.JSQLParserException; -import net.sf.jsqlparser.parser.CCJSqlParserManager; -import net.sf.jsqlparser.statement.Statement; -import net.sf.jsqlparser.statement.create.table.ColDataType; -import net.sf.jsqlparser.statement.create.table.ColumnDefinition; -import net.sf.jsqlparser.statement.create.table.CreateTable; -import org.apache.commons.collections.CollectionUtils; -import org.apache.commons.lang3.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -166,7 +165,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) {