145 lines
5.1 KiB
Python
145 lines
5.1 KiB
Python
"""VPS Agent 采集模块
|
||
|
||
采集本机资源、服务器信息与基础安全状态。
|
||
依赖 psutil + 少量系统命令（ufw/iptables/sshd_config）。
|
||
"""
|
||
|
||
import platform
|
||
import socket
|
||
import subprocess
|
||
import time
|
||
|
||
import psutil
|
||
|
||
|
||
def collect_metrics() -> dict:
|
||
"""资源使用率（CPU/内存/磁盘/网络/负载/运行时长）"""
|
||
cpu = psutil.cpu_percent(interval=1)
|
||
mem = psutil.virtual_memory()
|
||
disk = psutil.disk_usage("/")
|
||
net = psutil.net_io_counters()
|
||
try:
|
||
load_1m = psutil.getloadavg()[0]
|
||
except (AttributeError, OSError):
|
||
load_1m = None
|
||
return {
|
||
"cpu_pct": round(cpu, 1),
|
||
"mem_pct": round(mem.percent, 1),
|
||
"disk_pct": round(disk.percent, 1),
|
||
"net_in_mb": round(net.bytes_recv / 1024 / 1024, 2),
|
||
"net_out_mb": round(net.bytes_sent / 1024 / 1024, 2),
|
||
"load_1m": round(load_1m, 2) if load_1m is not None else None,
|
||
"uptime_sec": int(time.time() - psutil.boot_time()),
|
||
}
|
||
|
||
|
||
def _get_ip() -> str:
|
||
"""获取本机出口 IP（不实际发包）"""
|
||
try:
|
||
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
s.connect(("8.8.8.8", 80))
|
||
ip = s.getsockname()[0]
|
||
s.close()
|
||
return ip
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def collect_server_info() -> dict:
|
||
"""服务器基础信息"""
|
||
os_name = platform.system()
|
||
try:
|
||
with open("/etc/os-release") as f:
|
||
for line in f:
|
||
if line.startswith("PRETTY_NAME="):
|
||
os_name = line.split("=", 1)[1].strip().strip('"')
|
||
break
|
||
except Exception:
|
||
pass
|
||
return {
|
||
"hostname": socket.gethostname(),
|
||
"os": os_name,
|
||
"kernel": platform.release(),
|
||
"cpu_cores": psutil.cpu_count(),
|
||
"mem_total_gb": round(psutil.virtual_memory().total / 1024**3, 1),
|
||
"disk_total_gb": round(psutil.disk_usage("/").total / 1024**3, 1),
|
||
"public_ip": _get_ip(),
|
||
"status": "online",
|
||
}
|
||
|
||
|
||
def _check_ssh_config() -> dict:
|
||
"""检查 SSH 配置（root 登录 / 密码登录）"""
|
||
path = "/etc/ssh/sshd_config"
|
||
permit_root = "unknown"
|
||
password_auth = "unknown"
|
||
try:
|
||
with open(path) as f:
|
||
for line in f:
|
||
low = line.strip().lower()
|
||
if low.startswith("permitrootlogin"):
|
||
parts = low.split()
|
||
permit_root = parts[1] if len(parts) > 1 else "unknown"
|
||
elif low.startswith("passwordauthentication"):
|
||
parts = low.split()
|
||
password_auth = parts[1] if len(parts) > 1 else "unknown"
|
||
except Exception:
|
||
return {"check_item": "ssh_config", "status": "unknown",
|
||
"detail": "无法读取 sshd_config", "suggestion": None}
|
||
|
||
status = "pass"
|
||
suggestions = []
|
||
if permit_root == "yes":
|
||
status = "warn"
|
||
suggestions.append("设置 PermitRootLogin no 或 prohibit-password")
|
||
if password_auth == "yes":
|
||
status = "warn"
|
||
suggestions.append("关闭密码登录 PasswordAuthentication no，改用密钥")
|
||
return {
|
||
"check_item": "ssh_config",
|
||
"status": status,
|
||
"detail": f"PermitRootLogin={permit_root}, PasswordAuthentication={password_auth}",
|
||
"suggestion": "；".join(suggestions) if suggestions else None,
|
||
}
|
||
|
||
|
||
def _check_firewall() -> dict:
|
||
"""检查防火墙状态"""
|
||
try:
|
||
out = subprocess.run(["ufw", "status"], capture_output=True, text=True, timeout=5)
|
||
if "Status: active" in out.stdout:
|
||
return {"check_item": "firewall", "status": "pass",
|
||
"detail": "ufw 已启用", "suggestion": None}
|
||
return {"check_item": "firewall", "status": "warn",
|
||
"detail": "ufw 未启用", "suggestion": "启用防火墙 ufw enable"}
|
||
except Exception:
|
||
try:
|
||
out = subprocess.run(["iptables", "-L", "-n"], capture_output=True, text=True, timeout=5)
|
||
if out.stdout.strip():
|
||
return {"check_item": "firewall", "status": "pass",
|
||
"detail": "iptables 有规则", "suggestion": None}
|
||
except Exception:
|
||
pass
|
||
return {"check_item": "firewall", "status": "unknown",
|
||
"detail": "未检测到 ufw/iptables", "suggestion": None}
|
||
|
||
|
||
def _check_listening_ports() -> dict:
|
||
"""统计监听端口"""
|
||
try:
|
||
conns = psutil.net_connections(kind="inet")
|
||
listening = sorted({c.laddr.port for c in conns if c.status == "LISTEN"})
|
||
detail = f"开放端口 {len(listening)} 个: {','.join(map(str, listening[:15]))}"
|
||
status = "pass" if len(listening) < 10 else "warn"
|
||
suggestion = "端口偏多，建议关闭非必要端口" if status == "warn" else None
|
||
return {"check_item": "listening_ports", "status": status,
|
||
"detail": detail, "suggestion": suggestion}
|
||
except Exception as e: # noqa: BLE001
|
||
return {"check_item": "listening_ports", "status": "unknown",
|
||
"detail": str(e), "suggestion": None}
|
||
|
||
|
||
def collect_security() -> list:
|
||
"""基础安全检查项"""
|
||
return [_check_ssh_config(), _check_firewall(), _check_listening_ports()]
|