diff --git a/server/api/servers.py b/server/api/servers.py index fed27930..e70b274e 100644 --- a/server/api/servers.py +++ b/server/api/servers.py @@ -442,6 +442,75 @@ async def get_agent_install_cmd( } +# ── Agent Upgrade ── + +@router.post("/{id}/upgrade-agent", response_model=dict) +async def upgrade_agent( + request: Request, + id: int, + admin: Admin = Depends(get_current_admin), + service: ServerService = Depends(get_server_service), + db: AsyncSession = Depends(get_db), +): + """Upgrade Nexus Agent on the managed server via SSH. + + Downloads the latest agent.py from the central server and restarts + the nexus-agent systemd service. Does NOT touch config.json. + """ + from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command + import shlex + + server = await service.get_server(id) + if not server: + raise HTTPException(status_code=404, detail="Server not found") + + base_url = (settings.API_BASE_URL or "").strip().rstrip("/") + if not base_url: + raise HTTPException( + status_code=400, + detail="NEXUS_API_BASE_URL 未配置，无法构建下载地址", + ) + + agent_url = f"{base_url}/agent/agent.py" + install_dir = "/opt/nexus-agent" + + # Download new agent.py → restart service + upgrade_cmd = ( + f"curl -fsSL {shlex.quote(agent_url)} -o {install_dir}/agent.py " + f"&& systemctl restart nexus-agent " + f"&& sleep 2 " + f"&& systemctl is-active nexus-agent " + f"&& echo 'upgrade_ok'" + ) + + try: + result = await exec_ssh_command(server, upgrade_cmd, timeout=60) + except Exception as e: + raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}") + + success = result["exit_code"] == 0 and "upgrade_ok" in result["stdout"] + + if not success: + raise HTTPException( + status_code=400, + detail=f"升级失败 (exit {result['exit_code']}): {result['stderr'][:300]}", + ) + + # Audit + ip_address = request.client.host if request.client else "" + audit_repo = AuditLogRepositoryImpl(db) + await audit_repo.create(AuditLog( + admin_username=admin.username, + action="upgrade_agent", + target_type="server", + target_id=id, + detail=f"Agent 升级: {server.name} ({server.domain})", + ip_address=ip_address, + )) + + return {"success": True, "server_id": id, "server_name": server.name, "stdout": result["stdout"][:500]} + + # ── Health Check ── @router.post("/check", response_model=dict) diff --git a/server/api/settings.py b/server/api/settings.py index 3522d80c..5a67ae5a 100644 --- a/server/api/settings.py +++ b/server/api/settings.py @@ -280,6 +280,198 @@ async def delete_preset( )) +# ── Alert History ── + +alert_history_router = APIRouter(prefix="/api/alert-history", tags=["alerts"]) + + +@alert_history_router.get("/", response_model=dict) +async def list_alert_history( + server_id: Optional[int] = None, + alert_type: Optional[str] = None, + is_recovery: Optional[bool] = None, + date_from: Optional[str] = None, + date_to: Optional[str] = None, + limit: int = 50, + offset: int = 0, + admin: Admin = Depends(get_current_admin), + db: AsyncSession = Depends(get_db), +): + """Paginated alert/recovery history with filters.""" + from sqlalchemy import select, func + from server.domain.models import AlertLog, Server + from datetime import datetime, timezone + + filters = [] + if server_id: + filters.append(AlertLog.server_id == server_id) + if alert_type: + filters.append(AlertLog.alert_type == alert_type) + if is_recovery is not None: + filters.append(AlertLog.is_recovery == is_recovery) + if date_from: + try: + dt = datetime.fromisoformat(date_from) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + filters.append(AlertLog.created_at >= dt.replace(tzinfo=None)) + except ValueError: + pass + if date_to: + try: + dt = datetime.fromisoformat(date_to) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + filters.append(AlertLog.created_at <= dt.replace(tzinfo=None)) + except ValueError: + pass + + count_q = select(func.count(AlertLog.id)) + data_q = ( + select(AlertLog, Server.name.label("srv_name")) + .join(Server, AlertLog.server_id == Server.id, isouter=True) + .order_by(AlertLog.created_at.desc()) + .offset(offset).limit(limit) + ) + if filters: + count_q = count_q.where(*filters) + data_q = data_q.where(*filters) + + total = int((await db.execute(count_q)).scalar() or 0) + rows = (await db.execute(data_q)).all() + + def _to_dict(log, srv_name): + return { + "id": log.id, + "server_id": log.server_id, + "server_name": log.server_name or srv_name or f"#{log.server_id}", + "alert_type": log.alert_type, + "value": log.value, + "is_recovery": log.is_recovery, + "created_at": str(log.created_at) if log.created_at else None, + } + + return { + "items": [_to_dict(log, sn) for log, sn in rows], + "total": total, "limit": limit, "offset": offset, + } + + +@alert_history_router.get("/stats", response_model=dict) +async def alert_stats( + days: int = 7, + admin: Admin = Depends(get_current_admin), + db: AsyncSession = Depends(get_db), +): + """Top alerting servers + daily counts for the last N days.""" + from sqlalchemy import select, func, text + from server.domain.models import AlertLog + + # Top 5 servers by alert count + top_q = ( + select(AlertLog.server_id, AlertLog.server_name, func.count(AlertLog.id).label("cnt")) + .where(AlertLog.is_recovery == False) + .group_by(AlertLog.server_id, AlertLog.server_name) + .order_by(func.count(AlertLog.id).desc()) + .limit(5) + ) + top_rows = (await db.execute(top_q)).all() + + # Daily counts + daily_q = text( + f"SELECT DATE(created_at) as day, " + f"SUM(is_recovery=0) as alerts, SUM(is_recovery=1) as recoveries " + f"FROM alert_logs " + f"WHERE created_at >= DATE_SUB(NOW(), INTERVAL {int(days)} DAY) " + f"GROUP BY DATE(created_at) ORDER BY day" + ) + daily_rows = (await db.execute(daily_q)).all() + + return { + "top_servers": [{"server_id": r.server_id, "server_name": r.server_name or f"#{r.server_id}", "count": r.cnt} for r in top_rows], + "daily": [{"day": str(r.day), "alerts": int(r.alerts), "recoveries": int(r.recoveries)} for r in daily_rows], + } + + +# ── Telegram Test + Chat ID Detection ── + +@router.post("/telegram/test", response_model=dict) +async def telegram_test_send( + admin: Admin = Depends(get_current_admin), +): + """Send a test Telegram message to verify Bot Token + Chat ID configuration.""" + from server.infrastructure.telegram import send_telegram + from server.config import settings as _settings + + if not _settings.TELEGRAM_BOT_TOKEN: + raise HTTPException(status_code=400, detail="TELEGRAM_BOT_TOKEN 未配置") + if not _settings.TELEGRAM_CHAT_ID: + raise HTTPException(status_code=400, detail="TELEGRAM_CHAT_ID 未配置，请先检测并填写") + + from datetime import datetime, timezone + now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") + ok = await send_telegram( + f"✅ Nexus 测试消息\n" + f"Bot 配置正确，告警通知已就绪\n" + f"时间: {now}" + ) + if ok: + return {"success": True} + raise HTTPException(status_code=502, detail="Telegram API 返回错误，请检查 Bot Token 和 Chat ID") + + +@router.get("/telegram/chats", response_model=dict) +async def telegram_get_chats( + admin: Admin = Depends(get_current_admin), +): + """Call getUpdates to detect recent chats and extract unique chat_ids. + + User must send at least one message to the Bot before calling this. + Returns up to 5 distinct chats from recent messages. + """ + import httpx + from server.config import settings as _settings + from server.infrastructure.telegram import TELEGRAM_API_BASE + + token = _settings.TELEGRAM_BOT_TOKEN + if not token: + raise HTTPException(status_code=400, detail="TELEGRAM_BOT_TOKEN 未配置") + + try: + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.get( + f"{TELEGRAM_API_BASE}/bot{token}/getUpdates", + params={"limit": 50, "allowed_updates": ["message", "channel_post"]}, + ) + if resp.status_code != 200: + raise HTTPException(status_code=502, detail=f"Telegram API 返回 {resp.status_code}") + data = resp.json() + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=502, detail=f"请求失败: {e}") + + # Extract unique chats + seen: set[int] = set() + chats: list[dict] = [] + for update in data.get("result", []): + msg = update.get("message") or update.get("channel_post") or {} + chat = msg.get("chat", {}) + cid = chat.get("id") + if cid and cid not in seen: + seen.add(cid) + chats.append({ + "id": cid, + "type": chat.get("type", ""), + "title": chat.get("title") or chat.get("first_name") or "", + "username": chat.get("username") or "", + }) + if len(chats) >= 5: + break + + return {"chats": chats, "count": len(chats)} + + # ── IP Allowlist ── class IpAllowlistRequest(BaseModel): diff --git a/server/api/websocket.py b/server/api/websocket.py index 09ae33b9..233951e4 100644 --- a/server/api/websocket.py +++ b/server/api/websocket.py @@ -324,6 +324,24 @@ def _should_push_telegram(alert_key: str) -> bool: # ── Broadcast Functions (called from other modules) ── +async def _save_alert_log(server_id: int, server_name: str, alert_type: str, value: str, is_recovery: bool): + """Persist alert / recovery event to MySQL alert_logs table.""" + try: + from server.infrastructure.database.session import AsyncSessionLocal + from server.domain.models import AlertLog + async with AsyncSessionLocal() as session: + session.add(AlertLog( + server_id=server_id, + server_name=server_name or None, + alert_type=alert_type, + value=value, + is_recovery=is_recovery, + )) + await session.commit() + except Exception as e: + logger.debug("Failed to save alert log: %s", e) + + async def broadcast_alert(server_id: int, alert_type: str, alert_value: float, server_name: str = ""): """Broadcast alert to all WebSocket clients + Telegram @@ -344,6 +362,9 @@ async def broadcast_alert(server_id: int, alert_type: str, alert_value: float, s # Redis Pub/Sub (multi-worker) — local clients via subscriber only (C9) await _dispatch_ws_message(msg) + # Persist to alert_logs + await _save_alert_log(server_id, server_name, alert_type, f"{alert_value:.1f}%", is_recovery=False) + # Telegram push — deduplicated per (server_id, alert_type) with cooldown alert_key = f"alert:{server_id}:{alert_type}" if _should_push_telegram(alert_key): @@ -369,6 +390,9 @@ async def broadcast_recovery(server_id: int, metric: str, value: float, server_n await _dispatch_ws_message(msg) + # Persist recovery to alert_logs + await _save_alert_log(server_id, server_name, metric, f"{value:.1f}%", is_recovery=True) + # Recovery always sends Telegram + clears cooldown for next alert cycle alert_key = f"alert:{server_id}:{metric}" _last_alert_time.pop(alert_key, None) # Clear cooldown diff --git a/server/domain/models/__init__.py b/server/domain/models/__init__.py index 3f2769b1..d002a0de 100644 --- a/server/domain/models/__init__.py +++ b/server/domain/models/__init__.py @@ -338,6 +338,27 @@ class DbCredential(Base): # Web SSH 会话 & 命令日志 # ────────────────────────────────────────────── +class AlertLog(Base): + """Persistent alert & recovery history.""" + __tablename__ = "alert_logs" + + id = Column(Integer, primary_key=True, autoincrement=True) + server_id = Column(Integer, ForeignKey("servers.id", ondelete="CASCADE"), nullable=False) + server_name = Column(String(100), nullable=True, comment="冗余服务器名，便于列表显示") + alert_type = Column(String(20), nullable=False, comment="cpu|mem|disk|time_drift|system") + value = Column(String(50), nullable=True, comment="指标值（百分比或描述）") + is_recovery = Column(Boolean, default=False, comment="True=恢复通知") + created_at = Column(DateTime, default=_utcnow) + + server = relationship("Server") + + __table_args__ = ( + Index("idx_alert_logs_server_id", "server_id"), + Index("idx_alert_logs_created_at", "created_at"), + Index("idx_alert_logs_type", "alert_type"), + ) + + class SshSession(Base): """Web SSH session lifecycle tracking""" __tablename__ = "ssh_sessions" diff --git a/server/infrastructure/database/migrations.py b/server/infrastructure/database/migrations.py index ce575210..77cc2ee1 100644 --- a/server/infrastructure/database/migrations.py +++ b/server/infrastructure/database/migrations.py @@ -112,6 +112,19 @@ async def run_schema_migrations(): "ALTER TABLE push_schedules ADD COLUMN sync_mode VARCHAR(20) DEFAULT 'incremental' COMMENT '同步模式'", "ALTER TABLE admins ADD COLUMN token_version INT DEFAULT 0 COMMENT '令牌版本(重用时递增使旧token失效)'", "ALTER TABLE script_executions ADD COLUMN credential_id INT NULL COMMENT 'DB凭据替换$DB_*变量'", + # Alert log table + """CREATE TABLE IF NOT EXISTS `alert_logs` ( + `id` INT AUTO_INCREMENT PRIMARY KEY, + `server_id` INT NOT NULL, + `server_name` VARCHAR(100) NULL, + `alert_type` VARCHAR(20) NOT NULL, + `value` VARCHAR(50) NULL, + `is_recovery` TINYINT(1) DEFAULT 0, + `created_at` DATETIME DEFAULT CURRENT_TIMESTAMP, + INDEX `idx_alert_logs_server_id` (`server_id`), + INDEX `idx_alert_logs_created_at` (`created_at`), + INDEX `idx_alert_logs_type` (`alert_type`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""", # Script-schedule columns (push_schedules extended to support script execution) "ALTER TABLE push_schedules ADD COLUMN schedule_type VARCHAR(20) DEFAULT 'push' COMMENT '调度类型: push|script'", "ALTER TABLE push_schedules ADD COLUMN script_id INT NULL COMMENT '脚本库引用'", diff --git a/server/main.py b/server/main.py index 445ce346..3fc496f8 100644 --- a/server/main.py +++ b/server/main.py @@ -50,6 +50,7 @@ from server.api.settings import ( preset_router, audit_router, retry_router, + alert_history_router, ) from server.api.websocket import router as websocket_router from server.api.health import router as health_router @@ -365,6 +366,7 @@ app.include_router(settings_router) app.include_router(schedule_router) app.include_router(preset_router) app.include_router(audit_router) +app.include_router(alert_history_router) app.include_router(retry_router) # WebSocket + Health check routers (new in v6.0) diff --git a/web/agent/agent.py b/web/agent/agent.py index 87dd38cd..920f7b5e 100644 --- a/web/agent/agent.py +++ b/web/agent/agent.py @@ -45,6 +45,11 @@ CENTRAL_URL = config.get("central", {}).get("url", "") CENTRAL_API_KEY = config.get("central", {}).get("api_key", "") HEARTBEAT_INTERVAL = config.get("heartbeat_interval", 60) +# --- IP allowlist (only accept requests from trusted IPs) --- +# In config.json: "allowed_ips": ["1.2.3.4", "10.0.0.0/8"] +# Empty list = allow all (default for backward compat) +_ALLOWED_IPS: list[str] = config.get("allowed_ips") or [] + # --- Logging --- log_file = config.get("log_file", "/var/log/nexus-agent.log") @@ -65,7 +70,35 @@ logger = logging.getLogger("nexus-agent") app = FastAPI(title="Nexus Agent", version="2.0.0") -def verify_api_key(x_api_key: str = Header(default="")): +def _ip_allowed(client_ip: str) -> bool: + """Check client IP against allowed_ips list (supports CIDR and exact match).""" + if not _ALLOWED_IPS: + return True # Empty = allow all + import ipaddress + try: + addr = ipaddress.ip_address(client_ip) + for entry in _ALLOWED_IPS: + entry = entry.strip() + try: + if "/" in entry: + if addr in ipaddress.ip_network(entry, strict=False): + return True + elif addr == ipaddress.ip_address(entry): + return True + except ValueError: + if client_ip == entry: + return True + except ValueError: + return client_ip in _ALLOWED_IPS + return False + + +def verify_api_key(request: Request, x_api_key: str = Header(default="")): + # IP allowlist check first + client_ip = request.client.host if request.client else "" + if not _ip_allowed(client_ip): + logger.warning("Blocked request from %s (not in allowed_ips)", client_ip) + raise HTTPException(status_code=403, detail="IP not allowed") if not secrets.compare_digest(x_api_key or "", API_KEY): raise HTTPException(status_code=401, detail="Invalid API key") @@ -73,8 +106,11 @@ def verify_api_key(x_api_key: str = Header(default="")): # --- Health Endpoint --- @app.get("/health") -async def health_check(): +async def health_check(request: Request): """Health check endpoint — called by Nexus central server""" + client_ip = request.client.host if request.client else "" + if not _ip_allowed(client_ip): + raise HTTPException(status_code=403, detail="IP not allowed") try: cpu_percent = psutil.cpu_percent(interval=0.5) memory = psutil.virtual_memory() diff --git a/web/agent/install.sh b/web/agent/install.sh index 8947fd28..cb960149 100644 --- a/web/agent/install.sh +++ b/web/agent/install.sh @@ -66,17 +66,22 @@ curl -fsSL "${WEB_URL}/agent/agent.py" -o "$INSTALL_DIR/agent.py" || { # 4. Config echo "[4/5] Write config..." +# Extract central IP from URL for IP allowlist +CENTRAL_HOST=$(echo "${CENTRAL_URL}" | sed 's~https\?://~~' | cut -d'/' -f1 | cut -d':' -f1) + cat > "$INSTALL_DIR/config.json" << EOF { "server_id": ${SERVER_ID}, "central": { "url": "${CENTRAL_URL}", "api_key": "${API_KEY}" }, "api_key": "${API_KEY}", "heartbeat_interval": 60, + "allowed_ips": ["${CENTRAL_HOST}"], "log_file": "/var/log/nexus-agent.log", "log_level": "INFO", "server": { "host": "127.0.0.1", "port": ${AGENT_PORT} } } EOF +echo " IP allowlist: ${CENTRAL_HOST}" # 5. systemd + firewall + start echo "[5/5] Setup service + firewall + start..." diff --git a/web/app/alerts.html b/web/app/alerts.html new file mode 100644 index 00000000..1ef9e961 --- /dev/null +++ b/web/app/alerts.html @@ -0,0 +1,186 @@ +Nexus — 告警中心 + + +
+
+
+ +

告警中心

+
+
+ + + + + 至 + + +
+
+
+ + +
+ + + + + +
+ + + + + + + + + + + + + +
状态类型服务器数值时间
加载中...
+
+ + + +
+
+ + + + diff --git a/web/app/layout.js b/web/app/layout.js index 86c388e4..fb3ed612 100644 --- a/web/app/layout.js +++ b/web/app/layout.js @@ -19,6 +19,7 @@ function initLayout(activeNav) { const bottomNav = [ { id:'commands', icon:'💻', label:'命令日志', href:'/app/commands.html' }, + { id:'alerts', icon:'🔔', label:'告警中心', href:'/app/alerts.html' }, { id:'audit', icon:'📋', label:'审计日志', href:'/app/audit.html' }, { id:'settings', icon:'⚙️', label:'设置', href:'/app/settings.html' }, ]; diff --git a/web/app/servers.html b/web/app/servers.html index 77ea3ebe..4df74ac9 100644 --- a/web/app/servers.html +++ b/web/app/servers.html @@ -314,6 +314,10 @@
${remoteBtn} + ${d.is_online?``:''} 安装约需 30-60 秒；安装完成后 Agent 发送首次心跳约需再等 60 秒，状态徽章将自动更新为 ● Agent 在线
-
-

Telegram 通知

+
+

Telegram 通知

加载中...
+ +
+ + + 检测前请先向 Bot 发一条消息 +
+ +
@@ -574,6 +588,62 @@ if(key)saveSetting(key); }); + // ── Telegram Test + Chat ID Detection ── + async function telegramTest(){ + const btn=document.getElementById('telegramTestBtn'); + btn.disabled=true;btn.textContent='发送中...'; + try{ + const r=await apiFetch(API+'/settings/telegram/test',{method:'POST',headers:apiHeadersJSON()}); + if(!r)return; + const d=await r.json(); + if(d.success)toast('success','Telegram 测试消息发送成功！'); + else toast('error','发送失败: '+(d.detail||d.error||'未知错误')); + }catch(e){toast('error','请求失败: '+e.message)} + finally{btn.disabled=false;btn.textContent='📨 发送测试消息'} + } + + async function telegramGetChats(){ + const btn=document.getElementById('telegramChatBtn'); + btn.disabled=true;btn.textContent='检测中...'; + try{ + const r=await apiFetch(API+'/settings/telegram/chats'); + if(!r)return; + const d=await r.json(); + const listEl=document.getElementById('telegramChatList'); + if(!d.chats||d.chats.length===0){ + toast('warning','未检测到消息，请先向 Bot 发一条消息后再试'); + listEl.classList.add('hidden');return; + } + listEl.classList.remove('hidden'); + listEl.innerHTML='

检测到以下对话，点击填入 Chat ID：

'+ + d.chats.map(c=>`
+
+ ${esc(c.title||c.username||'私聊')} + ${esc(c.type)} + ${esc(String(c.id))} +
+ +
`).join(''); + toast('success',`检测到 ${d.chats.length} 个对话`); + }catch(e){toast('error','检测失败: '+e.message)} + finally{btn.disabled=false;btn.textContent='🔍 检测 Chat ID'} + } + + async function fillChatId(chatId){ + // Save chat_id to settings + const r=await apiFetch(API+'/settings/telegram_chat_id',{method:'PUT',headers:apiHeadersJSON(),body:JSON.stringify({value:chatId})}); + if(r&&r.ok){ + toast('success','Chat ID 已保存: '+chatId); + // Update the input field in telegramSection + const input=document.getElementById('set_telegram_chat_id'); + if(input)input.value=chatId; + loadSettings(); + }else toast('error','保存失败'); + } + // ── Notification Toggles ── const NOTIFY_ITEMS = [ { key:'notify_alert_cpu', label:'CPU 超阈值告警', desc:'CPU 使用率超过设定阈值时推送', group:'服务器资源' },