Nexus Agent
f6e8b29640
终端独立 terminal_search 上下文（10条），与 servers_search、push_search 隔离。 Co-authored-by: Cursor <cursoragent@cursor.com>
2223 lines
80 KiB
Python
2223 lines
80 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 asyncio
|
||
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, AddByIpRequest, AddByIpsBatchRequest, AddByIpsBatchResult, AddByIpsBatchItemResult,
|
||
BatchCategoryUpdate, BatchJobStarted, BatchJobStatus, ServerSearchHistoryUpdate,
|
||
)
|
||
from server.application.server_batch_common import (
|
||
install_error_msg as _install_error_msg,
|
||
sudo_wrap as _sudo_wrap,
|
||
)
|
||
from server.application.server_connectivity import (
|
||
count_monitored_connectivity,
|
||
item_matches_online_filter,
|
||
)
|
||
from server.application.services.server_batch_service import start_batch_job, get_batch_job, list_batch_jobs
|
||
from server.application.services.server_service import ServerService
|
||
from server.application.services.sync_service import SyncService
|
||
from server.domain.models import Server, Admin, AuditLog, PendingServer
|
||
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
|
||
import shlex
|
||
|
||
|
||
|
||
logger = logging.getLogger("nexus.servers")
|
||
|
||
router = APIRouter(prefix="/api/servers", tags=["servers"])
|
||
|
||
REDIS_KEY_PREFIX = "heartbeat:"
|
||
|
||
|
||
async def _schedule_server_onboarding(server_ids: list[int], operator: str) -> int | None:
|
||
"""Fire-and-forget batch job: sudoers + detect-path for newly created servers."""
|
||
if not server_ids:
|
||
return None
|
||
from server.application.services.server_onboarding_service import schedule_server_onboarding
|
||
|
||
job = await schedule_server_onboarding(server_ids, operator=operator or "admin")
|
||
if not job:
|
||
return None
|
||
job_id = job.get("job_id")
|
||
return int(job_id) if job_id is not None else None
|
||
|
||
|
||
def _server_has_agent_monitoring(server: Server) -> bool:
|
||
"""True when this server is expected to run Nexus Agent (key or version on file)."""
|
||
return bool((server.agent_api_key or "").strip() or (server.agent_version or "").strip())
|
||
|
||
# ── CRUD ──
|
||
|
||
@router.get("/", response_model=dict)
|
||
async def list_servers(
|
||
category: Optional[str] = Query(None, description="Filter by category"),
|
||
platform_id: Optional[int] = Query(None, description="Filter by platform ID"),
|
||
node_id: Optional[int] = Query(None, description="Filter by node ID"),
|
||
search: Optional[str] = Query(None, description="Search by name or domain (substring)"),
|
||
sort_by: Optional[str] = Query(None, description="Sort field: name/domain/status/heartbeat/agent_version/category/id"),
|
||
sort_order: Optional[str] = Query("asc", description="Sort direction: asc or desc"),
|
||
is_online: Optional[bool] = Query(None, description="Filter by online status"),
|
||
target_path_unset: Optional[bool] = Query(
|
||
None, description="Filter unset target paths: true=only unset, false=exclude unset",
|
||
),
|
||
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.
|
||
|
||
Static columns sort in MySQL; status/heartbeat/agent_version sort after
|
||
Redis overlay so order matches displayed values.
|
||
"""
|
||
from server.application.services.server_list_sort import (
|
||
LIVE_SORT_FIELDS,
|
||
MAX_LIVE_SORT_ROWS,
|
||
sort_server_items,
|
||
)
|
||
from server.infrastructure.database.server_repo import ServerRepositoryImpl
|
||
|
||
offset = (page - 1) * per_page
|
||
repo = ServerRepositoryImpl(db)
|
||
redis = get_redis()
|
||
order = sort_order if sort_order in ("asc", "desc") else "asc"
|
||
|
||
use_live_path = sort_by in LIVE_SORT_FIELDS or is_online is not None
|
||
|
||
if use_live_path:
|
||
all_servers, db_total = await repo.get_filtered(
|
||
category,
|
||
platform_id,
|
||
node_id,
|
||
search=search,
|
||
is_online=None,
|
||
target_path_unset=target_path_unset,
|
||
max_rows=MAX_LIVE_SORT_ROWS,
|
||
)
|
||
if db_total > MAX_LIVE_SORT_ROWS:
|
||
logger.warning(
|
||
"Live sort capped at %s servers (total matching=%s)",
|
||
MAX_LIVE_SORT_ROWS,
|
||
db_total,
|
||
)
|
||
result = await _build_server_list_with_overlay(redis, all_servers)
|
||
if is_online is not None:
|
||
result = [s for s in result if item_matches_online_filter(s, is_online)]
|
||
total = len(result)
|
||
if sort_by in LIVE_SORT_FIELDS:
|
||
result = sort_server_items(result, sort_by, order)
|
||
result = result[offset : offset + per_page]
|
||
else:
|
||
page_servers, total = await repo.get_paginated(
|
||
category,
|
||
platform_id,
|
||
node_id,
|
||
offset,
|
||
per_page,
|
||
search=search,
|
||
sort_by=sort_by,
|
||
sort_order=order,
|
||
is_online=None,
|
||
target_path_unset=target_path_unset,
|
||
)
|
||
result = await _build_server_list_with_overlay(redis, page_servers)
|
||
|
||
return {
|
||
"items": result,
|
||
"total": total,
|
||
"page": page,
|
||
"per_page": per_page,
|
||
"pages": (total + per_page - 1) // per_page if total > 0 else 1,
|
||
}
|
||
|
||
|
||
# ── 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
|
||
|
||
# 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
|
||
|
||
redis = get_redis()
|
||
try:
|
||
fleet = await count_monitored_connectivity(redis, db, exclude_unset_target_path=False)
|
||
main_scope = await count_monitored_connectivity(redis, db, exclude_unset_target_path=True)
|
||
except Exception as e:
|
||
logger.warning(f"Failed to count monitored connectivity in server_stats: {e}")
|
||
fleet = {"monitored": 0, "online": 0, "offline": 0}
|
||
main_scope = {"monitored": 0, "online": 0, "offline": 0}
|
||
|
||
online = fleet["online"]
|
||
offline = fleet["offline"]
|
||
monitored = fleet["monitored"]
|
||
unknown = max(0, total - monitored)
|
||
|
||
from server.infrastructure.database.server_repo import _target_path_unset_filter
|
||
|
||
unset_path_result = await db.execute(
|
||
select(func.count(Server.id)).where(_target_path_unset_filter(True))
|
||
)
|
||
unset_path = int(unset_path_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": offline,
|
||
"unknown": unknown,
|
||
"online_list": main_scope["online"],
|
||
"offline_list": main_scope["offline"],
|
||
"unset_path": unset_path,
|
||
"alerts": alerts,
|
||
"categories": categories,
|
||
}
|
||
|
||
|
||
@router.get("/fleet-metrics", response_model=dict)
|
||
async def fleet_metrics(
|
||
hours: int = 24,
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Fleet-wide CPU/mem/disk trend samples for Dashboard (B-02)."""
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from server.infrastructure.database.fleet_metric_repo import (
|
||
FLUSH_INTERVAL_MINUTES,
|
||
FleetMetricRepositoryImpl,
|
||
)
|
||
|
||
from server.domain.models import AlertLog
|
||
from server.utils.display_time import to_naive_utc
|
||
from server.utils.fleet_alert_buckets import attach_alerts_to_fleet_points
|
||
|
||
bounded = max(1, min(int(hours), 24 * 7))
|
||
since = datetime.now(timezone.utc) - timedelta(hours=bounded)
|
||
since_naive = to_naive_utc(since)
|
||
repo = FleetMetricRepositoryImpl(db)
|
||
rows = await repo.list_since(since)
|
||
points = [
|
||
{
|
||
"recorded_at": str(r.recorded_at) if r.recorded_at else None,
|
||
"online_count": r.online_count,
|
||
"sample_count": r.sample_count,
|
||
"cpu_avg": r.cpu_avg,
|
||
"mem_avg": r.mem_avg,
|
||
"disk_avg": r.disk_avg,
|
||
}
|
||
for r in rows
|
||
]
|
||
|
||
alert_result = await db.execute(
|
||
select(AlertLog)
|
||
.where(AlertLog.created_at >= since_naive)
|
||
.order_by(AlertLog.created_at.asc())
|
||
)
|
||
alert_rows = list(alert_result.scalars().all())
|
||
points = attach_alerts_to_fleet_points(
|
||
points,
|
||
alert_rows,
|
||
interval_minutes=FLUSH_INTERVAL_MINUTES,
|
||
)
|
||
|
||
return {
|
||
"hours": bounded,
|
||
"interval_minutes": FLUSH_INTERVAL_MINUTES,
|
||
"points": points,
|
||
}
|
||
|
||
|
||
# ── Sync Logs ──
|
||
|
||
@router.get("/categories", response_model=dict)
|
||
async def server_categories(
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Return distinct categories with counts and platform list for server filtering."""
|
||
from sqlalchemy import func as sqlfunc
|
||
from server.domain.models import Platform
|
||
|
||
# Categories with server counts
|
||
cat_result = await db.execute(
|
||
select(Server.category, sqlfunc.count(Server.id))
|
||
.group_by(Server.category)
|
||
)
|
||
categories = []
|
||
for cat, cnt in cat_result.all():
|
||
categories.append({"name": cat or "uncategorized", "count": cnt})
|
||
|
||
# Platforms
|
||
plat_result = await db.execute(select(Platform).order_by(Platform.id))
|
||
platforms = []
|
||
for p in plat_result.scalars().all():
|
||
platforms.append({"id": p.id, "name": p.name, "category": p.category or ""})
|
||
|
||
return {"categories": categories, "platforms": platforms}
|
||
|
||
|
||
@router.get("/search-history", response_model=dict)
|
||
async def get_server_search_history(
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Return servers page search history for the current admin."""
|
||
from server.infrastructure.database.admin_ui_preference_repo import AdminUiPreferenceRepositoryImpl
|
||
from server.utils.server_search_history import (
|
||
SERVERS_SEARCH_CONTEXT,
|
||
parse_search_state,
|
||
)
|
||
|
||
repo = AdminUiPreferenceRepositoryImpl(db)
|
||
raw = await repo.get(admin.id, SERVERS_SEARCH_CONTEXT)
|
||
return parse_search_state(raw)
|
||
|
||
|
||
@router.put("/search-history", response_model=dict)
|
||
async def update_server_search_history(
|
||
payload: ServerSearchHistoryUpdate,
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Record a search query or replace full history for the current admin."""
|
||
from server.infrastructure.database.admin_ui_preference_repo import AdminUiPreferenceRepositoryImpl
|
||
from server.utils.server_search_history import (
|
||
SERVERS_SEARCH_CONTEXT,
|
||
apply_search_record,
|
||
apply_search_replace,
|
||
parse_search_state,
|
||
)
|
||
|
||
repo = AdminUiPreferenceRepositoryImpl(db)
|
||
state = parse_search_state(await repo.get(admin.id, SERVERS_SEARCH_CONTEXT))
|
||
if payload.history is not None:
|
||
state = apply_search_replace(history=payload.history, last=payload.last)
|
||
else:
|
||
state = apply_search_record(state, payload.query or "")
|
||
await repo.upsert(admin.id, SERVERS_SEARCH_CONTEXT, state)
|
||
return state
|
||
|
||
|
||
@router.get("/push-search-history", response_model=dict)
|
||
async def get_push_search_history(
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Return push page server search history (max 5) for the current admin."""
|
||
from server.infrastructure.database.admin_ui_preference_repo import AdminUiPreferenceRepositoryImpl
|
||
from server.utils.server_search_history import (
|
||
MAX_PUSH_SEARCH_HISTORY,
|
||
PUSH_SEARCH_CONTEXT,
|
||
parse_search_state,
|
||
)
|
||
|
||
repo = AdminUiPreferenceRepositoryImpl(db)
|
||
raw = await repo.get(admin.id, PUSH_SEARCH_CONTEXT)
|
||
return parse_search_state(raw, max_history=MAX_PUSH_SEARCH_HISTORY)
|
||
|
||
|
||
@router.put("/push-search-history", response_model=dict)
|
||
async def update_push_search_history(
|
||
payload: ServerSearchHistoryUpdate,
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Record a push page search query or replace full history (max 5)."""
|
||
from server.infrastructure.database.admin_ui_preference_repo import AdminUiPreferenceRepositoryImpl
|
||
from server.utils.server_search_history import (
|
||
MAX_PUSH_SEARCH_HISTORY,
|
||
PUSH_SEARCH_CONTEXT,
|
||
apply_search_record,
|
||
apply_search_replace,
|
||
parse_search_state,
|
||
)
|
||
|
||
repo = AdminUiPreferenceRepositoryImpl(db)
|
||
state = parse_search_state(
|
||
await repo.get(admin.id, PUSH_SEARCH_CONTEXT),
|
||
max_history=MAX_PUSH_SEARCH_HISTORY,
|
||
)
|
||
if payload.history is not None:
|
||
state = apply_search_replace(
|
||
history=payload.history,
|
||
last=payload.last,
|
||
max_history=MAX_PUSH_SEARCH_HISTORY,
|
||
)
|
||
else:
|
||
state = apply_search_record(
|
||
state,
|
||
payload.query or "",
|
||
max_history=MAX_PUSH_SEARCH_HISTORY,
|
||
)
|
||
await repo.upsert(admin.id, PUSH_SEARCH_CONTEXT, state)
|
||
return state
|
||
|
||
|
||
@router.get("/terminal-search-history", response_model=dict)
|
||
async def get_terminal_search_history(
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Return terminal page server search history (max 10) for the current admin."""
|
||
from server.infrastructure.database.admin_ui_preference_repo import AdminUiPreferenceRepositoryImpl
|
||
from server.utils.server_search_history import (
|
||
MAX_TERMINAL_SEARCH_HISTORY,
|
||
TERMINAL_SEARCH_CONTEXT,
|
||
parse_search_state,
|
||
)
|
||
|
||
repo = AdminUiPreferenceRepositoryImpl(db)
|
||
raw = await repo.get(admin.id, TERMINAL_SEARCH_CONTEXT)
|
||
return parse_search_state(raw, max_history=MAX_TERMINAL_SEARCH_HISTORY)
|
||
|
||
|
||
@router.put("/terminal-search-history", response_model=dict)
|
||
async def update_terminal_search_history(
|
||
payload: ServerSearchHistoryUpdate,
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Record a terminal page search query or replace full history (max 10)."""
|
||
from server.infrastructure.database.admin_ui_preference_repo import AdminUiPreferenceRepositoryImpl
|
||
from server.utils.server_search_history import (
|
||
MAX_TERMINAL_SEARCH_HISTORY,
|
||
TERMINAL_SEARCH_CONTEXT,
|
||
apply_search_record,
|
||
apply_search_replace,
|
||
parse_search_state,
|
||
)
|
||
|
||
repo = AdminUiPreferenceRepositoryImpl(db)
|
||
state = parse_search_state(
|
||
await repo.get(admin.id, TERMINAL_SEARCH_CONTEXT),
|
||
max_history=MAX_TERMINAL_SEARCH_HISTORY,
|
||
)
|
||
if payload.history is not None:
|
||
state = apply_search_replace(
|
||
history=payload.history,
|
||
last=payload.last,
|
||
max_history=MAX_TERMINAL_SEARCH_HISTORY,
|
||
)
|
||
else:
|
||
state = apply_search_record(
|
||
state,
|
||
payload.query or "",
|
||
max_history=MAX_TERMINAL_SEARCH_HISTORY,
|
||
)
|
||
await repo.upsert(admin.id, TERMINAL_SEARCH_CONTEXT, state)
|
||
return state
|
||
|
||
|
||
@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|cancelled"),
|
||
sync_mode: Optional[str] = Query(None, description="incremental|full|overwrite|checksum|command"),
|
||
trigger_type: Optional[str] = Query(None, description="manual|schedule|retry|batch"),
|
||
page: int = Query(1, ge=1, description="Page number"),
|
||
per_page: int = Query(20, ge=1, le=100, description="Items per page"),
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Get paginated sync logs across all servers with optional filters."""
|
||
import math
|
||
from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl
|
||
repo = SyncLogRepositoryImpl(db)
|
||
offset = (page - 1) * per_page
|
||
rows, total = await repo.get_all_paginated(
|
||
server_id=server_id, status=status, sync_mode=sync_mode,
|
||
trigger_type=trigger_type, offset=offset, limit=per_page,
|
||
)
|
||
items = [_sync_log_to_dict(log, server_name=name) for log, name in rows]
|
||
pages = math.ceil(total / per_page) if total > 0 else 1
|
||
return {"items": items, "total": total, "page": page, "per_page": per_page, "pages": pages}
|
||
|
||
|
||
# ── CSV Batch Import ──
|
||
|
||
MAX_IMPORT_FILE_SIZE = 1_048_576 # 1 MB
|
||
MAX_IMPORT_ROWS = 500
|
||
MAX_BATCH_IP_LINES = 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("/meta/api_base_url")
|
||
async def get_api_base_url(admin: Admin = Depends(get_current_admin)):
|
||
"""Return the configured API_BASE_URL (from .env / settings).
|
||
|
||
Used by the frontend to determine if Agent install commands can be generated.
|
||
"""
|
||
return {"api_base_url": settings.API_BASE_URL or ""}
|
||
|
||
|
||
@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 as enc_err:
|
||
raise HTTPException(status_code=400, detail="CSV 编码不支持，请使用 UTF-8 或 GBK") from enc_err
|
||
|
||
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
|
||
|
||
onboarding_job_id = await _schedule_server_onboarding(result.created_ids, admin.username)
|
||
if onboarding_job_id is not None:
|
||
result.onboarding_job_id = onboarding_job_id
|
||
|
||
# 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 jobs (background, same UX as script library) ──
|
||
|
||
async def _start_server_batch(
|
||
op: str,
|
||
server_ids: list[int],
|
||
admin: Admin,
|
||
params: dict | None = None,
|
||
) -> BatchJobStarted:
|
||
if op in ("install-agent", "upgrade-agent", "uninstall-agent"):
|
||
base_url = (settings.API_BASE_URL or "").strip().rstrip("/")
|
||
if not base_url:
|
||
raise HTTPException(status_code=400, detail="NEXUS_API_BASE_URL 未配置")
|
||
try:
|
||
started = await start_batch_job(
|
||
op=op,
|
||
server_ids=server_ids,
|
||
operator=admin.username,
|
||
params=params,
|
||
)
|
||
except ValueError as e:
|
||
raise HTTPException(status_code=400, detail=str(e)) from e
|
||
return BatchJobStarted(**started)
|
||
|
||
|
||
@router.get("/batch/jobs/{job_id}", response_model=BatchJobStatus)
|
||
async def get_server_batch_job(
|
||
job_id: int,
|
||
admin: Admin = Depends(get_current_admin),
|
||
):
|
||
"""Poll server batch job status and per-server results."""
|
||
detail = await get_batch_job(job_id)
|
||
if not detail:
|
||
raise HTTPException(status_code=404, detail="任务不存在或已过期")
|
||
return BatchJobStatus(**detail)
|
||
|
||
|
||
@router.get("/batch/jobs", response_model=dict)
|
||
async def list_server_batch_jobs(
|
||
limit: int = Query(50, ge=1, le=200),
|
||
offset: int = Query(0, ge=0),
|
||
op: str | None = Query(None, max_length=50),
|
||
admin: Admin = Depends(get_current_admin),
|
||
):
|
||
"""List persisted server batch job history (MySQL)."""
|
||
return await list_batch_jobs(limit=limit, offset=offset, op=op or None)
|
||
|
||
|
||
@router.post("/batch/category", response_model=BatchJobStarted)
|
||
async def batch_update_category(
|
||
payload: BatchCategoryUpdate,
|
||
admin: Admin = Depends(get_current_admin),
|
||
):
|
||
"""Queue batch category update (background)."""
|
||
category = (payload.category or "").strip()
|
||
return await _start_server_batch(
|
||
"category",
|
||
payload.server_ids,
|
||
admin,
|
||
params={"category": category},
|
||
)
|
||
|
||
|
||
@router.post("/batch/install-agent", response_model=BatchJobStarted)
|
||
async def batch_install_agent(
|
||
payload: BatchAgentAction,
|
||
admin: Admin = Depends(get_current_admin),
|
||
):
|
||
"""Queue Agent install on multiple servers (background)."""
|
||
return await _start_server_batch("install-agent", payload.server_ids, admin)
|
||
|
||
|
||
@router.post("/batch/upgrade-agent", response_model=BatchJobStarted)
|
||
async def batch_upgrade_agent(
|
||
payload: BatchAgentAction,
|
||
admin: Admin = Depends(get_current_admin),
|
||
):
|
||
"""Queue Agent upgrade on multiple servers (background)."""
|
||
return await _start_server_batch("upgrade-agent", payload.server_ids, admin)
|
||
|
||
|
||
@router.post("/batch/detect-path", response_model=BatchJobStarted)
|
||
async def batch_detect_path(
|
||
payload: BatchAgentAction,
|
||
admin: Admin = Depends(get_current_admin),
|
||
):
|
||
"""Queue SSH target_path detection (background)."""
|
||
return await _start_server_batch("detect-path", payload.server_ids, admin)
|
||
|
||
|
||
@router.post("/batch/uninstall-agent", response_model=BatchJobStarted)
|
||
async def batch_uninstall_agent(
|
||
payload: BatchAgentAction,
|
||
admin: Admin = Depends(get_current_admin),
|
||
):
|
||
"""Queue Agent uninstall on multiple servers (background)."""
|
||
return await _start_server_batch("uninstall-agent", payload.server_ids, admin)
|
||
|
||
|
||
# ── Credential polling: add-by-ip + pending failure queue ──
|
||
|
||
|
||
def _pending_to_dict(row: PendingServer) -> dict:
|
||
return {
|
||
"id": row.id,
|
||
"name": row.name,
|
||
"domain": row.domain,
|
||
"port": row.port,
|
||
"agent_port": row.agent_port,
|
||
"target_path": row.target_path or "",
|
||
"category": row.category,
|
||
"platform_id": row.platform_id,
|
||
"node_id": row.node_id,
|
||
"attempts": row.attempts or 0,
|
||
"last_error": row.last_error or "",
|
||
"last_attempt_at": str(row.last_attempt_at) if row.last_attempt_at else None,
|
||
"created_at": str(row.created_at) if row.created_at else None,
|
||
}
|
||
|
||
|
||
async def _unique_server_name(db: AsyncSession, base: str) -> str:
|
||
from sqlalchemy import select as sa_select
|
||
|
||
candidate = base.strip()[:100] or "server"
|
||
result = await db.execute(sa_select(Server.name).where(Server.name == candidate))
|
||
if not result.scalar_one_or_none():
|
||
return candidate
|
||
for i in range(2, 1000):
|
||
name = f"{candidate[:90]}-{i}"
|
||
result = await db.execute(sa_select(Server.name).where(Server.name == name))
|
||
if not result.scalar_one_or_none():
|
||
return name
|
||
raise HTTPException(500, "无法生成唯一服务器名称")
|
||
|
||
|
||
async def _server_exists_for_domain(db: AsyncSession, domain: str) -> bool:
|
||
from sqlalchemy import select as sa_select
|
||
|
||
result = await db.execute(sa_select(Server.id).where(Server.domain == domain).limit(1))
|
||
return result.scalar_one_or_none() is not None
|
||
|
||
|
||
async def _create_server_from_poll(
|
||
db: AsyncSession,
|
||
service: ServerService,
|
||
*,
|
||
poll_match,
|
||
name: str,
|
||
domain: str,
|
||
port: int,
|
||
agent_port: int,
|
||
target_path: str | None,
|
||
category: str | None,
|
||
platform_id: int | None,
|
||
node_id: int | None,
|
||
admin: Admin,
|
||
request: Request,
|
||
) -> Server:
|
||
import secrets
|
||
from datetime import datetime, timezone
|
||
|
||
from server.infrastructure.database.crypto import encrypt_value
|
||
|
||
unique_name = await _unique_server_name(db, name)
|
||
data: dict = {
|
||
"name": unique_name,
|
||
"domain": domain,
|
||
"port": port,
|
||
"username": poll_match.username,
|
||
"auth_method": poll_match.auth_method,
|
||
"agent_port": agent_port,
|
||
"target_path": target_path or "",
|
||
"category": category,
|
||
"platform_id": platform_id,
|
||
"node_id": node_id,
|
||
"connectivity": "ok",
|
||
"last_checked_at": datetime.now(timezone.utc),
|
||
"agent_api_key": f"nxs-{secrets.token_urlsafe(32)}",
|
||
}
|
||
if poll_match.auth_method == "password":
|
||
data["preset_id"] = poll_match.preset_id
|
||
data["password"] = encrypt_value(poll_match.password)
|
||
else:
|
||
data["ssh_key_preset_id"] = poll_match.ssh_key_preset_id
|
||
data["ssh_key_private"] = encrypt_value(poll_match.ssh_key_private)
|
||
data["ssh_key_public"] = poll_match.ssh_key_public or ""
|
||
data["ssh_key_configured"] = True
|
||
|
||
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"凭据轮询添加 名称={created.name} 地址={created.domain}:{created.port} "
|
||
f"预设={poll_match.preset_name} 用户={poll_match.username}"
|
||
),
|
||
ip_address=ip_address,
|
||
))
|
||
return created
|
||
|
||
|
||
@router.get("/pending", response_model=dict)
|
||
async def list_pending_servers(
|
||
search: Optional[str] = Query(None, description="Filter by name or domain (substring)"),
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""List servers that failed SSH credential polling."""
|
||
from server.infrastructure.database.pending_server_repo import PendingServerRepositoryImpl
|
||
|
||
repo = PendingServerRepositoryImpl(db)
|
||
rows = await repo.get_all()
|
||
items = [_pending_to_dict(r) for r in rows]
|
||
q = (search or "").strip().lower()
|
||
if q:
|
||
items = [
|
||
i for i in items
|
||
if q in (i.get("name") or "").lower() or q in (i.get("domain") or "").lower()
|
||
]
|
||
return {"items": items, "total": len(items)}
|
||
|
||
|
||
async def _try_add_by_ip(
|
||
request: Request,
|
||
*,
|
||
domain: str,
|
||
port: int,
|
||
agent_port: int,
|
||
target_path: Optional[str],
|
||
category: Optional[str],
|
||
platform_id: Optional[int],
|
||
node_id: Optional[int],
|
||
name: Optional[str],
|
||
admin: Admin,
|
||
service: ServerService,
|
||
db: AsyncSession,
|
||
) -> AddByIpsBatchItemResult:
|
||
"""Poll credential presets for one host; create server or enqueue pending."""
|
||
from server.application.services.credential_poller import (
|
||
format_poll_errors,
|
||
poll_credentials,
|
||
)
|
||
from server.infrastructure.database.pending_server_repo import PendingServerRepositoryImpl
|
||
|
||
domain = domain.strip()
|
||
if await _server_exists_for_domain(db, domain):
|
||
return AddByIpsBatchItemResult(
|
||
domain=domain,
|
||
status="skipped",
|
||
message="已存在于服务器列表",
|
||
)
|
||
|
||
display_name = (name or domain).strip()
|
||
poll = await poll_credentials(db, domain, port)
|
||
|
||
if poll.success and poll.match:
|
||
created = await _create_server_from_poll(
|
||
db,
|
||
service,
|
||
poll_match=poll.match,
|
||
name=display_name,
|
||
domain=domain,
|
||
port=port,
|
||
agent_port=agent_port,
|
||
target_path=target_path,
|
||
category=category,
|
||
platform_id=platform_id,
|
||
node_id=node_id,
|
||
admin=admin,
|
||
request=request,
|
||
)
|
||
return AddByIpsBatchItemResult(
|
||
domain=domain,
|
||
status="created",
|
||
server_id=created.id,
|
||
matched_preset=poll.match.preset_name,
|
||
matched_username=poll.match.username,
|
||
)
|
||
|
||
err_text = format_poll_errors(poll.errors)
|
||
pending_repo = PendingServerRepositoryImpl(db)
|
||
pending = await pending_repo.record_failure(
|
||
name=display_name,
|
||
domain=domain,
|
||
port=port,
|
||
agent_port=agent_port,
|
||
target_path=target_path,
|
||
category=category,
|
||
platform_id=platform_id,
|
||
node_id=node_id,
|
||
last_error=err_text,
|
||
)
|
||
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="add_server_by_ip_failed",
|
||
target_type="pending_server",
|
||
target_id=pending.id,
|
||
detail=f"凭据轮询失败 地址={domain}:{port}",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
from server.api.schemas import CredentialPollErrorOut
|
||
|
||
return AddByIpsBatchItemResult(
|
||
domain=domain,
|
||
status="pending",
|
||
pending_id=pending.id,
|
||
message=err_text,
|
||
errors=[
|
||
CredentialPollErrorOut(
|
||
preset_type=e.preset_type,
|
||
preset_name=e.preset_name,
|
||
username=e.username,
|
||
error=e.error,
|
||
)
|
||
for e in poll.errors
|
||
],
|
||
)
|
||
|
||
|
||
@router.post("/add-by-ip", response_model=dict)
|
||
async def add_server_by_ip(
|
||
request: Request,
|
||
payload: AddByIpRequest,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Add server by IP only — poll all credential presets until SSH login succeeds."""
|
||
item = await _try_add_by_ip(
|
||
request,
|
||
domain=payload.domain,
|
||
port=payload.port,
|
||
agent_port=payload.agent_port,
|
||
target_path=payload.target_path,
|
||
category=payload.category,
|
||
platform_id=payload.platform_id,
|
||
node_id=payload.node_id,
|
||
name=payload.name,
|
||
admin=admin,
|
||
service=service,
|
||
db=db,
|
||
)
|
||
|
||
if item.status == "skipped":
|
||
raise HTTPException(409, f"服务器 {item.domain} 已存在于列表中")
|
||
|
||
if item.status == "created":
|
||
onboarding_job_id = None
|
||
if item.server_id is not None:
|
||
onboarding_job_id = await _schedule_server_onboarding([item.server_id], admin.username)
|
||
body: dict = {
|
||
"success": True,
|
||
"server": {"id": item.server_id, "domain": item.domain},
|
||
"matched_preset": item.matched_preset,
|
||
"matched_username": item.matched_username,
|
||
}
|
||
if onboarding_job_id is not None:
|
||
body["onboarding_job_id"] = onboarding_job_id
|
||
body["onboarding_label"] = "新服务器初始化"
|
||
return body
|
||
|
||
return {
|
||
"success": False,
|
||
"pending_id": item.pending_id,
|
||
"errors": [e.model_dump() for e in (item.errors or [])],
|
||
"message": item.message,
|
||
}
|
||
|
||
|
||
@router.post("/add-by-ips-batch", response_model=AddByIpsBatchResult)
|
||
async def add_servers_by_ips_batch(
|
||
request: Request,
|
||
payload: AddByIpsBatchRequest,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Batch add servers: one IP/domain per line, credential preset polling, failures → pending."""
|
||
from server.application.services.credential_poller import parse_batch_ip_lines
|
||
|
||
try:
|
||
parsed = parse_batch_ip_lines(payload.text, max_lines=MAX_BATCH_IP_LINES)
|
||
except ValueError as e:
|
||
raise HTTPException(400, str(e)) from e
|
||
|
||
entries = parsed.entries
|
||
if not entries:
|
||
raise HTTPException(400, "未解析到有效 IP 或域名")
|
||
|
||
items: list[AddByIpsBatchItemResult] = []
|
||
created = pending = skipped = 0
|
||
|
||
for entry in entries:
|
||
item = await _try_add_by_ip(
|
||
request,
|
||
domain=entry.domain,
|
||
port=payload.port,
|
||
agent_port=payload.agent_port,
|
||
target_path=payload.target_path,
|
||
category=payload.category,
|
||
platform_id=payload.platform_id,
|
||
node_id=payload.node_id,
|
||
name=entry.name,
|
||
admin=admin,
|
||
service=service,
|
||
db=db,
|
||
)
|
||
items.append(item)
|
||
if item.status == "created":
|
||
created += 1
|
||
elif item.status == "pending":
|
||
pending += 1
|
||
else:
|
||
skipped += 1
|
||
|
||
onboarding_ids = [i.server_id for i in items if i.status == "created" and i.server_id is not None]
|
||
onboarding_job_id = await _schedule_server_onboarding(onboarding_ids, admin.username)
|
||
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="add_servers_by_ips_batch",
|
||
target_type="server",
|
||
target_id=None,
|
||
detail=(
|
||
f"批量 IP 添加 输入{parsed.input_lines}行 去重{parsed.duplicates_removed} "
|
||
f"实处理{len(entries)} 成功{created} 待连接{pending} 跳过{skipped}"
|
||
),
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
return AddByIpsBatchResult(
|
||
total=len(entries),
|
||
created=created,
|
||
pending=pending,
|
||
skipped=skipped,
|
||
input_lines=parsed.input_lines,
|
||
duplicates_removed=parsed.duplicates_removed,
|
||
items=items,
|
||
onboarding_job_id=onboarding_job_id,
|
||
)
|
||
|
||
|
||
@router.post("/pending/{pending_id}/retry", response_model=dict)
|
||
async def retry_pending_server(
|
||
pending_id: int,
|
||
request: Request,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Re-poll all credential presets for a pending server."""
|
||
from server.application.services.credential_poller import (
|
||
format_poll_errors,
|
||
poll_credentials,
|
||
)
|
||
from server.infrastructure.database.pending_server_repo import PendingServerRepositoryImpl
|
||
|
||
pending_repo = PendingServerRepositoryImpl(db)
|
||
pending = await pending_repo.get_by_id(pending_id)
|
||
if not pending:
|
||
raise HTTPException(404, "待连接服务器不存在")
|
||
|
||
if await _server_exists_for_domain(db, pending.domain):
|
||
await pending_repo.delete(pending_id)
|
||
raise HTTPException(409, f"服务器 {pending.domain} 已存在于列表中，已从失败列表移除")
|
||
|
||
poll = await poll_credentials(db, pending.domain, pending.port or 22)
|
||
|
||
if poll.success and poll.match:
|
||
created = await _create_server_from_poll(
|
||
db,
|
||
service,
|
||
poll_match=poll.match,
|
||
name=pending.name,
|
||
domain=pending.domain,
|
||
port=pending.port or 22,
|
||
agent_port=pending.agent_port or 8601,
|
||
target_path=pending.target_path,
|
||
category=pending.category,
|
||
platform_id=pending.platform_id,
|
||
node_id=pending.node_id,
|
||
admin=admin,
|
||
request=request,
|
||
)
|
||
await pending_repo.delete(pending_id)
|
||
onboarding_job_id = await _schedule_server_onboarding([created.id], admin.username)
|
||
body: dict = {
|
||
"success": True,
|
||
"server": _server_to_dict(created),
|
||
"matched_preset": poll.match.preset_name,
|
||
"matched_username": poll.match.username,
|
||
}
|
||
if onboarding_job_id is not None:
|
||
body["onboarding_job_id"] = onboarding_job_id
|
||
body["onboarding_label"] = "新服务器初始化"
|
||
return body
|
||
|
||
err_text = format_poll_errors(poll.errors)
|
||
pending.attempts = (pending.attempts or 0) + 1
|
||
pending.last_error = err_text
|
||
from datetime import datetime, timezone
|
||
|
||
pending.last_attempt_at = datetime.now(timezone.utc)
|
||
await pending_repo.update(pending)
|
||
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="retry_pending_server_failed",
|
||
target_type="pending_server",
|
||
target_id=pending.id,
|
||
detail=f"凭据轮询重试失败 地址={pending.domain}:{pending.port or 22} 第{pending.attempts}次",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
return {
|
||
"success": False,
|
||
"pending_id": pending.id,
|
||
"errors": [{"preset_type": e.preset_type, "preset_name": e.preset_name, "username": e.username, "error": e.error} for e in poll.errors],
|
||
"message": err_text,
|
||
}
|
||
|
||
|
||
@router.delete("/pending/{pending_id}", status_code=204)
|
||
async def delete_pending_server(
|
||
pending_id: int,
|
||
request: Request,
|
||
admin: Admin = Depends(get_current_admin),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
from server.infrastructure.database.pending_server_repo import PendingServerRepositoryImpl
|
||
|
||
repo = PendingServerRepositoryImpl(db)
|
||
pending = await repo.get_by_id(pending_id)
|
||
if not pending:
|
||
raise HTTPException(404, "待连接服务器不存在")
|
||
if not await repo.delete(pending_id):
|
||
raise HTTPException(404, "待连接服务器不存在")
|
||
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="delete_pending_server",
|
||
target_type="pending_server",
|
||
target_id=pending_id,
|
||
detail=f"删除待连接 地址={pending.domain}:{pending.port or 22}",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
|
||
@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)
|
||
|
||
redis = get_redis()
|
||
try:
|
||
heartbeat = await redis.hgetall(f"{REDIS_KEY_PREFIX}{id}")
|
||
_apply_heartbeat_overlay(server_data, server, heartbeat or None)
|
||
except Exception as e:
|
||
logger.warning(f"Redis read failed for server {id}: {e}")
|
||
server_data["status"] = _derive_server_status(
|
||
server_data.get("is_online"),
|
||
server.agent_version,
|
||
has_agent=_server_has_agent_monitoring(server),
|
||
)
|
||
|
||
return server_data
|
||
|
||
|
||
@router.get("/{id}/metrics", response_model=dict)
|
||
async def get_server_metrics(
|
||
id: int,
|
||
hours: int = 24,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Per-server CPU/mem trend samples (10min flush, 7d retention)."""
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from server.infrastructure.database.server_metric_repo import (
|
||
FLUSH_INTERVAL_MINUTES,
|
||
ServerMetricRepositoryImpl,
|
||
)
|
||
from server.utils.server_metrics import merge_offline_segments
|
||
|
||
server = await service.get_server(id)
|
||
if not server:
|
||
raise HTTPException(status_code=404, detail="Server not found")
|
||
|
||
bounded = max(1, min(int(hours), 24 * 7))
|
||
since = datetime.now(timezone.utc) - timedelta(hours=bounded)
|
||
repo = ServerMetricRepositoryImpl(db)
|
||
rows = await repo.list_since(id, since)
|
||
points = [
|
||
{
|
||
"recorded_at": str(r.recorded_at) if r.recorded_at else None,
|
||
"is_online": bool(r.is_online),
|
||
"cpu_pct": r.cpu_pct,
|
||
"mem_pct": r.mem_pct,
|
||
}
|
||
for r in rows
|
||
]
|
||
offline_segments = merge_offline_segments(points)
|
||
|
||
return {
|
||
"server_id": id,
|
||
"hours": bounded,
|
||
"interval_minutes": FLUSH_INTERVAL_MINUTES,
|
||
"points": points,
|
||
"offline_segments": offline_segments,
|
||
}
|
||
|
||
|
||
@router.get("/{id}/files-capability", response_model=dict)
|
||
async def get_files_capability(
|
||
id: int,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
):
|
||
"""Probe remote sudo capability for file manager (chmod/chown/write)."""
|
||
from server.infrastructure.ssh.files_capability import probe_files_capability
|
||
|
||
server = await service.get_server(id)
|
||
if not server:
|
||
raise HTTPException(status_code=404, detail="Server not found")
|
||
return await probe_files_capability(server)
|
||
|
||
|
||
@router.post("/{id}/setup-files-sudo", response_model=dict)
|
||
async def setup_files_sudo(
|
||
id: int,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
):
|
||
"""Auto-configure passwordless sudoers for file manager + rsync push on non-root servers."""
|
||
from server.application.services.files_sudoers_service import (
|
||
NEXUS_FILES_SUDOERS_PATH,
|
||
install_nexus_files_sudoers,
|
||
)
|
||
|
||
server = await service.get_server(id)
|
||
if not server:
|
||
raise HTTPException(status_code=404, detail="Server not found")
|
||
|
||
ssh_user = (server.username or "root").strip() or "root"
|
||
if ssh_user == "root":
|
||
return {"ok": True, "message": "root 用户无需配置 sudoers，已拥有完整权限"}
|
||
|
||
outcome = await install_nexus_files_sudoers(server)
|
||
if outcome.get("action") == "already_ok":
|
||
return {
|
||
"ok": True,
|
||
"message": f"sudo -n rsync 已可用，无需重复配置（{NEXUS_FILES_SUDOERS_PATH}）",
|
||
}
|
||
if not outcome.get("ok"):
|
||
raise HTTPException(
|
||
status_code=500,
|
||
detail=outcome.get("detail") or f"sudoers 配置失败：{outcome.get('action')}",
|
||
)
|
||
|
||
return {
|
||
"ok": True,
|
||
"message": (
|
||
f"sudoers 已配置：{NEXUS_FILES_SUDOERS_PATH}。"
|
||
"文件管理与文件推送可使用 sudo -n 自动提权。"
|
||
),
|
||
}
|
||
|
||
|
||
@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
|
||
|
||
from server.utils.files_elevation import apply_files_elevation_payload
|
||
|
||
data = payload.model_dump(exclude_none=True)
|
||
apply_files_elevation_payload(data)
|
||
|
||
# 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.username and (not data.get("username") or data.get("username") == "root"):
|
||
data["username"] = preset.username
|
||
# 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 ssh_key_preset.username and (not data.get("username") or data.get("username") == "root"):
|
||
data["username"] = ssh_key_preset.username
|
||
# 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"名称={created.name} 域名={created.domain}",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
resp = _server_to_dict(created)
|
||
onboarding_job_id = await _schedule_server_onboarding([created.id], admin.username)
|
||
if onboarding_job_id is not None:
|
||
resp["onboarding_job_id"] = onboarding_job_id
|
||
resp["onboarding_label"] = "新服务器初始化"
|
||
return resp
|
||
|
||
|
||
@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")
|
||
|
||
from server.utils.files_elevation import apply_files_elevation_payload
|
||
|
||
# If preset_id is provided (and not None), resolve the preset password
|
||
update_data = payload.model_dump(exclude_unset=True)
|
||
apply_files_elevation_payload(update_data)
|
||
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 preset.username:
|
||
update_data["username"] = preset.username
|
||
|
||
# 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
|
||
if ssh_key_preset.username:
|
||
update_data["username"] = ssh_key_preset.username
|
||
|
||
# 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"名称={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"名称={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"为 {server.name} 生成新 Agent API Key",
|
||
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
|
||
|
||
server = await service.get_server(id)
|
||
if not server:
|
||
raise HTTPException(status_code=404, detail="Server not found")
|
||
|
||
from server.application.server_batch_common import agent_install_skip_result
|
||
from server.utils.agent_version import get_central_agent_version
|
||
|
||
skip = await agent_install_skip_result(
|
||
server=server,
|
||
server_id=id,
|
||
server_name=server.name or str(id),
|
||
central_version=get_central_agent_version(),
|
||
)
|
||
if skip is not None:
|
||
return {
|
||
"success": True,
|
||
"server_id": id,
|
||
"server_name": server.name,
|
||
"stdout": skip.get("stdout", ""),
|
||
"skipped": True,
|
||
}
|
||
|
||
# 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 "").strip()
|
||
if not api_key:
|
||
import secrets
|
||
api_key = f"nxs-{secrets.token_urlsafe(32)}"
|
||
server.agent_api_key = api_key
|
||
await service.update_server(server)
|
||
|
||
agent_port = server.agent_port or 8601
|
||
ssh_user = (server.username or "root").strip()
|
||
|
||
# Build install command — download first, then execute (avoids pipe hiding curl errors)
|
||
install_cmd = (
|
||
f"TMP=$(mktemp) && curl -fsSL {shlex.quote(base_url)}/agent/install.sh -o \"$TMP\" && bash \"$TMP\" -- "
|
||
f"--url {shlex.quote(base_url)} --key {shlex.quote(api_key)} "
|
||
f"--id {id} --port {shlex.quote(str(agent_port))}"
|
||
f"; RC=$?; rm -f \"$TMP\"; exit $RC"
|
||
)
|
||
|
||
# For non-root SSH users, pre-configure passwordless sudo so install.sh
|
||
# can run apt-get / systemctl / mkdir /opt without prompting.
|
||
install_cmd = _sudo_wrap(install_cmd, ssh_user)
|
||
|
||
# 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}") from e
|
||
|
||
if result["exit_code"] != 0:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"安装失败: {_install_error_msg(result.get('stdout', ''), result.get('stderr', ''), result['exit_code'])}",
|
||
)
|
||
|
||
# Check for "Status: FAILED" in stdout (install.sh exits 0 but service may not start)
|
||
if "FAILED" in result.get("stdout", ""):
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="安装完成但 Agent 启动失败，请检查子服务器日志: journalctl -u nexus-agent",
|
||
)
|
||
|
||
# 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}/uninstall-agent", response_model=dict)
|
||
async def uninstall_agent_remote(
|
||
request: Request,
|
||
id: int,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Uninstall Nexus Agent on the managed server via SSH.
|
||
|
||
Stops the systemd service, removes the install directory and log file,
|
||
then clears the agent metadata (is_online, agent_version) in the database.
|
||
"""
|
||
from server.infrastructure.ssh.asyncssh_pool import exec_ssh_command
|
||
|
||
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 未配置，无法下载卸载脚本。",
|
||
)
|
||
|
||
ssh_user = (server.username or "root").strip()
|
||
|
||
# Build uninstall command — download uninstall.sh then execute
|
||
uninstall_cmd = (
|
||
f"TMP=$(mktemp) && curl -fsSL {shlex.quote(base_url)}/agent/uninstall.sh -o \"$TMP\" "
|
||
f"&& bash \"$TMP\" ; RC=$?; rm -f \"$TMP\"; exit $RC"
|
||
)
|
||
uninstall_cmd = _sudo_wrap(uninstall_cmd, ssh_user)
|
||
|
||
# Execute via SSH
|
||
try:
|
||
result = await exec_ssh_command(server, uninstall_cmd, timeout=60)
|
||
except Exception as e:
|
||
raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}") from e
|
||
|
||
if result["exit_code"] != 0:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=f"卸载失败: {_install_error_msg(result.get('stdout', ''), result.get('stderr', ''), result['exit_code'])}",
|
||
)
|
||
|
||
# Clear agent metadata: Redis (real-time source) + MySQL (history)
|
||
try:
|
||
redis = get_redis()
|
||
await redis.delete(f"{REDIS_KEY_PREFIX}{id}")
|
||
except Exception:
|
||
logger.warning(f"Failed to clear Redis heartbeat for server {id}", exc_info=True)
|
||
server.is_online = False
|
||
server.agent_version = None
|
||
server.agent_api_key = None
|
||
await db.commit()
|
||
|
||
# Audit
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="uninstall_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"查看 {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
|
||
|
||
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 未配置，无法构建下载地址",
|
||
)
|
||
|
||
from server.utils.agent_deploy import (
|
||
agent_python_version_check_cmd,
|
||
agent_venv_python,
|
||
build_agent_upgrade_cmd,
|
||
build_agent_upgrade_rollback_cmd,
|
||
)
|
||
|
||
install_dir = "/opt/nexus-agent"
|
||
venv_python = agent_venv_python(install_dir)
|
||
ssh_user = (server.username or "root").strip()
|
||
agent_port = server.agent_port or 8601
|
||
|
||
# Step 1: Check Python version in venv
|
||
pyver_cmd = agent_python_version_check_cmd(venv_python)
|
||
try:
|
||
pyver = await exec_ssh_command(server, pyver_cmd, timeout=15)
|
||
except Exception as e:
|
||
raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}") from 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}"
|
||
),
|
||
)
|
||
|
||
upgrade_cmd = build_agent_upgrade_cmd(
|
||
base_url=base_url,
|
||
ssh_user=ssh_user,
|
||
install_dir=install_dir,
|
||
agent_port=agent_port,
|
||
)
|
||
|
||
try:
|
||
result = await exec_ssh_command(server, upgrade_cmd, timeout=120)
|
||
except Exception as e:
|
||
raise HTTPException(status_code=502, detail=f"SSH 连接失败: {e}") from e
|
||
|
||
success = result["exit_code"] == 0 and "upgrade_ok" in result["stdout"]
|
||
|
||
if not success:
|
||
# Rollback: restore backup and restart
|
||
rollback_cmd = build_agent_upgrade_rollback_cmd(
|
||
ssh_user=ssh_user,
|
||
install_dir=install_dir,
|
||
)
|
||
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"升级失败: {_install_error_msg(result.get('stdout', ''), result.get('stderr', ''), result['exit_code'])}，已回滚至旧版本",
|
||
)
|
||
|
||
# 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]}
|
||
|
||
|
||
# ── Agent Diagnose ──
|
||
|
||
@router.post("/{id}/agent-diagnose", response_model=dict)
|
||
async def agent_diagnose(
|
||
request: Request,
|
||
id: int,
|
||
admin: Admin = Depends(get_current_admin),
|
||
service: ServerService = Depends(get_server_service),
|
||
db: AsyncSession = Depends(get_db),
|
||
):
|
||
"""Diagnose Nexus Agent: central Redis/MySQL heartbeat + remote SSH probe."""
|
||
from server.application.agent_diagnose_service import diagnose_server_record
|
||
|
||
server = await service.get_server(id)
|
||
if not server:
|
||
raise HTTPException(status_code=404, detail="服务器不存在")
|
||
|
||
redis = get_redis()
|
||
heartbeat = await redis.hgetall(f"{REDIS_KEY_PREFIX}{id}")
|
||
result = await diagnose_server_record(server, heartbeat or None, ssh=True, timeout=45)
|
||
|
||
ip_address = request.client.host if request.client else ""
|
||
audit_repo = AuditLogRepositoryImpl(db)
|
||
central = result.get("central_status", "unknown")
|
||
remote = result.get("remote") or {}
|
||
ssh_ok = remote.get("ssh_ok")
|
||
await audit_repo.create(AuditLog(
|
||
admin_username=admin.username,
|
||
action="agent_diagnose",
|
||
target_type="server",
|
||
target_id=id,
|
||
detail=f"Agent 诊断: {server.name} 中心={central} SSH={'✓' if ssh_ok else '✗'}",
|
||
ip_address=ip_address,
|
||
))
|
||
|
||
return result
|
||
|
||
|
||
# ── Health Check ──
|
||
|
||
@router.post("/check", response_model=BatchJobStarted)
|
||
async def check_servers(
|
||
payload: ServerCheck,
|
||
admin: Admin = Depends(get_current_admin),
|
||
):
|
||
"""Queue SSH health check for selected servers (background)."""
|
||
return await _start_server_batch("health-check", payload.server_ids, admin)
|
||
|
||
|
||
|
||
@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 ──
|
||
|
||
async def _build_server_list_with_overlay(redis, servers: list) -> list[dict]:
|
||
"""Batch Redis heartbeat overlay for server list rows."""
|
||
if not servers:
|
||
return []
|
||
pipe = redis.pipeline()
|
||
for server in servers:
|
||
pipe.hgetall(f"{REDIS_KEY_PREFIX}{server.id}")
|
||
heartbeats = await pipe.execute()
|
||
result: list[dict] = []
|
||
for server, heartbeat in zip(servers, heartbeats):
|
||
server_data = _server_to_dict(server)
|
||
try:
|
||
_apply_heartbeat_overlay(server_data, server, heartbeat or None)
|
||
except Exception as e:
|
||
logger.warning(f"Redis read failed for server {server.id}: {e}")
|
||
server_data["status"] = _derive_server_status(
|
||
server_data.get("is_online"),
|
||
server.agent_version,
|
||
has_agent=_server_has_agent_monitoring(server),
|
||
)
|
||
result.append(server_data)
|
||
return result
|
||
|
||
|
||
def _derive_server_status(is_online: bool | None, agent_version: str | None, *, has_agent: bool = False) -> str:
|
||
"""Map live/DB fields to frontend status: online | offline | unknown."""
|
||
has_agent = has_agent or bool(agent_version and str(agent_version).strip())
|
||
if has_agent or is_online:
|
||
return "online" if is_online else "offline"
|
||
return "unknown"
|
||
|
||
|
||
def _apply_heartbeat_overlay(server_data: dict, server: Server, heartbeat: dict | None) -> None:
|
||
"""Overlay Redis heartbeat onto server_data and refresh derived status."""
|
||
has_agent = _server_has_agent_monitoring(server)
|
||
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"
|
||
elif has_agent:
|
||
server_data["is_online"] = False
|
||
server_data["_source"] = "redis_miss"
|
||
server_data["status"] = _derive_server_status(
|
||
server_data.get("is_online"),
|
||
server_data.get("agent_version") or server.agent_version,
|
||
has_agent=has_agent,
|
||
)
|
||
|
||
|
||
def _server_to_dict(server: Server) -> dict:
|
||
"""Convert Server model to API response dict (hide sensitive fields)"""
|
||
from server.utils.files_elevation import get_files_elevation
|
||
|
||
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,
|
||
"files_elevation": get_files_elevation(server).value,
|
||
"connectivity": server.connectivity,
|
||
"ssh_key_configured": server.ssh_key_configured,
|
||
"is_online": server.is_online,
|
||
"status": _derive_server_status(
|
||
server.is_online,
|
||
server.agent_version,
|
||
has_agent=_server_has_agent_monitoring(server),
|
||
),
|
||
"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,
|
||
"files_skipped": log.files_skipped,
|
||
"bytes_transferred": log.bytes_transferred,
|
||
"duration_seconds": log.duration_seconds,
|
||
"error_message": log.error_message,
|
||
"diff_summary": log.diff_summary,
|
||
"push_permission": log.push_permission,
|
||
"started_at": str(log.started_at) if log.started_at else None,
|
||
"finished_at": str(log.finished_at) if log.finished_at else None,
|
||
} |