V5.0 Agent架构重构:轻量Agent注入MC进程 + WebSocket实时数据回传

架构变更:
- ServerGuard 保持独立外部进程(Spring Boot),Agent 为轻量 Paper 插件
- Agent 通过 WebSocket (ws://host:8080/agent) 实时推送 Bukkit 事件/指标/告警至 ServerGuard
- ServerGuard 通过 WebSocket 下发防御指令(ban/kick/whitelist),Agent 端通过 InternalCommandExecutor 直接执行
- 新增 AgentMessage 通信协议(14种消息类型 + 10种命令类型)
- Agent 端即时拦截(超高速移动/超远攻击/命令注入/广告/钓鱼)+ Server 端深度 AI 分析

新增文件:
- agent/AgentMessage.java — 通信协议定义
- server/AgentWebSocketServer.java — Spring WebSocket 服务端
- server/AgentCommandDispatcher.java — 指令下发调度器
- config/WebSocketConfig.java — WebSocket 端点配置
- plugin/bridge/AgentWebSocketClient.java — Agent WebSocket 客户端

重写文件:
- AluerPlugin.java — 轻量 Agent(移除 Spring Boot 嵌入)
- 所有 9 个 EventListener — 数据通过 WebSocket 推送
- AluerCommandExecutor.java — 精简版命令执行器
- ServerGuardService.java — Agent WebSocket 事件处理集成
- ServerGuardApplication.java — 还原为独立引擎入口

技术栈:pom.xml 新增 spring-boot-starter-websocket
278 项全量测试通过,0 失败
This commit is contained in:
Peijun Zhao
2026-05-16 16:14:48 +08:00
parent 030ced87f9
commit 92a590c36a
20 changed files with 1538 additions and 1220 deletions
@@ -3,11 +3,13 @@ package com.aluer.service;
import com.aluer.ai.*;
import com.aluer.alert.EmailAlertService;
import com.aluer.config.ServerGuardConfig;
import com.aluer.monitor.*;
import com.aluer.model.AlertEvent;
import com.aluer.model.AlertType;
import com.aluer.model.MetricsData;
import com.aluer.plugin.bridge.DataBridge;
import com.aluer.plugin.bridge.InternalCommandExecutor;
import com.aluer.monitor.ProcessMonitor;
import com.aluer.monitor.ResourceMonitor;
import com.aluer.server.AgentWebSocketServer;
import com.google.gson.JsonObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -15,12 +17,21 @@ import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Service;
import java.time.Instant;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* ServerGuard 核心调度服务
*
* 负责:
* 1. 外部监控模式:进程保活 + 资源监控 + 日志分析(传统 external mode)
* 2. Agent 模式:接收 Paper 插件 Agent 通过 WebSocket 推送的实时事件/指标/告警
* 3. 告警处理管道(AI 分析 + 自动执行 + 邮件通知),两种模式共用
*/
@Service
public class ServerGuardService implements CommandLineRunner {
private static final Logger logger = LoggerFactory.getLogger(ServerGuardService.class);
@@ -28,8 +39,6 @@ public class ServerGuardService implements CommandLineRunner {
private final ServerGuardConfig config;
private final ProcessMonitor processMonitor;
private final ResourceMonitor resourceMonitor;
private final ConnectionMonitor connectionMonitor;
private final LogMonitor logMonitor;
private final AdaptiveThreshold adaptiveThreshold;
private final AnomalyDetector anomalyDetector;
private final TimeSeriesPredictor timeSeriesPredictor;
@@ -37,15 +46,14 @@ public class ServerGuardService implements CommandLineRunner {
private final EmailAlertService emailAlertService;
private final DeepSeekClient deepSeekClient;
private final AutoExecutor autoExecutor;
private final RconClient rconClient;
/** 插件模式下的数据桥接(external 模式下为 null) */
/** Agent WebSocket 服务器(接收 Paper 插件 Agent 的实时数据推送) */
@Autowired(required = false)
private DataBridge dataBridge;
private AgentWebSocketServer agentServer;
/** 插件模式下的内部命令执行器(external 模式下为 null) */
/** Agent 命令调度器(通过 WebSocket 向 Agent 下发防御指令) */
@Autowired(required = false)
private InternalCommandExecutor internalCommandExecutor;
private AgentCommandDispatcher agentCommandDispatcher;
@Value("${server.port:8080}")
private int serverPort;
@@ -53,25 +61,23 @@ public class ServerGuardService implements CommandLineRunner {
private final ScheduledExecutorService scheduler;
private final AtomicBoolean running = new AtomicBoolean(true);
/** Agent 推送的事件/告警计数 */
private final AtomicBoolean agentHandlersRegistered = new AtomicBoolean(false);
public ServerGuardService(
ServerGuardConfig config,
ProcessMonitor processMonitor,
ResourceMonitor resourceMonitor,
ConnectionMonitor connectionMonitor,
LogMonitor logMonitor,
AdaptiveThreshold adaptiveThreshold,
AnomalyDetector anomalyDetector,
TimeSeriesPredictor timeSeriesPredictor,
AttackDetector attackDetector,
EmailAlertService emailAlertService,
DeepSeekClient deepSeekClient,
AutoExecutor autoExecutor,
RconClient rconClient) {
AutoExecutor autoExecutor) {
this.config = config;
this.processMonitor = processMonitor;
this.resourceMonitor = resourceMonitor;
this.connectionMonitor = connectionMonitor;
this.logMonitor = logMonitor;
this.adaptiveThreshold = adaptiveThreshold;
this.anomalyDetector = anomalyDetector;
this.timeSeriesPredictor = timeSeriesPredictor;
@@ -79,7 +85,6 @@ public class ServerGuardService implements CommandLineRunner {
this.emailAlertService = emailAlertService;
this.deepSeekClient = deepSeekClient;
this.autoExecutor = autoExecutor;
this.rconClient = rconClient;
this.scheduler = Executors.newScheduledThreadPool(4, r -> {
Thread t = new Thread(r, "serverguard-scheduler");
@@ -93,114 +98,134 @@ public class ServerGuardService implements CommandLineRunner {
printBanner();
validateConfiguration();
startMonitoring();
// 插件模式:注册 DataBridge 告警/指标处理器
if (config.isPluginMode() && dataBridge != null) {
dataBridge.setAlertHandler(this::handleAlert);
dataBridge.setMetricsHandler(metrics -> {
deepSeekClient.addMetrics(metrics);
adaptiveThreshold.addMetric(metrics);
anomalyDetector.addMetric(metrics);
timeSeriesPredictor.addMetric(metrics);
});
// 启动 DataBridge 包速率定时更新
scheduler.scheduleAtFixedRate(() -> dataBridge.updatePacketRate(), 1, 1, TimeUnit.SECONDS);
logger.info("DataBridge alert/metrics handlers registered — plugin mode active");
// 注册 Agent WebSocket 事件处理器
if (agentServer != null) {
registerAgentHandlers();
agentHandlersRegistered.set(true);
logger.info("Agent WebSocket handlers registered — ready to receive agent data");
}
// 启动监控(external 模式使用传统监控,agent 模式使用 agent 推送 + 传统指标补充)
startMonitoring();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
logger.info("Shutdown signal received, gracefully stopping...");
running.set(false);
scheduler.shutdown();
try {
if (!scheduler.awaitTermination(8, TimeUnit.SECONDS)) {
scheduler.shutdownNow();
}
if (!scheduler.awaitTermination(8, TimeUnit.SECONDS)) scheduler.shutdownNow();
} catch (InterruptedException e) {
scheduler.shutdownNow();
Thread.currentThread().interrupt();
}
emailAlertService.shutdown();
if (rconClient.isConnected()) {
rconClient.close();
}
logger.info("Aluer ServerGuard stopped");
}, "serverguard-shutdown"));
}
// ─── Agent 事件处理器注册 ──────────────────────────
private void registerAgentHandlers() {
// 事件处理器:将 Agent 推送的原始事件转换为指标数据
agentServer.setEventHandler((agentId, data) -> {
try {
String eventType = data.get("eventType") != null ? data.get("eventType").getAsString() : "";
JsonObject eventData = data.getAsJsonObject("data");
// 将事件数据注入 AI 分析引擎的指标流
MetricsData metrics = buildMetricsFromAgentEvent(eventType, eventData);
if (metrics != null) {
deepSeekClient.addMetrics(metrics);
adaptiveThreshold.addMetric(metrics);
anomalyDetector.addMetric(metrics);
}
} catch (Exception e) {
logger.debug("Failed to process agent event: {}", e.getMessage());
}
});
// 指标处理器:Agent 定期推送的 TPS/包速率等指标
agentServer.setMetricsHandler((agentId, data) -> {
try {
MetricsData metrics = new MetricsData();
metrics.setTps(data.get("tps") != null ? data.get("tps").getAsDouble() : 20.0);
metrics.setOnlinePlayers(data.get("onlinePlayers") != null ? data.get("onlinePlayers").getAsInt() : 0);
metrics.setConnections(metrics.getOnlinePlayers());
deepSeekClient.addMetrics(metrics);
adaptiveThreshold.addMetric(metrics);
anomalyDetector.addMetric(metrics);
timeSeriesPredictor.addMetric(metrics);
} catch (Exception e) {
logger.debug("Failed to process agent metrics: {}", e.getMessage());
}
});
// 告警处理器:Agent 即时检测到的安全告警
agentServer.setAlertHandler((agentId, data) -> {
try {
AlertEvent alert = new AlertEvent();
String alertType = data.get("alertType") != null ? data.get("alertType").getAsString() : "SECURITY_OTHER";
// 映射告警类型字符串到 AlertType 枚举
try {
alert.setType(AlertType.valueOf(alertType));
} catch (IllegalArgumentException e) {
alert.setType(AlertType.SECURITY_OTHER);
}
alert.setMessage(data.get("message") != null ? data.get("message").getAsString() : "");
alert.setConfidence(data.get("confidence") != null ? data.get("confidence").getAsDouble() : 0.5);
alert.setSource(data.get("source") != null ? data.get("source").getAsString() : agentId);
alert.setTimestamp(Instant.now().toEpochMilli());
handleAlert(alert);
} catch (Exception e) {
logger.debug("Failed to process agent alert: {}", e.getMessage());
}
});
}
/** 从 Agent 事件构建 MetricsData 快照 */
private MetricsData buildMetricsFromAgentEvent(String eventType, JsonObject data) {
// 跟踪玩家在线数变化
if ("PLAYER_JOIN".equals(eventType) || "PLAYER_QUIT".equals(eventType)) {
MetricsData m = new MetricsData();
m.setTps(20.0); // 由定期 metrics 推送更新精确值
m.setConnections(1); // 连接变化
return m;
}
return null;
}
// ─── Banner ────────────────────────────────────────
private void printBanner() {
boolean hasAgent = agentServer != null;
logger.info("");
logger.info("╔══════════════════════════════════════════════════════════════╗");
logger.info("║ Aluer ServerGuard v4.0.0 ║");
logger.info("║ Aluer ServerGuard v5.0.0 ║");
logger.info("║ AI-Powered Minecraft Server Protection ║");
logger.info("║ Mode: " + String.format("%-39s", config.isPluginMode() ? "Paper Plugin (Embedded)" : "External Monitor") + "║");
logger.info("║ Mode: " + String.format("%-39s", hasAgent ? "Agent Server (WebSocket)" : "External Monitor") + "║");
logger.info("╠══════════════════════════════════════════════════════════════╣");
logger.info("║ Java: {} ║", String.format("%-36s", System.getProperty("java.version")));
logger.info("║ Available CPUs: {}{} ║",
String.format("%-32s", Runtime.getRuntime().availableProcessors()),
"");
logger.info("║ Max Memory: {}{} ║",
String.format("%-32s", (Runtime.getRuntime().maxMemory() / (1024 * 1024)) + " MB"),
"");
logger.info("╠══════════════════════════════════════════════════════════════╣");
logger.info("║ RCON: {}:{} {}{} ║",
config.getMinecraft().getRcon().getHost(),
config.getMinecraft().getRcon().getPort(),
config.getMinecraft().getRcon().isEnabled() ? "enabled" : "disabled",
" ");
logger.info("║ Web Console: http://0.0.0.0:{}{} ║", serverPort, " ");
if (hasAgent) {
logger.info("║ Agent WS: ws://0.0.0.0:{}/agent{} ║", serverPort, " ");
}
logger.info("║ DeepSeek AI: {}{} ║",
deepSeekClient.isEnabled() ? "enabled (" + config.getAi().getDeepseek().getModel() + ")" : "disabled",
deepSeekClient.isEnabled() ? "" : " ");
logger.info("║ Web Console: http://0.0.0.0:{}{} ║",
serverPort,
" ");
logger.info("╠══════════════════════════════════════════════════════════════╣");
logger.info("║ Monitor: {}s interval{}{} ║",
config.getMinecraft().getCheckIntervalSeconds(),
" ",
"");
logger.info("║ Auto-Execute: {}{} ║",
config.getAi().getDeepseek().getAutoExecute().isEnabled() ? "enabled" : "disabled",
" ");
logger.info("║ Auto-Heal: {}{} ║",
config.getSecurity().getSelfHealing().isEnabled() ? "enabled" : "disabled",
" ");
logger.info("╚══════════════════════════════════════════════════════════════╝");
logger.info("");
logger.info("Type 'help' in the shell for available commands.");
logger.info("Access the web console at http://localhost:{}/", serverPort);
logger.info("");
}
private void validateConfiguration() {
int warnings = 0;
if (!deepSeekClient.isEnabled()) {
logger.warn("CONFIG: DeepSeek AI is disabled. Set DEEPSEEK_API_KEY to enable AI features.");
warnings++;
}
String rconPassword = config.getMinecraft().getRcon().getPassword();
if (rconPassword == null || rconPassword.isEmpty()) {
logger.warn("CONFIG: RCON password is not set. Set RCON_PASSWORD env var.");
warnings++;
}
if (!config.getAlert().getEmail().getUsername().isEmpty()) {
if (config.getAlert().getEmail().getTo() == null || config.getAlert().getEmail().getTo().isEmpty()) {
logger.warn("CONFIG: Email alert enabled but no recipients configured.");
warnings++;
}
}
if (config.getSecurity().getHostEnforcement().isDryRun()) {
logger.info("CONFIG: Host enforcement is in dry-run mode (safe for first deployment).");
}
if (config.getSecurity().getSelfHealing().isDryRun()) {
logger.info("CONFIG: Self-healing is in dry-run mode (safe for first deployment).");
}
if (warnings > 0) {
logger.info("Configuration check complete: {} warning(s) found.", warnings);
} else {
@@ -208,27 +233,30 @@ public class ServerGuardService implements CommandLineRunner {
}
}
// ─── 监控循环 ──────────────────────────────────────
private void startMonitoring() {
int intervalSeconds = config.getMinecraft().getCheckIntervalSeconds();
int interval = config.getMinecraft().getCheckIntervalSeconds();
scheduler.scheduleAtFixedRate(this::checkProcess, 0, intervalSeconds, TimeUnit.SECONDS);
scheduler.scheduleAtFixedRate(this::collectAndAnalyzeMetrics, 5, intervalSeconds, TimeUnit.SECONDS);
if (deepSeekClient.isEnabled() && config.getAi().getDeepseek().isAutoAnalyzeAlerts()) {
int analysisInterval = config.getAi().getDeepseek().getAnalysisIntervalSeconds();
scheduler.scheduleAtFixedRate(this::runDeepSeekAnalysis, analysisInterval, analysisInterval, TimeUnit.SECONDS);
// 进程保活(external 模式使用)
if (!config.isPluginMode()) {
scheduler.scheduleAtFixedRate(this::checkProcess, 0, interval, TimeUnit.SECONDS);
}
logger.info("Monitoring started with {}s interval", intervalSeconds);
scheduler.scheduleAtFixedRate(this::collectAndAnalyzeMetrics, 5, interval, TimeUnit.SECONDS);
if (deepSeekClient.isEnabled() && config.getAi().getDeepseek().isAutoAnalyzeAlerts()) {
int aiInterval = config.getAi().getDeepseek().getAnalysisIntervalSeconds();
scheduler.scheduleAtFixedRate(this::runDeepSeekAnalysis, aiInterval, aiInterval, TimeUnit.SECONDS);
}
logger.info("Monitoring started with {}s interval", interval);
}
private void checkProcess() {
try {
AlertEvent event = processMonitor.checkAndRestart();
if (event != null) {
logger.warn("Process alert: {}", event.getMessage());
handleAlert(event);
}
if (event != null) handleAlert(event);
} catch (Exception e) {
logger.error("Process check failed: {}", e.getMessage());
}
@@ -236,58 +264,30 @@ public class ServerGuardService implements CommandLineRunner {
private void collectAndAnalyzeMetrics() {
try {
// 插件模式:从 DataBridge 采集实时指标
if (config.isPluginMode() && dataBridge != null) {
MetricsData pluginMetrics = dataBridge.buildMetricsSnapshot();
deepSeekClient.addMetrics(pluginMetrics);
adaptiveThreshold.addMetric(pluginMetrics);
anomalyDetector.addMetric(pluginMetrics);
timeSeriesPredictor.addMetric(pluginMetrics);
}
MetricsData data = resourceMonitor.collectMetrics();
deepSeekClient.addMetrics(data);
adaptiveThreshold.addMetric(data);
anomalyDetector.addMetric(data);
timeSeriesPredictor.addMetric(data);
List<AlertEvent> resourceAlerts = resourceMonitor.checkThresholds(data);
for (AlertEvent alert : resourceAlerts) {
handleAlert(alert);
}
AlertEvent connectionAlert = connectionMonitor.checkConnectionFlood();
if (connectionAlert != null) {
handleAlert(connectionAlert);
}
List<AlertEvent> logAlerts = logMonitor.analyzeLogs();
for (AlertEvent alert : logAlerts) {
handleAlert(alert);
}
for (AlertEvent alert : resourceAlerts) handleAlert(alert);
if (config.getAi().isEnabled() && config.getAi().isUseIsolationForest()) {
AlertEvent aiAlert = anomalyDetector.detectAnomaly(data);
if (aiAlert != null) {
handleAlert(aiAlert);
}
if (aiAlert != null) handleAlert(aiAlert);
}
if (config.getAi().isEnabled() && config.getAi().isUsePrediction()) {
TimeSeriesPredictor.PredictionResult prediction = timeSeriesPredictor.predict();
if (prediction != null && prediction.isWillBreachThreshold() && prediction.getConfidence() > 0.6) {
logger.warn("TPS Prediction: {} (confidence: {:.0f}%) - {}",
prediction.getPredictedValue(),
prediction.getConfidence() * 100,
prediction.getRecommendation());
logger.warn("TPS Prediction: {} (confidence: {:.0f}%) — {}",
prediction.getPredictedValue(), prediction.getConfidence() * 100, prediction.getRecommendation());
}
}
List<AlertEvent> attackAlerts = attackDetector.analyzeThreats();
for (AlertEvent alert : attackAlerts) {
handleAlert(alert);
}
for (AlertEvent alert : attackAlerts) handleAlert(alert);
} catch (Exception e) {
logger.error("Metrics collection failed: {}", e.getMessage());
@@ -295,19 +295,13 @@ public class ServerGuardService implements CommandLineRunner {
}
private void runDeepSeekAnalysis() {
if (!deepSeekClient.isEnabled()) {
return;
}
if (!deepSeekClient.isEnabled()) return;
try {
deepSeekClient.getServerHealthReportAsync().thenAccept(healthReport -> {
if (healthReport == null) {
return;
}
if (healthReport == null) return;
logger.info("=== DeepSeek AI Health Report ===");
logger.info("Status: {}", healthReport.getSeverity());
logger.info("Summary: {}", healthReport.getDescription());
if ("critical".equals(healthReport.getSeverity()) || "warning".equals(healthReport.getSeverity())) {
for (String action : healthReport.getRecommendedActions()) {
logger.warn("Recommendation: {}", action);
@@ -319,56 +313,32 @@ public class ServerGuardService implements CommandLineRunner {
}
}
// ─── 告警处理管道 ──────────────────────────────────
private void handleAlert(AlertEvent alert) {
deepSeekClient.addAlert(alert);
logger.warn("Alert [{}]: {} (confidence: {:.0f}%)",
alert.getType(), alert.getMessage(), alert.getConfidence() * 100);
logger.warn("Alert [{}]: {} (confidence: {:.0f}%)", alert.getType(), alert.getMessage(), alert.getConfidence() * 100);
if (deepSeekClient.isEnabled() && config.getAi().getDeepseek().isAutoAnalyzeAlerts()) {
deepSeekClient.analyzeAlertAsync(alert).thenAccept(analysis -> {
if (analysis != null) {
logger.info(" [DeepSeek AI Analysis]");
logger.info(" Severity: {}", analysis.getSeverity());
logger.info(" Root Cause: {}", analysis.getRootCause());
if (!analysis.getRecommendedActions().isEmpty()) {
logger.info(" Recommended Actions:");
for (String action : analysis.getRecommendedActions()) {
logger.info(" - {}", action);
}
}
logger.info(" [DeepSeek AI Analysis] severity={}, rootCause={}", analysis.getSeverity(), analysis.getRootCause());
alert.setRootCause(analysis.getRootCause());
alert.setSuggestedAction(String.join("; ", analysis.getRecommendedActions()));
autoExecutor.executeAutoAction(analysis, alert);
// Agent 模式下通过 WebSocket 下发命令,external 模式下通过 RCON
if (agentCommandDispatcher != null && agentServer != null && agentServer.getConnectedAgentCount() > 0) {
agentCommandDispatcher.dispatchAutoAction(analysis, alert, agentServer);
} else {
autoExecutor.executeAutoAction(analysis, alert);
}
}
});
} else {
if (alert.getRootCause() != null) {
logger.info(" Root Cause: {}", alert.getRootCause());
}
if (alert.getSuggestedAction() != null) {
logger.info(" Suggested Action: {}", alert.getSuggestedAction());
}
}
emailAlertService.sendAlert(alert);
if (alert.getType().name().contains("CONNECTION_FLOOD") ||
alert.getType().name().contains("LOG_ATTACK")) {
if (alert.getConfidence() > 0.8) {
logger.info("High confidence attack alert — triggering auto defense");
}
}
}
public boolean isRunning() {
return running.get();
}
public void stop() {
running.set(false);
}
public boolean isRunning() { return running.get(); }
public void stop() { running.set(false); }
}