You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2026/08/04 16:31:03 UTC
(streampark) branch dev-2.1.8 updated: [Improve] Refine permission checks and resource scope validation (#4468)
This is an automated email from the ASF dual-hosted git repository.
wolfboys pushed a commit to branch dev-2.1.8
in repository https://gitbox.apache.org/repos/asf/streampark.git
The following commit(s) were added to refs/heads/dev-2.1.8 by this push:
new cb74b6849 [Improve] Refine permission checks and resource scope validation (#4468)
cb74b6849 is described below
commit cb74b6849e24da7730f394c6868218a7ebe14065
Author: Tamer <ta...@gmail.com>
AuthorDate: Wed Aug 5 00:30:57 2026 +0800
[Improve] Refine permission checks and resource scope validation (#4468)
* Add missing permission checks to FlinkClusterController
check/start/shutdown/delete on flink/cluster/* had no @RequiresPermissions,
letting any logged-in user (regardless of team/role) start, stop or delete
arbitrary Flink clusters, and letting check() SSRF-probe any address the
caller supplies. Add cluster:create (check), cluster:update (start/shutdown)
and a new cluster:delete permission point (delete), seeded only for the
team-admin role to match existing cluster:create/update seeding. list/get/
page/remoteUrl are intentionally left unchanged since they are used by all
team members to populate the REMOTE-cluster dropdown when creating/editing
Applications.
* Fix cross-team secret exposure in VariableController
showOriginal/updateVariable/deleteVariable only checked @RequiresPermissions
without verifying the target Variable's real teamId, and updateVariable/
deleteVariable trusted the client-supplied Variable object (id/teamId) as-is.
This let an attacker read (via showOriginal, which bypasses dataMasking),
modify, or delete another team's variables, which may hold secret values.
Add a checkTeamAccess() helper that re-fetches the record and verifies
ADMIN-or-team-membership against the persisted teamId, apply it in
updateVariable/deleteVariable (now operating on the re-fetched, trusted
record), and add a new getOriginal() service method used by showOriginal
instead of the unchecked getById().
* Require team:view permission on /team/list
teamList had no @RequiresPermissions and TeamServiceImpl.page() passed the
request straight to the mapper with no per-user filtering, letting any
logged-in user enumerate every team on the platform. The team:view
permission point (t_menu id 110504) already exists in the seed data but was
never referenced by any controller, so no new SQL is needed; by default only
platform ADMIN has it (MenuServiceImpl.findUserPermissions() grants ADMIN
every permission unconditionally). The "switch team" dropdown used by
regular users calls a separate /member/teams endpoint and is unaffected.
* Require auth on proxyFlinkAssets and add cluster ownership check to proxy/cluster
proxyFlinkAssets had no @RequiresPermissions while /proxy/** is "anon" in
ShiroConfig's filter chain (JWTFilter is skipped for that URL prefix), so
Shiro evaluates the annotation against a fully anonymous Subject and this
endpoint was reachable by unauthenticated callers, exposing the same
flink/cluster/history/yarn reverse-proxy primitive (internal JobManager/YARN
RM REST APIs) as the already-protected proxyFlink. Add @RequiresPermissions
("app:view"), matching proxyFlink. Verified this does not break same-origin
sub-resource loading of the embedded Flink Web UI: Shiro here uses its
default session-based Subject binding (no stateless/SessionStorageEvaluator
override in this codebase), so the JSESSIONID cookie set on any earlier
authenticated request is sent automatically by the browser on same-origin
navigations/asset requests even without a custom Authorization header,
exactly how the pre-existing proxyFlink annotation already works today.
Also add checkProxyCluster() to proxyCluster, which previously did zero
ownership checks (FlinkCluster has no team_id column, so ownership is
derived indirectly: ADMIN, the cluster creator, or any user whose team owns
an Application bound to this cluster is allowed).
* Add functional authorization to Alert endpoints and FlinkEnvController
AlertController had no @RequiresPermissions on any endpoint, and
AlertConfigServiceImpl.page()/listWithOutPage returned every user's alert
configs (which can hold webhook/DingTalk/Feishu credentials) with no
per-caller filtering. FlinkEnvController's create/update/delete/sync/
validity/default had no authorization either, letting any logged-in user
modify the Flink runtime environments used by every team.
AlertConfig has no team_id column (bound to an individual user), so
authorization is scoped by ownership (creator userId or platform ADMIN)
rather than team membership: added a checkOwnership() helper and new
createAlertConfig/updateAlertConfig/getAlertConfig/listByCurrentUser service
methods that enforce it and never trust a client-supplied userId. Added four
new permission points (alert:view/add/update/delete) plus three for
FlinkEnv (env:create/update/delete), seeded only for team-admin
(role_id=100002), matching the existing convention for this settings area;
list/get/page endpoints used by all team members (e.g. the Flink-version
dropdown when creating an Application) are intentionally left unchanged.
Also includes a data-h2.sql fix that was dropped from the earlier
FlinkClusterController permission-check commit due to a tooling mishap: the
cluster:delete permission point insert only landed in mysql-data.sql/
pgsql-data.sql/the upgrade scripts, not data-h2.sql. It is restored here.
* Fix IDOR in ProjectController.buildLog
buildLog validated @PermissionScope(team = "#teamId") against the
client-supplied teamId parameter, never checking it against the actual
project's real team. An attacker could pass a teamId for a team they belong
to (satisfying the membership check) together with an arbitrary project id
from another team, and read that project's build log (which may contain
git credentials/build args). Remove the unenforceable teamId parameter and
@PermissionScope, and push a real check into
ProjectServiceImpl.getBuildLog(): fetch the project from the database and
verify ADMIN-or-team-membership against its persisted teamId before serving
the log. Ran the existing ProjectServiceImplTest (module-name validation
tests, unrelated to this change but covering the same class) - all 7 tests
still pass.
* Restrict H2 console to admin JWT and warn on default datasource credentials
/h2-console/** was mapped to "anon" in ShiroConfig's filter chain, meaning
JWTFilter never ran for it: the H2 web console (raw SQL access to the whole
application database) was reachable without authentication. The RCE itself
was already closed by spring.h2.console.settings.web-allow-others=false, but
the console page/endpoint remained anonymously reachable, which is a
meaningful residual risk for same-origin attacks. Add AdminOnlyFilter, a
JWTFilter subclass that additionally requires the authenticated user to be
UserType.ADMIN, and map /h2-console/** to it instead of anon.
Also add a startup warning in SpringProperties when the H2 datasource is
left on its built-in default credentials (admin/streampark), since those
same credentials gate both the primary datasource and (indirectly) console
access. Kept the default value itself unchanged to avoid a breaking change
for existing deployments; only a log warning was added, nudging operators
towards setting datasource.username/password in config.yaml.
---
.../src/main/assembly/script/data/mysql-data.sql | 16 ++++
.../src/main/assembly/script/data/pgsql-data.sql | 16 ++++
.../main/assembly/script/upgrade/mysql/2.1.8.sql | 18 ++++
.../main/assembly/script/upgrade/pgsql/2.1.8.sql | 18 ++++
.../console/base/config/SpringProperties.java | 10 +++
.../console/core/controller/AlertController.java | 21 +++--
.../core/controller/FlinkClusterController.java | 4 +
.../core/controller/FlinkEnvController.java | 9 ++
.../console/core/controller/ProjectController.java | 5 +-
.../console/core/controller/ProxyController.java | 1 +
.../core/controller/VariableController.java | 2 +-
.../console/core/service/VariableService.java | 9 ++
.../core/service/alert/AlertConfigService.java | 37 +++++++++
.../service/alert/impl/AlertConfigServiceImpl.java | 95 +++++++++++++++++++++-
.../core/service/impl/ProjectServiceImpl.java | 38 ++++++++-
.../core/service/impl/ProxyServiceImpl.java | 39 +++++++++
.../core/service/impl/VariableServiceImpl.java | 55 ++++++++++++-
.../system/authentication/AdminOnlyFilter.java | 74 +++++++++++++++++
.../console/system/authentication/ShiroConfig.java | 10 ++-
.../console/system/controller/TeamController.java | 1 +
.../src/main/resources/db/data-h2.sql | 16 ++++
21 files changed, 478 insertions(+), 16 deletions(-)
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/data/mysql-data.sql b/streampark-console/streampark-console-service/src/main/assembly/script/data/mysql-data.sql
index bc9cb2b5f..0a38b271d 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/data/mysql-data.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/data/mysql-data.sql
@@ -125,6 +125,14 @@ insert into `t_menu` values (130101, 130100, 'view', null, null, 'setting:view',
insert into `t_menu` values (130102, 130100, 'setting update', null, null, 'setting:update', null, '1', 1, null, now(), now());
insert into `t_menu` values (130401, 130400, 'add cluster', '/setting/add_cluster', 'setting/FlinkCluster/AddCluster', 'cluster:create', '', '0', 0, null, now(), now());
insert into `t_menu` values (130402, 130400, 'edit cluster', '/setting/edit_cluster', 'setting/FlinkCluster/EditCluster', 'cluster:update', '', '0', 0, null, now(), now());
+insert into `t_menu` values (130403, 130400, 'delete cluster', null, null, 'cluster:delete', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130201, 130200, 'alert view', null, null, 'alert:view', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130202, 130200, 'alert add', null, null, 'alert:add', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130203, 130200, 'alert update', null, null, 'alert:update', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130204, 130200, 'alert delete', null, null, 'alert:delete', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130301, 130300, 'env create', null, null, 'env:create', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130302, 130300, 'env update', null, null, 'env:update', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130303, 130300, 'env delete', null, null, 'env:delete', null, '1', 1, null, now(), now());
insert into `t_menu` values (130501, 130500, 'link view', null, null, 'externalLink:view', null, '1', 1, null, now(), now());
insert into `t_menu` values (130502, 130500, 'link create', null, null, 'externalLink:create', null, '1', 1, null, now(), now());
insert into `t_menu` values (130503, 130500, 'link update', null, null, 'externalLink:update', null, '1', 1, null, now(), now());
@@ -220,6 +228,14 @@ insert into `t_role_menu` (role_id, menu_id) values (100002, 130300);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130400);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130401);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130402);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130403);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130201);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130202);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130203);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130204);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130301);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130302);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130303);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130500);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130501);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130502);
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/data/pgsql-data.sql b/streampark-console/streampark-console-service/src/main/assembly/script/data/pgsql-data.sql
index fb5e92d78..c191a68d2 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/data/pgsql-data.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/data/pgsql-data.sql
@@ -119,6 +119,14 @@ insert into "public"."t_menu" values (130101, 130100, 'view', null, null, 'setti
insert into "public"."t_menu" values (130102, 130100, 'setting update', null, null, 'setting:update', null, '1', '1', null, now(), now());
insert into "public"."t_menu" values (130401, 130400, 'add cluster', '/setting/add_cluster', 'setting/FlinkCluster/AddCluster', 'cluster:create', '', '0', '0', null, now(), now());
insert into "public"."t_menu" values (130402, 130400, 'edit cluster', '/setting/edit_cluster', 'setting/FlinkCluster/EditCluster', 'cluster:update', '', '0', '0', null, now(), now());
+insert into "public"."t_menu" values (130403, 130400, 'delete cluster', null, null, 'cluster:delete', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130201, 130200, 'alert view', null, null, 'alert:view', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130202, 130200, 'alert add', null, null, 'alert:add', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130203, 130200, 'alert update', null, null, 'alert:update', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130204, 130200, 'alert delete', null, null, 'alert:delete', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130301, 130300, 'env create', null, null, 'env:create', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130302, 130300, 'env update', null, null, 'env:update', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130303, 130300, 'env delete', null, null, 'env:delete', null, '1', '1', null, now(), now());
insert into "public"."t_menu" values (130501, 130500, 'link view', null, null, 'externalLink:view', null, '1', '1', null, now(), now());
insert into "public"."t_menu" values (130502, 130500, 'link create', null, null, 'externalLink:create', null, '1', '1', null, now(), now());
insert into "public"."t_menu" values (130503, 130500, 'link update', null, null, 'externalLink:update', null, '1', '1', null, now(), now());
@@ -214,6 +222,14 @@ insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130300);
insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130400);
insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130401);
insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130402);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130403);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130201);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130202);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130203);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130204);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130301);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130302);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130303);
insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130500);
insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130501);
insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130502);
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.8.sql b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.8.sql
index d4c853d95..351e182e4 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.8.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.8.sql
@@ -19,3 +19,21 @@ use streampark;
ALTER TABLE `t_user`
MODIFY COLUMN `password` varchar(255) collate utf8mb4_general_ci not null comment 'password';
+
+insert into `t_menu` values (130403, 130400, 'delete cluster', null, null, 'cluster:delete', null, '1', 1, null, now(), now());
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130403);
+
+insert into `t_menu` values (130201, 130200, 'alert view', null, null, 'alert:view', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130202, 130200, 'alert add', null, null, 'alert:add', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130203, 130200, 'alert update', null, null, 'alert:update', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130204, 130200, 'alert delete', null, null, 'alert:delete', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130301, 130300, 'env create', null, null, 'env:create', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130302, 130300, 'env update', null, null, 'env:update', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130303, 130300, 'env delete', null, null, 'env:delete', null, '1', 1, null, now(), now());
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130201);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130202);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130203);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130204);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130301);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130302);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130303);
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.8.sql b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.8.sql
index b3263a5db..f39efa01c 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.8.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.8.sql
@@ -17,3 +17,21 @@
ALTER TABLE "public"."t_user"
ALTER COLUMN "password" TYPE varchar(255) collate "pg_catalog"."default";
+
+insert into "public"."t_menu" values (130403, 130400, 'delete cluster', null, null, 'cluster:delete', null, '1', '1', null, now(), now());
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130403);
+
+insert into "public"."t_menu" values (130201, 130200, 'alert view', null, null, 'alert:view', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130202, 130200, 'alert add', null, null, 'alert:add', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130203, 130200, 'alert update', null, null, 'alert:update', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130204, 130200, 'alert delete', null, null, 'alert:delete', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130301, 130300, 'env create', null, null, 'env:create', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130302, 130300, 'env update', null, null, 'env:update', null, '1', '1', null, now(), now());
+insert into "public"."t_menu" values (130303, 130300, 'env delete', null, null, 'env:delete', null, '1', '1', null, now(), now());
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130201);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130202);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130203);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130204);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130301);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130302);
+insert into "public"."t_role_menu" (role_id, menu_id) values (100002, 130303);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/config/SpringProperties.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/config/SpringProperties.java
index 92151af31..2c4e5dff6 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/config/SpringProperties.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/config/SpringProperties.java
@@ -104,6 +104,16 @@ public class SpringProperties {
String userName = userConfig.getProperty("spring.datasource.username", "admin");
String password = userConfig.getProperty("spring.datasource.password", "streampark");
+ if ("admin".equals(userName) && "streampark".equals(password)) {
+ log.warn(
+ "[StreamPark] The H2 datasource is using the built-in default credentials "
+ + "(username=\"admin\"). These same credentials protect the H2 web console at "
+ + "/h2-console (now restricted to platform administrators). Please set "
+ + "\"datasource.username\" and \"datasource.password\" in conf/config.yaml to a "
+ + "strong, unique value before running this instance outside of local "
+ + "development.");
+ }
+
springConfig.put("spring.jpa.database-platform", "org.hibernate.dialect.H2Dialect");
springConfig.put("spring.datasource.driver-class-name", "org.h2.Driver");
springConfig.put("spring.datasource.username", userName);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/AlertController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/AlertController.java
index 1de322c69..b70f122e4 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/AlertController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/AlertController.java
@@ -27,6 +27,8 @@ import org.apache.streampark.console.core.entity.AlertConfig;
import org.apache.streampark.console.core.service.alert.AlertConfigService;
import org.apache.streampark.console.core.service.alert.AlertService;
+import org.apache.shiro.authz.annotation.RequiresPermissions;
+
import com.baomidou.mybatisplus.core.metadata.IPage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@@ -55,30 +57,35 @@ public class AlertController {
@Autowired private AlertService alertService;
@PostMapping(value = "/add")
+ @RequiresPermissions("alert:add")
public RestResponse createAlertConfig(@RequestBody AlertConfigWithParams params) {
- boolean save = alertConfigService.save(AlertConfig.of(params));
+ boolean save = alertConfigService.createAlertConfig(AlertConfig.of(params));
return RestResponse.success(save);
}
@PostMapping(value = "/exists")
+ @RequiresPermissions("alert:add")
public RestResponse verifyAlertConfig(@RequestBody AlertConfigWithParams params) {
boolean exist = alertConfigService.exist(AlertConfig.of(params));
return RestResponse.success(exist);
}
@PostMapping(value = "/update")
+ @RequiresPermissions("alert:update")
public RestResponse updateAlertConfig(@RequestBody AlertConfigWithParams params) {
- boolean update = alertConfigService.updateById(AlertConfig.of(params));
+ boolean update = alertConfigService.updateAlertConfig(AlertConfig.of(params));
return RestResponse.success(update);
}
@PostMapping("/get")
+ @RequiresPermissions("alert:view")
public RestResponse getAlertConfig(@RequestBody AlertConfigWithParams params) {
- AlertConfig alertConfig = alertConfigService.getById(params.getId());
+ AlertConfig alertConfig = alertConfigService.getAlertConfig(params.getId());
return RestResponse.success(AlertConfigWithParams.of(alertConfig));
}
@PostMapping(value = "/list")
+ @RequiresPermissions("alert:view")
public RestResponse alertConfigsPaginationList(
@RequestBody AlertConfigWithParams params, RestRequest request) {
IPage<AlertConfigWithParams> page = alertConfigService.page(params, request);
@@ -86,12 +93,14 @@ public class AlertController {
}
@PostMapping(value = "/listWithOutPage")
+ @RequiresPermissions("alert:view")
public RestResponse alertConfigsList() {
- List<AlertConfig> page = alertConfigService.list();
+ List<AlertConfig> page = alertConfigService.listByCurrentUser();
return RestResponse.success(page);
}
@DeleteMapping("/delete")
+ @RequiresPermissions("alert:delete")
public RestResponse deleteAlertConfig(
@RequestParam("id") @NotNull(message = "config id must not be null") Long id) {
boolean result = alertConfigService.deleteById(id);
@@ -99,6 +108,7 @@ public class AlertController {
}
@PostMapping("/send")
+ @RequiresPermissions("alert:view")
public RestResponse sendAlert(Long id) throws AlertException {
AlertTemplate alertTemplate = new AlertTemplate();
alertTemplate.setTitle("Notify: StreamPark alert job for test");
@@ -113,7 +123,8 @@ public class AlertController {
alertTemplate.setEndTime(DateUtils.format(date, DateUtils.fullFormat(), TimeZone.getDefault()));
alertTemplate.setDuration("-");
boolean alert =
- alertService.alert(AlertConfigWithParams.of(alertConfigService.getById(id)), alertTemplate);
+ alertService.alert(
+ AlertConfigWithParams.of(alertConfigService.getAlertConfig(id)), alertTemplate);
return RestResponse.success(alert);
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkClusterController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkClusterController.java
index dc522ec10..4dd6c4749 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkClusterController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkClusterController.java
@@ -66,6 +66,7 @@ public class FlinkClusterController {
}
@PostMapping("check")
+ @RequiresPermissions("cluster:create")
public RestResponse check(FlinkCluster cluster) {
ResponseResult checkResult = flinkClusterService.check(cluster);
return RestResponse.success(checkResult);
@@ -92,18 +93,21 @@ public class FlinkClusterController {
}
@PostMapping("start")
+ @RequiresPermissions("cluster:update")
public RestResponse start(Long id) {
flinkClusterService.start(id);
return RestResponse.success();
}
@PostMapping("shutdown")
+ @RequiresPermissions("cluster:update")
public RestResponse shutdown(Long id) {
flinkClusterService.shutdown(id);
return RestResponse.success();
}
@PostMapping("delete")
+ @RequiresPermissions("cluster:delete")
public RestResponse delete(Long id) {
flinkClusterService.delete(id);
return RestResponse.success();
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkEnvController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkEnvController.java
index d7eda1757..4c2733e5b 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkEnvController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkEnvController.java
@@ -23,6 +23,8 @@ import org.apache.streampark.console.base.exception.ApiDetailException;
import org.apache.streampark.console.core.entity.FlinkEnv;
import org.apache.streampark.console.core.service.FlinkEnvService;
+import org.apache.shiro.authz.annotation.RequiresPermissions;
+
import com.baomidou.mybatisplus.core.metadata.IPage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@@ -54,12 +56,14 @@ public class FlinkEnvController {
}
@PostMapping("check")
+ @RequiresPermissions("env:create")
public RestResponse check(FlinkEnv version) {
Integer checkResp = flinkEnvService.check(version);
return RestResponse.success(checkResp);
}
@PostMapping("create")
+ @RequiresPermissions("env:create")
public RestResponse create(FlinkEnv version) {
try {
flinkEnvService.create(version);
@@ -77,12 +81,14 @@ public class FlinkEnvController {
}
@PostMapping("sync")
+ @RequiresPermissions("env:update")
public RestResponse sync(Long id) throws Exception {
flinkEnvService.syncConf(id);
return RestResponse.success();
}
@PostMapping("update")
+ @RequiresPermissions("env:update")
public RestResponse update(FlinkEnv version) throws Exception {
try {
flinkEnvService.update(version);
@@ -93,18 +99,21 @@ public class FlinkEnvController {
}
@PostMapping("delete")
+ @RequiresPermissions("env:delete")
public RestResponse delete(Long id) {
flinkEnvService.delete(id);
return RestResponse.success();
}
@PostMapping("validity")
+ @RequiresPermissions("env:update")
public RestResponse validity(FlinkEnv version) {
flinkEnvService.validity(version.getId());
return RestResponse.success(true);
}
@PostMapping("default")
+ @RequiresPermissions("env:update")
public RestResponse setDefault(Long id) {
flinkEnvService.setDefault(id);
return RestResponse.success();
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProjectController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProjectController.java
index 33a3b5828..22124aac4 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProjectController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProjectController.java
@@ -84,11 +84,8 @@ public class ProjectController {
@PostMapping("buildlog")
@RequiresPermissions("project:build")
- @PermissionScope(team = "#teamId")
public RestResponse buildLog(
- Long id,
- @RequestParam(value = "startOffset", required = false) Long startOffset,
- Long teamId) {
+ Long id, @RequestParam(value = "startOffset", required = false) Long startOffset) {
return projectService.getBuildLog(id, startOffset);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProxyController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProxyController.java
index f36155604..d6cb20caf 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProxyController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProxyController.java
@@ -41,6 +41,7 @@ public class ProxyController {
@Autowired private ProxyService proxyService;
@GetMapping("{type}/{id}/assets/**")
+ @RequiresPermissions("app:view")
public ResponseEntity<?> proxyFlinkAssets(
HttpServletRequest request, @PathVariable("type") String type, @PathVariable("id") Long id)
throws Exception {
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/VariableController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/VariableController.java
index 9017243ff..26edbddb2 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/VariableController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/VariableController.java
@@ -106,7 +106,7 @@ public class VariableController {
@PostMapping("showOriginal")
@RequiresPermissions("variable:show_original")
public RestResponse showOriginal(@RequestParam Long id) {
- Variable v = this.variableService.getById(id);
+ Variable v = this.variableService.getOriginal(id);
return RestResponse.success(v);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/VariableService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/VariableService.java
index c255b8254..1a91c74d8 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/VariableService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/VariableService.java
@@ -72,4 +72,13 @@ public interface VariableService extends IService<Variable> {
IPage<Application> dependAppsPage(Variable variable, RestRequest request);
void updateVariable(Variable variable);
+
+ /**
+ * Get a variable with its original (unmasked) value, after verifying the current user belongs to
+ * the team that actually owns it.
+ *
+ * @param id variable id
+ * @return the variable, or {@code null} if it doesn't exist
+ */
+ Variable getOriginal(Long id);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/AlertConfigService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/AlertConfigService.java
index 582d78fda..2f052cd95 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/AlertConfigService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/AlertConfigService.java
@@ -25,10 +25,47 @@ import org.apache.streampark.console.core.entity.AlertConfig;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.IService;
+import java.util.List;
+
public interface AlertConfigService extends IService<AlertConfig> {
IPage<AlertConfigWithParams> page(AlertConfigWithParams params, RestRequest request);
boolean exist(AlertConfig alertConfig);
boolean deleteById(Long id) throws AlertException;
+
+ /**
+ * Create a new alert config. Ownership (userId) is always forced to the currently logged-in user,
+ * regardless of what the client submitted in the request body.
+ *
+ * @param alertConfig the alert config to create.
+ * @return {@code true} if saved successfully.
+ */
+ boolean createAlertConfig(AlertConfig alertConfig);
+
+ /**
+ * Update an existing alert config, after verifying the current user is the owner (creator) of the
+ * target record, or the platform administrator.
+ *
+ * @param alertConfig the alert config to update.
+ * @return {@code true} if updated successfully.
+ */
+ boolean updateAlertConfig(AlertConfig alertConfig);
+
+ /**
+ * Get a single alert config, after verifying the current user is the owner (creator) of the
+ * target record, or the platform administrator.
+ *
+ * @param id alert config id.
+ * @return the alert config, or {@code null} if it doesn't exist.
+ */
+ AlertConfig getAlertConfig(Long id);
+
+ /**
+ * List every alert config owned by the currently logged-in user, or every alert config in the
+ * system if the current user is the platform administrator.
+ *
+ * @return the list of alert configs.
+ */
+ List<AlertConfig> listByCurrentUser();
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/AlertConfigServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/AlertConfigServiceImpl.java
index 218f45dfd..899360275 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/AlertConfigServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/AlertConfigServiceImpl.java
@@ -19,13 +19,18 @@ package org.apache.streampark.console.core.service.alert.impl;
import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.base.exception.AlertException;
+import org.apache.streampark.console.base.exception.ApiAlertException;
+import org.apache.streampark.console.base.exception.PermissionDeniedException;
import org.apache.streampark.console.base.mybatis.pager.MybatisPager;
import org.apache.streampark.console.core.bean.AlertConfigWithParams;
import org.apache.streampark.console.core.entity.AlertConfig;
import org.apache.streampark.console.core.entity.Application;
+import org.apache.streampark.console.core.enums.UserType;
import org.apache.streampark.console.core.mapper.AlertConfigMapper;
import org.apache.streampark.console.core.service.ApplicationService;
+import org.apache.streampark.console.core.service.ServiceHelper;
import org.apache.streampark.console.core.service.alert.AlertConfigService;
+import org.apache.streampark.console.system.entity.User;
import org.apache.commons.collections.CollectionUtils;
@@ -39,6 +44,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
+import java.util.List;
import java.util.stream.Collectors;
@Service
@@ -49,11 +55,50 @@ public class AlertConfigServiceImpl extends ServiceImpl<AlertConfigMapper, Alert
@Autowired private ApplicationService applicationService;
+ @Autowired private ServiceHelper serviceHelper;
+
+ /**
+ * Verify that the currently logged-in user is either the platform administrator or the same user
+ * as {@code targetUserId}. AlertConfig records are bound to an individual user (t_alert_config
+ * has no team_id column), so this is the closest equivalent of the team-ownership checks used
+ * elsewhere in the codebase for team-scoped entities.
+ *
+ * @param targetUserId the real, persisted userId of the record being accessed (or the userId a
+ * query is being explicitly scoped to). {@code null} means "no explicit scope requested".
+ * @return the currently logged-in user, for convenience.
+ */
+ private User checkOwnership(Long targetUserId) {
+ User currentUser = serviceHelper.getLoginUser();
+ ApiAlertException.throwIfNull(currentUser, "Permission denied, please login first.");
+ if (currentUser.getUserType() == UserType.ADMIN) {
+ return currentUser;
+ }
+ if (targetUserId != null && !currentUser.getUserId().equals(targetUserId)) {
+ throw new PermissionDeniedException(
+ "Permission denied, this alert config was not created by the current user.");
+ }
+ return currentUser;
+ }
+
@Override
public IPage<AlertConfigWithParams> page(AlertConfigWithParams params, RestRequest request) {
+ // Never allow an unscoped query to leak every user's alert configs: if the caller didn't
+ // specify a userId, default to the current user's own id (unless the current user is the
+ // platform administrator). If the caller DID specify a userId, verify it's their own (or
+ // they're the administrator) before honoring it.
+ Long userId = params.getUserId();
+ if (userId == null) {
+ User currentUser = checkOwnership(null);
+ if (currentUser.getUserType() != UserType.ADMIN) {
+ userId = currentUser.getUserId();
+ }
+ } else {
+ checkOwnership(userId);
+ }
+
// build query conditions
LambdaQueryWrapper<AlertConfig> wrapper = new LambdaQueryWrapper<>();
- wrapper.eq(params.getUserId() != null, AlertConfig::getUserId, params.getUserId());
+ wrapper.eq(userId != null, AlertConfig::getUserId, userId);
Page<AlertConfig> page = MybatisPager.getPage(request);
IPage<AlertConfig> resultPage = getBaseMapper().selectPage(page, wrapper);
@@ -77,6 +122,9 @@ public class AlertConfigServiceImpl extends ServiceImpl<AlertConfigMapper, Alert
@Override
public boolean deleteById(Long id) throws AlertException {
+ AlertConfig persisted = getById(id);
+ ApiAlertException.throwIfTrue(persisted == null, "Sorry, the alert config does not exist.");
+ checkOwnership(persisted.getUserId());
long count =
applicationService.count(
new LambdaQueryWrapper<Application>().eq(id != null, Application::getAlertId, id));
@@ -88,4 +136,49 @@ public class AlertConfigServiceImpl extends ServiceImpl<AlertConfigMapper, Alert
}
return removeById(id);
}
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public boolean createAlertConfig(AlertConfig alertConfig) {
+ User currentUser = serviceHelper.getLoginUser();
+ ApiAlertException.throwIfNull(currentUser, "Permission denied, please login first.");
+ // Always attribute the new alert config to the current user, ignoring whatever userId the
+ // client submitted, to prevent creating alert configs under someone else's identity.
+ alertConfig.setUserId(currentUser.getUserId());
+ return save(alertConfig);
+ }
+
+ @Override
+ @Transactional(rollbackFor = Exception.class)
+ public boolean updateAlertConfig(AlertConfig alertConfig) {
+ ApiAlertException.throwIfTrue(
+ alertConfig.getId() == null, "Sorry, the alert config id cannot be null.");
+ AlertConfig persisted = getById(alertConfig.getId());
+ ApiAlertException.throwIfTrue(persisted == null, "Sorry, the alert config does not exist.");
+ checkOwnership(persisted.getUserId());
+ // Never allow the client to reassign ownership of an existing alert config.
+ alertConfig.setUserId(persisted.getUserId());
+ return updateById(alertConfig);
+ }
+
+ @Override
+ public AlertConfig getAlertConfig(Long id) {
+ AlertConfig alertConfig = getById(id);
+ if (alertConfig == null) {
+ return null;
+ }
+ checkOwnership(alertConfig.getUserId());
+ return alertConfig;
+ }
+
+ @Override
+ public List<AlertConfig> listByCurrentUser() {
+ User currentUser = serviceHelper.getLoginUser();
+ ApiAlertException.throwIfNull(currentUser, "Permission denied, please login first.");
+ if (currentUser.getUserType() == UserType.ADMIN) {
+ return list();
+ }
+ return list(
+ new LambdaQueryWrapper<AlertConfig>().eq(AlertConfig::getUserId, currentUser.getUserId()));
+ }
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java
index 303966958..7b6bc1021 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java
@@ -28,6 +28,7 @@ import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.base.domain.RestResponse;
import org.apache.streampark.console.base.exception.ApiAlertException;
import org.apache.streampark.console.base.exception.ApiDetailException;
+import org.apache.streampark.console.base.exception.PermissionDeniedException;
import org.apache.streampark.console.base.mybatis.pager.MybatisPager;
import org.apache.streampark.console.base.util.CommonUtils;
import org.apache.streampark.console.base.util.FileUtils;
@@ -40,11 +41,16 @@ import org.apache.streampark.console.core.entity.Project;
import org.apache.streampark.console.core.enums.BuildState;
import org.apache.streampark.console.core.enums.GitAuthorizedError;
import org.apache.streampark.console.core.enums.ReleaseState;
+import org.apache.streampark.console.core.enums.UserType;
import org.apache.streampark.console.core.mapper.ProjectMapper;
import org.apache.streampark.console.core.service.ApplicationService;
import org.apache.streampark.console.core.service.ProjectService;
+import org.apache.streampark.console.core.service.ServiceHelper;
import org.apache.streampark.console.core.task.FlinkAppHttpWatcher;
import org.apache.streampark.console.core.task.ProjectBuildTask;
+import org.apache.streampark.console.system.entity.Member;
+import org.apache.streampark.console.system.entity.User;
+import org.apache.streampark.console.system.service.MemberService;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.MemorySize;
@@ -89,6 +95,10 @@ public class ProjectServiceImpl extends ServiceImpl<ProjectMapper, Project>
@Autowired private FlinkAppHttpWatcher flinkAppHttpWatcher;
+ @Autowired private ServiceHelper serviceHelper;
+
+ @Autowired private MemberService memberService;
+
@Value("${streampark.project.max-build:6}")
public Long maxProjectBuildNum;
@@ -446,6 +456,10 @@ public class ProjectServiceImpl extends ServiceImpl<ProjectMapper, Project>
@Override
public RestResponse getBuildLog(Long id, Long startOffset) {
+ Project project = getById(id);
+ ApiAlertException.throwIfNull(project, "Project not found, id: " + id);
+ checkProjectTeamAccess(project);
+
File logFile = Paths.get(getBuildLogPath(id)).toFile();
if (!logFile.exists()) {
String errorMsg =
@@ -453,7 +467,7 @@ public class ProjectServiceImpl extends ServiceImpl<ProjectMapper, Project>
log.warn(errorMsg);
return RestResponse.success().data(errorMsg);
}
- boolean isBuilding = this.getById(id).getBuildState() == 0;
+ boolean isBuilding = project.getBuildState() == 0;
byte[] fileContent;
long endOffset = 0L;
boolean readFinished = true;
@@ -486,4 +500,26 @@ public class ProjectServiceImpl extends ServiceImpl<ProjectMapper, Project>
private String getBuildLogPath(Long projectId) {
return String.format("%s/%s/build.log", Workspace.PROJECT_BUILD_LOG_PATH(), projectId);
}
+
+ /**
+ * Verify that the currently logged-in user is allowed to access build artifacts (e.g. the build
+ * log) of the given project, by checking membership against the project's REAL, persisted teamId
+ * - never a client-supplied one, which is how this method previously could be bypassed (see
+ * buildLog's now-removed {@code @PermissionScope(team = "#teamId")}, which only validated the
+ * caller-supplied teamId and never cross-checked it against the actual project).
+ *
+ * @param project the project fetched from the database.
+ */
+ private void checkProjectTeamAccess(Project project) {
+ User currentUser = serviceHelper.getLoginUser();
+ ApiAlertException.throwIfNull(currentUser, "Permission denied, please login first.");
+ if (currentUser.getUserType() == UserType.ADMIN) {
+ return;
+ }
+ Member member = memberService.findByUserId(project.getTeamId(), currentUser.getUserId());
+ if (member == null) {
+ throw new PermissionDeniedException(
+ "Permission denied, only members of this project's team can access this resource.");
+ }
+ }
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProxyServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProxyServiceImpl.java
index 24aba76ef..2c1740465 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProxyServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProxyServiceImpl.java
@@ -33,6 +33,7 @@ import org.apache.streampark.console.core.service.ServiceHelper;
import org.apache.streampark.console.core.task.FlinkK8sWatcherWrapper;
import org.apache.streampark.console.system.authentication.JWTUtil;
import org.apache.streampark.console.system.entity.Member;
+import org.apache.streampark.console.system.entity.Team;
import org.apache.streampark.console.system.entity.User;
import org.apache.streampark.console.system.service.MemberService;
import org.apache.streampark.console.system.service.UserService;
@@ -70,6 +71,7 @@ import java.io.InputStream;
import java.net.URI;
import java.security.PrivilegedExceptionAction;
import java.util.Enumeration;
+import java.util.List;
@Slf4j
@Service
@@ -183,6 +185,7 @@ public class ProxyServiceImpl implements ProxyService {
if (cluster == null) {
return builder.body("The cluster not found.");
}
+ checkProxyCluster(cluster);
String url = cluster.getAddress();
if (StringUtils.isBlank(url)) {
return builder.body("The cluster address is invalid.");
@@ -219,6 +222,42 @@ public class ProxyServiceImpl implements ProxyService {
checkProxyApp(app);
}
+ /**
+ * Verify the current user is allowed to access the proxied Web UI / REST API of the given Flink
+ * cluster. {@code t_flink_cluster} has no {@code team_id} column, so unlike {@link
+ * #checkProxyApp(Application)} this cannot compare against a real team id directly. Ownership is
+ * therefore derived indirectly: the platform administrator, the cluster's creator, or any user
+ * who belongs to a team that has at least one application bound to this cluster (via {@code
+ * Application#flinkClusterId}) is allowed through. Everyone else is denied.
+ *
+ * @param cluster the cluster being proxied, fetched from the database.
+ */
+ public void checkProxyCluster(FlinkCluster cluster) {
+ String token = serviceHelper.getAuthorization();
+ Long userId = token == null ? null : JWTUtil.getUserId(token);
+ if (userId == null) {
+ throw new PermissionDeniedException("Permission denied, please login first.");
+ }
+ if (userId.equals(cluster.getUserId())) {
+ return;
+ }
+ User user = userService.getById(userId);
+ if (user != null && user.getUserType() == UserType.ADMIN) {
+ return;
+ }
+ List<Team> userTeams = memberService.findUserTeams(userId);
+ boolean accessible =
+ userTeams.stream()
+ .anyMatch(
+ team ->
+ applicationService.getByTeamId(team.getId()).stream()
+ .anyMatch(app -> cluster.getId().equals(app.getFlinkClusterId())));
+ if (!accessible) {
+ throw new PermissionDeniedException(
+ "Permission denied, this cluster is not accessible from any team the current user belongs to.");
+ }
+ }
+
private HttpEntity<?> getRequestEntity(HttpServletRequest request, String url) throws Exception {
HttpHeaders headers = new HttpHeaders();
Enumeration<String> headerNames = request.getHeaderNames();
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/VariableServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/VariableServiceImpl.java
index 62fcc7655..e1dc13ad9 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/VariableServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/VariableServiceImpl.java
@@ -20,16 +20,21 @@ package org.apache.streampark.console.core.service.impl;
import org.apache.streampark.common.util.DeflaterUtils;
import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.base.exception.ApiAlertException;
+import org.apache.streampark.console.base.exception.PermissionDeniedException;
import org.apache.streampark.console.base.mybatis.pager.MybatisPager;
import org.apache.streampark.console.core.entity.Application;
import org.apache.streampark.console.core.entity.FlinkSql;
import org.apache.streampark.console.core.entity.Variable;
import org.apache.streampark.console.core.enums.ReleaseState;
+import org.apache.streampark.console.core.enums.UserType;
import org.apache.streampark.console.core.mapper.VariableMapper;
import org.apache.streampark.console.core.service.ApplicationService;
import org.apache.streampark.console.core.service.FlinkSqlService;
import org.apache.streampark.console.core.service.ServiceHelper;
import org.apache.streampark.console.core.service.VariableService;
+import org.apache.streampark.console.system.entity.Member;
+import org.apache.streampark.console.system.entity.User;
+import org.apache.streampark.console.system.service.MemberService;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
@@ -72,6 +77,8 @@ public class VariableServiceImpl extends ServiceImpl<VariableMapper, Variable>
@Autowired private ServiceHelper serviceHelper;
+ @Autowired private MemberService memberService;
+
@Override
@Transactional(rollbackFor = Exception.class)
public void createVariable(Variable variable) {
@@ -89,10 +96,15 @@ public class VariableServiceImpl extends ServiceImpl<VariableMapper, Variable>
@Override
@Transactional(rollbackFor = Exception.class)
public void deleteVariable(Variable variable) {
- if (isDependByApplications(variable)) {
+ ApiAlertException.throwIfTrue(
+ variable.getId() == null, "Sorry, the variable id cannot be null.");
+ Variable persisted = this.baseMapper.selectById(variable.getId());
+ ApiAlertException.throwIfTrue(persisted == null, "Sorry, the variable does not exist.");
+ checkTeamAccess(persisted.getTeamId());
+ if (isDependByApplications(persisted)) {
throw new ApiAlertException("Sorry, the variable is actually used.");
}
- this.removeById(variable);
+ this.removeById(persisted.getId());
}
@Override
@@ -132,9 +144,14 @@ public class VariableServiceImpl extends ServiceImpl<VariableMapper, Variable>
if (findVariable == null) {
throw new ApiAlertException("Sorry, the variable does not exist.");
}
+ // Verify against the REAL, persisted teamId - never trust variable.getTeamId() from the
+ // client request, which can be spoofed to any team the attacker happens to belong to.
+ checkTeamAccess(findVariable.getTeamId());
if (!findVariable.getVariableCode().equals(variable.getVariableCode())) {
throw new ApiAlertException("Sorry, the variable code cannot be updated.");
}
+ // A variable can never be moved to a different team via update.
+ variable.setTeamId(findVariable.getTeamId());
variable.setModifyTime(new Date());
this.baseMapper.updateById(variable);
@@ -217,6 +234,40 @@ public class VariableServiceImpl extends ServiceImpl<VariableMapper, Variable>
return restore;
}
+ /**
+ * Verify that the currently logged-in user is allowed to operate on a variable that really
+ * belongs to {@code teamId} (fetched from the database, never trusted from client input). Mirrors
+ * the same ADMIN-bypass / team-membership check performed by {@code PermissionAspect} for the
+ * {@code #project.teamId} style scopes, but is applied explicitly here because {@code
+ * showOriginal}/{@code updateVariable}/{@code deleteVariable} only carry the variable id (or a
+ * client-supplied, therefore untrustworthy, teamId) rather than a nested object SpEL can resolve
+ * against.
+ *
+ * @param teamId the real teamId of the variable being accessed, as read from the database.
+ */
+ private void checkTeamAccess(Long teamId) {
+ User currentUser = serviceHelper.getLoginUser();
+ ApiAlertException.throwIfNull(currentUser, "Permission denied, please login first.");
+ if (currentUser.getUserType() == UserType.ADMIN) {
+ return;
+ }
+ Member member = memberService.findByUserId(teamId, currentUser.getUserId());
+ if (member == null) {
+ throw new PermissionDeniedException(
+ "Permission denied, only members of this team can access this variable.");
+ }
+ }
+
+ @Override
+ public Variable getOriginal(Long id) {
+ Variable variable = this.baseMapper.selectById(id);
+ if (variable == null) {
+ return null;
+ }
+ checkTeamAccess(variable.getTeamId());
+ return variable;
+ }
+
private boolean isDependByApplications(Variable variable) {
return CollectionUtils.isNotEmpty(getDependApplicationsByCode(variable));
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/AdminOnlyFilter.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/AdminOnlyFilter.java
new file mode 100644
index 000000000..0aacbd40e
--- /dev/null
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/AdminOnlyFilter.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.console.system.authentication;
+
+import org.apache.streampark.console.core.enums.UserType;
+import org.apache.streampark.console.system.entity.User;
+import org.apache.streampark.console.system.service.UserService;
+
+import org.apache.shiro.SecurityUtils;
+import org.apache.shiro.authz.UnauthorizedException;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletRequest;
+
+/**
+ * A JWT filter variant that, on top of requiring a valid platform JWT (inherited from {@link
+ * JWTFilter}), additionally requires the authenticated user to be a platform administrator ({@link
+ * UserType#ADMIN}).
+ *
+ * <p>This exists to guard highly sensitive, non-REST endpoints (such as the embedded H2 database
+ * console) that cannot be protected with the usual method-level {@code @RequiresPermissions}
+ * annotation because they aren't Controller methods.
+ */
+@Slf4j
+@Component
+public class AdminOnlyFilter extends JWTFilter {
+
+ @Autowired private UserService userService;
+
+ @Override
+ protected boolean isAccessAllowed(
+ ServletRequest request, ServletResponse response, Object mappedValue)
+ throws UnauthorizedException {
+ boolean authenticated = super.isAccessAllowed(request, response, mappedValue);
+ if (!authenticated) {
+ return false;
+ }
+ try {
+ Long userId = JWTUtil.getUserId((String) SecurityUtils.getSubject().getPrincipal());
+ User user = userId == null ? null : userService.getById(userId);
+ if (user == null || user.getUserType() != UserType.ADMIN) {
+ log.warn(
+ "Denied non-admin access to an admin-only resource, userId={}, uri={}",
+ userId,
+ ((HttpServletRequest) request).getRequestURI());
+ return false;
+ }
+ return true;
+ } catch (Exception e) {
+ log.warn("Failed to verify admin-only access.", e);
+ return false;
+ }
+ }
+}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/ShiroConfig.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/ShiroConfig.java
index 3ccf84a9e..42823a571 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/ShiroConfig.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/ShiroConfig.java
@@ -33,17 +33,23 @@ import java.util.LinkedHashMap;
public class ShiroConfig {
@Bean
- public ShiroFilterFactoryBean shiroFilterFactoryBean(SecurityManager securityManager) {
+ public ShiroFilterFactoryBean shiroFilterFactoryBean(
+ SecurityManager securityManager, AdminOnlyFilter adminOnlyFilter) {
ShiroFilterFactoryBean shiroFilterFactoryBean = new ShiroFilterFactoryBean();
shiroFilterFactoryBean.setSecurityManager(securityManager);
LinkedHashMap<String, Filter> filters = new LinkedHashMap<>();
filters.put("jwt", new JWTFilter());
+ filters.put("adminJwt", adminOnlyFilter);
shiroFilterFactoryBean.setFilters(filters);
LinkedHashMap<String, String> filterChainDefinitionMap = new LinkedHashMap<>();
filterChainDefinitionMap.put("/actuator/**", "anon");
- filterChainDefinitionMap.put("/h2-console/**", "anon");
+ // The H2 web console gives raw SQL access to the entire application database. It must never
+ // be anonymously reachable. This is defense-in-depth on top of
+ // `spring.h2.console.settings.web-allow-others=false`, which already restricts the console to
+ // same-origin connections: only an authenticated platform administrator may open it.
+ filterChainDefinitionMap.put("/h2-console/**", "adminJwt");
filterChainDefinitionMap.put("/passport/**", "anon");
filterChainDefinitionMap.put("/systemName", "anon");
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/TeamController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/TeamController.java
index ea45e1a32..131c6367e 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/TeamController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/TeamController.java
@@ -46,6 +46,7 @@ public class TeamController {
@Autowired private TeamService teamService;
@PostMapping("list")
+ @RequiresPermissions("team:view")
public RestResponse teamList(RestRequest restRequest, Team team) {
IPage<Team> teamList = teamService.page(team, restRequest);
return RestResponse.success(teamList);
diff --git a/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql b/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql
index 480b4e959..7765f128f 100644
--- a/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql
+++ b/streampark-console/streampark-console-service/src/main/resources/db/data-h2.sql
@@ -111,6 +111,14 @@ insert into `t_menu` values (130101, 130100, 'view', null, null, 'setting:view',
insert into `t_menu` values (130102, 130100, 'setting update', null, null, 'setting:update', null, '1', 1, null, now(), now());
insert into `t_menu` values (130401, 130400, 'add cluster', '/setting/add_cluster', 'setting/FlinkCluster/AddCluster', 'cluster:create', '', '0', 0, null, now(), now());
insert into `t_menu` values (130402, 130400, 'edit cluster', '/setting/edit_cluster', 'setting/FlinkCluster/EditCluster', 'cluster:update', '', '0', 0, null, now(), now());
+insert into `t_menu` values (130403, 130400, 'delete cluster', null, null, 'cluster:delete', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130201, 130200, 'alert view', null, null, 'alert:view', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130202, 130200, 'alert add', null, null, 'alert:add', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130203, 130200, 'alert update', null, null, 'alert:update', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130204, 130200, 'alert delete', null, null, 'alert:delete', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130301, 130300, 'env create', null, null, 'env:create', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130302, 130300, 'env update', null, null, 'env:update', null, '1', 1, null, now(), now());
+insert into `t_menu` values (130303, 130300, 'env delete', null, null, 'env:delete', null, '1', 1, null, now(), now());
insert into `t_menu` values (130501, 130500, 'link view', null, null, 'externalLink:view', null, '1', 1, null, now(), now());
insert into `t_menu` values (130502, 130500, 'link create', null, null, 'externalLink:create', null, '1', 1, null, now(), now());
insert into `t_menu` values (130503, 130500, 'link update', null, null, 'externalLink:update', null, '1', 1, null, now(), now());
@@ -206,6 +214,14 @@ insert into `t_role_menu` (role_id, menu_id) values (100002, 130300);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130400);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130401);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130402);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130403);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130201);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130202);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130203);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130204);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130301);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130302);
+insert into `t_role_menu` (role_id, menu_id) values (100002, 130303);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130500);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130501);
insert into `t_role_menu` (role_id, menu_id) values (100002, 130502);