e689b999e6
被控端：新增生成一次性配对码并弹窗展示，支持倒计时与复制；控制端：新增输入配对码兑换绑定，刷新设备列表，并在设备列表中显示在线状态和点击连接。同时添加统一 401 拦截器，令牌失效时触发重新激活。
465 lines
22 KiB
Java
465 lines
22 KiB
Java
package com.ttstd.signaling.handler;
|
||
|
||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||
import com.ttstd.signaling.manager.ConnectionRequestManager;
|
||
import com.ttstd.signaling.manager.SessionManager;
|
||
import com.ttstd.signaling.manager.SignalMetrics;
|
||
import com.ttstd.signaling.model.AuthPrincipal;
|
||
import com.ttstd.signaling.model.DeviceBinding;
|
||
import com.ttstd.signaling.model.DeviceType;
|
||
import com.ttstd.signaling.model.SignalMessage;
|
||
import com.ttstd.signaling.security.AuthHandshakeInterceptor;
|
||
import com.ttstd.signaling.service.AccountService;
|
||
import com.ttstd.signaling.service.AuditService;
|
||
import com.ttstd.signaling.service.BindingService;
|
||
import com.ttstd.signaling.service.DeviceIdentityService;
|
||
import jakarta.annotation.PostConstruct;
|
||
import org.slf4j.Logger;
|
||
import org.slf4j.LoggerFactory;
|
||
import org.springframework.stereotype.Component;
|
||
import org.springframework.web.socket.CloseStatus;
|
||
import org.springframework.web.socket.TextMessage;
|
||
import org.springframework.web.socket.WebSocketSession;
|
||
import org.springframework.web.socket.handler.TextWebSocketHandler;
|
||
|
||
import java.io.IOException;
|
||
import java.util.HashMap;
|
||
import java.util.List;
|
||
import java.util.Map;
|
||
import java.util.Set;
|
||
|
||
@Component
|
||
public class SignalWebSocketHandler extends TextWebSocketHandler {
|
||
|
||
private static final Logger logger = LoggerFactory.getLogger(SignalWebSocketHandler.class);
|
||
private static final ObjectMapper objectMapper = new ObjectMapper();
|
||
|
||
/** 因鉴权/封禁被关闭连接时使用的关闭码 */
|
||
private static final CloseStatus CLOSE_UNAUTHORIZED = new CloseStatus(4001, "UNAUTHORIZED");
|
||
private static final CloseStatus CLOSE_FORCED_LOGOUT = new CloseStatus(4003, "FORCED_LOGOUT");
|
||
|
||
private final SessionManager sessionManager;
|
||
private final ConnectionRequestManager connectionRequestManager;
|
||
private final SignalMetrics metrics;
|
||
private final AccountService accountService;
|
||
private final DeviceIdentityService deviceIdentityService;
|
||
private final BindingService bindingService;
|
||
private final AuditService auditService;
|
||
|
||
public SignalWebSocketHandler(SessionManager sessionManager,
|
||
ConnectionRequestManager connectionRequestManager,
|
||
SignalMetrics metrics,
|
||
AccountService accountService,
|
||
DeviceIdentityService deviceIdentityService,
|
||
BindingService bindingService,
|
||
AuditService auditService) {
|
||
this.sessionManager = sessionManager;
|
||
this.connectionRequestManager = connectionRequestManager;
|
||
this.metrics = metrics;
|
||
this.accountService = accountService;
|
||
this.deviceIdentityService = deviceIdentityService;
|
||
this.bindingService = bindingService;
|
||
this.auditService = auditService;
|
||
this.connectionRequestManager.setSender(this::sendToDevice);
|
||
}
|
||
|
||
/**
|
||
* 注册会话失效回调：账号被封禁 / 会话被踢出 / 设备被禁用时立即断开对应连接。
|
||
*/
|
||
@PostConstruct
|
||
void registerRevocationListener() {
|
||
AccountService.SessionRevocationListener listener = this::forceDisconnect;
|
||
accountService.setRevocationListener(listener);
|
||
deviceIdentityService.setRevocationListener(listener);
|
||
}
|
||
|
||
/**
|
||
* 强制断开指定主体的连接。
|
||
*
|
||
* @param principalId 主体 ID
|
||
* @param sessionId 指定登录会话；为 null 表示断开该主体全部连接
|
||
*/
|
||
private void forceDisconnect(String principalId, String sessionId, String reason) {
|
||
Set<String> deviceIds = sessionManager.getDeviceIdsByPrincipal(principalId);
|
||
for (String deviceId : deviceIds) {
|
||
AuthPrincipal principal = sessionManager.getPrincipal(deviceId);
|
||
if (principal == null) {
|
||
continue;
|
||
}
|
||
// 指定了会话时只断开该会话，避免误伤同账号其他端
|
||
if (sessionId != null && !sessionId.equals(principal.sessionId())) {
|
||
continue;
|
||
}
|
||
WebSocketSession session = sessionManager.getSession(deviceId);
|
||
if (session == null || !session.isOpen()) {
|
||
continue;
|
||
}
|
||
Map<String, Object> notice = new HashMap<>();
|
||
notice.put("type", "FORCE_LOGOUT");
|
||
notice.put("payload", reason == null ? "会话已失效" : reason);
|
||
sendToSession(session, notice);
|
||
try {
|
||
session.close(CLOSE_FORCED_LOGOUT);
|
||
} catch (IOException e) {
|
||
logger.warn("强制断开连接 {} 失败: {}", deviceId, e.getMessage());
|
||
}
|
||
logger.info("已强制断开连接: deviceId={} principal={} reason={}",
|
||
deviceId, principalId, reason);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 向指定设备回送服务端事件（错误/超时通知等）。
|
||
*/
|
||
private void sendToDevice(String deviceId, Object message) {
|
||
WebSocketSession session = sessionManager.getSession(deviceId);
|
||
if (session == null || !session.isOpen()) {
|
||
logger.warn("无法向 {} 发送事件：连接不在线", deviceId);
|
||
return;
|
||
}
|
||
sendToSession(session, message);
|
||
}
|
||
|
||
/**
|
||
* 连接建立后立即以握手阶段裁定的身份完成注册，无需客户端再发 REGISTER。
|
||
*/
|
||
@Override
|
||
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
|
||
AuthPrincipal principal = principalOf(session);
|
||
if (principal == null) {
|
||
// 正常情况下握手拦截器已拦截，此处为纵深防御
|
||
logger.warn("连接 {} 缺少鉴权主体，立即关闭", session.getId());
|
||
session.close(CLOSE_UNAUTHORIZED);
|
||
return;
|
||
}
|
||
|
||
sessionManager.registerDevice(principal, session);
|
||
metrics.incRegistered();
|
||
metrics.recordSessionCount(sessionManager.getOnlineCount());
|
||
|
||
Map<String, Object> response = new HashMap<>();
|
||
response.put("type", "REGISTER_SUCCESS");
|
||
response.put("deviceId", principal.deviceId());
|
||
// 与既有信令消息协议一致：下发的设备ID也写入 fromDeviceId，便于客户端统一解析。
|
||
response.put("fromDeviceId", principal.deviceId());
|
||
response.put("deviceType", principal.deviceType().name());
|
||
response.put("displayName", principal.displayName());
|
||
sendToSession(session, response);
|
||
|
||
logger.info("连接已建立并注册: {} ({})", principal.deviceId(), principal.deviceType());
|
||
}
|
||
|
||
@Override
|
||
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
|
||
AuthPrincipal principal = principalOf(session);
|
||
if (principal == null) {
|
||
session.close(CLOSE_UNAUTHORIZED);
|
||
return;
|
||
}
|
||
|
||
String payload = message.getPayload();
|
||
logger.debug("收到消息: {}", payload);
|
||
metrics.incMessage();
|
||
|
||
try {
|
||
SignalMessage signalMessage = objectMapper.readValue(payload, SignalMessage.class);
|
||
String type = signalMessage.getType();
|
||
|
||
if (type == null) {
|
||
logger.warn("消息缺少 type 字段");
|
||
return;
|
||
}
|
||
|
||
// 关键：发送方身份一律以服务端鉴权结果覆盖，忽略客户端自报值，
|
||
// 防止伪造 fromDeviceId 冒充他人。
|
||
String claimedFrom = signalMessage.getFromDeviceId();
|
||
if (claimedFrom != null && !claimedFrom.equals(principal.deviceId())) {
|
||
logger.warn("客户端自报 fromDeviceId={} 与鉴权身份 {} 不一致，已强制覆盖",
|
||
claimedFrom, principal.deviceId());
|
||
}
|
||
signalMessage.setFromDeviceId(principal.deviceId());
|
||
signalMessage.setDeviceType(principal.deviceType().name());
|
||
|
||
switch (type.toUpperCase()) {
|
||
case "REGISTER":
|
||
// 身份已在握手阶段确定，REGISTER 仅作兼容响应
|
||
handleLegacyRegister(session, principal);
|
||
break;
|
||
case "DEVICE_LIST":
|
||
handleDeviceList(session, principal);
|
||
break;
|
||
case "PING":
|
||
// 客户端心跳保活消息，无需处理，仅用于防止中间代理因空闲超时断开连接
|
||
break;
|
||
case "OFFER":
|
||
metrics.incOffer();
|
||
// 绑定/黑名单前置校验：仅允许已绑定且未被拉黑的主控端发起
|
||
if (principal.deviceType() == DeviceType.CONTROLLER) {
|
||
String targetUid = signalMessage.getToDeviceId();
|
||
if (targetUid == null || targetUid.isEmpty()) {
|
||
sendError(session, "缺少目标设备ID");
|
||
metrics.incBlockedOffer();
|
||
break;
|
||
}
|
||
if (!bindingService.isBound(targetUid, principal.principalId())) {
|
||
sendError(session, "未与该设备建立绑定关系，无法发起连接");
|
||
metrics.incBlockedOffer();
|
||
auditOfferBlocked(targetUid, principal.principalId(), "NOT_BOUND");
|
||
break;
|
||
}
|
||
if (bindingService.isBlacklisted(targetUid, principal.principalId())) {
|
||
sendError(session, "该设备已拒绝来自你的连接");
|
||
metrics.incBlockedOffer();
|
||
auditOfferBlocked(targetUid, principal.principalId(), "BLACKLISTED");
|
||
break;
|
||
}
|
||
}
|
||
// 连接请求：统一经 ConnectionRequestManager 做校验/去重/待确认跟踪后再转发
|
||
handleConnectionRequest(signalMessage);
|
||
break;
|
||
case "ANSWER":
|
||
metrics.incAnswer();
|
||
// 被控端已接受：清理待确认状态并转发 Answer 给主控端
|
||
connectionRequestManager.completePendingRequest(
|
||
signalMessage.getToDeviceId(), signalMessage.getFromDeviceId());
|
||
forwardMessage(signalMessage, true);
|
||
break;
|
||
case "CONNECTION_REJECTED":
|
||
metrics.incRejected();
|
||
// 被控端已拒绝：清理待确认状态并转发拒绝通知给主控端
|
||
connectionRequestManager.completePendingRequest(
|
||
signalMessage.getToDeviceId(), signalMessage.getFromDeviceId());
|
||
forwardMessage(signalMessage, false);
|
||
break;
|
||
case "ICE_CANDIDATE":
|
||
case "CONTROL_COMMAND":
|
||
// 普通信令：目标不在线时静默丢弃，不返回提示
|
||
forwardMessage(signalMessage, false);
|
||
break;
|
||
default:
|
||
// 其他消息类型直接转发，目标不在线时静默丢弃
|
||
forwardMessage(signalMessage, false);
|
||
break;
|
||
}
|
||
} catch (Exception e) {
|
||
logger.error("处理消息出错: {}", e.getMessage(), e);
|
||
}
|
||
}
|
||
|
||
@Override
|
||
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
|
||
logger.info("连接已关闭: {} ({})", session.getId(), status);
|
||
sessionManager.unregisterSession(session);
|
||
metrics.recordSessionCount(sessionManager.getOnlineCount());
|
||
}
|
||
|
||
@Override
|
||
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
|
||
// EOFException / 连接被对端重置 属于正常网络事件（客户端网络中断、代理空闲超时等），
|
||
// 并非服务器错误，降级为 DEBUG 记录，避免误导。
|
||
boolean benign = exception instanceof java.io.EOFException
|
||
|| (exception.getMessage() != null
|
||
&& (exception.getMessage().contains("Connection reset")
|
||
|| exception.getMessage().contains("Broken pipe")
|
||
|| exception.getMessage().contains("An established connection")));
|
||
if (benign) {
|
||
logger.debug("传输层关闭（客户端断开），会话 {}: {}",
|
||
session.getId(), exception.getMessage());
|
||
} else {
|
||
logger.warn("传输层错误，会话 {}: {}", session.getId(), exception.getMessage());
|
||
}
|
||
sessionManager.unregisterSession(session);
|
||
metrics.recordSessionCount(sessionManager.getOnlineCount());
|
||
// 连接已不可用时无需再以 SERVER_ERROR(1011) 关闭；仅在仍打开时温和关闭。
|
||
if (session.isOpen()) {
|
||
session.close(CloseStatus.GOING_AWAY);
|
||
}
|
||
}
|
||
|
||
private AuthPrincipal principalOf(WebSocketSession session) {
|
||
Object attr = session.getAttributes().get(AuthHandshakeInterceptor.ATTR_PRINCIPAL);
|
||
return attr instanceof AuthPrincipal principal ? principal : null;
|
||
}
|
||
|
||
/**
|
||
* 兼容旧客户端的 REGISTER：不再接受客户端自报身份，仅回显服务端裁定结果。
|
||
*/
|
||
private void handleLegacyRegister(WebSocketSession session, AuthPrincipal principal) {
|
||
Map<String, Object> response = new HashMap<>();
|
||
response.put("type", "REGISTER_SUCCESS");
|
||
response.put("deviceId", principal.deviceId());
|
||
response.put("fromDeviceId", principal.deviceId());
|
||
response.put("deviceType", principal.deviceType().name());
|
||
response.put("displayName", principal.displayName());
|
||
sendToSession(session, response);
|
||
}
|
||
|
||
/**
|
||
* 设备列表（绑定视图）。
|
||
*
|
||
* <p>为防止枚举被控端，服务端不再向任意主控端返回全局在线设备清单。
|
||
* 主控端仅能看到「自己已绑定」的设备及其在线状态；被控端此项为空列表。
|
||
*/
|
||
private void handleDeviceList(WebSocketSession session, AuthPrincipal principal) {
|
||
Map<String, Object> response = new HashMap<>();
|
||
response.put("type", "DEVICE_LIST");
|
||
|
||
if (principal.deviceType() == DeviceType.CONTROLLER) {
|
||
List<Map<String, Object>> devices = bindingService.listByUser(principal.principalId())
|
||
.stream()
|
||
.filter(b -> b.getStatus() == DeviceBinding.BindingStatus.ACTIVE)
|
||
.map(b -> {
|
||
Map<String, Object> item = new HashMap<>();
|
||
item.put("deviceUid", b.getDeviceUid());
|
||
item.put("alias", b.getAlias());
|
||
item.put("role", b.getRole().name());
|
||
item.put("online", sessionManager.isDeviceOnline(b.getDeviceUid()));
|
||
return item;
|
||
})
|
||
.toList();
|
||
response.put("boundDevices", devices);
|
||
} else {
|
||
// 被控端无需浏览主控端列表
|
||
response.put("boundDevices", List.of());
|
||
}
|
||
sendToSession(session, response);
|
||
}
|
||
|
||
/** 记录被拦截的 OFFER（未绑定/被拉黑），便于审计与风控。 */
|
||
private void auditOfferBlocked(String deviceUid, String userId, String reason) {
|
||
try {
|
||
auditService.recordUser(userId, AuditService.ACTION_OFFER_BLOCKED,
|
||
AuditService.RESULT_SUCCESS, null, "device=" + deviceUid + " reason=" + reason);
|
||
} catch (Exception e) {
|
||
logger.warn("审计 OFFER 拦截记录失败: {}", e.getMessage());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 处理主控端发来的连接请求（OFFER）：先做设备类型校验与去重，
|
||
* 再登记“待被控端确认”状态，最后转发给被控端。
|
||
*/
|
||
private void handleConnectionRequest(SignalMessage message) {
|
||
String fromDeviceId = message.getFromDeviceId();
|
||
String toDeviceId = message.getToDeviceId();
|
||
|
||
// 1. 校验：仅允许 CONTROLLER -> CONTROLLED
|
||
String error = connectionRequestManager.validateOffer(fromDeviceId, toDeviceId);
|
||
if (error != null) {
|
||
logger.warn("非法连接请求 {} -> {}: {}", fromDeviceId, toDeviceId, error);
|
||
Map<String, Object> response = new HashMap<>();
|
||
response.put("type", "REQUEST_ERROR");
|
||
response.put("toDeviceId", toDeviceId);
|
||
response.put("payload", error);
|
||
sendToDevice(fromDeviceId, response);
|
||
return;
|
||
}
|
||
|
||
// 鉴权类型分类：动态验证码(CODE) / 固定密码(PASSWORD) / 免密(NONE)。
|
||
// 服务器只做分类与转发，真实校验在被控端完成（服务器不保存任何密钥）。
|
||
String authType = message.getAuthType();
|
||
if (authType != null && !authType.trim().isEmpty()) {
|
||
if ("CODE".equalsIgnoreCase(authType)) {
|
||
logger.info("连接请求 {} -> {} 使用【动态验证码】鉴权", fromDeviceId, toDeviceId);
|
||
} else if ("PASSWORD".equalsIgnoreCase(authType)) {
|
||
logger.info("连接请求 {} -> {} 使用【固定密码】鉴权", fromDeviceId, toDeviceId);
|
||
} else if ("NONE".equalsIgnoreCase(authType)) {
|
||
// 显式标记为免密，便于被控端识别
|
||
message.setAuthType("NONE");
|
||
logger.info("连接请求 {} -> {} 使用【免密连接】鉴权（被控端手动确认）", fromDeviceId, toDeviceId);
|
||
} else {
|
||
logger.warn("连接请求 {} -> {} 携带未知鉴权类型 authType={}", fromDeviceId, toDeviceId, authType);
|
||
}
|
||
} else {
|
||
// 未携带 authType 也视为免密连接（向后兼容），规范化为 NONE
|
||
message.setAuthType("NONE");
|
||
logger.info("连接请求 {} -> {} 未携带鉴权，按【免密连接】处理（被控端手动确认）", fromDeviceId, toDeviceId);
|
||
}
|
||
|
||
// 2. 去重：短时间内重复 OFFER 直接忽略，避免被控端反复弹窗
|
||
if (connectionRequestManager.isDuplicateOffer(fromDeviceId, toDeviceId)) {
|
||
logger.info("重复连接请求 {} -> {} 已忽略", fromDeviceId, toDeviceId);
|
||
return;
|
||
}
|
||
|
||
// 3. 目标不在线：立即回送 TARGET_OFFLINE，且不登记待确认（避免等待超时）
|
||
if (!sessionManager.isDeviceOnline(toDeviceId)) {
|
||
logger.warn("目标被控端 {} 不在线，无法投递连接请求", toDeviceId);
|
||
notifySenderTargetOffline(message, toDeviceId);
|
||
return;
|
||
}
|
||
|
||
// 4. 登记待确认并转发 OFFER 给被控端
|
||
connectionRequestManager.registerPendingRequest(fromDeviceId, toDeviceId);
|
||
forwardMessage(message, false);
|
||
}
|
||
|
||
private void forwardMessage(SignalMessage message, boolean notifyOffline) {
|
||
String toDeviceId = message.getToDeviceId();
|
||
if (toDeviceId == null) {
|
||
logger.warn("无法转发消息：toDeviceId 为空");
|
||
return;
|
||
}
|
||
|
||
WebSocketSession targetSession = sessionManager.getSession(toDeviceId);
|
||
if (targetSession == null || !targetSession.isOpen()) {
|
||
logger.warn("目标设备 {} 不在线", toDeviceId);
|
||
// 仅 OFFER / ANSWER 在目标不在线时回送 TARGET_OFFLINE，其余类型静默丢弃
|
||
if (notifyOffline) {
|
||
notifySenderTargetOffline(message, toDeviceId);
|
||
}
|
||
return;
|
||
}
|
||
|
||
try {
|
||
String jsonMessage = objectMapper.writeValueAsString(message);
|
||
targetSession.sendMessage(new TextMessage(jsonMessage));
|
||
logger.debug("已转发 {}：{} -> {}", message.getType(), message.getFromDeviceId(), toDeviceId);
|
||
} catch (IOException e) {
|
||
logger.error("转发消息到 {} 失败: {}", toDeviceId, e.getMessage());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 目标被控端不在线时，向发送方（主控端）回送 TARGET_OFFLINE 提醒。
|
||
*/
|
||
private void notifySenderTargetOffline(SignalMessage message, String offlineDeviceId) {
|
||
String fromDeviceId = message.getFromDeviceId();
|
||
if (fromDeviceId == null) {
|
||
logger.warn("无法通知离线状态：fromDeviceId 为空");
|
||
return;
|
||
}
|
||
|
||
WebSocketSession senderSession = sessionManager.getSession(fromDeviceId);
|
||
if (senderSession == null || !senderSession.isOpen()) {
|
||
logger.warn("发送方 {} 会话不存在，无法通知目标离线", fromDeviceId);
|
||
return;
|
||
}
|
||
|
||
Map<String, Object> response = new HashMap<>();
|
||
response.put("type", "TARGET_OFFLINE");
|
||
response.put("toDeviceId", offlineDeviceId);
|
||
response.put("payload", "目标被控端不在线，请确认设备已开启并连接到信令服务器");
|
||
|
||
sendToSession(senderSession, response);
|
||
logger.info("已通知发送方 {} 目标 {} 不在线", fromDeviceId, offlineDeviceId);
|
||
}
|
||
|
||
private void sendToSession(WebSocketSession session, Object data) {
|
||
try {
|
||
String json = objectMapper.writeValueAsString(data);
|
||
session.sendMessage(new TextMessage(json));
|
||
} catch (IOException e) {
|
||
logger.error("向会话 {} 发送消息失败: {}", session.getId(), e.getMessage());
|
||
}
|
||
}
|
||
|
||
/** 向发送方回送一条 REQUEST_ERROR 业务提示（不关闭连接）。 */
|
||
private void sendError(WebSocketSession session, String message) {
|
||
Map<String, Object> response = new HashMap<>();
|
||
response.put("type", "REQUEST_ERROR");
|
||
response.put("payload", message);
|
||
sendToSession(session, response);
|
||
}
|
||
}
|