jiazhizhong/kafka-console-ui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

接口权限过滤.

Browse Source
This commit is contained in:
许晓东
2023-05-18 22:56:00 +08:00
parent 7e98a58f60
commit 571efe6ddc
30 changed files with 463 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
src/main/java/com/xuxd/kafka/console/aspect/PermissionAspect.java Normal file
View File

@@ -0,0 +1,127 @@
package com.xuxd.kafka.console.aspect;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.Credentials;
import com.xuxd.kafka.console.beans.dos.SysUserDO;
import com.xuxd.kafka.console.cache.RolePermCache;
import com.xuxd.kafka.console.config.AuthConfig;
import com.xuxd.kafka.console.dao.SysPermissionMapper;
import com.xuxd.kafka.console.dao.SysRoleMapper;
import com.xuxd.kafka.console.dao.SysUserMapper;
import com.xuxd.kafka.console.exception.UnAuthorizedException;
import com.xuxd.kafka.console.filter.CredentialsContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.aspectj.lang.JoinPoint;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Before;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import java.lang.reflect.Method;
import java.util.*;
import java.util.stream.Collectors;
/**
* @author: xuxd
* @date: 2023/5/17 22:32
**/
@Slf4j
@Order(1)
@Aspect
@Component
public class PermissionAspect {
private Map<String, Set<String>> permMap = new HashMap<>();
private final AuthConfig authConfig;
private final SysUserMapper userMapper;
private final SysRoleMapper roleMapper;
private final SysPermissionMapper permissionMapper;
private final RolePermCache rolePermCache;
public PermissionAspect(AuthConfig authConfig,
SysUserMapper userMapper,
SysRoleMapper roleMapper,
SysPermissionMapper permissionMapper,
RolePermCache rolePermCache) {
this.authConfig = authConfig;
this.userMapper = userMapper;
this.roleMapper = roleMapper;
this.permissionMapper = permissionMapper;
this.rolePermCache = rolePermCache;
}
@Pointcut("@annotation(com.xuxd.kafka.console.aspect.annotation.Permission)")
private void pointcut() {
}
@Before(value = "pointcut()")
public void before(JoinPoint joinPoint) {
if (!authConfig.isEnable()) {
return;
}
MethodSignature signature = (MethodSignature) joinPoint.getSignature();
Method method = signature.getMethod();
Permission permission = method.getAnnotation(Permission.class);
if (permission == null) {
return;
}
String[] value = permission.value();
if (value == null || value.length == 0) {
return;
}
String name = method.getName() + "@" + method.hashCode();
Map<String, Set<String>> pm = checkPermMap(name, value);
Set<String> allowPermSet = pm.get(name);
if (allowPermSet == null) {
log.error("解析权限出现意外啦!!!");
return;
}
Credentials credentials = CredentialsContext.get();
if (credentials == null || credentials.isInvalid()) {
throw new UnAuthorizedException("credentials is invalid");
}
QueryWrapper<SysUserDO> queryWrapper = new QueryWrapper<>();
queryWrapper.eq("username", credentials.getUsername());
SysUserDO userDO = userMapper.selectOne(queryWrapper);
if (userDO == null) {
throw new UnAuthorizedException(credentials.getUsername() + ":" + allowPermSet);
}
String roleIds = userDO.getRoleIds();
List<Long> roleIdList = Arrays.stream(roleIds.split(",")).map(String::trim).filter(StringUtils::isNotEmpty).map(Long::valueOf).collect(Collectors.toList());
for (Long roleId : roleIdList) {
Set<String> permSet = rolePermCache.getRolePermCache().getOrDefault(roleId, Collections.emptySet());
for (String p : allowPermSet) {
if (permSet.contains(p)) {
return;
}
}
}
throw new UnAuthorizedException(credentials.getUsername() + ":" + allowPermSet);
}
private Map<String, Set<String>> checkPermMap(String methodName, String[] value) {
if (!permMap.containsKey(methodName)) {
Map<String, Set<String>> map = new HashMap<>(permMap);
map.put(methodName, new HashSet<>(Arrays.asList(value)));
permMap = map;
return map;
}
return permMap;
}
}

17
src/main/java/com/xuxd/kafka/console/aspect/annotation/Permission.java Normal file
View File

@@ -0,0 +1,17 @@
package com.xuxd.kafka.console.aspect.annotation;
import java.lang.annotation.*;
/**
* 权限注解，开启认证的时候拥有该权限的用户才能访问对应接口.
*
* @author: xuxd
* @date: 2023/5/17 22:30
**/
@Inherited
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface Permission {
String[] value() default {};
}

22
src/main/java/com/xuxd/kafka/console/beans/RolePermUpdateEvent.java Normal file
View File

@@ -0,0 +1,22 @@
package com.xuxd.kafka.console.beans;
import lombok.Getter;
import lombok.Setter;
import lombok.ToString;
import org.springframework.context.ApplicationEvent;
/**
* @author: xuxd
* @date: 2023/5/18 15:49
**/
@ToString
public class RolePermUpdateEvent extends ApplicationEvent {
@Getter
@Setter
private boolean reload = false;
public RolePermUpdateEvent(Object source) {
super(source);
}
}

91
src/main/java/com/xuxd/kafka/console/cache/RolePermCache.java vendored Normal file
View File

@@ -0,0 +1,91 @@
package com.xuxd.kafka.console.cache;
import com.xuxd.kafka.console.beans.RolePermUpdateEvent;
import com.xuxd.kafka.console.beans.dos.SysPermissionDO;
import com.xuxd.kafka.console.beans.dos.SysRoleDO;
import com.xuxd.kafka.console.dao.SysPermissionMapper;
import com.xuxd.kafka.console.dao.SysRoleMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
/**
* @author: xuxd
* @date: 2023/5/18 15:47
**/
@DependsOn("dataInit")
@Slf4j
@Component
public class RolePermCache implements ApplicationListener<RolePermUpdateEvent>, SmartInitializingSingleton {
private Map<Long, SysPermissionDO> permCache = new HashMap<>();
private Map<Long, Set<String>> rolePermCache = new HashMap<>();
private final SysPermissionMapper permissionMapper;
private final SysRoleMapper roleMapper;
public RolePermCache(SysPermissionMapper permissionMapper, SysRoleMapper roleMapper) {
this.permissionMapper = permissionMapper;
this.roleMapper = roleMapper;
}
@Override
public void onApplicationEvent(RolePermUpdateEvent event) {
log.info("更新角色权限信息：{}", event);
if (event.isReload()) {
this.loadPermCache();
}
refresh();
}
public Map<Long, SysPermissionDO> getPermCache() {
return permCache;
}
public Map<Long, Set<String>> getRolePermCache() {
return rolePermCache;
}
private void refresh() {
List<SysRoleDO> roleDOS = roleMapper.selectList(null);
Map<Long, Set<String>> tmp = new HashMap<>();
for (SysRoleDO roleDO : roleDOS) {
String permissionIds = roleDO.getPermissionIds();
if (StringUtils.isEmpty(permissionIds)) {
continue;
}
List<Long> list = Arrays.stream(permissionIds.split(",")).map(String::trim).filter(StringUtils::isNotEmpty).map(Long::valueOf).collect(Collectors.toList());
Set<String> permSet = tmp.getOrDefault(roleDO.getId(), new HashSet<>());
for (Long permId : list) {
SysPermissionDO permissionDO = permCache.get(permId);
if (permissionDO != null) {
permSet.add(permissionDO.getPermission());
}
}
tmp.put(roleDO.getId(), permSet);
}
rolePermCache = tmp;
}
private void loadPermCache() {
List<SysPermissionDO> roleDOS = permissionMapper.selectList(null);
Map<Long, SysPermissionDO> map = roleDOS.stream().collect(Collectors.toMap(SysPermissionDO::getId, Function.identity(), (e1, e2) -> e1));
permCache = map;
}
@Override
public void afterSingletonsInstantiated() {
this.loadPermCache();
this.refresh();
}
}

11
src/main/java/com/xuxd/kafka/console/controller/AclAuthController.java
View File

@@ -1,5 +1,6 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.AclEntry;
import com.xuxd.kafka.console.beans.dto.AddAuthDTO;
import com.xuxd.kafka.console.beans.dto.ConsumerAuthDTO;
@@ -28,6 +29,7 @@ public class AclAuthController {
@Autowired
private AclService aclService;
@Permission({"acl:authority:detail", "acl:sasl-scram:detail"})
@PostMapping("/detail")
public Object getAclDetailList(@RequestBody QueryAclDTO param) {
return aclService.getAclDetailList(param.toEntry());
@@ -38,11 +40,13 @@ public class AclAuthController {
return aclService.getOperationList();
}
@Permission("acl:authority")
@PostMapping("/list")
public Object getAclList(@RequestBody QueryAclDTO param) {
return aclService.getAclList(param.toEntry());
}
@Permission({"acl:authority:add-principal", "acl:authority:add", "acl:sasl-scram:add-auth"})
@PostMapping
public Object addAcl(@RequestBody AddAuthDTO param) {
return aclService.addAcl(param.toAclEntry());
@@ -54,6 +58,7 @@ public class AclAuthController {
* @param param entry.topic && entry.username must.
* @return
*/
@Permission({"acl:authority:producer", "acl:sasl-scram:producer"})
@PostMapping("/producer")
public Object addProducerAcl(@RequestBody ProducerAuthDTO param) {
@@ -66,6 +71,7 @@ public class AclAuthController {
* @param param entry.topic && entry.groupId entry.username must.
* @return
*/
@Permission({"acl:authority:consumer", "acl:sasl-scram:consumer"})
@PostMapping("/consumer")
public Object addConsumerAcl(@RequestBody ConsumerAuthDTO param) {
@@ -78,6 +84,7 @@ public class AclAuthController {
* @param entry entry
* @return
*/
@Permission({"acl:authority:clean", "acl:sasl-scram:pure"})
@DeleteMapping
public Object deleteAclByUser(@RequestBody AclEntry entry) {
return aclService.deleteAcl(entry);
@@ -89,6 +96,7 @@ public class AclAuthController {
* @param param entry.username
* @return
*/
@Permission({"acl:authority:clean", "acl:sasl-scram:pure"})
@DeleteMapping("/user")
public Object deleteAclByUser(@RequestBody DeleteAclDTO param) {
return aclService.deleteUserAcl(param.toUserEntry());
@@ -100,6 +108,7 @@ public class AclAuthController {
* @param param entry.topic && entry.username must.
* @return
*/
@Permission({"acl:authority:clean", "acl:sasl-scram:pure"})
@DeleteMapping("/producer")
public Object deleteProducerAcl(@RequestBody ProducerAuthDTO param) {
@@ -112,6 +121,7 @@ public class AclAuthController {
* @param param entry.topic && entry.groupId entry.username must.
* @return
*/
@Permission({"acl:authority:clean", "acl:sasl-scram:pure"})
@DeleteMapping("/consumer")
public Object deleteConsumerAcl(@RequestBody ConsumerAuthDTO param) {
@@ -124,6 +134,7 @@ public class AclAuthController {
* @param param acl principal.
* @return true or false.
*/
@Permission({"acl:authority:clean", "acl:sasl-scram:pure"})
@DeleteMapping("/clear")
public Object clearAcl(@RequestBody DeleteAclDTO param) {
return aclService.clearAcl(param.toUserEntry());

6
src/main/java/com/xuxd/kafka/console/controller/AclUserController.java
View File

@@ -1,5 +1,6 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.AclEntry;
import com.xuxd.kafka.console.beans.AclUser;
import com.xuxd.kafka.console.service.AclService;
@@ -26,27 +27,32 @@ public class AclUserController {
@Autowired
private AclService aclService;
@Permission("acl:sasl-scram")
@GetMapping
public Object getUserList() {
return aclService.getUserList();
}
@Permission({"acl:sasl-scram:add-update", "acl:sasl-scram:add-auth"})
@PostMapping
public Object addOrUpdateUser(@RequestBody AclUser user) {
return aclService.addOrUpdateUser(user.getUsername(), user.getPassword());
}
@Permission({"acl:sasl-scram:del", "acl:sasl-scram:pure"})
@DeleteMapping
public Object deleteUser(@RequestBody AclUser user) {
return aclService.deleteUser(user.getUsername());
}
@Permission({"acl:sasl-scram:del", "acl:sasl-scram:pure"})
@DeleteMapping("/auth")
public Object deleteUserAndAuth(@RequestBody AclUser user) {
return aclService.deleteUserAndAuth(user.getUsername());
}
@Permission("acl:sasl-scram:detail")
@GetMapping("/detail")
public Object getUserDetail(@RequestParam String username) {
return aclService.getUserDetail(username);

4
src/main/java/com/xuxd/kafka/console/controller/ClientQuotaController.java
View File

@@ -1,5 +1,6 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.ResponseData;
import com.xuxd.kafka.console.beans.dto.AlterClientQuotaDTO;
import com.xuxd.kafka.console.beans.dto.QueryClientQuotaDTO;
@@ -21,11 +22,13 @@ public class ClientQuotaController {
this.clientQuotaService = clientQuotaService;
}
@Permission({"quota:user", "quota:client", "quota:user-client"})
@PostMapping("/list")
public Object getClientQuotaConfigs(@RequestBody QueryClientQuotaDTO request) {
return clientQuotaService.getClientQuotaConfigs(request.getTypes(), request.getNames());
}
@Permission({"quota:user:add", "quota:client:add", "quota:user-client:add", "quota:edit"})
@PostMapping
public Object alterClientQuotaConfigs(@RequestBody AlterClientQuotaDTO request) {
if (request.getTypes().size() != 2) {
@@ -38,6 +41,7 @@ public class ClientQuotaController {
return clientQuotaService.alterClientQuotaConfigs(request);
}
@Permission("quota:del")
@DeleteMapping
public Object deleteClientQuotaConfigs(@RequestBody AlterClientQuotaDTO request) {
if (request.getTypes().size() != 2) {

4
src/main/java/com/xuxd/kafka/console/controller/ClusterController.java
View File

@@ -1,5 +1,6 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.dto.ClusterInfoDTO;
import com.xuxd.kafka.console.service.ClusterService;
import org.springframework.beans.factory.annotation.Autowired;
@@ -34,16 +35,19 @@ public class ClusterController {
return clusterService.getClusterInfoList();
}
@Permission("op:cluster-switch:add")
@PostMapping("/info")
public Object addClusterInfo(@RequestBody ClusterInfoDTO dto) {
return clusterService.addClusterInfo(dto.to());
}
@Permission("op:cluster-switch:del")
@DeleteMapping("/info")
public Object deleteClusterInfo(@RequestBody ClusterInfoDTO dto) {
return clusterService.deleteClusterInfo(dto.getId());
}
@Permission("op:cluster-switch:edit")
@PutMapping("/info")
public Object updateClusterInfo(@RequestBody ClusterInfoDTO dto) {
return clusterService.updateClusterInfo(dto.to());

10
src/main/java/com/xuxd/kafka/console/controller/ConfigController.java
View File

@@ -1,5 +1,6 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.ResponseData;
import com.xuxd.kafka.console.beans.dto.AlterConfigDTO;
import com.xuxd.kafka.console.beans.enums.AlterType;
@@ -41,46 +42,55 @@ public class ConfigController {
return ResponseData.create().data(configMap).success();
}
@Permission("topic:property-config")
@GetMapping("/topic")
public Object getTopicConfig(String topic) {
return configService.getTopicConfig(topic);
}
@Permission("topic:property-config:edit")
@PostMapping("/topic")
public Object setTopicConfig(@RequestBody AlterConfigDTO dto) {
return configService.alterTopicConfig(dto.getEntity(), dto.to(), AlterType.SET);
}
@Permission("topic:property-config:del")
@DeleteMapping("/topic")
public Object deleteTopicConfig(@RequestBody AlterConfigDTO dto) {
return configService.alterTopicConfig(dto.getEntity(), dto.to(), AlterType.DELETE);
}
@Permission("cluster:property-config")
@GetMapping("/broker")
public Object getBrokerConfig(String brokerId) {
return configService.getBrokerConfig(brokerId);
}
@Permission("cluster:edit")
@PostMapping("/broker")
public Object setBrokerConfig(@RequestBody AlterConfigDTO dto) {
return configService.alterBrokerConfig(dto.getEntity(), dto.to(), AlterType.SET);
}
@Permission("cluster:edit")
@DeleteMapping("/broker")
public Object deleteBrokerConfig(@RequestBody AlterConfigDTO dto) {
return configService.alterBrokerConfig(dto.getEntity(), dto.to(), AlterType.DELETE);
}
@Permission("cluster:log-config")
@GetMapping("/broker/logger")
public Object getBrokerLoggerConfig(String brokerId) {
return configService.getBrokerLoggerConfig(brokerId);
}
@Permission("cluster:edit")
@PostMapping("/broker/logger")
public Object setBrokerLoggerConfig(@RequestBody AlterConfigDTO dto) {
return configService.alterBrokerLoggerConfig(dto.getEntity(), dto.to(), AlterType.SET);
}
@Permission("cluster:edit")
@DeleteMapping("/broker/logger")
public Object deleteBrokerLoggerConfig(@RequestBody AlterConfigDTO dto) {
return configService.alterBrokerLoggerConfig(dto.getEntity(), dto.to(), AlterType.DELETE);

30
src/main/java/com/xuxd/kafka/console/controller/ConsumerController.java
View File

@@ -1,28 +1,20 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.ResponseData;
import com.xuxd.kafka.console.beans.dto.AddSubscriptionDTO;
import com.xuxd.kafka.console.beans.dto.QueryConsumerGroupDTO;
import com.xuxd.kafka.console.beans.dto.ResetOffsetDTO;
import com.xuxd.kafka.console.service.ConsumerService;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.OffsetResetStrategy;
import org.apache.kafka.common.ConsumerGroupState;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.*;
import java.util.*;
/**
* kafka-console-ui.
@@ -51,26 +43,34 @@ public class ConsumerController {
return consumerService.getConsumerGroupList(groupIdList, stateSet);
}
@Permission("group:del")
@DeleteMapping("/group")
public Object deleteConsumerGroup(@RequestParam String groupId) {
return consumerService.deleteConsumerGroup(groupId);
}
@Permission("group:client")
@GetMapping("/member")
public Object getConsumerMembers(@RequestParam String groupId) {
return consumerService.getConsumerMembers(groupId);
}
@Permission("group:consumer-detail")
@GetMapping("/detail")
public Object getConsumerDetail(@RequestParam String groupId) {
return consumerService.getConsumerDetail(groupId);
}
@Permission("group:add")
@PostMapping("/subscription")
public Object addSubscription(@RequestBody AddSubscriptionDTO subscriptionDTO) {
return consumerService.addSubscription(subscriptionDTO.getGroupId(), subscriptionDTO.getTopic());
}
@Permission({"group:consumer-detail:min",
"group:consumer-detail:last",
"group:consumer-detail:timestamp",
"group:consumer-detail:any"})
@PostMapping("/reset/offset")
public Object restOffset(@RequestBody ResetOffsetDTO offsetDTO) {
ResponseData res = ResponseData.create().failed("unknown");
@@ -78,7 +78,7 @@ public class ConsumerController {
case ResetOffsetDTO.Level.TOPIC:
switch (offsetDTO.getType()) {
case ResetOffsetDTO.Type
.EARLIEST:
.EARLIEST:
res = consumerService.resetOffsetToEndpoint(offsetDTO.getGroupId(), offsetDTO.getTopic(), OffsetResetStrategy.EARLIEST);
break;
case ResetOffsetDTO.Type.LATEST:
@@ -94,7 +94,7 @@ public class ConsumerController {
case ResetOffsetDTO.Level.PARTITION:
switch (offsetDTO.getType()) {
case ResetOffsetDTO.Type
.SPECIAL:
.SPECIAL:
res = consumerService.resetPartitionToTargetOffset(offsetDTO.getGroupId(), new TopicPartition(offsetDTO.getTopic(), offsetDTO.getPartition()), offsetDTO.getOffset());
break;
default:
@@ -118,11 +118,13 @@ public class ConsumerController {
return consumerService.getSubscribeTopicList(groupId);
}
@Permission({"topic:consumer-detail"})
@GetMapping("/topic/subscribed")
public Object getTopicSubscribedByGroups(@RequestParam String topic) {
return consumerService.getTopicSubscribedByGroups(topic);
}
@Permission("group:offset-partition")
@GetMapping("/offset/partition")
public Object getOffsetPartition(@RequestParam String groupId) {
return consumerService.getOffsetPartition(groupId);

7
src/main/java/com/xuxd/kafka/console/controller/MessageController.java
View File

@@ -1,5 +1,6 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.QueryMessage;
import com.xuxd.kafka.console.beans.ResponseData;
import com.xuxd.kafka.console.beans.SendMessage;
@@ -24,16 +25,19 @@ public class MessageController {
@Autowired
private MessageService messageService;
@Permission("message:search-time")
@PostMapping("/search/time")
public Object searchByTime(@RequestBody QueryMessageDTO dto) {
return messageService.searchByTime(dto.toQueryMessage());
}
@Permission("message:search-offset")
@PostMapping("/search/offset")
public Object searchByOffset(@RequestBody QueryMessageDTO dto) {
return messageService.searchByOffset(dto.toQueryMessage());
}
@Permission("message:detail")
@PostMapping("/search/detail")
public Object searchDetail(@RequestBody QueryMessageDTO dto) {
return messageService.searchDetail(dto.toQueryMessage());
@@ -44,16 +48,19 @@ public class MessageController {
return messageService.deserializerList();
}
@Permission("message:send")
@PostMapping("/send")
public Object send(@RequestBody SendMessage message) {
return messageService.send(message);
}
@Permission("message:resend")
@PostMapping("/resend")
public Object resend(@RequestBody SendMessage message) {
return messageService.resend(message);
}
@Permission("message:del")
@DeleteMapping
public Object delete(@RequestBody List<QueryMessage> messages) {
if (CollectionUtils.isEmpty(messages)) {

14
src/main/java/com/xuxd/kafka/console/controller/OperationController.java
View File

@@ -1,5 +1,6 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.TopicPartition;
import com.xuxd.kafka.console.beans.dto.BrokerThrottleDTO;
import com.xuxd.kafka.console.beans.dto.ProposedAssignmentDTO;
@@ -8,13 +9,7 @@ import com.xuxd.kafka.console.beans.dto.SyncDataDTO;
import com.xuxd.kafka.console.service.OperationService;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.bind.annotation.*;
/**
* kafka-console-ui.
@@ -51,26 +46,31 @@ public class OperationController {
return operationService.deleteAlignmentById(id);
}
@Permission({"topic:partition-detail:preferred", "op:replication-preferred"})
@PostMapping("/replication/preferred")
public Object electPreferredLeader(@RequestBody ReplicationDTO dto) {
return operationService.electPreferredLeader(dto.getTopic(), dto.getPartition());
}
@Permission("op:config-throttle")
@PostMapping("/broker/throttle")
public Object configThrottle(@RequestBody BrokerThrottleDTO dto) {
return operationService.configThrottle(dto.getBrokerList(), dto.getUnit().toKb(dto.getThrottle()));
}
@Permission("op:remove-throttle")
@DeleteMapping("/broker/throttle")
public Object removeThrottle(@RequestBody BrokerThrottleDTO dto) {
return operationService.removeThrottle(dto.getBrokerList());
}
@Permission("op:replication-update-detail")
@GetMapping("/replication/reassignments")
public Object currentReassignments() {
return operationService.currentReassignments();
}
@Permission("op:replication-update-detail:cancel")
@DeleteMapping("/replication/reassignments")
public Object cancelReassignment(@RequestBody TopicPartition partition) {
return operationService.cancelReassignment(new org.apache.kafka.common.TopicPartition(partition.getTopic(), partition.getPartition()));

20
src/main/java/com/xuxd/kafka/console/controller/TopicController.java
View File

@@ -1,23 +1,19 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.ReplicaAssignment;
import com.xuxd.kafka.console.beans.dto.AddPartitionDTO;
import com.xuxd.kafka.console.beans.dto.NewTopicDTO;
import com.xuxd.kafka.console.beans.dto.TopicThrottleDTO;
import com.xuxd.kafka.console.beans.enums.TopicType;
import com.xuxd.kafka.console.service.TopicService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
/**
* kafka-console-ui.
@@ -37,26 +33,31 @@ public class TopicController {
return topicService.getTopicNameList(false);
}
@Permission("topic:load")
@GetMapping("/list")
public Object getTopicList(@RequestParam(required = false) String topic, @RequestParam String type) {
return topicService.getTopicList(topic, TopicType.valueOf(type.toUpperCase()));
}
@Permission({"topic:batch-del", "topic:del"})
@DeleteMapping
public Object deleteTopic(@RequestBody List<String> topics) {
return topicService.deleteTopics(topics);
}
@Permission("topic:partition-detail")
@GetMapping("/partition")
public Object getTopicPartitionInfo(@RequestParam String topic) {
return topicService.getTopicPartitionInfo(topic.trim());
}
@Permission("topic:add")
@PostMapping("/new")
public Object createNewTopic(@RequestBody NewTopicDTO topicDTO) {
return topicService.createTopic(topicDTO.toNewTopic());
}
@Permission("topic:partition-add")
@PostMapping("/partition/new")
public Object addPartition(@RequestBody AddPartitionDTO partitionDTO) {
String topic = partitionDTO.getTopic().trim();
@@ -79,16 +80,19 @@ public class TopicController {
return topicService.getCurrentReplicaAssignment(topic);
}
@Permission({"topic:replication-modify", "op:replication-reassign"})
@PostMapping("/replica/assignment")
public Object updateReplicaAssignment(@RequestBody ReplicaAssignment assignment) {
return topicService.updateReplicaAssignment(assignment);
}
@Permission("topic:replication-sync-throttle")
@PostMapping("/replica/throttle")
public Object configThrottle(@RequestBody TopicThrottleDTO dto) {
return topicService.configThrottle(dto.getTopic(), dto.getPartitions(), dto.getOperation());
}
@Permission("topic:send-count")
@GetMapping("/send/stats")
public Object sendStats(@RequestParam String topic) {
return topicService.sendStats(topic);

12
src/main/java/com/xuxd/kafka/console/controller/UserManageController.java
View File

@@ -1,6 +1,7 @@
package com.xuxd.kafka.console.controller;
import com.xuxd.kafka.console.aspect.annotation.ControllerLog;
import com.xuxd.kafka.console.aspect.annotation.Permission;
import com.xuxd.kafka.console.beans.Credentials;
import com.xuxd.kafka.console.beans.dto.SysPermissionDTO;
import com.xuxd.kafka.console.beans.dto.SysRoleDTO;
@@ -24,12 +25,14 @@ public class UserManageController {
this.userManageService = userManageService;
}
@Permission({"user-manage:user:add", "user-manage:user:change-role", "user-manage:user:reset-pass"})
@ControllerLog("新增/更新用户")
@PostMapping("/user")
public Object addOrUpdateUser(@RequestBody SysUserDTO userDTO) {
return userManageService.addOrUpdateUser(userDTO);
}
@Permission("user-manage:role:save")
@ControllerLog("新增/更新角色")
@PostMapping("/role")
public Object addOrUpdateRole(@RequestBody SysRoleDTO roleDTO) {
@@ -42,43 +45,50 @@ public class UserManageController {
return userManageService.addPermission(permissionDTO);
}
@Permission("user-manage:role:save")
@ControllerLog("更新角色")
@PutMapping("/role")
public Object updateRole(@RequestBody SysRoleDTO roleDTO) {
return userManageService.updateRole(roleDTO);
}
@Permission({"user-manage:role"})
@GetMapping("/role")
public Object selectRole() {
return userManageService.selectRole();
}
@Permission({"user-manage:permission"})
@GetMapping("/permission")
public Object selectPermission() {
return userManageService.selectPermission();
}
@Permission({"user-manage:user"})
@GetMapping("/user")
public Object selectUser() {
return userManageService.selectUser();
}
@Permission("user-manage:role:del")
@ControllerLog("删除角色")
@DeleteMapping("/role")
public Object deleteRole(@RequestParam Long id) {
return userManageService.deleteRole(id);
}
@Permission("user-manage:user:del")
@ControllerLog("删除用户")
@DeleteMapping("/user")
public Object deleteUser(@RequestParam Long id) {
return userManageService.deleteUser(id);
}
@Permission("user-manage:setting")
@ControllerLog("更新密码")
@PostMapping("/user/password")
public Object updatePassword(@RequestBody SysUserDTO userDTO, HttpServletRequest request) {
Credentials credentials = (Credentials)request.getAttribute("credentials");
Credentials credentials = (Credentials) request.getAttribute("credentials");
if (credentials != null && !credentials.isInvalid()) {
userDTO.setUsername(credentials.getUsername());
}

11
src/main/java/com/xuxd/kafka/console/dao/init/DataInit.java
View File

@@ -1,11 +1,13 @@
package com.xuxd.kafka.console.dao.init;
import com.xuxd.kafka.console.beans.RolePermUpdateEvent;
import com.xuxd.kafka.console.config.AuthConfig;
import com.xuxd.kafka.console.dao.SysPermissionMapper;
import com.xuxd.kafka.console.dao.SysRoleMapper;
import com.xuxd.kafka.console.dao.SysUserMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
@@ -33,17 +35,21 @@ public class DataInit implements SmartInitializingSingleton {
private final SqlParse sqlParse;
private final ApplicationEventPublisher publisher;
public DataInit(AuthConfig authConfig,
SysUserMapper userMapper,
SysRoleMapper roleMapper,
SysPermissionMapper permissionMapper,
DataSource dataSource) {
DataSource dataSource,
ApplicationEventPublisher publisher) {
this.authConfig = authConfig;
this.userMapper = userMapper;
this.roleMapper = roleMapper;
this.permissionMapper = permissionMapper;
this.dataSource = dataSource;
this.publisher = publisher;
this.sqlParse = new SqlParse();
}
@@ -69,6 +75,9 @@ public class DataInit implements SmartInitializingSingleton {
if (permCount == null || permCount == 0) {
initData(connection, SqlParse.PERM_TABLE);
}
RolePermUpdateEvent event = new RolePermUpdateEvent(this);
event.setReload(true);
publisher.publishEvent(event);
} catch (SQLException e) {
throw new RuntimeException(e);
}

15
src/main/java/com/xuxd/kafka/console/interceptor/GlobalExceptionHandler.java → src/main/java/com/xuxd/kafka/console/exception/GlobalExceptionHandler.java
View File

@@ -1,11 +1,14 @@
package com.xuxd.kafka.console.interceptor;
package com.xuxd.kafka.console.exception;
import com.xuxd.kafka.console.beans.ResponseData;
import javax.servlet.http.HttpServletRequest;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.ControllerAdvice;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.ResponseStatus;
import javax.servlet.http.HttpServletRequest;
/**
* kafka-console-ui.
@@ -17,6 +20,14 @@ import org.springframework.web.bind.annotation.ResponseBody;
@ControllerAdvice(basePackages = "com.xuxd.kafka.console.controller")
public class GlobalExceptionHandler {
@ResponseStatus(code = HttpStatus.FORBIDDEN)
@ExceptionHandler(value = UnAuthorizedException.class)
@ResponseBody
public Object unAuthorizedExceptionHandler(HttpServletRequest req, Exception ex) throws Exception {
log.error("unAuthorized: {}", ex.getMessage());
return ResponseData.create().failed("UnAuthorized: " + ex.getMessage());
}
@ExceptionHandler(value = Exception.class)
@ResponseBody
public Object exceptionHandler(HttpServletRequest req, Exception ex) throws Exception {

12
src/main/java/com/xuxd/kafka/console/exception/UnAuthorizedException.java Normal file
View File

@@ -0,0 +1,12 @@
package com.xuxd.kafka.console.exception;
/**
* @author: xuxd
* @date: 2023/5/17 23:08
**/
public class UnAuthorizedException extends RuntimeException{
public UnAuthorizedException(String message) {
super(message);
}
}

9
src/main/java/com/xuxd/kafka/console/interceptor/AuthFilter.java → src/main/java/com/xuxd/kafka/console/filter/AuthFilter.java
View File

@@ -1,4 +1,4 @@
package com.xuxd.kafka.console.interceptor;
package com.xuxd.kafka.console.filter;
import com.xuxd.kafka.console.beans.Credentials;
import com.xuxd.kafka.console.config.AuthConfig;
@@ -60,6 +60,11 @@ public class AuthFilter implements Filter {
}
request.setAttribute("credentials", credentials);
filterChain.doFilter(servletRequest, servletResponse);
try {
CredentialsContext.set(credentials);
filterChain.doFilter(servletRequest, servletResponse);
}finally {
CredentialsContext.remove();
}
}
}

2
src/main/java/com/xuxd/kafka/console/interceptor/ContextSetFilter.java → src/main/java/com/xuxd/kafka/console/filter/ContextSetFilter.java
View File

@@ -1,4 +1,4 @@
package com.xuxd.kafka.console.interceptor;
package com.xuxd.kafka.console.filter;
import com.xuxd.kafka.console.beans.ResponseData;
import com.xuxd.kafka.console.beans.dos.ClusterInfoDO;

23
src/main/java/com/xuxd/kafka/console/filter/CredentialsContext.java Normal file
View File

@@ -0,0 +1,23 @@
package com.xuxd.kafka.console.filter;
import com.xuxd.kafka.console.beans.Credentials;
/**
* @author: xuxd
* @date: 2023/5/17 23:02
**/
public class CredentialsContext {
private static final ThreadLocal<Credentials> CREDENTIALS = new ThreadLocal<>();
public static void set(Credentials credentials) {
CREDENTIALS.set(credentials);
}
public static Credentials get() {
return CREDENTIALS.get();
}
public static void remove() {
CREDENTIALS.remove();
}
}

29
src/main/java/com/xuxd/kafka/console/service/impl/AuthServiceImpl.java
View File

@@ -4,12 +4,11 @@ import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.xuxd.kafka.console.beans.Credentials;
import com.xuxd.kafka.console.beans.LoginResult;
import com.xuxd.kafka.console.beans.ResponseData;
import com.xuxd.kafka.console.beans.dos.SysPermissionDO;
import com.xuxd.kafka.console.beans.dos.SysRoleDO;
import com.xuxd.kafka.console.beans.dos.SysUserDO;
import com.xuxd.kafka.console.beans.dto.LoginUserDTO;
import com.xuxd.kafka.console.cache.RolePermCache;
import com.xuxd.kafka.console.config.AuthConfig;
import com.xuxd.kafka.console.dao.SysPermissionMapper;
import com.xuxd.kafka.console.dao.SysRoleMapper;
import com.xuxd.kafka.console.dao.SysUserMapper;
import com.xuxd.kafka.console.service.AuthService;
@@ -17,11 +16,11 @@ import com.xuxd.kafka.console.utils.AuthUtil;
import com.xuxd.kafka.console.utils.UUIDStrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.stereotype.Service;
import java.util.*;
import java.util.function.Function;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
/**
@@ -30,26 +29,24 @@ import java.util.stream.Collectors;
**/
@Slf4j
@Service
public class AuthServiceImpl implements AuthService, SmartInitializingSingleton {
public class AuthServiceImpl implements AuthService {
private final SysUserMapper userMapper;
private final SysRoleMapper roleMapper;
private final SysPermissionMapper permissionMapper;
private final AuthConfig authConfig;
private final Map<Long, SysPermissionDO> permCache = new HashMap<>();
private final RolePermCache rolePermCache;
public AuthServiceImpl(SysUserMapper userMapper,
SysRoleMapper roleMapper,
SysPermissionMapper permissionMapper,
AuthConfig authConfig) {
AuthConfig authConfig,
RolePermCache rolePermCache) {
this.userMapper = userMapper;
this.roleMapper = roleMapper;
this.permissionMapper = permissionMapper;
this.authConfig = authConfig;
this.rolePermCache = rolePermCache;
}
@Override
@@ -81,7 +78,7 @@ public class AuthServiceImpl implements AuthService, SmartInitializingSingleton
List<Long> permIds = Arrays.stream(permissionIds.split(",")).map(String::trim).
filter(StringUtils::isNotEmpty).map(Long::valueOf).collect(Collectors.toList());
permIds.forEach(id -> {
String permission = permCache.get(id).getPermission();
String permission = rolePermCache.getPermCache().get(id).getPermission();
if (StringUtils.isNotEmpty(permission)) {
permissions.add(permission);
} else {
@@ -96,10 +93,4 @@ public class AuthServiceImpl implements AuthService, SmartInitializingSingleton
return ResponseData.create().data(loginResult).success();
}
@Override
public void afterSingletonsInstantiated() {
List<SysPermissionDO> roleDOS = permissionMapper.selectList(null);
Map<Long, SysPermissionDO> map = roleDOS.stream().collect(Collectors.toMap(SysPermissionDO::getId, Function.identity(), (e1, e2) -> e1));
permCache.putAll(map);
}
}

11
src/main/java/com/xuxd/kafka/console/service/impl/UserManageServiceImpl.java
View File

@@ -2,6 +2,7 @@ package com.xuxd.kafka.console.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.xuxd.kafka.console.beans.ResponseData;
import com.xuxd.kafka.console.beans.RolePermUpdateEvent;
import com.xuxd.kafka.console.beans.dos.SysPermissionDO;
import com.xuxd.kafka.console.beans.dos.SysRoleDO;
import com.xuxd.kafka.console.beans.dos.SysUserDO;
@@ -20,6 +21,7 @@ import com.xuxd.kafka.console.utils.UUIDStrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import java.util.*;
@@ -40,12 +42,16 @@ public class UserManageServiceImpl implements UserManageService {
private final SysPermissionMapper permissionMapper;
private final ApplicationEventPublisher publisher;
public UserManageServiceImpl(ObjectProvider<SysUserMapper> userMapper,
ObjectProvider<SysRoleMapper> roleMapper,
ObjectProvider<SysPermissionMapper> permissionMapper) {
ObjectProvider<SysPermissionMapper> permissionMapper,
ApplicationEventPublisher publisher) {
this.userMapper = userMapper.getIfAvailable();
this.roleMapper = roleMapper.getIfAvailable();
this.permissionMapper = permissionMapper.getIfAvailable();
this.publisher = publisher;
}
@Override
@@ -62,6 +68,7 @@ public class UserManageServiceImpl implements UserManageService {
} else {
roleMapper.updateById(roleDO);
}
publisher.publishEvent(new RolePermUpdateEvent(this));
return ResponseData.create().success();
}
@@ -187,6 +194,7 @@ public class UserManageServiceImpl implements UserManageService {
@Override
public ResponseData updateRole(SysRoleDTO roleDTO) {
roleMapper.updateById(roleDTO.toDO());
publisher.publishEvent(new RolePermUpdateEvent(this));
return ResponseData.create().success();
}
@@ -199,6 +207,7 @@ public class UserManageServiceImpl implements UserManageService {
return ResponseData.create().failed("存在用户被分配为当前角色，不允许删除");
}
roleMapper.deleteById(id);
publisher.publishEvent(new RolePermUpdateEvent(this));
return ResponseData.create().success();
}