Skip to content

[Console] Add Spark application team-level permission checks #4395

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
shangeyao wants to merge 1 commit into from
+43 −7
Closed

[Console] Add Spark application team-level permission checks #4395

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 20 additions & 5 deletions ...ole-service/src/main/java/org/apache/streampark/console/core/aspect/PermissionAspect.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,10 @@
import org.apache.streampark.console.base.exception.ApiAlertException;
import org.apache.streampark.console.core.annotation.Permission;
import org.apache.streampark.console.core.entity.FlinkApplication;
import org.apache.streampark.console.core.entity.SparkApplication;
import org.apache.streampark.console.core.enums.UserTypeEnum;
import org.apache.streampark.console.core.service.application.FlinkApplicationManageService;
import org.apache.streampark.console.core.service.application.SparkApplicationManageService;
import org.apache.streampark.console.core.util.ServiceHelper;
import org.apache.streampark.console.system.entity.Member;
import org.apache.streampark.console.system.entity.User;
Expand Down Expand Up @@ -55,6 +57,9 @@ public class PermissionAspect {
@Autowired
private FlinkApplicationManageService applicationManageService;

@Autowired
private SparkApplicationManageService sparkApplicationManageService;

@Pointcut("@annotation(org.apache.streampark.console.core.annotation.Permission)")
public void permissionPointcut() {
}
Expand Down Expand Up @@ -89,19 +94,29 @@ public RestResponse permissionAction(ProceedingJoinPoint joinPoint) throws Throw
Long appId = getId(joinPoint, methodSignature, permission.app());
if (appId != null) {
FlinkApplication app = applicationManageService.getById(appId);
ApiAlertException.throwIfTrue(app == null, "Invalid operation, application is null");
if (!currentUser.getUserId().equals(app.getUserId())) {
Member member = memberService.getByTeamIdUserName(app.getTeamId(), currentUser.getUsername());
if (app != null) {
checkAppTeamAccess(currentUser, app.getUserId(), app.getTeamId());
} else {
SparkApplication sparkApp = sparkApplicationManageService.getById(appId);
ApiAlertException.throwIfTrue(
member == null,
"Permission denied, this job not created by the current user, And the job cannot be found in the current user's team.");
sparkApp == null, "Invalid operation, application is null");
checkAppTeamAccess(currentUser, sparkApp.getUserId(), sparkApp.getTeamId());
}
}
}

return (RestResponse) joinPoint.proceed();
}

private void checkAppTeamAccess(User currentUser, Long ownerId, Long teamId) {
if (!currentUser.getUserId().equals(ownerId)) {
Member member = memberService.getByTeamIdUserName(teamId, currentUser.getUsername());
ApiAlertException.throwIfTrue(
member == null,
"Permission denied, this job not created by the current user, And the job cannot be found in the current user's team.");
}
}

private Long getId(ProceedingJoinPoint joinPoint, MethodSignature methodSignature, String expr) {
if (StringUtils.isEmpty(expr)) {
return null;
Expand Down
25 changes: 23 additions & 2 deletions ...c/main/java/org/apache/streampark/console/core/controller/SparkApplicationController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.apache.streampark.console.base.domain.RestResponse;
import org.apache.streampark.console.base.exception.InternalException;
import org.apache.streampark.console.core.annotation.AppChangeEvent;
import org.apache.streampark.console.core.annotation.Permission;
import org.apache.streampark.console.core.entity.ApplicationLog;
import org.apache.streampark.console.core.entity.FlinkApplicationBackup;
import org.apache.streampark.console.core.entity.SparkApplication;
Expand Down Expand Up @@ -75,19 +76,22 @@ public class SparkApplicationController {
private ResourceService resourceService;

@PostMapping("get")
@Permission(app = "#app.id")
@RequiresPermissions("app:detail")
public RestResponse get(SparkApplication app) {
SparkApplication application = applicationManageService.getApp(app.getId());
return RestResponse.success(application);
}

@Permission(team = "#app.teamId")
@PostMapping("create")
@RequiresPermissions("app:create")
public RestResponse create(SparkApplication app) throws IOException {
boolean saved = applicationManageService.create(app);
return RestResponse.success(saved);
}

@Permission(app = "#app.id", team = "#app.teamId")
@PostMapping("copy")
@RequiresPermissions("app:copy")
public RestResponse copy(SparkApplication app) throws IOException {
Expand All @@ -96,6 +100,7 @@ public RestResponse copy(SparkApplication app) throws IOException {
}

@AppChangeEvent
@Permission(app = "#app.id")
@PostMapping("update")
@RequiresPermissions("app:update")
public RestResponse update(SparkApplication app) {
Expand All @@ -104,12 +109,14 @@ public RestResponse update(SparkApplication app) {
}

@PostMapping("dashboard")
@Permission(team = "#teamId")
public RestResponse dashboard(Long teamId) {
Map<String, Serializable> dashboardMap = applicationInfoService.getDashboardDataMap(teamId);
return RestResponse.success(dashboardMap);
}

@PostMapping("list")
@Permission(team = "#app.teamId")
@RequiresPermissions("app:view")
public RestResponse list(SparkApplication app, RestRequest request) {
IPage<SparkApplication> applicationList = applicationManageService.page(app, request);
Expand All @@ -118,27 +125,31 @@ public RestResponse list(SparkApplication app, RestRequest request) {

@AppChangeEvent
@PostMapping("mapping")
@Permission(app = "#app.id")
@RequiresPermissions("app:mapping")
public RestResponse mapping(SparkApplication app) {
boolean flag = applicationManageService.mapping(app);
return RestResponse.success(flag);
}

@AppChangeEvent
@Permission(app = "#app.id")
@PostMapping("revoke")
@RequiresPermissions("app:release")
public RestResponse revoke(SparkApplication app) {
applicationActionService.revoke(app.getId());
return RestResponse.success();
}

@Permission(app = "#app.id", team = "#app.teamId")
@PostMapping("check/start")
@RequiresPermissions("app:start")
public RestResponse checkStart(SparkApplication app) {
AppExistsStateEnum stateEnum = applicationInfoService.checkStart(app.getId());
return RestResponse.success(stateEnum.get());
}

@Permission(app = "#app.id", team = "#app.teamId")
@PostMapping("start")
@RequiresPermissions("app:start")
public RestResponse start(SparkApplication app) {
Expand All @@ -150,6 +161,7 @@ public RestResponse start(SparkApplication app) {
}
}

@Permission(app = "#app.id", team = "#app.teamId")
@PostMapping("cancel")
@RequiresPermissions("app:cancel")
public RestResponse cancel(SparkApplication app) throws Exception {
Expand All @@ -158,13 +170,15 @@ public RestResponse cancel(SparkApplication app) throws Exception {
}

@AppChangeEvent
@Permission(app = "#app.id")
@PostMapping("clean")
@RequiresPermissions("app:clean")
public RestResponse clean(SparkApplication app) {
applicationManageService.clean(app);
return RestResponse.success(true);
}

@Permission(app = "#app.id")
@PostMapping("forcedStop")
@RequiresPermissions("app:cancel")
public RestResponse forcedStop(SparkApplication app) {
Expand All @@ -178,12 +192,14 @@ public RestResponse yarn() {
}

@PostMapping("name")
@Permission(app = "#app.id", team = "#app.teamId")
public RestResponse yarnName(SparkApplication app) {
String yarnName = applicationInfoService.getYarnName(app.getConfig());
return RestResponse.success(yarnName);
}

@PostMapping("check/name")
@Permission(app = "#app.id", team = "#app.teamId")
public RestResponse checkName(SparkApplication app) {
AppExistsStateEnum exists = applicationInfoService.checkExists(app);
return RestResponse.success(exists.get());
Expand All @@ -196,31 +212,36 @@ public RestResponse readConf(SparkApplication app) throws IOException {
}

@PostMapping("backups")
@Permission(app = "#backUp.appId", team = "#backUp.teamId")
public RestResponse backups(FlinkApplicationBackup backUp, RestRequest request) {
IPage<FlinkApplicationBackup> backups = backUpService.getPage(backUp, request);
return RestResponse.success(backups);
}

@PostMapping("opt_log")
@Permission(app = "#applicationLog.appId", team = "#applicationLog.teamId")
public RestResponse optionlog(ApplicationLog applicationLog, RestRequest request) {
IPage<ApplicationLog> applicationList = applicationLogService.getPage(applicationLog, request);
return RestResponse.success(applicationList);
}

@Permission(app = "#applicationLog.appId", team = "#applicationLog.teamId")
@PostMapping("delete/opt_log")
@RequiresPermissions("app:delete")
public RestResponse deleteOperationLog(Long id) {
Boolean deleted = applicationLogService.removeById(id);
public RestResponse deleteOperationLog(ApplicationLog applicationLog) {
Boolean deleted = applicationLogService.delete(applicationLog);
return RestResponse.success(deleted);
}

@Permission(app = "#app.id", team = "#app.teamId")
@PostMapping("delete")
@RequiresPermissions("app:delete")
public RestResponse delete(SparkApplication app) throws InternalException {
Boolean deleted = applicationManageService.remove(app.getId());
return RestResponse.success(deleted);
}

@Permission(app = "#backUp.appId")
@PostMapping("delete/bak")
public RestResponse deleteBak(FlinkApplicationBackup backUp) throws InternalException {
Boolean deleted = backUpService.removeById(backUp.getId());
Expand Down
Loading