Your Name
a1b0b2c514
- terminal.html: left sidebar (default open), right server panel with server list - servers.py: fix SSH key double-encryption in update_server, auth method switch cleanup - servers.html: auth switch clears opposite credentials (key↔password) - deploy/pre_deploy_check.sh: 3-gate pre-deploy check (changelog/audit/test) - mcp/Nexus_server.py: deploy() runs gate check before git pull+restart - docs/audit/: audit record template + today's audit results - CLAUDE.md: add gate enforcement rules and progress bar discipline Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
1173 lines
45 KiB
Python
1173 lines
45 KiB
Python
"""Nexus — Servers API Routes (CRUD + heartbeat + health check + push)
|
||
Presentation layer — receives HTTP requests, delegates to ServerService/SyncService.
|
||
|
||
Live server status comes from Redis (real-time), base info from MySQL.
|
||
All write operations require JWT authentication and produce audit logs.
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
from typing import Optional
|
||
|
||
import bcrypt
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, Request
|
||
|
||
from server.api.dependencies import get_server_service, get_sync_service, get_db
|
||
from server.api.auth_jwt import get_current_admin
|
||
from server.api.schemas import (
|
||
ServerCreate, ServerUpdate, ServerCheck, ApiKeyRevealRequest,
|
||
ServerImportResult, BatchAgentAction, BatchAgentResult, BatchAgentResultItem,
|
||
)
|
||
from server.application.services.server_service import ServerService
|
||
from server.application.services.sync_service import SyncService
|
||
from server.domain.models import Server, SyncLog, Admin, AuditLog
|
||
from server.config import settings
|
||
from server.infrastructure.redis.client import get_redis
|
||
from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
logger = logging.getLogger("nexus.servers")
|
||
|
||
router = APIRouter(prefix="/api/servers", tags=["servers"])
|
||
|
||
REDIS_KEY_PREFIX = "heartbeat:"
|
||
|
||
|
||
# ── CRUD ──
|
||
|
||
@router.get("/", response_model=dict)
|
||
async def list_servers(
|
||
category: Optional[str] = Query(None, description="Filter by category"),
|
||
page: int = Query(1, ge=1, description="Page number"),
|
||
per_page: int = Query(50, ge=1, le=200, description="Items per page"),
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""List servers with live status from Redis (DB-level pagination)
|
||
|
||
Base info (name/IP/category) from MySQL with DB-level pagination,
|
||
real-time status from Redis. If Redis heartbeat key exists,
|
||
override is_online/system_info/last_heartbeat.
|
||
"""
|
||
from server.infrastructure.database.server_repo import ServerRepositoryImpl
|
||
|
||
offset = (page - 1) * per_page
|
||
repo = ServerRepositoryImpl(db)
|
||
page_servers, total = await repo.get_paginated(category, offset, per_page)
|
||
redis = get_redis()
|
||
|
||
result = []
|
||
for server in page_servers:
|
||
server_data = _server_to_dict(server)
|
||
|
||
# Overlay Redis heartbeat data (real-time)
|
||
try:
|
||
heartbeat = await redis.hgetall(f"{REDIS_KEY_PREFIX}{server.id}")
|
||
if heartbeat:
|
||
server_data["is_online"] = heartbeat.get("is_online") == "True"
|
||
if heartbeat.get("system_info"):
|
||
try:
|
||
server_data["system_info"] = json.loads(heartbeat.get("system_info", "{}"))
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass # Keep MySQL value
|
||
server_data["last_heartbeat"] = heartbeat.get("last_heartbeat", server_data.get("last_heartbeat"))
|
||
server_data["agent_version"] = heartbeat.get("agent_version", server_data.get("agent_version", ""))
|
||
# Time drift detection fields
|
||
try:
|
||
server_data["time_drift_seconds"] = float(heartbeat.get("time_drift_seconds", 0))
|
||
except (ValueError, TypeError):
|
||
server_data["time_drift_seconds"] = 0.0
|
||
server_data["drift_level"] = heartbeat.get("drift_level", "ok")
|
||
server_data["_source"] = "redis"
|
||
except Exception as e:
|
||
logger.warning(f"Redis read failed for server {server.id}: {e}")
|
||
|
||
result.append(server_data)
|
||
|
||
return {
|
||
"items": result,
|
||
"total": total,
|
||
"page": page,
|
||
"per_page": per_page,
|
||
"pages": (total + per_page - 1) // per_page,
|
||
}
|
||
|
||
|
||
# ── Dashboard Stats (must be before /{id} to avoid path collision) ──
|
||
|
||
@router.get("/stats", response_model=dict)
|
||
async def server_stats(
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Aggregated server stats for dashboard — SQL aggregation + Redis alert count"""
|
||
from sqlalchemy import func, text
|
||
|
||
# Total + category breakdown in a single query
|
||
result = await db.execute(
|
||
select(Server.category, func.count(Server.id))
|
||
.group_by(Server.category)
|
||
)
|
||
categories = {}
|
||
total = 0
|
||
for cat, cnt in result.all():
|
||
key = cat or "uncategorized"
|
||
categories[key] = cnt
|
||
total += cnt
|
||
|
||
# Online count from Redis heartbeats (real-time); MySQL is stale between flushes
|
||
redis = get_redis()
|
||
online = 0
|
||
try:
|
||
cursor = 0
|
||
while True:
|
||
cursor, keys = await redis.scan(cursor, match="heartbeat:*", count=200)
|
||
for key in keys:
|
||
data = await redis.hgetall(key)
|
||
if data.get("is_online") == "True":
|
||
online += 1
|
||
if cursor == 0:
|
||
break
|
||
except Exception as e:
|
||
logger.warning(f"Failed to count Redis heartbeats in server_stats: {e}")
|
||
result = await db.execute(
|
||
select(func.count(Server.id)).where(Server.is_online == True)
|
||
)
|
||
online = int(result.scalar() or 0)
|
||
|
||
alerts = 0
|
||
try:
|
||
cursor = 0
|
||
while True:
|
||
cursor, keys = await redis.scan(cursor, match="alerts:*", count=200)
|
||
alerts += len(keys)
|
||
if cursor == 0:
|
||
break
|
||
except Exception as e:
|
||
logger.warning(f"Failed to count Redis alerts in server_stats: {e}")
|
||
|
||
return {
|
||
"total": total,
|
||
"online": online,
|
||
"offline": max(0, total - online),
|
||
"alerts": alerts,
|
||
"categories": categories,
|
||
}
|
||
|
||
|
||
# ── Sync Logs ──
|
||
|
||
@router.get("/logs", response_model=dict)
|
||
async def get_all_sync_logs(
|
||
server_id: Optional[int] = Query(None, description="Filter by server ID"),
|
||
status: Optional[str] = Query(None, description="success|failed|running"),
|
||
sync_mode: Optional[str] = Query(None, description="incremental|full|overwrite|checksum|command"),
|
||
trigger_type: Optional[str] = Query(None, description="manual|schedule|retry|batch"),
|
||
offset: int = Query(0, ge=0),
|
||
limit: int = Query(50, ge=1, le=500),
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Get paginated sync logs across all servers with optional filters."""
|
||
from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl
|
||
repo = SyncLogRepositoryImpl(db)
|
||
rows, total = await repo.get_all_paginated(
|
||
server_id=server_id, status=status, sync_mode=sync_mode,
|
||
trigger_type=trigger_type, offset=offset, limit=limit,
|
||
)
|
||
items = [_sync_log_to_dict(log, server_name=name) for log, name in rows]
|
||
return {"items": items, "total": total, "offset": offset, "limit": limit}
|
||
|
||
|
||
# ── CSV Batch Import ──
|
||
|
||
MAX_IMPORT_FILE_SIZE = 1_048_576 # 1 MB
|
||
MAX_IMPORT_ROWS = 500
|
||
|
||
# CSV column mapping: CSV header (Chinese) → Server model field
|
||
CSV_COLUMNS = [
|
||
("名称", "name", True),
|
||
("地址", "domain", True),
|
||
("SSH端口", "port", False),
|
||
("用户名", "username", False),
|
||
("认证方式", "auth_method", False),
|
||
("密码", "password", False),
|
||
("密钥路径", "ssh_key_path", False),
|
||
("私钥内容", "ssh_key_private", False),
|
||
("分类", "category", False),
|
||
("目标路径", "target_path", False),
|
||
("备注", "description", False),
|
||
]
|
||
|
||
# Columns excluded from the template but still accepted on import (backward compat)
|
||
_CSV_IMPORT_ONLY_COLUMNS = [
|
||
("Agent端口", "agent_port", False),
|
||
("平台", "platform_name", False),
|
||
("节点", "node_name", False),
|
||
]
|
||
|
||
|
||
def _sanitize_csv_cell(value: str) -> str:
|
||
"""Sanitize a CSV cell to prevent CSV injection (Excel formula injection).
|
||
|
||
Strips leading dangerous characters: = + - @ \t \r
|
||
that could be interpreted as formulas by spreadsheet software.
|
||
"""
|
||
if not value:
|
||
return value
|
||
# Remove leading whitespace and dangerous formula prefix characters
|
||
stripped = value.lstrip()
|
||
if stripped and stripped[0] in ("=", "+", "-", "@", "\t", "\r"):
|
||
# Prepend a single quote to defang the formula (standard CSV injection defense)
|
||
return "'" + value
|
||
return value
|
||
|
||
|
||
@router.get("/import/template")
|
||
async def download_import_template(
|
||
admin: Admin = Depends(get_current_admin),
|
||
):
|
||
"""Download CSV import template with headers and one example row."""
|
||
from fastapi.responses import StreamingResponse
|
||
import io
|
||
|
||
headers = ",".join(col[0] for col in CSV_COLUMNS)
|
||
example = "web-01,192.168.1.10,22,root,password,your-password,,,商城,/www/wwwroot,Web服务器"
|
||
csv_content = f"{headers}\n{example}\n" # BOM for Excel Chinese compatibility
|
||
|
||
return StreamingResponse(
|
||
io.BytesIO(csv_content.encode("utf-8")),
|
||
media_type="text/csv; charset=utf-8",
|
||
headers={"Content-Disposition": "attachment; filename=servers_import_template.csv"},
|
||
)
|
||
|
||
|
||
@router.post("/import", response_model=ServerImportResult)
|
||
async def import_servers(
|
||
request: Request,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Import servers from a CSV file (multipart/form-data upload).
|
||
|
||
Each row is processed independently — a failure in one row does not affect others.
|
||
Duplicate domains (already in DB) are skipped, not counted as errors.
|
||
"""
|
||
from server.infrastructure.database.crypto import encrypt_value
|
||
from server.infrastructure.database.server_repo import ServerRepositoryImpl
|
||
import csv
|
||
import io
|
||
import secrets
|
||
|
||
# Read the uploaded file
|
||
form = await request.form()
|
||
file = form.get("file")
|
||
if not file or not hasattr(file, "read"):
|
||
raise HTTPException(status_code=400, detail="请上传 CSV 文件")
|
||
|
||
content = await file.read()
|
||
if len(content) > MAX_IMPORT_FILE_SIZE:
|
||
raise HTTPException(status_code=400, detail=f"文件大小超过限制 ({MAX_IMPORT_FILE_SIZE // 1024}KB)")
|
||
|
||
# Decode CSV
|
||
try:
|
||
text = content.decode("utf-8-sig") # Handle BOM
|
||
except UnicodeDecodeError:
|
||
try:
|
||
text = content.decode("gbk") # Fallback for Chinese Excel
|
||
except UnicodeDecodeError:
|
||
raise HTTPException(status_code=400, detail="CSV 编码不支持，请使用 UTF-8 或 GBK")
|
||
|
||
reader = csv.DictReader(io.StringIO(text))
|
||
if not reader.fieldnames:
|
||
raise HTTPException(status_code=400, detail="CSV 文件为空或缺少表头")
|
||
|
||
# Validate headers — at least name and domain must exist
|
||
header_set = set(reader.fieldnames)
|
||
required_headers = {col[0] for col in CSV_COLUMNS if col[2]}
|
||
missing = required_headers - header_set
|
||
if missing:
|
||
raise HTTPException(status_code=400, detail=f"CSV 缺少必填列: {', '.join(missing)}")
|
||
|
||
# Build column mapping: CSV header → field name (template columns + backward-compat columns)
|
||
col_map = {}
|
||
all_columns = CSV_COLUMNS + _CSV_IMPORT_ONLY_COLUMNS
|
||
for cn_name, field_name, _ in all_columns:
|
||
if cn_name in header_set:
|
||
col_map[cn_name] = field_name
|
||
|
||
# Load existing domains for duplicate check
|
||
server_repo = ServerRepositoryImpl(db)
|
||
existing_servers = await server_repo.get_all()
|
||
existing_domains = {s.domain for s in existing_servers}
|
||
|
||
# Load platform/node lookup by name for mapping
|
||
from server.domain.models import Platform, Node
|
||
platform_map = {} # name → id
|
||
node_map = {} # name → id
|
||
try:
|
||
result = await db.execute(select(Platform))
|
||
for p in result.scalars().all():
|
||
platform_map[p.name] = p.id
|
||
result = await db.execute(select(Node))
|
||
for n in result.scalars().all():
|
||
node_map[n.name] = n.id
|
||
except Exception as e:
|
||
logger.warning(f"Platform/node lookup failed during CSV import: {e}")
|
||
|
||
result = ServerImportResult()
|
||
row_num = 1 # Header is row 0
|
||
|
||
for row in reader:
|
||
row_num += 1
|
||
if row_num - 1 > MAX_IMPORT_ROWS:
|
||
result.errors.append({"row": row_num, "name": "", "domain": "", "error": f"超过最大行数限制 ({MAX_IMPORT_ROWS})"})
|
||
break
|
||
|
||
# Map CSV columns to server data
|
||
data = {}
|
||
for cn_name, field_name, _ in all_columns:
|
||
if cn_name in col_map:
|
||
value = row.get(cn_name, "").strip()
|
||
if value:
|
||
value = _sanitize_csv_cell(value)
|
||
data[field_name] = value
|
||
|
||
# Validate required fields
|
||
name = data.get("name", "").strip()
|
||
domain = data.get("domain", "").strip()
|
||
if not name or not domain:
|
||
result.failed += 1
|
||
result.errors.append({"row": row_num, "name": name, "domain": domain, "error": "名称和地址为必填项"})
|
||
continue
|
||
|
||
# Duplicate check
|
||
if domain in existing_domains:
|
||
result.skipped += 1
|
||
continue
|
||
|
||
# Parse numeric fields
|
||
try:
|
||
if "port" in data:
|
||
data["port"] = int(data["port"])
|
||
if "agent_port" in data:
|
||
data["agent_port"] = int(data["agent_port"])
|
||
except ValueError:
|
||
result.failed += 1
|
||
result.errors.append({"row": row_num, "name": name, "domain": domain, "error": "端口号必须为数字"})
|
||
continue
|
||
|
||
# Set defaults
|
||
data.setdefault("port", 22)
|
||
data.setdefault("username", "root")
|
||
data.setdefault("auth_method", "password")
|
||
data.setdefault("agent_port", 8601)
|
||
|
||
# Validate auth_method
|
||
if data["auth_method"] not in ("password", "key"):
|
||
result.failed += 1
|
||
result.errors.append({"row": row_num, "name": name, "domain": domain, "error": f"认证方式无效: {data['auth_method']}"})
|
||
continue
|
||
|
||
# Resolve platform/node by name
|
||
if data.get("platform_name") and data["platform_name"] in platform_map:
|
||
data["platform_id"] = platform_map[data["platform_name"]]
|
||
if data.get("node_name") and data["node_name"] in node_map:
|
||
data["node_id"] = node_map[data["node_name"]]
|
||
# Remove helper fields that are not Server model columns
|
||
data.pop("platform_name", None)
|
||
data.pop("node_name", None)
|
||
|
||
# Encrypt sensitive fields
|
||
try:
|
||
if data.get("password"):
|
||
data["password"] = encrypt_value(data["password"])
|
||
if data.get("ssh_key_private"):
|
||
data["ssh_key_private"] = encrypt_value(data["ssh_key_private"])
|
||
data["ssh_key_configured"] = True
|
||
|
||
# Auto-generate Agent API Key
|
||
data["agent_api_key"] = f"nxs-{secrets.token_urlsafe(32)}"
|
||
|
||
server = Server(**data)
|
||
created = await service.create_server(server)
|
||
existing_domains.add(domain)
|
||
|
||
result.created += 1
|
||
result.created_ids.append(created.id)
|
||
except Exception as e:
|
||
result.failed += 1
|
||
result.errors.append({"row": row_num, "name": name, "domain": domain, "error": str(e)[:200]})
|
||
logger.warning(f"CSV import row {row_num} failed: {e}")
|
||
|
||
result.total = result.created + result.skipped + result.failed
|
||
|
||
# Audit log
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="import_servers",
|
||
target_type="server",
|
||
target_id=0,
|
||
detail=f"CSV导入: 成功{result.created} 跳过{result.skipped} 失败{result.failed}",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
return result
|
||
|
||
|
||
# ── Batch Agent Install/Upgrade ──
|
||
|
||
@router.post("/batch/install-agent", response_model=BatchAgentResult)
|
||
async def batch_install_agent(
|
||
request: Request,
|
||
payload: BatchAgentAction,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Install Nexus Agent on multiple servers via SSH (concurrent, max 5 parallel)."""
|
||
import asyncio
|
||
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
|
||
import shlex
|
||
|
||
base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
|
||
if not base_url:
|
||
raise HTTPException(status_code=400, detail="NEXUS_API_BASE_URL 未配置")
|
||
|
||
sem = asyncio.Semaphore(5)
|
||
results = []
|
||
|
||
async def _install_one(sid: int):
|
||
async with sem:
|
||
try:
|
||
server = await service.get_server(sid)
|
||
if not server:
|
||
return BatchAgentResultItem(server_id=sid, server_name="", success=False, error="服务器不存在")
|
||
api_key = server.agent_api_key or settings.API_KEY
|
||
if not api_key:
|
||
return BatchAgentResultItem(server_id=sid, server_name=server.name, success=False, error="请先生成 Agent API Key")
|
||
install_cmd = (
|
||
f"curl -fsSL {shlex.quote(base_url)}/agent/install.sh | bash -s -- "
|
||
f"--url {shlex.quote(base_url)} --key {shlex.quote(api_key)} "
|
||
f"--id {sid} --port {shlex.quote(str(server.agent_port or 8601))}"
|
||
)
|
||
r = await exec_ssh_command(server, install_cmd, timeout=120)
|
||
ok = r["exit_code"] == 0
|
||
return BatchAgentResultItem(
|
||
server_id=sid, server_name=server.name, success=ok,
|
||
stdout=r["stdout"][:1000] if ok else "",
|
||
error=r["stderr"][:300] if not ok else "",
|
||
)
|
||
except Exception as e:
|
||
return BatchAgentResultItem(server_id=sid, server_name="", success=False, error=str(e)[:200])
|
||
|
||
items = await asyncio.gather(*[_install_one(sid) for sid in payload.server_ids])
|
||
results = list(items)
|
||
|
||
# Audit
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
ok_count = sum(1 for r in results if r.success)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="batch_install_agent",
|
||
target_type="server",
|
||
target_id=0,
|
||
detail=f"批量安装Agent: {ok_count}/{len(results)} 成功",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
return BatchAgentResult(results=results)
|
||
|
||
|
||
@router.post("/batch/upgrade-agent", response_model=BatchAgentResult)
|
||
async def batch_upgrade_agent(
|
||
request: Request,
|
||
payload: BatchAgentAction,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Upgrade Nexus Agent on multiple servers via SSH (concurrent, max 5 parallel)."""
|
||
import asyncio
|
||
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
|
||
import shlex
|
||
|
||
base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
|
||
if not base_url:
|
||
raise HTTPException(status_code=400, detail="NEXUS_API_BASE_URL 未配置")
|
||
|
||
sem = asyncio.Semaphore(5)
|
||
results = []
|
||
|
||
async def _upgrade_one(sid: int):
|
||
async with sem:
|
||
try:
|
||
server = await service.get_server(sid)
|
||
if not server:
|
||
return BatchAgentResultItem(server_id=sid, server_name="", success=False, error="服务器不存在")
|
||
|
||
agent_url = f"{base_url}/agent/agent.py"
|
||
install_dir = "/opt/nexus-agent"
|
||
venv_python = f"{install_dir}/.venv/bin/python"
|
||
|
||
upgrade_cmd = (
|
||
f"{venv_python} -c 'import sys; v=sys.version_info; sys.exit(0 if v.major==3 and v.minor>=10 else 1)' "
|
||
f"&& {venv_python} -m pip install -q fastapi==0.115.6 uvicorn==0.34.0 httpx==0.28.1 psutil python-multipart==0.0.19 "
|
||
f"&& cp {install_dir}/agent.py {install_dir}/agent.py.bak "
|
||
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'"
|
||
)
|
||
r = await exec_ssh_command(server, upgrade_cmd, timeout=120)
|
||
ok = r["exit_code"] == 0 and "upgrade_ok" in r["stdout"]
|
||
|
||
if not ok:
|
||
# Attempt rollback
|
||
try:
|
||
await exec_ssh_command(
|
||
server,
|
||
f"cp {install_dir}/agent.py.bak {install_dir}/agent.py && systemctl restart nexus-agent",
|
||
timeout=30,
|
||
)
|
||
except Exception as rb_err:
|
||
logger.warning(f"Rollback failed for server {sid}: {rb_err}")
|
||
|
||
return BatchAgentResultItem(
|
||
server_id=sid, server_name=server.name, success=ok,
|
||
stdout=r["stdout"][:500] if ok else "",
|
||
error=r["stderr"][:300] if not ok else "",
|
||
)
|
||
except Exception as e:
|
||
return BatchAgentResultItem(server_id=sid, server_name="", success=False, error=str(e)[:200])
|
||
|
||
items = await asyncio.gather(*[_upgrade_one(sid) for sid in payload.server_ids])
|
||
results = list(items)
|
||
|
||
# Audit
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
ok_count = sum(1 for r in results if r.success)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="batch_upgrade_agent",
|
||
target_type="server",
|
||
target_id=0,
|
||
detail=f"批量升级Agent: {ok_count}/{len(results)} 成功",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
return BatchAgentResult(results=results)
|
||
|
||
@router.get("/{id}", response_model=dict)
|
||
async def get_server(
|
||
id: int,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
):
|
||
"""Get a single server by ID with live status from Redis"""
|
||
server = await service.get_server(id)
|
||
if not server:
|
||
raise HTTPException(status_code=404, detail="Server not found")
|
||
|
||
server_data = _server_to_dict(server)
|
||
|
||
# Overlay Redis heartbeat data
|
||
redis = get_redis()
|
||
try:
|
||
heartbeat = await redis.hgetall(f"{REDIS_KEY_PREFIX}{id}")
|
||
if heartbeat:
|
||
server_data["is_online"] = heartbeat.get("is_online") == "True"
|
||
if heartbeat.get("system_info"):
|
||
try:
|
||
server_data["system_info"] = json.loads(heartbeat.get("system_info", "{}"))
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
server_data["last_heartbeat"] = heartbeat.get("last_heartbeat", server_data.get("last_heartbeat"))
|
||
server_data["agent_version"] = heartbeat.get("agent_version", server_data.get("agent_version", ""))
|
||
try:
|
||
server_data["time_drift_seconds"] = float(heartbeat.get("time_drift_seconds", 0))
|
||
except (ValueError, TypeError):
|
||
server_data["time_drift_seconds"] = 0.0
|
||
server_data["drift_level"] = heartbeat.get("drift_level", "ok")
|
||
server_data["_source"] = "redis"
|
||
except Exception as e:
|
||
logger.warning(f"Redis read failed for server {id}: {e}")
|
||
|
||
return server_data
|
||
|
||
|
||
@router.post("/", response_model=dict, status_code=201)
|
||
async def create_server(
|
||
request: Request,
|
||
payload: ServerCreate,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Create a new server (sensitive fields encrypted at rest)"""
|
||
from server.infrastructure.database.crypto import encrypt_value, decrypt_value
|
||
from server.infrastructure.database.password_preset_repo import PasswordPresetRepositoryImpl
|
||
from server.infrastructure.database.ssh_key_preset_repo import SshKeyPresetRepositoryImpl
|
||
|
||
data = payload.model_dump(exclude_none=True)
|
||
|
||
# If preset_id is provided, resolve the preset password (server-side, no re-auth needed)
|
||
if data.get("preset_id") and data.get("auth_method") == "password" and not data.get("password"):
|
||
preset_repo = PasswordPresetRepositoryImpl(db)
|
||
preset = await preset_repo.get_by_id(data["preset_id"])
|
||
if preset:
|
||
data["password"] = decrypt_value(preset.encrypted_pw)
|
||
# If preset not found, fall through — user may provide password manually
|
||
|
||
# If ssh_key_preset_id is provided, resolve the preset private key (server-side, no re-auth needed)
|
||
if data.get("ssh_key_preset_id") and data.get("auth_method") == "key" and not data.get("ssh_key_private"):
|
||
ssh_key_preset_repo = SshKeyPresetRepositoryImpl(db)
|
||
ssh_key_preset = await ssh_key_preset_repo.get_by_id(data["ssh_key_preset_id"])
|
||
if ssh_key_preset:
|
||
data["ssh_key_private"] = decrypt_value(ssh_key_preset.encrypted_private_key)
|
||
if ssh_key_preset.public_key and not data.get("ssh_key_public"):
|
||
data["ssh_key_public"] = ssh_key_preset.public_key
|
||
# If preset not found, fall through — user may provide key manually
|
||
|
||
# Encrypt sensitive fields before storing
|
||
if data.get("password"):
|
||
data["password"] = encrypt_value(data["password"])
|
||
if data.get("ssh_key_private"):
|
||
data["ssh_key_private"] = encrypt_value(data["ssh_key_private"])
|
||
data["ssh_key_configured"] = True
|
||
|
||
# Auto-generate Agent API Key — used by /install-agent, no need to expose to frontend
|
||
import secrets
|
||
data["agent_api_key"] = f"nxs-{secrets.token_urlsafe(32)}"
|
||
|
||
server = Server(**data)
|
||
created = await service.create_server(server)
|
||
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="create_server",
|
||
target_type="server",
|
||
target_id=created.id,
|
||
detail=f"name={created.name} domain={created.domain}",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
return _server_to_dict(created)
|
||
|
||
|
||
@router.put("/{id}", response_model=dict)
|
||
async def update_server(
|
||
request: Request,
|
||
id: int,
|
||
payload: ServerUpdate,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Update an existing server (sensitive fields encrypted at rest)"""
|
||
from server.infrastructure.database.crypto import encrypt_value, decrypt_value
|
||
from server.infrastructure.database.password_preset_repo import PasswordPresetRepositoryImpl
|
||
from server.infrastructure.database.ssh_key_preset_repo import SshKeyPresetRepositoryImpl
|
||
|
||
server = await service.get_server(id)
|
||
if not server:
|
||
raise HTTPException(status_code=404, detail="Server not found")
|
||
|
||
# If preset_id is provided (and not None), resolve the preset password
|
||
update_data = payload.model_dump(exclude_unset=True)
|
||
if "preset_id" in update_data and update_data["preset_id"] is not None and not update_data.get("password"):
|
||
preset_repo = PasswordPresetRepositoryImpl(db)
|
||
preset = await preset_repo.get_by_id(update_data["preset_id"])
|
||
if preset:
|
||
update_data["password"] = decrypt_value(preset.encrypted_pw)
|
||
|
||
# If ssh_key_preset_id is provided, resolve the preset private key
|
||
if "ssh_key_preset_id" in update_data and update_data["ssh_key_preset_id"] is not None and not update_data.get("ssh_key_private"):
|
||
ssh_key_preset_repo = SshKeyPresetRepositoryImpl(db)
|
||
ssh_key_preset = await ssh_key_preset_repo.get_by_id(update_data["ssh_key_preset_id"])
|
||
if ssh_key_preset:
|
||
update_data["ssh_key_private"] = decrypt_value(ssh_key_preset.encrypted_private_key)
|
||
if ssh_key_preset.public_key:
|
||
update_data["ssh_key_public"] = ssh_key_preset.public_key
|
||
|
||
# Explicit allowlist to prevent mass-assignment to sensitive columns
|
||
# agent_api_key is excluded — must be set via dedicated /agent-key endpoint
|
||
ALLOWED_FIELDS = {
|
||
"name", "domain", "port", "username", "auth_method",
|
||
"password", "ssh_key_path", "ssh_key_private", "ssh_key_public",
|
||
"agent_port", "preset_id", "ssh_key_preset_id", "description", "target_path", "category",
|
||
"platform_id", "node_id", "protocols", "extra_attrs",
|
||
}
|
||
|
||
# When auth_method changes, clear the opposite side's credentials
|
||
new_auth = update_data.get("auth_method")
|
||
if new_auth == "password":
|
||
# Switching to password: clear key-side fields if not explicitly provided
|
||
update_data.setdefault("ssh_key_private", None)
|
||
update_data.setdefault("ssh_key_path", None)
|
||
update_data.setdefault("ssh_key_public", None)
|
||
update_data.setdefault("ssh_key_preset_id", None)
|
||
# Also clear ssh_key_configured on the server object
|
||
server.ssh_key_configured = False
|
||
elif new_auth == "key":
|
||
# Switching to key: clear password-side fields if not explicitly provided
|
||
update_data.setdefault("password", None)
|
||
update_data.setdefault("preset_id", None)
|
||
|
||
for key, value in update_data.items():
|
||
if key not in ALLOWED_FIELDS:
|
||
continue
|
||
# Encrypt sensitive fields on update
|
||
if key == "password" and value:
|
||
value = encrypt_value(value)
|
||
elif key == "ssh_key_private" and value:
|
||
value = encrypt_value(value)
|
||
server.ssh_key_configured = True # Mark key as configured when private key is set
|
||
setattr(server, key, value)
|
||
updated = await service.update_server(server)
|
||
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="update_server",
|
||
target_type="server",
|
||
target_id=id,
|
||
detail=f"name={updated.name}",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
return _server_to_dict(updated)
|
||
|
||
|
||
@router.delete("/{id}", status_code=204)
|
||
async def delete_server(
|
||
request: Request,
|
||
id: int,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Delete a server"""
|
||
server = await service.get_server(id)
|
||
if not server:
|
||
raise HTTPException(status_code=404, detail="Server not found")
|
||
result = await service.delete_server(id)
|
||
if not result:
|
||
raise HTTPException(status_code=404, detail="Server not found")
|
||
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="delete_server",
|
||
target_type="server",
|
||
target_id=id,
|
||
detail=f"name={server.name}",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
|
||
# ── Agent API Key ──
|
||
|
||
@router.post("/{id}/agent-key", response_model=dict)
|
||
async def generate_agent_api_key(
|
||
request: Request,
|
||
id: int,
|
||
payload: ApiKeyRevealRequest,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Generate a new per-server Agent API key (re-auth required)"""
|
||
# Re-authenticate: require current password (same pattern as reveal_api_key)
|
||
from server.infrastructure.database.admin_repo import AdminRepositoryImpl
|
||
|
||
admin_repo = AdminRepositoryImpl(db)
|
||
current = await admin_repo.get_by_id(admin.id)
|
||
if not current or not bcrypt.checkpw(
|
||
payload.current_password.encode(),
|
||
current.password_hash.encode(),
|
||
):
|
||
raise HTTPException(status_code=400, detail="当前密码错误")
|
||
|
||
server = await service.get_server(id)
|
||
if not server:
|
||
raise HTTPException(status_code=404, detail="Server not found")
|
||
|
||
import secrets
|
||
new_key = f"nxs-{secrets.token_urlsafe(32)}"
|
||
server.agent_api_key = new_key
|
||
await service.update_server(server)
|
||
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="generate_agent_key",
|
||
target_type="server",
|
||
target_id=id,
|
||
detail=f"Generated new Agent API key for {server.name}",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
return {
|
||
"server_id": id,
|
||
"agent_api_key": new_key,
|
||
"agent_api_key_preview": f"{new_key[:12]}...",
|
||
"display_once": True,
|
||
"message": "完整密钥仅在本响应中返回一次，请立即复制保存",
|
||
}
|
||
|
||
|
||
# ── Agent Install ──
|
||
|
||
@router.post("/{id}/install-agent", response_model=dict)
|
||
async def install_agent_remote(
|
||
request: Request,
|
||
id: int,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Install Nexus Agent on the managed server via SSH.
|
||
|
||
Uses the server's stored SSH credentials to connect and run install.sh.
|
||
Requires NEXUS_API_BASE_URL to be configured (used in the install command).
|
||
"""
|
||
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
|
||
from server.infrastructure.database.crypto import decrypt_value
|
||
import shlex
|
||
|
||
server = await service.get_server(id)
|
||
if not server:
|
||
raise HTTPException(status_code=404, detail="Server not found")
|
||
|
||
# Validate prerequisites
|
||
base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
|
||
if not base_url:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="NEXUS_API_BASE_URL 未配置，无法生成安装命令。请在系统设置中配置主站对外 URL。",
|
||
)
|
||
|
||
api_key = server.agent_api_key or settings.API_KEY
|
||
if not api_key:
|
||
raise HTTPException(status_code=400, detail="请先为该服务器生成 Agent API Key")
|
||
|
||
agent_port = server.agent_port or 8601
|
||
|
||
# Build install command (all variables shell-escaped to prevent injection)
|
||
install_cmd = (
|
||
f"curl -fsSL {shlex.quote(base_url)}/agent/install.sh | bash -s -- "
|
||
f"--url {shlex.quote(base_url)} --key {shlex.quote(api_key)} "
|
||
f"--id {id} --port {shlex.quote(str(agent_port))}"
|
||
)
|
||
|
||
# Execute via SSH
|
||
try:
|
||
result = await exec_ssh_command(server, install_cmd, timeout=120)
|
||
except Exception as e:
|
||
raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}")
|
||
|
||
if result["exit_code"] != 0:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"安装脚本失败 (exit {result['exit_code']}): {result['stderr'][:500]}",
|
||
)
|
||
|
||
# Audit
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="install_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"][:3000],
|
||
"exit_code": result["exit_code"],
|
||
}
|
||
|
||
|
||
@router.post("/{id}/agent-install-cmd", response_model=dict)
|
||
async def get_agent_install_cmd(
|
||
id: int,
|
||
payload: "ApiKeyRevealRequest",
|
||
request: Request,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Return the curl install command for this server (re-auth required).
|
||
|
||
Security: Only exposes per-server agent_api_key after password verification.
|
||
Never falls back to global API_KEY — that would expose all servers' auth.
|
||
"""
|
||
# Re-authenticate: require current password (same pattern as reveal_api_key)
|
||
from server.infrastructure.database.admin_repo import AdminRepositoryImpl
|
||
|
||
admin_repo = AdminRepositoryImpl(db)
|
||
current = await admin_repo.get_by_id(admin.id)
|
||
if not current or not bcrypt.checkpw(
|
||
payload.current_password.encode(),
|
||
current.password_hash.encode(),
|
||
):
|
||
raise HTTPException(status_code=400, detail="当前密码错误")
|
||
|
||
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("/")
|
||
agent_port = server.agent_port or 8601
|
||
|
||
# Only use per-server key — NEVER expose global API_KEY via this endpoint
|
||
api_key = server.agent_api_key
|
||
has_key = bool(api_key)
|
||
cmd = None
|
||
if base_url and api_key:
|
||
import shlex
|
||
cmd = (
|
||
f"curl -fsSL {shlex.quote(base_url)}/agent/install.sh | bash -s -- "
|
||
f"--url {shlex.quote(base_url)} --key {shlex.quote(api_key)} "
|
||
f"--id {id} --port {shlex.quote(str(agent_port))}"
|
||
)
|
||
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="reveal_install_cmd",
|
||
target_type="server",
|
||
target_id=id,
|
||
detail=f"Revealed install command for {server.name}",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
return {
|
||
"server_id": id,
|
||
"server_name": server.name,
|
||
"base_url": base_url,
|
||
"has_agent_key": has_key,
|
||
"agent_port": agent_port,
|
||
"install_cmd": cmd,
|
||
"is_online": server.is_online,
|
||
}
|
||
|
||
|
||
# ── 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"
|
||
venv_python = f"{install_dir}/.venv/bin/python"
|
||
|
||
# Step 1: Check Python version in venv
|
||
pyver_cmd = (
|
||
f"{venv_python} -c '"
|
||
"import sys; v=sys.version_info; "
|
||
"print(f\"py_ok_{v.major}.{v.minor}.{v.micro}\") "
|
||
"if v.major==3 and v.minor>=10 else sys.exit(1)'"
|
||
)
|
||
try:
|
||
pyver = await exec_ssh_command(server, pyver_cmd, timeout=15)
|
||
except Exception as e:
|
||
raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}")
|
||
|
||
if pyver["exit_code"] != 0 or "py_ok_" not in pyver["stdout"]:
|
||
ver_info = pyver["stderr"].replace("\n", "; ").strip()[:200] if pyver["stderr"] else "venv 不存在或 Python 版本 < 3.10"
|
||
raise HTTPException(
|
||
status_code=424,
|
||
detail=(
|
||
f"子机 Python 版本不满足要求（需要 3.10+），请手动修复后重试。\n"
|
||
f" venv 路径: {venv_python}\n"
|
||
f" 详情: {ver_info}"
|
||
),
|
||
)
|
||
|
||
# Step 2: Auto-update pip dependencies (locked versions)
|
||
pip_cmd = (
|
||
f"{venv_python} -m pip install -q "
|
||
f"fastapi==0.115.6 uvicorn==0.34.0 httpx==0.28.1 psutil python-multipart==0.0.19"
|
||
)
|
||
|
||
# Step 3: Backup current agent.py
|
||
backup_cmd = f"cp {install_dir}/agent.py {install_dir}/agent.py.bak"
|
||
|
||
# Step 4: pip install + backup + download new agent.py → restart service
|
||
upgrade_cmd = (
|
||
f"{pip_cmd} "
|
||
f"&& {backup_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=120)
|
||
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:
|
||
# Rollback: restore backup and restart
|
||
rollback_cmd = (
|
||
f"cp {install_dir}/agent.py.bak {install_dir}/agent.py "
|
||
f"&& systemctl restart nexus-agent"
|
||
)
|
||
try:
|
||
await exec_ssh_command(server, rollback_cmd, timeout=30)
|
||
except Exception as rb_err:
|
||
logger.warning(f"Rollback failed for server {id}: {rb_err}")
|
||
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)
|
||
async def check_servers(
|
||
payload: ServerCheck,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
):
|
||
"""Check health status of specified servers via SSH probe (no Agent port required)"""
|
||
return await service.check_all_servers(payload.server_ids)
|
||
|
||
|
||
|
||
@router.get("/{id}/logs", response_model=list)
|
||
async def get_server_logs(
|
||
id: int,
|
||
limit: int = Query(50, ge=1, le=200),
|
||
admin: Admin = Depends(get_current_admin),
|
||
sync_service: SyncService = Depends(get_sync_service),
|
||
):
|
||
"""Get sync logs for a specific server"""
|
||
logs = await sync_service.get_sync_logs(id, limit)
|
||
return [_sync_log_to_dict(log) for log in logs]
|
||
|
||
|
||
# ── Helper functions ──
|
||
|
||
def _server_to_dict(server: Server) -> dict:
|
||
"""Convert Server model to API response dict (hide sensitive fields)"""
|
||
return {
|
||
"id": server.id,
|
||
"name": server.name,
|
||
"domain": server.domain,
|
||
"port": server.port,
|
||
"username": server.username,
|
||
"auth_method": server.auth_method,
|
||
"password_set": bool(server.password),
|
||
"ssh_key_path": server.ssh_key_path,
|
||
"ssh_key_public": server.ssh_key_public or "",
|
||
"ssh_key_private_set": bool(server.ssh_key_private),
|
||
"agent_port": server.agent_port,
|
||
"agent_api_key": (server.agent_api_key[:8] + "...") if server.agent_api_key and len(server.agent_api_key) > 8 else (server.agent_api_key or ""),
|
||
"agent_api_key_set": bool(server.agent_api_key),
|
||
"preset_id": server.preset_id,
|
||
"ssh_key_preset_id": server.ssh_key_preset_id,
|
||
"description": server.description,
|
||
"target_path": server.target_path,
|
||
"category": server.category,
|
||
"platform_id": server.platform_id,
|
||
"node_id": server.node_id,
|
||
"protocols": server.protocols,
|
||
"extra_attrs": server.extra_attrs,
|
||
"connectivity": server.connectivity,
|
||
"ssh_key_configured": server.ssh_key_configured,
|
||
"is_online": server.is_online,
|
||
"last_heartbeat": str(server.last_heartbeat) if server.last_heartbeat else None,
|
||
"last_checked_at": str(server.last_checked_at) if server.last_checked_at else None,
|
||
"system_info": server.system_info,
|
||
"agent_version": server.agent_version,
|
||
"created_at": str(server.created_at) if server.created_at else None,
|
||
"updated_at": str(server.updated_at) if server.updated_at else None,
|
||
}
|
||
|
||
|
||
def _sync_log_to_dict(log, server_name: str = None) -> dict:
|
||
"""Convert SyncLog model to API response dict"""
|
||
return {
|
||
"id": log.id,
|
||
"server_id": log.server_id,
|
||
"server_name": server_name,
|
||
"source_path": log.source_path,
|
||
"target_path": log.target_path,
|
||
"trigger_type": log.trigger_type,
|
||
"operator": log.operator,
|
||
"status": log.status,
|
||
"sync_mode": log.sync_mode,
|
||
"files_total": log.files_total,
|
||
"files_transferred": log.files_transferred,
|
||
"duration_seconds": log.duration_seconds,
|
||
"error_message": log.error_message,
|
||
"started_at": str(log.started_at) if log.started_at else None,
|
||
"finished_at": str(log.finished_at) if log.finished_at else None,
|
||
} |