From 516f20374c302e1f016f96e09dbe3182e2dae26e Mon Sep 17 00:00:00 2001 From: zhugongping Date: Thu, 23 Jul 2026 08:45:52 +0800 Subject: [PATCH 01/10] =?UTF-8?q?20260723:bug=E6=8F=90=E4=BA=A4=EF=BC=8CNP?= =?UTF-8?q?E=E9=80=A0=E6=88=90=E6=B0=B4=E4=BD=8D=E8=AE=A1=E8=BF=94?= =?UTF-8?q?=E5=B7=A5=EF=BC=8C=E9=80=A0=E6=88=90=E5=85=AC=E5=8F=B8=E5=90=8D?= =?UTF-8?q?=E8=AA=89=E5=8F=97=E6=8D=9F=EF=BC=8C=E9=99=8D=E4=BD=8E=E9=94=99?= =?UTF-8?q?=E8=AF=AF=EF=BC=8C1=E3=80=81=E7=94=A8AOP=E7=9B=91=E6=8E=A7?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E4=B8=8A=E6=8A=A5=E9=93=BE=E8=B7=AF=EF=BC=9B?= =?UTF-8?q?2=E3=80=81=E4=BD=BF=E7=94=A8=E8=87=AA=E5=8A=A8=E5=8C=96?= =?UTF-8?q?=E8=87=AA=E6=B5=8B=E6=A1=86=E6=9E=B6=E5=9B=9E=E5=BD=92=E4=B8=BB?= =?UTF-8?q?=E6=B5=81=E7=A8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../java/com/fastbee/iot/service/impl/DeviceServiceImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java index 49de5187..59373496 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/DeviceServiceImpl.java @@ -720,7 +720,7 @@ public class DeviceServiceImpl implements IDeviceService { params.put("{" + refIdentifier + "}", refValue); } else { // 纵向计算物模型 - if (Integer.valueOf(2).equals(thingModels.getIsCalculated())) { + if (Integer.valueOf(2).equals(thingModels.getIsCalculated()) && Objects.nonNull(cacheIdentifier)) { params.put("{" + refIdentifier + "}", cacheIdentifier.getValue()); } else { params.put("{" + refIdentifier + "}", "0"); From 9d0f392bd73bd72856f836ff62ea43fb1797e2e0 Mon Sep 17 00:00:00 2001 From: zhugongping Date: Thu, 23 Jul 2026 10:43:08 +0800 Subject: [PATCH 02/10] =?UTF-8?q?20260723:sip=E6=94=BB=E5=87=BB=E8=A7=A3?= =?UTF-8?q?=E5=86=B3=EF=BC=8C1.=20=E5=9F=BA=E4=BA=8E=20Redis=20=E7=9A=84IP?= =?UTF-8?q?=E9=80=9F=E7=8E=87=E9=99=90=E5=88=B6=202.=20=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E9=BB=91=E5=90=8D=E5=8D=95=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bin/{ => script}/send_modbus_and_register.py | 0 bin/script/send_sip_register_attack.py | 43 +++++++ .../src/main/resources/application-dev.yml | 6 + .../src/main/resources/application-prod.yml | 6 + .../src/main/resources/application-test.yml | 10 +- .../controller/sip/SipRelationController.java | 15 +++ fastbee-server/sip-server/pom.xml | 6 + .../sip/handler/req/RegisterReqHandler.java | 69 +++++++--- .../sip/service/ISipSecurityService.java | 60 +++++++++ .../service/impl/SipSecurityServiceImpl.java | 121 ++++++++++++++++++ 10 files changed, 314 insertions(+), 22 deletions(-) rename bin/{ => script}/send_modbus_and_register.py (100%) create mode 100644 bin/script/send_sip_register_attack.py create mode 100644 fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/ISipSecurityService.java create mode 100644 fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/impl/SipSecurityServiceImpl.java diff --git a/bin/send_modbus_and_register.py b/bin/script/send_modbus_and_register.py similarity index 100% rename from bin/send_modbus_and_register.py rename to bin/script/send_modbus_and_register.py diff --git a/bin/script/send_sip_register_attack.py b/bin/script/send_sip_register_attack.py new file mode 100644 index 00000000..b0796010 --- /dev/null +++ b/bin/script/send_sip_register_attack.py @@ -0,0 +1,43 @@ +#!/usr/bin/env python3 +"""本地 SIP 攻击模拟""" + +import socket + +def send_sip_register(target_ip, target_port, sip_id, local_ip="127.0.0.1", local_port=50600): + message = f"""REGISTER sip:{target_ip} SIP/2.0\r +Via: SIP/2.0/UDP {local_ip}:{local_port};rport;branch=z9hG4bK-test-{sip_id}\r +From: ;tag=12345\r +To: \r +Call-ID: test-{sip_id}@{local_ip}\r +CSeq: 1 REGISTER\r +Contact: \r +Max-Forwards: 70\r +User-Agent: Python-Test\r +Expires: 3600\r +Content-Length: 0\r +\r +""" + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.settimeout(2) + + try: + sock.sendto(message.encode(), (target_ip, target_port)) + response, _ = sock.recvfrom(4096) + return response.decode() + except socket.timeout: + return "Timeout" + finally: + sock.close() + +IP_POOL = ("1.113.12.33","1.113.12.34","1.113.12.35") + +if __name__ == "__main__": + TARGET_IP = "127.0.0.1" # 本地测试 + TARGET_PORT = 5061 # SIP端口 + + # 模拟攻击：连续发送20个请求 + for i in range(100): + sip_id = str(7000 + i) + response = send_sip_register(TARGET_IP, TARGET_PORT, sip_id) + print(f"[{sip_id}] {response.split(chr(13))[0] if response else 'No response'}") \ No newline at end of file diff --git a/fastbee-admin/src/main/resources/application-dev.yml b/fastbee-admin/src/main/resources/application-dev.yml index 5f9807f0..fe7c6cbc 100644 --- a/fastbee-admin/src/main/resources/application-dev.yml +++ b/fastbee-admin/src/main/resources/application-dev.yml @@ -144,6 +144,12 @@ sip: log: true zlmRecordPath: /opt/media/bin/www mp4_max_second: 30 # 视频录像时长，单位秒 + # SIP安全防护配置 + security: + enabled: true # 是否启用安全防护 + rate-limit: 30 # 每分钟最大请求数 + block-threshold: 50 # 触发封禁的请求次数 + block-duration: 30 # 封禁时长（分钟） # 日志配置 logging: diff --git a/fastbee-admin/src/main/resources/application-prod.yml b/fastbee-admin/src/main/resources/application-prod.yml index c936a670..7cf3f1de 100644 --- a/fastbee-admin/src/main/resources/application-prod.yml +++ b/fastbee-admin/src/main/resources/application-prod.yml @@ -139,6 +139,12 @@ sip: log: false zlmRecordPath: /opt/media/bin/www mp4MaxSecond: 30 # 视频录像时长，单位秒 + # SIP安全防护配置 + security: + enabled: true # 是否启用安全防护 + rate-limit: 10 # 每分钟最大请求数（生产环境更严格） + block-threshold: 20 # 触发封禁的请求次数 + block-duration: 10080 # 封禁时长（分钟）一个星期 # 日志配置 logging: diff --git a/fastbee-admin/src/main/resources/application-test.yml b/fastbee-admin/src/main/resources/application-test.yml index 7157c618..d1b61ac9 100644 --- a/fastbee-admin/src/main/resources/application-test.yml +++ b/fastbee-admin/src/main/resources/application-test.yml @@ -78,10 +78,10 @@ emqx: ApiSecret: tMbf5EEZFb08sut88i4tZscxTc9CE3gaRQewSOTRNvWL # sip 配置 sip: - enabled: false # 是否启用视频监控SIP，true为启用 + enabled: true # 是否启用视频监控SIP，true为启用 ## 本地调试时，绑定网卡局域网IP，设备在同一局域网，设备接入IP填写绑定IP ## 部署服务端时，默认绑定容器IP，设备接入IP填写服务器公网IP - ip: 47.121.180.229 # 容器部署容器IP：47.121.180.229；云服务器部署写公网IP；本地部署127.0.0.1；总原则：能被摄像机访问的地址 + ip: 127.0.0.1 # 容器部署容器IP：47.121.180.229；云服务器部署写公网IP；本地部署127.0.0.1；总原则：能被摄像机访问的地址 port: 5061 # SIP端口(保持默认) domain: 3402000000 # 由省级、市级、区级、基层编号组成 id: 34020000002000000001 # 同上，另外增加编号，(可保持默认) @@ -89,6 +89,12 @@ sip: log: false zlmRecordPath: /opt/media/bin/www mp4MaxSecond: 30 # 视频录像时长，单位秒 + # SIP安全防护配置 + security: + enabled: true # 是否启用安全防护 + rate-limit: 20 # 每分钟最大请求数（生产环境更严格） + block-threshold: 30 # 触发封禁的请求次数 + block-duration: 60 # 封禁时长（分钟） # 日志配置 logging: diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/sip/SipRelationController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/sip/SipRelationController.java index 6219fe2c..156d4a59 100644 --- a/fastbee-open-api/src/main/java/com/fastbee/controller/sip/SipRelationController.java +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/sip/SipRelationController.java @@ -6,12 +6,14 @@ import com.fastbee.common.core.page.TableDataInfo; import com.fastbee.common.utils.poi.ExcelUtil; import com.fastbee.iot.domain.SipRelation; import com.fastbee.iot.service.ISipRelationService; +import com.fastbee.sip.service.ISipSecurityService; import io.swagger.annotations.Api; import io.swagger.annotations.ApiOperation; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.web.bind.annotation.*; +import javax.annotation.Resource; import javax.servlet.http.HttpServletResponse; import java.util.List; @@ -28,6 +30,9 @@ public class SipRelationController extends BaseController { @Autowired private ISipRelationService sipRelationService; + @Resource + private ISipSecurityService sipSecurityService; + /** * 查询监控设备关联列表 */ @@ -98,4 +103,14 @@ public class SipRelationController extends BaseController { public AjaxResult remove(@PathVariable Long[] ids) { return toAjax(sipRelationService.deleteSipRelationByIds(ids)); } + + // ================================= 解禁IP ================================= + + @PreAuthorize("@ss.hasPermi('iot:relation:edit')") + @GetMapping("/unblockIp") + @ApiOperation("删除监控设备关联") + public AjaxResult unblockIp(@RequestParam("ip") String ip) { + sipSecurityService.unblockIp(ip); + return success(); + } } diff --git a/fastbee-server/sip-server/pom.xml b/fastbee-server/sip-server/pom.xml index 6d907ecc..b41feefa 100644 --- a/fastbee-server/sip-server/pom.xml +++ b/fastbee-server/sip-server/pom.xml @@ -63,6 +63,12 @@ jackson-databind + + + com.fastbee + fastbee-common + + diff --git a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/handler/req/RegisterReqHandler.java b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/handler/req/RegisterReqHandler.java index d8ee9b4f..921fc05d 100644 --- a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/handler/req/RegisterReqHandler.java +++ b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/handler/req/RegisterReqHandler.java @@ -16,6 +16,7 @@ import com.fastbee.sip.service.IMqttService; import com.fastbee.sip.service.ISipConfigService; import com.fastbee.sip.service.ISipDeviceChannelService; import com.fastbee.sip.service.ISipDeviceService; +import com.fastbee.sip.service.ISipSecurityService; import com.fastbee.sip.util.DigestAuthUtil; import gov.nist.javax.sip.address.AddressImpl; import gov.nist.javax.sip.address.SipUri; @@ -23,11 +24,11 @@ import gov.nist.javax.sip.header.Expires; import gov.nist.javax.sip.header.SIPDateHeader; import gov.nist.javax.sip.message.SIPRequest; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.springframework.util.ObjectUtils; -import org.springframework.util.StringUtils; import javax.sip.InvalidArgumentException; import javax.sip.RequestEvent; @@ -70,42 +71,70 @@ public class RegisterReqHandler extends ReqAbstractHandler implements Initializi @Autowired private SIPSender sipSender; + @Autowired + private ISipSecurityService sipSecurityService; + @Override public void processMsg(RequestEvent evt) { try { - log.info("收到注册请求，开始处理"); Request request = evt.getRequest(); - SIPRequest Sipreq = (SIPRequest) evt.getRequest(); + SIPRequest sipReq = (SIPRequest) evt.getRequest(); Response response; - // 注册标志 0：未携带授权头或者密码错误 1：注册成功 2：注销成功 - int registerFlag; + + // ========== 安全防护检查 ========== + // 提取客户端IP + ViaHeader viaHeader = (ViaHeader) request.getHeader(ViaHeader.NAME); + String clientIp = viaHeader.getReceived(); + + if (StringUtils.isEmpty(clientIp)) { + clientIp = viaHeader.getHost(); + } + + // 检查IP是否在黑名单中 + if (sipSecurityService.isBlocked(clientIp)) { + log.debug("[SIP安全] IP {} 已封禁，拒绝REGISTER请求", clientIp); + return; // 直接丢弃，不响应 + } + + // 检查速率限制 FromHeader fromHeader = (FromHeader) request.getHeader(FromHeader.NAME); AddressImpl address = (AddressImpl) fromHeader.getAddress(); SipUri uri = (SipUri) address.getURI(); String sipId = uri.getUser(); + + if (sipSecurityService.checkAndRecord(clientIp, sipId)) { + log.debug("[SIP安全] IP {} 请求超限，拒绝请求，sipId: {}", clientIp, sipId); + return; // 直接丢弃，让攻击者超时等待 + } + + log.debug("收到注册请求，开始处理，sipId: {}, IP: {}", sipId, clientIp); + // ========== 安全防护检查结束 ========== + + // 注册标志 0：未携带授权头或者密码错误 1：注册成功 2：注销成功 + int registerFlag; //取默认Sip配置 SipConfig sipConfig = sipConfigService.selectSipConfigBydeviceSipId(sipId); if (sipConfig != null) { - AuthorizationHeader authorhead = (AuthorizationHeader) request.getHeader(AuthorizationHeader.NAME); + AuthorizationHeader authorHead = (AuthorizationHeader) request.getHeader(AuthorizationHeader.NAME); // 校验密码是否正确 - if (authorhead == null && !ObjectUtils.isEmpty(sipConfig.getPassword())) { - log.info("未携带授权头 回复401，sipId：{}", sipId); + if (authorHead == null && !ObjectUtils.isEmpty(sipConfig.getPassword())) { + log.debug("未携带授权头 回复401，sipId：{}，IP：{}", sipId, clientIp); response = getMessageFactory().createResponse(Response.UNAUTHORIZED, request); new DigestAuthUtil().generateChallenge(getHeaderFactory(), response, sipConfig.getDomain()); - sipSender.transmitRequest(Sipreq.getLocalAddress().getHostAddress(), response); + sipSender.transmitRequest(sipReq.getLocalAddress().getHostAddress(), response); return; } - boolean pcheck = new DigestAuthUtil().doAuthenticatePlainTextPassword(request, + boolean preCheck = new DigestAuthUtil().doAuthenticatePlainTextPassword(request, sipConfig.getPassword()); - boolean syscheck = new DigestAuthUtil().doAuthenticatePlainTextPassword(request, + boolean sysCheck = new DigestAuthUtil().doAuthenticatePlainTextPassword(request, sysSipConfig.getPassword()); - if (!pcheck && !syscheck) { + if (!preCheck && !sysCheck) { // 注册失败 response = getMessageFactory().createResponse(Response.FORBIDDEN, request); response.setReasonPhrase("wrong password"); - log.info("[注册请求] 密码/SIP服务器ID错误, 回复403 sipId：" + sipId); + log.info("[注册请求] 密码/SIP服务器ID错误, 回复403 sipId：{}", sipId); getServerTransaction(evt).sendResponse(response); return; } @@ -122,7 +151,7 @@ public class RegisterReqHandler extends ReqAbstractHandler implements Initializi response.addHeader(request.getHeader(ContactHeader.NAME)); // 添加Expires头 response.addHeader(request.getExpires()); - ViaHeader viaHeader = (ViaHeader) request.getHeader(ViaHeader.NAME); + //ViaHeader viaHeader = (ViaHeader) request.getHeader(ViaHeader.NAME); String received = viaHeader.getReceived(); int rPort = viaHeader.getRPort(); // 本地模拟设备 received 为空 rPort 为 -1 @@ -153,20 +182,20 @@ public class RegisterReqHandler extends ReqAbstractHandler implements Initializi } device.setTransport(isTcp ? "TCP" : "UDP"); } - sipSender.transmitRequest(Sipreq.getLocalAddress().getHostAddress(), response); + sipSender.transmitRequest(sipReq.getLocalAddress().getHostAddress(), response); // 注册成功 if (registerFlag == 1) { - log.info("注册成功! sipId:" + device.getDeviceSipId()); + log.info("注册成功! sipId:{}", device.getDeviceSipId()); device.setRegistertime(DateUtils.getNowDate()); sipDeviceService.updateDevice(device); List channels = sipDeviceChannelService.selectSipDeviceChannelByDeviceSipId(device.getDeviceSipId()); - if (channels.size() > 0) { + if (!channels.isEmpty()) { Device iotdev = deviceService.selectDeviceBySerialNumber(device.getDeviceSipId()); if (iotdev == null) { //添加到统一设备管理 默认添加到admin账号下 int result = deviceService.insertDeviceAuto(device.getDeviceSipId(), 1L, sipConfig.getProductId()); if (result == 1) { - log.info("-----------sip设备认证成功,并自动添加设备到系统，SipId:" + device.getDeviceSipId() + "---------------"); + log.info("-----------sip设备认证成功,并自动添加设备到系统，SipId:{}---------------", device.getDeviceSipId()); } iotdev = new Device(); } @@ -177,10 +206,10 @@ public class RegisterReqHandler extends ReqAbstractHandler implements Initializi // 重新注册更新设备和通道，以免设备替换或更新后信息无法更新 onRegister(device); } else { - log.info("未找到改设备! deviceId:" + device.getDeviceId()); + log.info("未找到改设备! deviceId:{}", device.getDeviceId()); } } else if (registerFlag == 2) { - log.info("注销成功! deviceId:" + device.getDeviceId()); + log.info("注销成功! deviceId:{}", device.getDeviceId()); mqttService.publishStatus(device, 4); } diff --git a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/ISipSecurityService.java b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/ISipSecurityService.java new file mode 100644 index 00000000..4ccf9f16 --- /dev/null +++ b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/ISipSecurityService.java @@ -0,0 +1,60 @@ +package com.fastbee.sip.service; + +/** + * SIP安全防护服务接口 + * + * @author George + * @date 2026/7/23 + */ +public interface ISipSecurityService { + + /** + * Redis key前缀 - IP封禁黑名单 + */ + String BLOCK_PREFIX = "sip:block:"; + + /** + * Redis key前缀 - IP请求计数 + */ + String COUNT_PREFIX = "sip:count:"; + + /** + * 检查IP是否被封禁 + * + * @param ip 客户端IP + * @return true=已封禁, false=未封禁 + */ + boolean isBlocked(String ip); + + /** + * 记录请求并检查是否超过速率限制 + * + * @param ip 客户端IP + * @param sipId SIP设备ID + * @return true=超过限制触发封禁, false=正常 + */ + boolean checkAndRecord(String ip, String sipId); + + /** + * 手动封禁IP + * + * @param ip 客户端IP + * @param minutes 封禁时长（分钟） + */ + void blockIp(String ip, int minutes); + + /** + * 解除IP封禁 + * + * @param ip 客户端IP + */ + void unblockIp(String ip); + + /** + * 获取IP请求计数 + * + * @param ip 客户端IP + * @return 当前计数 + */ + long getRequestCount(String ip); +} \ No newline at end of file diff --git a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/impl/SipSecurityServiceImpl.java b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/impl/SipSecurityServiceImpl.java new file mode 100644 index 00000000..45e7c678 --- /dev/null +++ b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/impl/SipSecurityServiceImpl.java @@ -0,0 +1,121 @@ +package com.fastbee.sip.service.impl; + +import com.fastbee.common.core.redis.RedisCache; +import com.fastbee.sip.service.ISipSecurityService; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; + +import java.util.concurrent.TimeUnit; + +/** + * SIP安全防护服务实现 + * 基于Redis实现IP速率限制和自动黑名单 + * + * @author George + * @date 2026/7/23 + */ +@Slf4j +@Service +public class SipSecurityServiceImpl implements ISipSecurityService { + + @Autowired + private RedisCache redisCache; + + /** + * 速率限制：每分钟最大请求数 + */ + @Value("${sip.security.rate-limit:20}") + private int rateLimit; + + /** + * 触发封禁的失败次数阈值 + */ + @Value("${sip.security.block-threshold:30}") + private int blockThreshold; + + /** + * 默认封禁时长（分钟） + */ + @Value("${sip.security.block-duration:120}") + private int blockDuration; + + @Override + public boolean isBlocked(String ip) { + if (ip == null || ip.isEmpty()) { + return false; + } + String key = BLOCK_PREFIX + ip; + return Boolean.TRUE.equals(redisCache.hasKey(key)); + } + + @Override + public boolean checkAndRecord(String ip, String sipId) { + if (ip == null || ip.isEmpty()) { + return false; + } + + // 1. 检查是否已封禁 + if (isBlocked(ip)) { + log.warn("[SIP安全] IP {} 已在黑名单中，拒绝请求", ip); + return true; + } + + // 2. 记录请求计数（1分钟窗口） + String countKey = COUNT_PREFIX + ip; + long count = redisCache.incr(countKey, 1); + + // 首次访问设置过期时间 + if (count == 1) { + redisCache.expire(countKey, 60, TimeUnit.SECONDS); + } + + // 3. 检查是否超过速率限制 + if (count > rateLimit) { + log.warn("[SIP安全] IP {} 请求频率超限，当前: {}/min，限制: {}/min", ip, count, rateLimit); + + // 超过阈值则自动封禁 + if (count >= blockThreshold) { + blockIp(ip, blockDuration); + log.warn("[SIP安全] IP {} 触发自动封禁，时长: {}分钟", ip, blockDuration); + } + return true; + } + + return false; + } + + @Override + public void blockIp(String ip, int minutes) { + if (ip == null || ip.isEmpty()) { + return; + } + String key = BLOCK_PREFIX + ip; + redisCache.setCacheObject(key, System.currentTimeMillis(), minutes, TimeUnit.MINUTES); + log.info("[SIP安全] 封禁IP: {}，时长: {}分钟", ip, minutes); + } + + @Override + public void unblockIp(String ip) { + if (ip == null || ip.isEmpty()) { + return; + } + String key = BLOCK_PREFIX + ip; + redisCache.deleteObject(key); + log.info("[SIP安全] 解封IP: {}", ip); + } + + @Override + public long getRequestCount(String ip) { + if (ip == null || ip.isEmpty()) { + return 0; + } + String countKey = COUNT_PREFIX + ip; + Object obj = redisCache.getCacheObject(countKey); + if (obj == null) { + return 0; + } + return Long.parseLong(obj.toString()); + } +} \ No newline at end of file From ced3257a7999eb6cdf41cb8129e488607da6f64a Mon Sep 17 00:00:00 2001 From: zhugongping Date: Sun, 26 Jul 2026 09:38:34 +0800 Subject: [PATCH 03/10] =?UTF-8?q?20260726:mqtt=E5=AF=B9=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .claude/settings.local.json | 3 +- bin/script/send_modbus_and_register.py | 3 + docs/scripts/data_bridge_script.groovy | 138 ++++++++++++++++++ .../fastbee/iot/ruleEngine/MqttBridge.java | 25 +++- .../iot/ruleEngine/MqttClientFactory.java | 122 ++++++++++++---- .../iot/service/impl/BridgeServiceImpl.java | 19 ++- 6 files changed, 268 insertions(+), 42 deletions(-) create mode 100644 docs/scripts/data_bridge_script.groovy diff --git a/.claude/settings.local.json b/.claude/settings.local.json index ee3a2356..fa977bd3 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -27,7 +27,8 @@ "Bash(nvm use:*)", "Bash(npm run:*)", "mcp__filesystem__read_text_file", - "Bash(sips:*)" + "Bash(sips:*)", + "Bash(mysql -h localhost -u root -pfastbee fastbee -e \"SELECT id, name, config_json FROM bridge WHERE id = 9;\")" ] } } diff --git a/bin/script/send_modbus_and_register.py b/bin/script/send_modbus_and_register.py index b9d00023..138d2c46 100644 --- a/bin/script/send_modbus_and_register.py +++ b/bin/script/send_modbus_and_register.py @@ -12,6 +12,9 @@ MODBUS_DATA_HEX = ( "01030400110018AA3C" ) +# 重要-四川远德MQTT推送 +# "7e803042593632517e", +# "01033041011b7b0000000041eaafb742d19bd74513d545450dd9a4450ec53d00000000420ccf64420c74a9420c74a9c2d200009d99" # 深部位移计 # "7e80DD33D43DFC5B7e", # "0103204135317841ab8e39bf27bccd3f1fb19ac2b231bbc0b6f82040ae3207c3f9f80efa23", diff --git a/docs/scripts/data_bridge_script.groovy b/docs/scripts/data_bridge_script.groovy new file mode 100644 index 00000000..b056fa80 --- /dev/null +++ b/docs/scripts/data_bridge_script.groovy @@ -0,0 +1,138 @@ +// 数据桥接脚本 - FastBee物模型格式转第三方平台格式 +// 输入格式：[{"message":[{"id":"物模型标识","value":"值"...}],"sources":"..."}] +// 输出格式：{"设备编号":{"节点标识":{"字段":值}}} + +import cn.hutool.json.JSONArray; +import cn.hutool.json.JSONObject; +import cn.hutool.json.JSONUtil; +import cn.hutool.core.util.NumberUtil; + +// 在Groovy脚本中，msgContext对象直接可用 +// 获取上下文信息 +String topic = msgContext.getTopic(); +String payload = msgContext.getPayload(); +String serialNumber = msgContext.getSerialNumber(); + +msgContext.logger.info("数据桥接 - 接收到FastBee数据: 设备={}, payload={}", serialNumber, payload); + +// 数据转换处理 +try { + // 直接解析为JSONArray（实际格式是对象数组） + JSONArray inputArray = JSONUtil.parseArray(payload); + + if (inputArray == null || inputArray.isEmpty()) { + msgContext.logger.warn("数据桥接 - 输入数据为空"); + return; + } + + msgContext.logger.info("数据桥接 - 解析到 {} 个物模型项", inputArray.size()); + + // 准备输出数据结构 + JSONObject outputData = new JSONObject(); + JSONObject deviceData = new JSONObject(); + + // 初始化两个节点容器 + JSONObject L3_MS_1 = new JSONObject(); // 频率、温度、压力 + JSONObject S1_ZT_1 = new JSONObject(); // 经纬度、电压 + + // 遍历物模型数据，进行字段映射 + for (int i = 0; i < inputArray.size(); i++) { + JSONObject item = inputArray.getJSONObject(i); + String identifier = item.getStr("id"); + Object value = item.get("value"); + + if (identifier == null || value == null) { + msgContext.logger.debug("跳过空项: {}", item); + continue; + } + + msgContext.logger.debug("处理物模型项: id={}, value={}", identifier, value); + + switch (identifier) { + // 字段映射 - 频率通道 -> L3_MS_1 + case "F_CH1": + L3_MS_1.set("hz1", convertValue(value)) + break + case "F_CH2": + L3_MS_1.set("hz2", convertValue(value)) + break + case "F_CH3": + L3_MS_1.set("hz3", convertValue(value)) + break + // 字段映射 - 温度通道 -> L3_MS_1 + case "T_CH1": + L3_MS_1.set("wd1", convertValue(value)) + break + case "T_CH2": + L3_MS_1.set("wd2", convertValue(value)) + break + case "T_CH3": + L3_MS_1.set("wd3", convertValue(value)) + break + // 字段映射 - 压力 -> L3_MS_1 + case "KN": + L3_MS_1.set("value", convertValue(value)) + break + // 字段映射 - 经纬度、电压 -> S1_ZT_1 + case "Lont": + S1_ZT_1.set("lon", convertValue(value)) + break + case "Lat": + S1_ZT_1.set("lat", convertValue(value)) + break + case "Vol": + S1_ZT_1.set("ext_power_volt", convertValue(value)) + break + default: + msgContext.logger.debug("未映射的物模型标识: {}", identifier) + break; + } + } + + // 组装输出数据结构 + // 只有非空节点才添加到输出 + if (!L3_MS_1.isEmpty()) { + deviceData.set("L3_MS_1", L3_MS_1); + msgContext.logger.info("L3_MS_1 节点字段数: {}", L3_MS_1.size()); + } + if (!S1_ZT_1.isEmpty()) { + deviceData.set("S1_ZT_1", S1_ZT_1); + msgContext.logger.info("S1_ZT_1 节点字段数: {}", S1_ZT_1.size()); + } + + // 使用上下文中的设备编号 + if (serialNumber == null) { + throw new IllegalArgumentException("数据桥接失败：serialNumber 不能为空"); + } + outputData.set(serialNumber, deviceData); + + // 构建新的payload + String newPayload = outputData.toString(); + + msgContext.logger.info("数据桥接 - 转换完成: 设备={}, 输出payload={}", serialNumber, newPayload); + + // 返回新的数据 + msgContext.setTopic(topic); + msgContext.setPayload(newPayload); + +} catch (Exception e) { + msgContext.logger.error("数据桥接 - 转换异常: ", e); + // 转换失败时保持原数据 + msgContext.setTopic(topic); + msgContext.setPayload(payload); +} +// 执行Action动作参数(脚本由系统自动生成) +msgContext.setData("mqttBridgeID", 3); + +// 辅助方法：转换值为数字类型（保持精度） +Object convertValue(Object value) { + if (value == null) { + return null; + } + try { + String strValue = String.valueOf(value); + return Double.parseDouble(strValue); + } catch (Exception e) { + return value; + } +} \ No newline at end of file diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttBridge.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttBridge.java index 588a19be..214c66b4 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttBridge.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttBridge.java @@ -9,6 +9,7 @@ import com.fastbee.ruleEngine.context.MsgContext; import com.yomahub.liteflow.annotation.LiteflowComponent; import com.yomahub.liteflow.core.NodeComponent; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.lang3.StringUtils; import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; @@ -35,16 +36,28 @@ public class MqttBridge extends NodeComponent { if (bridge != null) { MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class); MqttAsyncClient client = MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), null); + + if (client == null) { + log.error("=>MQTT客户端创建失败: bridgeId={}", id); + return; + } + + // 优先使用bridge.route，如果为空再使用context.topic + String publishTopic = StringUtils.isNotBlank(bridge.getRoute()) + ? bridge.getRoute() + : cxt.getTopic(); + MqttMessage message = new MqttMessage(); message.setPayload(cxt.getPayload().getBytes()); + try { - client.publish(cxt.getTopic(), message); - log.info("=>桥接发布主题成功 topic={},message={}", cxt.getTopic(), message); + client.publish(publishTopic, message); + log.info("=>桥接发布主题成功: bridgeId={}, topic={}, payload={}", id, publishTopic, cxt.getPayload()); } catch (MqttException e) { - log.error("msg:" + e.getMessage()); - log.error("cause:" + e.getCause()); - log.error("reason:" + e.getReasonCode()); - log.error("=>桥接发布主题时发生错误 topic={},message={}", cxt.getTopic(), e.getMessage()); + log.error("msg:{}", e.getMessage()); + log.error("cause:{}", String.valueOf(e.getCause())); + log.error("reason:{}", e.getReasonCode()); + log.error("=>桥接发布主题失败: bridgeId={}, topic={}, error={}", id, publishTopic, e.getMessage()); } } } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java index d6bd83b9..f159920a 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/ruleEngine/MqttClientFactory.java @@ -14,50 +14,119 @@ public class MqttClientFactory { private static final Map CLIENT_CACHE = new ConcurrentHashMap<>(); private static final Map CLIENT_OPTIONS_CACHE = new ConcurrentHashMap<>(); + // 每个key独立的锁对象，避免同一bridge并发创建/重连时产生连接泄漏 + private static final Map KEY_LOCKS = new ConcurrentHashMap<>(); public static MqttAsyncClient instance(MqttClientConfig config, MqttCallbackonSuccess callback) throws MqttException { if (config.getKey() == null) { + log.error("=>MQTT配置key为空"); return null; } - MqttAsyncClient client = CLIENT_CACHE.get(config.getKey()); - if (client == null) { - log.debug("=>mqtt客户端：{} 不存在", config.getKey()); - client = instanceNew(config, callback); - CLIENT_CACHE.put(config.getKey(), client); - } else { - log.debug("=>mqtt客户端：{} 存在", config.getKey()); - MqttConnectOptions options = CLIENT_OPTIONS_CACHE.get(config.getKey()); - if (options != null && Objects.equals(options.getUserName(), config.getUsername()) && new String(options.getPassword()).equals(config.getPassword())) { - if (!client.isConnected()) { - client.reconnect(); + + Object lock = KEY_LOCKS.computeIfAbsent(config.getKey(), k -> new Object()); + synchronized (lock) { + MqttAsyncClient client = CLIENT_CACHE.get(config.getKey()); + if (client == null) { + log.info("=>MQTT客户端不存在，创建新客户端: {}", config.getKey()); + client = instanceNew(config, callback); + if (client != null && client.isConnected()) { + CLIENT_CACHE.put(config.getKey(), client); } } else { - log.debug("=>mqtt客户端：{} 配置不一致，重建", config.getKey()); - client = instanceNew(config, callback); - CLIENT_CACHE.put(config.getKey(), client); + log.debug("=>MQTT客户端已存在: {}", config.getKey()); + + // 检查配置是否一致 + MqttConnectOptions options = CLIENT_OPTIONS_CACHE.get(config.getKey()); + boolean configChanged = options == null + || !Objects.equals(options.getUserName(), config.getUsername()) + || !passwordEquals(options.getPassword(), config.getPassword()); + + if (configChanged) { + log.warn("=>MQTT配置已变化，重新创建客户端: {}", config.getKey()); + // 关闭旧连接 + try { + if (client.isConnected()) { + client.disconnect(); + } + client.close(); + } catch (MqttException e) { + log.warn("=>关闭旧客户端失败: {}", e.getMessage()); + } + // 创建新客户端 + client = instanceNew(config, callback); + if (client != null && client.isConnected()) { + CLIENT_CACHE.put(config.getKey(), client); + } + } else if (!client.isConnected()) { + log.warn("=>MQTT客户端未连接，尝试重连: {}", config.getKey()); + try { + client.reconnect(); + // 等待重连完成（最多10秒） + int waitCount = 0; + while (!client.isConnected() && waitCount < 100) { + Thread.sleep(100); + waitCount++; + } + if (client.isConnected()) { + log.info("=>MQTT客户端重连成功: {}", config.getKey()); + } else { + log.error("=>MQTT客户端重连超时: {}", config.getKey()); + return null; + } + } catch (Exception e) { + log.error("=>MQTT客户端重连失败: {}", e.getMessage()); + return null; + } + } } + return client; } - return client; + } + + private static boolean passwordEquals(char[] cached, String configured) { + String cachedStr = cached == null ? null : new String(cached); + return Objects.equals(cachedStr, configured); } public static MqttAsyncClient instanceNew(MqttClientConfig config, MqttCallbackonSuccess callback) { - MqttAsyncClient client = null; - log.debug("=>mqtt客户端config：{}", config); + MqttAsyncClient client; + log.info("=>创建MQTT客户端: hostUrl={}, clientId={}, username={}", config.getHostUrl(), config.getClientId(), config.getUsername()); try { client = new MqttAsyncClient(config.getHostUrl(), config.getClientId(), new MemoryPersistence()); + MqttConnectOptions options = new MqttConnectOptions(); options.setUserName(config.getUsername()); - options.setPassword(config.getPassword().toCharArray()); - options.setCleanSession(false); - options.setAutomaticReconnect(true); + if (config.getPassword() != null) { + options.setPassword(config.getPassword().toCharArray()); + } + + // 关键配置：避免会话冲突和超时 + options.setCleanSession(true); // 清除会话，避免ClientId冲突 + options.setConnectionTimeout(30); // 连接超时30秒 + options.setKeepAliveInterval(60); // 保活间隔60秒 + options.setAutomaticReconnect(true); // 自动重连 + options.setMaxReconnectDelay(60000); // 最大重连延迟60秒 + if (callback != null) { callback.setClient(client); } + + log.debug("=>开始连接MQTT Broker..."); IMqttToken token = client.connect(options, null, callback); - token.waitForCompletion(); + + // 等待连接完成，设置30秒超时 + token.waitForCompletion(30000); + log.info("=>MQTT客户端连接成功: {}", config.getKey()); + CLIENT_OPTIONS_CACHE.put(config.getKey(), options); + } catch (MqttException e) { - log.error("=>mqtt客户端创建错误，原因：{}", e.getMessage()); + log.error("=>MQTT客户端创建错误: reason={}, msg={}", e.getReasonCode(), e.getMessage()); + // 连接失败时返回null，不要返回未连接的客户端 + return null; + } catch (Exception e) { + log.error("=>MQTT客户端创建异常: {}", e.getMessage(), e); + return null; } return client; } @@ -70,7 +139,7 @@ public class MqttClientFactory { client.subscribe(topic, 0, new MessageArrivedHandler(topic)); } } catch (MqttException e) { - e.printStackTrace(); + log.error("=>MQTT订阅失败: key={}, topic={}, error={}", key, topic, e.getMessage()); } } @@ -78,7 +147,7 @@ public class MqttClientFactory { try { client.subscribe(topic, 0, new MessageArrivedHandler(topic)); } catch (MqttException e) { - e.printStackTrace(); + log.error("=>MQTT订阅失败: topic={}, error={}", topic, e.getMessage()); } } @@ -89,7 +158,7 @@ public class MqttClientFactory { client.unsubscribe(topic); } } catch (MqttException e) { - e.printStackTrace(); + log.error("=>MQTT取消订阅失败: key={}, topic={}, error={}", key, topic, e.getMessage()); } } @@ -104,6 +173,9 @@ public class MqttClientFactory { CLIENT_CACHE.remove(key); } catch (MqttException e) { throw new RuntimeException(e); + } finally { + CLIENT_OPTIONS_CACHE.remove(key); + KEY_LOCKS.remove(key); } } } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/BridgeServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/BridgeServiceImpl.java index a714fc56..0544a443 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/BridgeServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/service/impl/BridgeServiceImpl.java @@ -18,6 +18,7 @@ import com.fastbee.http.service.SuccessCondition; import com.fastbee.iot.domain.HttpClient; import com.fastbee.iot.domain.MqttClient; import com.fastbee.iot.domain.MultipleDataSource; +import com.fastbee.iot.model.MqttClientConfig; import com.fastbee.iot.ruleEngine.DataSourceService; import com.fastbee.iot.ruleEngine.MqttCallbackonSuccess; import com.fastbee.iot.ruleEngine.MqttClientFactory; @@ -65,8 +66,7 @@ public class BridgeServiceImpl extends ServiceImpl impleme MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class); if (config != null) { try { - MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), - new MqttCallbackonSuccess(bridge)); + MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), new MqttCallbackonSuccess(bridge)); } catch (MqttException e) { throw new RuntimeException(e); } @@ -146,8 +146,7 @@ public class BridgeServiceImpl extends ServiceImpl impleme MqttClient config = JSON.parseObject(add.getConfigJson(), MqttClient.class); if (config != null) { try { - MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), - new MqttCallbackonSuccess(add)); + MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), new MqttCallbackonSuccess(add)); } catch (MqttException e) { throw new RuntimeException(e); } @@ -217,9 +216,9 @@ public class BridgeServiceImpl extends ServiceImpl impleme } else if (bridge.getType() == 4) { MqttClient config = JSON.parseObject(bridge.getConfigJson(), MqttClient.class); if (config != null) { + MqttClientConfig clientConfig = mqttClientService.buildmqttclientconfig(config); try { - MqttAsyncClient client = MqttClientFactory.instance(mqttClientService.buildmqttclientconfig(config), - new MqttCallbackonSuccess(bridge)); + MqttAsyncClient client = MqttClientFactory.instance(clientConfig, new MqttCallbackonSuccess(bridge)); if (client != null && client.isConnected()) { bridge.setStatus("1"); result.set(1); @@ -227,11 +226,11 @@ public class BridgeServiceImpl extends ServiceImpl impleme bridge.setStatus("0"); result.set(0); } - if (client != null) { - client.disconnect().waitForCompletion(); - } } catch (MqttException e) { - log.error("=>mqtt客户端创建错误"); + log.error("=>MQTT客户端创建错误"); + } finally { + // 测试连接仅用于校验可用性，不长期占用缓存，避免下次真实发布时命中一个已断开的客户端 + MqttClientFactory.remove(clientConfig.getKey()); } } } else if (bridge.getType() == 5) { From be2e868ef5c72f426a2ff29722f5e8dda89d8162 Mon Sep 17 00:00:00 2001 From: zhugongping Date: Sun, 26 Jul 2026 10:00:09 +0800 Subject: [PATCH 04/10] =?UTF-8?q?20260726:=E5=9C=A8=E6=AC=A1=E8=B0=83?= =?UTF-8?q?=E4=B8=A5=E6=8E=A7IP=E6=94=BB=E5=87=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- fastbee-admin/src/main/resources/application-prod.yml | 4 ++-- .../com/fastbee/sip/service/impl/SipSecurityServiceImpl.java | 5 +++-- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/fastbee-admin/src/main/resources/application-prod.yml b/fastbee-admin/src/main/resources/application-prod.yml index 7cf3f1de..1fda0c71 100644 --- a/fastbee-admin/src/main/resources/application-prod.yml +++ b/fastbee-admin/src/main/resources/application-prod.yml @@ -142,8 +142,8 @@ sip: # SIP安全防护配置 security: enabled: true # 是否启用安全防护 - rate-limit: 10 # 每分钟最大请求数（生产环境更严格） - block-threshold: 20 # 触发封禁的请求次数 + rate-limit: 5 # 每分钟最大请求数（生产环境更严格） + block-threshold: 10 # 触发封禁的请求次数 block-duration: 10080 # 封禁时长（分钟）一个星期 # 日志配置 diff --git a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/impl/SipSecurityServiceImpl.java b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/impl/SipSecurityServiceImpl.java index 45e7c678..0560fc90 100644 --- a/fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/impl/SipSecurityServiceImpl.java +++ b/fastbee-server/sip-server/src/main/java/com/fastbee/sip/service/impl/SipSecurityServiceImpl.java @@ -62,13 +62,14 @@ public class SipSecurityServiceImpl implements ISipSecurityService { return true; } - // 2. 记录请求计数（1分钟窗口） + // 2. 记录请求计数（2分钟窗口） + // 2分钟5次请求之后就拒绝，10次请求后就封禁 String countKey = COUNT_PREFIX + ip; long count = redisCache.incr(countKey, 1); // 首次访问设置过期时间 if (count == 1) { - redisCache.expire(countKey, 60, TimeUnit.SECONDS); + redisCache.expire(countKey, 120, TimeUnit.SECONDS); } // 3. 检查是否超过速率限制 From 7877a3ccec989726907c9204199c735c56886e2e Mon Sep 17 00:00:00 2001 From: zhugongping Date: Tue, 28 Jul 2026 08:49:41 +0800 Subject: [PATCH 05/10] =?UTF-8?q?20260728:=E5=B9=B3=E5=8F=B0=E5=AF=B9?= =?UTF-8?q?=E6=8E=A5=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .claude/settings.local.json | 4 +- .../common/constant/FastBeeConstant.java | 5 + .../common/core/redis/RedisKeyBuilder.java | 4 + .../platform/PlatformInstanceController.java | 9 +- .../domain/PlatformDeviceMapping.java | 6 + .../iot/platform/flow/FlowContext.java | 13 ++ .../platform/flow/PlatformFlowExecutor.java | 84 +++++---- .../service/IPlatformInstanceService.java | 6 + .../PlatformDeviceMappingServiceImpl.java | 5 + .../impl/PlatformInstanceServiceImpl.java | 82 ++++++++- .../platform/PlatformDeviceMappingMapper.xml | 46 ++--- sql/platform-dcp-init-data.sql | 119 +++++++++--- sql/platform-integration.sql | 8 +- yack-web/src/api/platform/instance.js | 4 + .../platform/components/InterfaceDrawer.vue | 4 +- .../views/platform/device-mapping/index.vue | 172 ++++++++++++++++-- .../src/views/platform/instance/index.vue | 10 +- 17 files changed, 467 insertions(+), 114 deletions(-) diff --git a/.claude/settings.local.json b/.claude/settings.local.json index fa977bd3..683dd7bd 100644 --- a/.claude/settings.local.json +++ b/.claude/settings.local.json @@ -28,7 +28,9 @@ "Bash(npm run:*)", "mcp__filesystem__read_text_file", "Bash(sips:*)", - "Bash(mysql -h localhost -u root -pfastbee fastbee -e \"SELECT id, name, config_json FROM bridge WHERE id = 9;\")" + "Bash(mysql -h localhost -u root -pfastbee fastbee -e \"SELECT id, name, config_json FROM bridge WHERE id = 9;\")", + "Bash(mysql:*)", + "Bash(git:*)" ] } } diff --git a/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java b/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java index 48d88c51..d60c0b3a 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java +++ b/fastbee-common/src/main/java/com/fastbee/common/constant/FastBeeConstant.java @@ -299,6 +299,11 @@ public interface FastBeeConstant { */ String DATA_REPAIR_CONFIG_KEY = "data:repair:config"; + /** + * 平台对接KEYS + * DCP平台项目列表缓存key + */ + String DCP_PROJECT_INFO_KEY = "platform:dcp:projectInfoKey:"; } diff --git a/fastbee-common/src/main/java/com/fastbee/common/core/redis/RedisKeyBuilder.java b/fastbee-common/src/main/java/com/fastbee/common/core/redis/RedisKeyBuilder.java index 91114794..a5756857 100644 --- a/fastbee-common/src/main/java/com/fastbee/common/core/redis/RedisKeyBuilder.java +++ b/fastbee-common/src/main/java/com/fastbee/common/core/redis/RedisKeyBuilder.java @@ -161,6 +161,10 @@ public class RedisKeyBuilder { return FastBeeConstant.REDIS.MODBUS_TCP_RUNTIME + serialNumber; } + public static String buildDCPProjectInfoKey(String platformUserId) { + return FastBeeConstant.REDIS.DCP_PROJECT_INFO_KEY + platformUserId; + } + /** * 设备OTA升级实时数据 diff --git a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformInstanceController.java b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformInstanceController.java index 55923fa8..9fdf5bc7 100644 --- a/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformInstanceController.java +++ b/fastbee-open-api/src/main/java/com/fastbee/controller/platform/PlatformInstanceController.java @@ -68,9 +68,16 @@ public class PlatformInstanceController extends BaseController { @PreAuthorize("@ss.hasPermi('platform:instance:edit')") @PostMapping("/{id}/initialize") - @ApiOperation("初始化平台（执行初始化流程，获取userId/projectId等）") + @ApiOperation("初始化平台（执行初始化流程，获取userId）") public AjaxResult initialize(@PathVariable Long id) { instanceService.initialize(id); return success("初始化成功"); } + + @PreAuthorize("@ss.hasAnyPermi('platform:instance:list,platform:deviceMapping:list,platform:deviceMapping:add,platform:deviceMapping:edit')") + @GetMapping("/{id}/projects") + @ApiOperation("查询平台项目列表") + public AjaxResult getProjects(@PathVariable Long id) { + return success(instanceService.getProjectsByPlatformId(id)); + } } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java index fc3d64ab..7b592d15 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/domain/PlatformDeviceMapping.java @@ -28,6 +28,12 @@ public class PlatformDeviceMapping implements Serializable { @ApiModelProperty("所属平台ID") private Long platformId; + @ApiModelProperty("DCP项目ID,绑定页面用户自己选择") + private String platformProjectId; + + @ApiModelProperty("DCP项目名称") + private String platformProjectName; + @ApiModelProperty("FastBee设备ID") private Long deviceId; diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java index 087a25a2..0a992b79 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/FlowContext.java @@ -1,9 +1,11 @@ package com.fastbee.iot.platform.flow; import com.fastbee.iot.platform.domain.PlatformDeviceMapping; +import com.fastbee.iot.platform.domain.PlatformDeviceSubMapping; import com.fastbee.iot.platform.domain.PlatformInstance; import java.util.HashMap; +import java.util.List; import java.util.Map; /** @@ -19,6 +21,9 @@ public class FlowContext { * 绑定推送关系 */ private final PlatformDeviceMapping deviceMapping; + // 绑定推送子关系 + private List monitorSubMappings; + /** * 设备上报的原始 payload（实时推送时有值） */ @@ -107,6 +112,14 @@ public class FlowContext { this.currentMonitorIndex = currentMonitorIndex; } + public List getMonitorSubMappings() { + return monitorSubMappings; + } + + public void setMonitorSubMappings(List monitorSubMappings) { + this.monitorSubMappings = monitorSubMappings; + } + public String getSubMappingOnboardConfig() { return subMappingOnboardConfig; } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java index ff17374d..447a0e86 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/flow/PlatformFlowExecutor.java @@ -21,6 +21,7 @@ import com.fastbee.iot.platform.mapper.PlatformInterfaceMapper; import com.fastbee.iot.platform.service.IPlatformPushLogService; import com.dtflys.forest.http.ForestRequestType; import lombok.extern.slf4j.Slf4j; +import org.apache.commons.collections4.CollectionUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; @@ -293,6 +294,10 @@ public class PlatformFlowExecutor { String df = fd.getString("field"); if ("platform_device_id".equals(df)) return ctx.getDeviceMapping().getPlatformDeviceId(); if ("platform_point_id".equals(df)) return ctx.getDeviceMapping().getPlatformPointId(); + // 支持从设备绑定表读取项目信息 + if ("platform_project_id".equals(df)) { + return ctx.getDeviceMapping().getPlatformProjectId(); + } if ("serial_number".equals(df)) { // 网关入网时优先使用当前监测点序列号（每个 point/device 以监测点命名） if (StringUtils.hasText(ctx.getCurrentMonitorSerialNumber())) { @@ -321,7 +326,7 @@ public class PlatformFlowExecutor { case "cascadedSubDeviceSensorIds": return collectCascadedSubDeviceSensorIds(ctx); case "cascadedSubDeviceDepths": - return collectCascadedSubDeviceDepths(ctx, fd); + return collectCascadedSubDeviceDepths(ctx); case "referenceSubDeviceSensorIdPairs": return collectReferenceSubDeviceSensorIdPairs(ctx); default: @@ -403,20 +408,19 @@ public class PlatformFlowExecutor { private Object resolveOnboardConfigValue(FlowContext ctx, JSONObject fieldDef) { String name = fieldDef.getString("name"); - if (!fieldDef.containsKey("value")) { - throw new RuntimeException("入网参数 " + name + " 未配置值"); - } - Object value = fieldDef.get("value"); - if (value instanceof JSONArray && ctx.getCurrentMonitorIndex() != null) { - JSONArray values = (JSONArray) value; + List monitorSubMappings = ctx.getMonitorSubMappings(); + if (ctx.getCurrentMonitorIndex() != null && CollectionUtils.isNotEmpty(monitorSubMappings)) { int index = ctx.getCurrentMonitorIndex(); - if (index >= values.size()) { - throw new RuntimeException("入网参数 " + name + " 数组长度不足，当前子设备序号=" - + (index + 1) + "，数组长度=" + values.size()); + PlatformDeviceSubMapping platformDeviceSubMapping = monitorSubMappings.get(index); + JSONObject authConfig = StringUtils.hasText(platformDeviceSubMapping.getOnboardConfig()) + ? JSON.parseObject(platformDeviceSubMapping.getOnboardConfig()) : new JSONObject(); + String value = authConfig.getString(name); + if (value == null) { + throw new RuntimeException("入网失败,子设备" + platformDeviceSubMapping.getSubDeviceName() + "未配置值:" + name); } - return values.get(index); + return value; } - return value; + throw new RuntimeException("入网失败," + name + "未配置值"); } private String newGeneratedId() { @@ -555,21 +559,26 @@ public class PlatformFlowExecutor { private List collectCascadedSubDeviceSensorIds(FlowContext ctx) { List suffix = collectCascadedSubDeviceSuffix(ctx); Collections.reverse(suffix); - return suffix.stream() + List sensorIds = suffix.stream() .map(sub -> resolveOnboardSensorId(sub.getOnboardConfig(), sub.getSubSerialNumber())) - .filter(StringUtils::hasText) .collect(Collectors.toList()); + if (sensorIds.stream().anyMatch(id -> !StringUtils.hasText(id))) { + throw new RuntimeException("深部位移子设备缺少 sensorId"); + } + return sensorIds; } - private List collectCascadedSubDeviceDepths(FlowContext ctx, JSONObject fieldDef) { + private List collectCascadedSubDeviceDepths(FlowContext ctx) { List monitors = collectCascadedSubDeviceMonitors(ctx); int currentIndex = resolveCurrentMonitorIndex(ctx, monitors); - List allDepths = resolveConfiguredDepths(fieldDef); - if (allDepths.size() < monitors.size()) { - throw new RuntimeException("串联子设备深度数组配置数量不足，监测点数量=" - + monitors.size() + "，深度数量=" + allDepths.size()); + List suffix = new ArrayList<>(); + for (PlatformDeviceSubMapping monitor : monitors.subList(currentIndex, monitors.size())) { + Object depth = readSensorDepth(monitor); + if (depth == null) { + throw new RuntimeException("深部位移子设备 " + monitor.getSubSerialNumber() + " 缺少传感器深度"); + } + suffix.add(depth); } - List suffix = new ArrayList<>(allDepths.subList(currentIndex, monitors.size())); Collections.reverse(suffix); return suffix; } @@ -588,10 +597,13 @@ public class PlatformFlowExecutor { if (mapping.getId() == null) { throw new RuntimeException("串联子设备入网缺少映射ID"); } - List subMappings = subMappingMapper.selectByParentId(mapping.getId()); - List monitors = subMappings == null ? Collections.emptyList() : subMappings.stream() - .filter(PlatformDeviceSubMapping::isMonitor) - .collect(Collectors.toList()); + List monitors = ctx.getMonitorSubMappings(); + if (monitors == null) { + List subMappings = subMappingMapper.selectByParentId(mapping.getId()); + monitors = subMappings == null ? Collections.emptyList() : subMappings.stream() + .filter(PlatformDeviceSubMapping::isMonitor) + .collect(Collectors.toList()); + } if (monitors.isEmpty()) { throw new RuntimeException("串联子设备入网未配置监测点子设备"); } @@ -614,23 +626,15 @@ public class PlatformFlowExecutor { throw new RuntimeException("串联子设备入网无法确定当前监测点序号"); } - private List resolveConfiguredDepths(JSONObject fieldDef) { - if (!fieldDef.containsKey("value")) { - throw new RuntimeException("串联子设备未配置深度数组"); + private Object readSensorDepth(PlatformDeviceSubMapping mapping) { + if (mapping == null || !StringUtils.hasText(mapping.getOnboardConfig())) { + return null; } - Object value = fieldDef.get("value"); - List list = toList(value); - if (list == null && value instanceof String) { - try { - list = JSON.parseArray(((String) value).trim()); - } catch (Exception ignored) { - list = null; - } + try { + return JSON.parseObject(mapping.getOnboardConfig()).get("sensorDepth"); + } catch (Exception e) { + throw new RuntimeException("子设备 " + mapping.getSubSerialNumber() + " 入网参数不是有效JSON"); } - if (list == null || list.isEmpty()) { - throw new RuntimeException("串联子设备深度数组为空"); - } - return new ArrayList<>(list); } /** @@ -1208,7 +1212,7 @@ public class PlatformFlowExecutor { } /** 确保 Token 有效，过期则刷新 */ - private String ensureToken(PlatformInstance platform) { + public String ensureToken(PlatformInstance platform) { Date now = new Date(); // 提前 5 分钟视为过期 boolean expired = platform.getTokenExpiresAt() == null diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformInstanceService.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformInstanceService.java index e4682cde..4d95fa82 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformInstanceService.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/IPlatformInstanceService.java @@ -5,6 +5,7 @@ import com.fastbee.iot.platform.domain.PlatformInstance; import java.util.Date; import java.util.List; +import java.util.Map; public interface IPlatformInstanceService extends IService { @@ -18,4 +19,9 @@ public interface IPlatformInstanceService extends IService { /** 更新 extra_config 中的某个字段 */ void updateExtraConfig(Long platformId, String key, String value); + + /** + * 根据平台ID查询项目列表 + */ + List> getProjectsByPlatformId(Long platformId); } diff --git a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java index 3a17f13e..41816ccc 100644 --- a/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java +++ b/fastbee-service/fastbee-iot-service/src/main/java/com/fastbee/iot/platform/service/impl/PlatformDeviceMappingServiceImpl.java @@ -72,6 +72,10 @@ public class PlatformDeviceMappingServiceImpl extends ServiceImpl @@ -27,6 +34,8 @@ public class PlatformInstanceServiceImpl extends ServiceImpl selectList(PlatformInstance query) { @@ -75,4 +84,73 @@ public class PlatformInstanceServiceImpl extends ServiceImpl> getProjectsByPlatformId(Long platformId) { + PlatformInstance platform = this.getById(platformId); + if (platform == null) { + throw new RuntimeException("平台不存在: " + platformId); + } + + // 从extra_config读取userId + JSONObject extraConfig = JSON.parseObject(platform.getExtraConfig()); + String userId = extraConfig == null ? null : extraConfig.getString("userId"); + if (!StringUtils.hasText(userId)) { + throw new RuntimeException("平台未完成初始化，缺少userId。请先执行平台初始化流程。"); + } + + String key = RedisKeyBuilder.buildDCPProjectInfoKey(userId); + Object cacheObject = redisCache.getCacheObject(key); + if (cacheObject != null) { + return (List>) cacheObject; + } + + // 确保 token 有效（过期自动刷新） + String token = flowExecutor.ensureToken(platform); + String url = platform.getBaseUrl() + "/api/app/project/by-user-id/" + userId; + + HttpClientConfig config = HttpClientConfig.builder() + .url(url) + .method(ForestRequestType.GET) + .headers(Collections.singletonMap("Authorization", "Bearer " + token)) + .build(); + + ForestRequest request = HttpClientFactory.instance(config); + String responseBody = request.executeAsString(); + + if (!StringUtils.hasText(responseBody)) { + return Collections.emptyList(); + } + + List> result = parseProjectsResponse(responseBody); + redisCache.setCacheObject(key, result, 10, TimeUnit.MINUTES); + return result; + } + + @SuppressWarnings("unchecked") + private List> parseProjectsResponse(String responseBody) { + String trimmed = responseBody.trim(); + List> rawList; + + if (trimmed.startsWith("[")) { + rawList = JSON.parseArray(trimmed, Map.class); + } else { + JSONObject json = JSON.parseObject(trimmed); + Object data = json.get("data"); + if (data instanceof List) { + rawList = (List>) data; + } else { + return Collections.emptyList(); + } + } + + return rawList.stream() + .map(m -> { + Map item = new HashMap<>(); + item.put("id", m.get("id")); + item.put("name", m.get("name")); + return item; + }) + .collect(Collectors.toList()); + } } diff --git a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml index e17e37f9..50a774a6 100644 --- a/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml +++ b/fastbee-service/fastbee-iot-service/src/main/resources/mapper/iot/platform/PlatformDeviceMappingMapper.xml @@ -3,28 +3,30 @@ - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + +