code-image-backup-public/fastbee
1
0
Fork 0
mirror of https://gitee.com/beecue/fastbee.git synced 2026-11-07 01:18:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(规则脚本): 规则引擎 Groovy 脚本注入导致远程代码执行漏洞修复

Browse Source
This commit is contained in:
gx_ma
2026-10-01 03:05:41 +08:00
parent 284b714c1e
commit 890a338460
56 changed files with 2124 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
springboot/fastbee-server/base-server/src/main/java/com/fastbee/base/session/Session.java
View File

@@ -1,6 +1,7 @@
package com.fastbee.base.session;
import com.fastbee.common.core.protocol.Message;
import com.fastbee.common.core.domain.model.MqttClientPrincipal;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.utils.DateUtils;
@@ -80,6 +81,9 @@ public class Session {
/*服务协议类型 MQTT，TCP UDP COAP*/
private ServerType serverType;
/*认证主体，记录这条连接代表哪个平台用户，供 PUBLISH 阶段做设备归属校验*/
private MqttClientPrincipal principal;
public Session(){
}

69
springboot/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/acl/MqttAccessControl.java Normal file
View File

@@ -0,0 +1,69 @@
package com.fastbee.mqtt.acl;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.fastbee.common.core.domain.entity.SysUser;
import com.fastbee.common.core.domain.model.MqttClientPrincipal;
import com.fastbee.iot.domain.Device;
import com.fastbee.iot.domain.DeviceUser;
import com.fastbee.iot.mapper.DeviceMapper;
import com.fastbee.iot.mapper.DeviceUserMapper;
import com.fastbee.mqtt.auth.AuthService;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.*;
/** Reuse the platform's owner, tenant and shared-model permissions, with live login state. */
@Component
public class MqttAccessControl {
@Resource private AuthService authService;
@Resource private DeviceMapper deviceMapper;
@Resource private DeviceUserMapper deviceUserMapper;
public boolean canPublish(MqttClientPrincipal principal, String topic) { return canPublish(principal, topic, null); }
public boolean canPublish(MqttClientPrincipal principal, String topic, String payload) { return allowed(principal, topic, payload, false); }
public boolean canSubscribe(MqttClientPrincipal principal, String topic) { return allowed(principal, topic, null, true); }
private boolean allowed(MqttClientPrincipal principal, String topic, String payload, boolean subscribe) {
if (principal == null || principal.getClientType() == null) return false;
if (!principal.isWeb()) return true;
try {
SysUser user = authService.currentWebUser(principal);
if (user == null || topic == null) return false;
if (user.isAdmin()) return subscribe || topic.indexOf('#') < 0 && topic.indexOf('+') < 0;
String[] parts = topic.split("/", -1);
if (parts.length < 4 || !parts[0].isEmpty() || parts[1].isEmpty() || parts[2].isEmpty()
|| parts[1].contains("+") || parts[1].contains("#") || parts[2].contains("+") || parts[2].contains("#")) return false;
if (!subscribe && (topic.contains("+") || topic.contains("#"))) return false;
Device device = deviceMapper.selectDeviceBySerialNumber(parts[2]);
if (device == null || !String.valueOf(device.getProductId()).equals(parts[1])) return false;
if (Objects.equals(device.getUserId(), user.getUserId())) return true;
if (Objects.equals(device.getTenantId(), user.getUserId()) && user.getRoles() != null
&& user.getRoles().stream().anyMatch(r -> "tenant".equals(r.getRoleKey()))) return true;
DeviceUser share = deviceUserMapper.selectDeviceUserByDeviceIdAndUserId(device.getDeviceId(), user.getUserId());
if (share == null || "2".equals(share.getDelFlag())) return false;
if (Integer.valueOf(1).equals(share.getIsOwner())) return true;
// Viewing a shared device retains its established subscription behavior.
if (subscribe) return true;
Set<String> permissions = new HashSet<>();
if (share.getPerms() != null) for (String p : share.getPerms().split(",")) permissions.add(p.trim());
String suffix = String.join("/", Arrays.copyOfRange(parts, 3, parts.length));
if ("monitor/get".equals(suffix)) return permissions.contains("monitor");
if (!"property/get".equals(suffix) && !"function/get".equals(suffix)) return false;
if (payload == null) return false;
Object body = JSON.parse(payload);
JSONArray values = body instanceof JSONArray ? (JSONArray) body : new JSONArray(Collections.singletonList(body));
if (values.isEmpty()) return false;
for (Object value : values) {
if (!(value instanceof JSONObject)) return false;
String identifier = ((JSONObject) value).getString("id");
if (identifier == null || !permissions.contains(identifier)) return false;
}
return true;
} catch (RuntimeException e) {
// Malformed topics, revoked state or unavailable authorization storage must fail closed.
return false;
}
}
}

91
springboot/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/auth/AuthService.java
View File

@@ -1,6 +1,10 @@
package com.fastbee.mqtt.auth;
import com.fastbee.common.constant.CacheConstants;
import com.fastbee.common.constant.Constants;
import com.fastbee.common.core.domain.entity.SysUser;
import com.fastbee.common.core.domain.model.LoginUser;
import com.fastbee.common.core.domain.model.MqttClientPrincipal;
import com.fastbee.common.core.redis.RedisCache;
import com.fastbee.common.exception.ServiceException;
import com.fastbee.common.utils.StringUtils;
@@ -33,6 +37,8 @@ public class AuthService {
private RedisCache redisCache;
@Resource
private MqttClientConfig mqttConfig;
@Resource
private com.fastbee.system.service.ISysUserService userService;
// 令牌秘钥
@Value("${token.secret}")
private String secret;
@@ -41,57 +47,96 @@ public class AuthService {
/**
* MQTT客户端认证
*
* <p>web / phone 客户端使用平台登录 JWT 作为 MQTT 密码。此处不能只做签名校验：
* 必须回查 Redis 中的登录态并取出 userId，一方面让已注销/已过期的令牌立即失效，
* 另一方面让 PUBLISH 阶段能校验该用户对目标设备的归属权限。</p>
*
* @param clientId 客户端id
* @param username 用户名
* @param password 密码
* @return 结果
* @return 认证主体，认证失败返回 null
*/
public boolean auth(String clientId, String username, String password) {
public MqttClientPrincipal authenticate(String clientId, String username, String password) {
if (StringUtils.isEmpty(clientId) || StringUtils.isEmpty(username) || StringUtils.isEmpty(password)) {
log.error("=>客户端参数缺少,clientId:{},username:{},password:{}", clientId, username, password);
return false;
log.error("=>客户端参数缺少,clientId:{}", clientId);
return null;
}
try {
if (clientId.startsWith("server")) {
// 服务端认证：配置的账号密码认证
if (mqttConfig.getUsername().equals(username) && mqttConfig.getPassword().equals(password)) {
log.info("-----------服务端mqtt认证成功,clientId:" + clientId + "---------------");
return true;
return MqttClientPrincipal.of(clientId, MqttClientPrincipal.ClientType.SERVER);
} else {
ResponseEntity response = toolService.returnUnauthorized(new MqttAuthenticationModel(clientId, username, password), "mqtt账号和密码与认证服务器配置不匹配");
log.warn("=>服务端认证失败[{}]", response.getBody());
throw new ServiceException("服务端认证失败:"+response.getBody());
throw new ServiceException("服务端认证失败:" + response.getBody());
}
} else if (clientId.startsWith("web") || clientId.startsWith("phone")) {
// web端和移动端认证：token认证
String token = password;
if (StringUtils.isNotEmpty(token) && token.startsWith(Constants.TOKEN_PREFIX)) {
token = token.replace(Constants.TOKEN_PREFIX, "");
}
try {
Claims claims = Jwts.parser().setSigningKey(secret).parseClaimsJws(token).getBody();
log.info("-----------移动端/Web端mqtt认证成功,clientId:" + clientId + "---------------");
return true;
} catch (Exception ex) {
ResponseEntity response = toolService.returnUnauthorized(new MqttAuthenticationModel(clientId, username, password), ex.getMessage());
log.warn("=>移动端/Web端mqtt认证失败[{}]",response.getBody());
throw new ServiceException("移动端/Web端mqtt认证失败:"+response.getBody());
}
return authenticateWebClient(clientId, username, password);
} else {
// 设备端认证
ResponseEntity response = toolService.clientAuth(clientId, username, password);
if (response.getStatusCodeValue() == HttpStatus.OK.value()) {
return true;
return MqttClientPrincipal.of(clientId, MqttClientPrincipal.ClientType.DEVICE);
} else {
log.warn("=>设备端认证失败");
throw new ServiceException("设备端认证失败:"+response.getBody());
throw new ServiceException("设备端认证失败:" + response.getBody());
}
}
} catch (Exception e) {
log.error("=>客户端认证失败", e);
return false;
return null;
}
}
/**
* Web / 移动端认证：校验 JWT 并解析出平台用户身份
*/
private MqttClientPrincipal authenticateWebClient(String clientId, String username, String password) {
String token = password;
if (StringUtils.isNotEmpty(token) && token.startsWith(Constants.TOKEN_PREFIX)) {
token = token.replace(Constants.TOKEN_PREFIX, "");
}
Claims claims;
try {
claims = Jwts.parser().setSigningKey(secret).parseClaimsJws(token).getBody();
} catch (Exception ex) {
ResponseEntity response = toolService.returnUnauthorized(new MqttAuthenticationModel(clientId, username, password), ex.getMessage());
log.warn("=>移动端/Web端mqtt认证失败[{}]", response.getBody());
throw new ServiceException("移动端/Web端mqtt认证失败:" + response.getBody());
}
// 回查登录态：令牌签名合法但已注销/已过期时必须拒绝
String uuid = (String) claims.get(Constants.LOGIN_USER_KEY);
if (StringUtils.isEmpty(uuid)) {
log.warn("=>移动端/Web端mqtt认证失败[令牌中缺少登录标识]");
throw new ServiceException("移动端/Web端mqtt认证失败:令牌中缺少登录标识");
}
LoginUser loginUser = redisCache.getCacheObject(CacheConstants.LOGIN_TOKEN_KEY + uuid);
if (loginUser == null || loginUser.getUser() == null) {
log.warn("=>移动端/Web端mqtt认证失败[登录态已失效],clientId:{}", clientId);
throw new ServiceException("移动端/Web端mqtt认证失败:登录态已失效，请重新登录");
}
MqttClientPrincipal principal = MqttClientPrincipal.ofWeb(clientId, loginUser.getUser().getUserId(), false);
principal.setSessionId(uuid);
SysUser user = currentWebUser(principal);
if (user == null) throw new ServiceException("登录态不可用或必须先修改默认密码");
log.info("-----------移动端/Web端mqtt认证成功,clientId:{},userId:{}---------------", clientId, user.getUserId());
principal.setAdmin(user.isAdmin());
return principal;
}
/** Check every Web operation/delivery; a valid old TCP connection is not a valid login session. */
public SysUser currentWebUser(MqttClientPrincipal principal) {
if (principal == null || !principal.isWeb() || StringUtils.isEmpty(principal.getSessionId())) return null;
LoginUser login = redisCache.getCacheObject(CacheConstants.LOGIN_TOKEN_KEY + principal.getSessionId());
if (login == null || login.getUser() == null || !java.util.Objects.equals(principal.getUserId(), login.getUser().getUserId())
|| !"0".equals(login.getUser().getStatus()) || "2".equals(login.getUser().getDelFlag())
|| com.fastbee.common.utils.PasswordPolicy.requiresChange(login)) return null;
SysUser current = userService.selectUserById(principal.getUserId());
if (current == null || !"0".equals(current.getStatus()) || "2".equals(current.getDelFlag())
|| !java.util.Objects.equals(current.getPassword(), login.getUser().getPassword())) return null;
return current;
}
}

5
springboot/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/MqttConnect.java
View File

@@ -2,6 +2,7 @@ package com.fastbee.mqtt.handler;
import com.fastbee.common.constant.FastBeeConstant;
import com.fastbee.common.core.redis.RedisCache;
import com.fastbee.common.core.domain.model.MqttClientPrincipal;
import com.fastbee.common.enums.ServerType;
import com.fastbee.common.utils.DateUtils;
import com.fastbee.mqtt.annotation.Process;
@@ -126,11 +127,13 @@ public class MqttConnect implements MqttHandler {
String username = message.payload().userName();
String password = message.payload().passwordInBytes() == null ? null : new String(message.payload().passwordInBytes(), CharsetUtil.UTF_8);
/*验证失败，应答客户端*/
if (!authService.auth(clientId, username, password)) {
MqttClientPrincipal principal = authService.authenticate(clientId, username, password);
if (principal == null) {
MqttConnAckMessage connAckMessage = MqttMessageUtils.buildConntAckMessage(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false);
ResponseManager.responseMessage(session, connAckMessage, true);
return false;
}
session.setPrincipal(principal);
return true;
}

14
springboot/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/MqttPublish.java
View File

@@ -11,6 +11,7 @@ import com.fastbee.iot.ruleEngine.RuleProcess;
import com.fastbee.mq.redischannel.producer.MessageProducer;
import com.fastbee.mq.service.IDeviceReportMessageService;
import com.fastbee.mqtt.annotation.Process;
import com.fastbee.mqtt.acl.MqttAccessControl;
import com.fastbee.mqtt.handler.adapter.MqttHandler;
import com.fastbee.mqtt.manager.ClientManager;
import com.fastbee.mqtt.manager.ResponseManager;
@@ -52,20 +53,31 @@ public class MqttPublish implements MqttHandler {
@Resource
private RuleProcess ruleProcess;
@Resource
private MqttAccessControl mqttAccessControl;
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message) {
MqttPublishMessage publishMessage = (MqttPublishMessage) message;
/*获取客户端id*/
String clientId = AttributeUtils.getClientId(ctx.channel());
String topicName = publishMessage.variableHeader().topicName();
Session session = AttributeUtils.getSession(ctx.channel());
if (!mqttAccessControl.canPublish(session == null ? null : session.getPrincipal(), topicName,
publishMessage.content().toString(StandardCharsets.UTF_8))) {
// MQTT 3.1.1 has no negative PUBACK: terminate denied connections rather than silently hanging QoS1.
ctx.close();
return;
}
log.debug("=>***客户端[{}],主题[{}],推送消息[{}]", clientId, topicName,
ByteBufUtil.hexDump(publishMessage.content()));
// 以get结尾是模拟客户端数据,只转发消息
if (topicName.endsWith(FastBeeConstant.MQTT.PROPERTY_GET_SIMULATE)) {
callBack(session, publishMessage, clientId);
sendTestToMQ(publishMessage);
} else {
/*获取客户端session*/
Session session = AttributeUtils.getSession(ctx.channel());
/*发布权限校验：web端只能操作自己名下的设备*/
/*推送保留信息*/
pubRetain(publishMessage);
/*响应客户端消息到达Broker*/

20
springboot/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/handler/MqttSubscribe.java
View File

@@ -30,6 +30,8 @@ public class MqttSubscribe implements MqttHandler {
@Resource
private RedisCache redisCache;
@Resource
private com.fastbee.mqtt.acl.MqttAccessControl mqttAccessControl;
@Override
public void handler(ChannelHandlerContext ctx, MqttMessage message) {
@@ -40,6 +42,7 @@ public class MqttSubscribe implements MqttHandler {
public void subscribe(ChannelHandlerContext ctx, MqttSubscribeMessage message) {
/*获取session*/
Session session = AttributeUtils.getSession(ctx.channel());
if (session == null) { ctx.close(); return; }
/*获取客户端订阅的topic列表*/
List<MqttTopicSubscription> topList = message.payload().topicSubscriptions();
/*获取topicName列表*/
@@ -47,10 +50,19 @@ public class MqttSubscribe implements MqttHandler {
log.debug("=>客户端:{},订阅主题:{}", session.getClientId(), JSON.toJSONString(topicNameList));
if (!TopicsUtils.validTopicFilter(topicNameList)) {
log.error("=>订阅主题不合法:{}", JSON.toJSONString(topicNameList));
ctx.close();
return;
}
java.util.List<Integer> grants = new java.util.ArrayList<>();
java.util.List<MqttTopicSubscription> permitted = new java.util.ArrayList<>();
for (MqttTopicSubscription subscription : topList) {
boolean allowed = mqttAccessControl.canSubscribe(session.getPrincipal(), subscription.topicName());
grants.add(allowed ? subscription.qualityOfService().value() : 128);
if (allowed) permitted.add(subscription);
}
/*存储到本地topic缓存*/
topicNameList.forEach(topicName -> {
permitted.forEach(subscription -> {
String topicName = subscription.topicName();
ClientManager.push(topicName, session);
/*累计订阅数*/
redisCache.incr2(FastBeeConstant.REDIS.MESSAGE_SUBSCRIBE_TOTAL,-1L);
@@ -58,10 +70,12 @@ public class MqttSubscribe implements MqttHandler {
/*更新客户端ping*/
ClientManager.updatePing(session.getClientId());
/*应答客户端订阅成功*/
MqttSubAckMessage subAckMessage = MqttMessageUtils.buildSubAckMessage(message);
MqttSubAckMessage subAckMessage = new MqttSubAckMessage(
new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
MqttMessageIdVariableHeader.from(message.variableHeader().messageId()), new MqttSubAckPayload(grants));
ResponseManager.responseMessage(session, subAckMessage, true);
/*客户端订阅了遗留消息主题后，推送遗留消息给客户端*/
topList.forEach(topic -> {
permitted.forEach(topic -> {
retain(topic.topicName(), session, topic.qualityOfService());
});
}

6
springboot/fastbee-server/mqtt-broker/src/main/java/com/fastbee/mqtt/manager/ResponseManager.java
View File

@@ -53,6 +53,12 @@ public class ResponseManager {
*/
public static void publishClients(MqttPublishMessage msg, Session session) {
try {
if (session.getPrincipal() == null || session.getPrincipal().isWeb()
&& !com.fastbee.common.utils.spring.SpringUtils.getBean(com.fastbee.mqtt.acl.MqttAccessControl.class)
.canSubscribe(session.getPrincipal(), msg.variableHeader().topicName())) {
session.getHandlerContext().close();
return;
}
final Channel channel = session.getHandlerContext().channel();
MqttQoS qos = msg.fixedHeader().qosLevel();
ByteBuf sendBuf = msg.content().retainedDuplicate();