AI + VPS：用大模型实现 VPS 配置智能巡检与自动化报告

引言：巡检不该是"体力活"

你管理着多台 VPS，每次上线新服务或定期安全审计时，都需要做配置巡检：

SSH 端口是不是改成了非标准？
root 登录是否禁用了？
防火墙规则是否过于宽松？
哪些服务暴露在公网？
SSH 密钥认证是否强制启用？
内核参数是否符合安全基线？

传统做法是打开一台台服务器，手动执行 sshd_config 检查、iptables 查看、sysctl 扫描……十几条检查项，10 台服务器，半天就过去了。而且人总会漏看，尤其是当检查项超过 30 条时。

AI 配置智能巡检改变了这个局面：你用自然语言描述检查要求，LLM 自动采集所有服务器的配置数据，分析每项设置的安全性，生成结构化的巡检报告，并直接给出整改命令——你只需要复核确认，不需要亲自查每一项。

本文将带你从零构建一套 AI 配置智能巡检系统，覆盖数据采集、LLM 分析、报告生成和整改建议全链路。

一、系统架构

┌──────────────────────────────────────────────────────────────────────┐
│                     AI 配置智能巡检系统                               │
│                                                                      │
│  ┌──────────────┐    ┌──────────────┐    ┌──────────────────────┐   │
│  │  配置采集层   │ →→ │  LLM 分析引擎 │ →→ │    报告 & 整改层      │   │
│  │  Collector   │    │  (Ollama/    │    │   (Markdown/Telegram)│   │
│  │              │    │   Qwen2.5)   │    │                      │   │
│  └──────┬───────┘    └──────┬───────┘    └──────────┬───────────┘   │
│         │                   │                       │                │
│  ┌──────▼───────┐    ┌──────▼───────┐    ┌──────────▼───────────┐   │
│  │  多源配置数据  │    │  安全规则库   │    │    历史趋势 & 对比    │   │
│  │  (SSH/内核/  │    │  (CIS/自定   │    │    (JSON 存储 + 变更   │   │
│  │   网络/Docker)│    │   义规则)    │    │     可视化)           │   │
│  └──────────────┘    └──────────────┘    └──────────────────────┘   │
└──────────────────────────────────────────────────────────────────────┘
         │                   │                       │
         ▼                   ▼                       ▼
   ┌───────────┐        ┌───────────┐        ┌───────────┐
   │  服务器 A   │        │  服务器 B   │        │  服务器 C   │
   │  (生产)    │        │  (预发)    │        │  (开发)    │
   └───────────┘        └───────────┘        └───────────┘

核心流程

阶段	内容	输出
采集	SSH/Python 批量登录各节点，采集系统配置快照	原始配置 JSON
分析	LLM 对照安全基线规则，逐项评估风险等级	风险清单 + 整改建议
报告	生成 Markdown 巡检报告，按服务器分组汇总	结构化报告文件
推送	通过 Telegram/邮件推送摘要，支持交互式整改	通知消息

二、基础环境搭建

2.1 安装 Ollama 和本地模型

# 安装 Ollama
curl -fsSL https://ollama.com/install.sh | sh

# 拉取模型（巡检任务用 7B 模型即可，响应快）
ollama pull qwen2.5:7b
ollama pull nomic-embed-text   # 用于历史报告相似度检索

2.2 安装依赖

pip install ollama psutil python-dotenv paramiko pymonitor

2.3 项目结构

ai-config-inspector/
├── config_collector.py    # 多源配置采集器
├── safety_rules.py        # 安全规则库（CIS Benchmark + 自定义）
├── llm_analyzer.py        # LLM 分析引擎
├── report_generator.py    # 报告生成器
├── notifier.py            # 通知推送（Telegram/邮件）
├── history.py             # 历史趋势存储
├── inspector.py           # 主入口
└── config.json            # 服务器列表和检查项配置

三、配置数据采集层

配置采集是巡检的基础。我们需要从每个节点采集多个维度的配置数据：

3.1 多源采集器

# config_collector.py
import subprocess
import json
import psutil
from datetime import datetime
from pathlib import Path
import paramiko

class ConfigCollector:
    """多源配置采集器——支持本地和远程 SSH 采集"""

    # 采集项定义
    COLLECTION_ITEMS = {
        "ssh_config": {
            "cmd": "sudo cat /etc/ssh/sshd_config 2>/dev/null | grep -v '^#' | grep -v '^$'",
            "parser": "text"
        },
        "firewall_rules": {
            "cmd": "sudo iptables -L -n --line-numbers 2>/dev/null; sudo ip6tables -L -n 2>/dev/null",
            "parser": "text"
        },
        "sysctl_params": {
            "cmd": "sudo sysctl -a 2>/dev/null | grep -E '(net\\.ipv4|net\\.ipv6|kernel\\.)'",
            "parser": "keyvalue"
        },
        "listening_ports": {
            "cmd": "sudo ss -tlnp 2>/dev/null; sudo ss -ulnp 2>/dev/null",
            "parser": "text"
        },
        "user_accounts": {
            "cmd": "cat /etc/passwd | grep -v '/nologin' | grep -v '/false'",
            "parser": "text"
        },
        "docker_config": {
            "cmd": "docker info 2>/dev/null; docker ps --format '{{.Names}}\\t{{.Image}}\\t{{.Status}}\\t{{.Ports}}'",
            "parser": "text"
        },
        "nginx_config": {
            "cmd": "sudo nginx -T 2>/dev/null | head -200",
            "parser": "text"
        },
        "installed_packages": {
            "cmd": "dpkg -l 2>/dev/null | grep '^ii' | awk '{print $2}' | sort",
            "parser": "list"
        },
        "cron_jobs": {
            "cmd": "crontab -l 2>/dev/null; ls /etc/cron.d/",
            "parser": "text"
        },
        "kernel_modules": {
            "cmd": "lsmod | awk 'NR==1{print} NF>=3{print $1}'",
            "parser": "list"
        }
    }

    def collect_local(self) -> dict:
        """采集本地服务器配置"""
        snapshot = {
            "hostname": subprocess.getoutput("hostname"),
            "os": subprocess.getoutput("cat /etc/os-release | grep PRETTY_NAME"),
            "kernel": subprocess.getoutput("uname -r"),
            "uptime": psutil.boot_time(),
            "collected_at": datetime.now().isoformat(),
            "items": {}
        }

        for name, spec in self.COLLECTION_ITEMS.items():
            try:
                result = subprocess.run(
                    spec["cmd"], shell=True, capture_output=True, text=True, timeout=15
                )
                snapshot["items"][name] = {
                    "raw_output": result.stdout.strip(),
                    "exit_code": result.returncode,
                    "error": result.stderr.strip() if result.stderr else None
                }
            except subprocess.TimeoutExpired:
                snapshot["items"][name] = {"error": "采集超时", "raw_output": ""}
            except Exception as e:
                snapshot["items"][name] = {"error": str(e), "raw_output": ""}

        return snapshot

    def collect_remote(self, host: str, port: int = 22,
                       username: str = "root", key_path: str = None) -> dict:
        """通过 SSH 采集远程服务器配置"""
        client = paramiko.SSHClient()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        connect_kwargs = {
            "hostname": host, "port": port, "username": username,
            "timeout": 30, "allow_agent": True, "look_for_keys": True
        }
        if key_path:
            connect_kwargs["key_filename"] = key_path

        client.connect(**connect_kwargs)
        snapshot = {"hostname": host, "collected_at": datetime.now().isoformat(), "items": {}}

        for name, spec in self.COLLECTION_ITEMS.items():
            try:
                stdin, stdout, stderr = client.exec_command(spec["cmd"], timeout=20)
                snapshot["items"][name] = {
                    "raw_output": stdout.read().decode().strip(),
                    "error": stderr.read().decode().strip() if stderr.read() else None
                }
            except Exception as e:
                snapshot["items"][name] = {"error": str(e), "raw_output": ""}

        client.close()
        return snapshot

3.2 服务器配置

// config.json
{
  "servers": [
    {"name": "prod-web-01", "host": "10.0.1.10", "role": "生产 Web", "port": 22},
    {"name": "prod-db-01",  "host": "10.0.1.20", "role": "生产数据库", "port": 22},
    {"name": "staging-01",  "host": "10.0.2.10", "role": "预发环境",    "port": 22},
    {"name": "dev-01",      "host": "10.0.3.10", "role": "开发环境",    "port": 22}
  ],
  "ssh_key_path": "~/.ssh/id_ed25519",
  "inspection_profile": "strict"
}

四、安全规则库

光有数据不够——我们需要告诉 LLM 什么算安全、什么不算。规则库采用分层设计：

4.1 CIS Benchmark 基础规则

# safety_rules.py
"""
安全规则库 —— 基于 CIS Linux Benchmark + 自定义扩展
每条规则包含：ID、检查项、期望值、风险等级、描述、修复命令
"""
import json

SAFETY_RULES = [
    {
        "id": "SSH-001",
        "category": "ssh",
        "check": "PermitRootLogin",
        "expected": "no",
        "severity": "HIGH",
        "description": "禁止 root 直接 SSH 登录，使用普通用户 + sudo 替代",
        "fix_command": "sed -i 's/^#*PermitRootLogin.*/PermitRootLogin no/' /etc/ssh/sshd_config && systemctl reload sshd",
        "cis_ref": "CIS 5.2.6"
    },
    {
        "id": "SSH-002",
        "category": "ssh",
        "check": "PasswordAuthentication",
        "expected": "no",
        "severity": "HIGH",
        "description": "禁用密码登录，仅允许密钥认证",
        "fix_command": "sed -i 's/^#*PasswordAuthentication.*/PasswordAuthentication no/' /etc/ssh/sshd_config && systemctl reload sshd",
        "cis_ref": "CIS 5.2.8"
    },
    {
        "id": "SSH-003",
        "category": "ssh",
        "check": "Port",
        "expected_pattern": "^(?!22$)",
        "severity": "MEDIUM",
        "description": "SSH 端口建议使用非标准端口以减少暴力破解风险",
        "fix_command": "# 修改 /etc/ssh/sshd_config 中的 Port 为非常用端口",
        "cis_ref": "CIS 5.2.1"
    },
    {
        "id": "NET-001",
        "category": "network",
        "check": "net.ipv4.ip_forward",
        "expected": "0",
        "severity": "MEDIUM",
        "description": "非路由器的 VPS 应关闭 IP 转发，防止被利用作跳板",
        "fix_command": "sysctl -w net.ipv4.ip_forward=0 && echo 'net.ipv4.ip_forward=0' >> /etc/sysctl.conf",
        "cis_ref": "CIS 3.3.1"
    },
    {
        "id": "NET-002",
        "category": "network",
        "check": "net.ipv4.conf.all.accept_redirects",
        "expected": "0",
        "severity": "LOW",
        "description": "禁用 ICMP 重定向接收，防止路由劫持",
        "fix_command": "sysctl -w net.ipv4.conf.all.accept_redirects=0",
        "cis_ref": "CIS 3.3.2"
    },
    {
        "id": "NET-003",
        "category": "network",
        "check": "net.ipv4.conf.all.send_redirects",
        "expected": "0",
        "severity": "LOW",
        "description": "禁用发送 ICMP 重定向",
        "fix_command": "sysctl -w net.ipv4.conf.all.send_redirects=0",
        "cis_ref": "CIS 3.3.3"
    },
    {
        "id": "NET-004",
        "category": "network",
        "check": "net.ipv4.tcp_syncookies",
        "expected": "1",
        "severity": "MEDIUM",
        "description": "启用 SYN Cookies 防御 SYN Flood 攻击",
        "fix_command": "sysctl -w net.ipv4.tcp_syncookies=1",
        "cis_ref": "CIS 3.3.5"
    },
    {
        "id": "FILE-001",
        "category": "filesystem",
        "check": "world_writable_tmp",
        "expected": "no",
        "severity": "LOW",
        "description": "/tmp 应有 sticky bit（1777），防止用户删除他人文件",
        "fix_command": "chmod 1777 /tmp && chmod 1777 /var/tmp",
        "cis_ref": "CIS 1.5.1"
    },
    {
        "id": "CONTAINER-001",
        "category": "docker",
        "check": "privileged_containers",
        "expected": "0",
        "severity": "HIGH",
        "description": "禁止运行 privileged 模式的容器（权限等同于 root 主机）",
        "fix_command": "# 停止并重新部署非 privileged 容器",
        "cis_ref": "CIS 5.14"
    },
    {
        "id": "USER-001",
        "category": "users",
        "check": "empty_password_accounts",
        "expected": "0",
        "severity": "CRITICAL",
        "description": "系统中不应存在空密码账户",
        "fix_command": "# 使用 passwd -l <user> 锁定空密码账户",
        "cis_ref": "CIS 5.1.3"
    },
    {
        "id": "FIREWALL-001",
        "category": "firewall",
        "check": "default_deny_inbound",
        "expected": "yes",
        "severity": "HIGH",
        "description": "入站默认策略应为 DROP/DENY，仅开放必要端口",
        "fix_command": "# 配置 iptables 默认策略：iptables -P INPUT DROP",
        "cis_ref": "CIS 3.5.1"
    },
    {
        "id": "LOG-001",
        "category": "logging",
        "check": "auditd_enabled",
        "expected": "yes",
        "severity": "MEDIUM",
        "description": "应启用 auditd 审计守护进程以记录系统调用",
        "fix_command": "systemctl enable --now auditd",
        "cis_ref": "CIS 4.1"
    }
]

# 将规则转换为 LLM 可用的 prompt 格式
def build_rule_prompt(rules: list = None) -> str:
    """将安全规则库转换为 LLM 可理解的提示词"""
    if rules is None:
        rules = SAFETY_RULES

    lines = ["## 安全基线规则库", ""]
    for rule in rules:
        lines.append(f"### {rule['id']} [{rule['severity']}]")
        lines.append(f"- 类别: {rule['category']}")
        lines.append(f"- 检查项: {rule['check']}")
        lines.append(f"- 期望值: `{rule['expected']}`")
        lines.append(f"- 说明: {rule['description']}")
        lines.append(f"- 修复命令: `{rule['fix_command']}`")
        if rule.get("cis_ref"):
            lines.append(f"- CIS 引用: {rule['cis_ref']}")
        lines.append("")

    return "\n".join(lines)

五、LLM 分析引擎

这是系统的核心——让 LLM 理解配置数据并给出风险评估：

5.1 分析器实现

# llm_analyzer.py
import json
import ollama
from datetime import datetime
from safety_rules import SAFETY_RULES, build_rule_prompt

class ConfigAnalyzer:
    """LLM 配置分析引擎"""

    def __init__(self, model: str = "qwen2.5:7b", base_url: str = "http://localhost:11434"):
        self.model = model
        self.base_url = base_url
        self.rule_prompt = build_rule_prompt()

    def analyze_snapshot(self, snapshot: dict) -> dict:
        """分析单台服务器的配置快照"""
        hostname = snapshot.get("hostname", "unknown")
        collected_at = snapshot.get("collected_at", "")

        # 构建配置上下文
        config_context = self._build_config_context(snapshot)

        # LLM 分析提示词
        analysis_prompt = f"""你是一位资深 Linux 系统安全工程师，正在对一台 VPS 进行配置安全巡检。

## 服务器信息
- 主机名: {hostname}
- 采集时间: {collected_at}

## 安全基线规则
{self.rule_prompt}

## 当前配置数据
{config_context}

## 你的任务
请逐项对照安全规则检查上述配置数据，对每项规则给出：
1. **状态**: PASS（合规）/ FAIL（违规）/ SKIP（无法判断）
2. **风险等级**: CRITICAL / HIGH / MEDIUM / LOW / INFO
3. **问题描述**: 简要说明哪里有问题
4. **修复建议**: 具体的修复命令或操作步骤

请以 JSON 数组格式返回结果，格式如下：
[
  {{
    "rule_id": "SSH-001",
    "status": "FAIL",
    "severity": "HIGH",
    "description": "PermitRootLogin 设置为 yes，存在风险",
    "fix_command": "sed -i 's/^#*PermitRootLogin.*/PermitRootLogin no/' /etc/ssh/sshd_config && systemctl reload sshd",
    "cis_ref": "CIS 5.2.6"
  }}
]

只返回 JSON 数组，不要有其他文字。"""

        try:
            response = ollama.chat(
                model=self.model,
                messages=[{"role": "user", "content": analysis_prompt}],
                options={
                    "temperature": 0.1,
                    "num_predict": 2048
                }
            )
            result_text = response["message"]["content"]

            # 提取 JSON
            start = result_text.find("[")
            end = result_text.rfind("]") + 1
            if start >= 0 and end > start:
                results = json.loads(result_text[start:end])
            else:
                results = [{"rule_id": "ERROR", "status": "SKIP", "description": f"解析失败: {result_text[:200]}"}]

            return {
                "hostname": hostname,
                "collected_at": collected_at,
                "analyzed_at": datetime.now().isoformat(),
                "results": results,
                "summary": self._compute_summary(results)
            }

        except Exception as e:
            return {
                "hostname": hostname,
                "error": str(e),
                "collected_at": collected_at
            }

    def _build_config_context(self, snapshot: dict) -> str:
        """将原始配置数据转换为 LLM 可读的上下文"""
        lines = []
        items = snapshot.get("items", {})

        # SSH 配置
        ssh_raw = items.get("ssh_config", {}).get("raw_output", "")
        if ssh_raw:
            lines.append("### SSH 配置 (/etc/ssh/sshd_config)")
            lines.append(ssh_raw)
            lines.append("")

        # 内核参数
        sysctl_raw = items.get("sysctl_params", {}).get("raw_output", "")
        if sysctl_raw:
            lines.append("### 内核网络参数 (sysctl)")
            lines.append(sysctl_raw)
            lines.append("")

        # 监听端口
        ports_raw = items.get("listening_ports", {}).get("raw_output", "")
        if ports_raw:
            lines.append("### 监听端口")
            lines.append(ports_raw)
            lines.append("")

        # 用户账户
        users_raw = items.get("user_accounts", {}).get("raw_output", "")
        if users_raw:
            lines.append("### 活跃用户账户")
            lines.append(users_raw)
            lines.append("")

        # Docker 配置
        docker_raw = items.get("docker_config", {}).get("raw_output", "")
        if docker_raw:
            lines.append("### Docker 状态")
            lines.append(docker_raw)
            lines.append("")

        # 防火墙
        fw_raw = items.get("firewall_rules", {}).get("raw_output", "")
        if fw_raw:
            lines.append("### 防火墙规则 (iptables)")
            lines.append(fw_raw)
            lines.append("")

        return "\n".join(lines)

    def _compute_summary(self, results: list) -> dict:
        """计算汇总统计"""
        severity_counts = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0, "INFO": 0}
        status_counts = {"PASS": 0, "FAIL": 0, "SKIP": 0}
        failed_rules = []

        for r in results:
            sev = r.get("severity", "INFO")
            status = r.get("status", "SKIP")
            severity_counts[sev] = severity_counts.get(sev, 0) + 1
            status_counts[status] = status_counts.get(status, 0) + 1
            if status == "FAIL":
                failed_rules.append(r["rule_id"])

        # 计算安全评分（100 分制）
        total = len(results)
        passed = status_counts.get("PASS", 0)
        critical_penalty = severity_counts.get("CRITICAL", 0) * 15
        high_penalty = severity_counts.get("HIGH", 0) * 10
        medium_penalty = severity_counts.get("MEDIUM", 0) * 5
        low_penalty = severity_counts.get("LOW", 0) * 2

        score = max(0, 100 - critical_penalty - high_penalty - medium_penalty - low_penalty)

        return {
            "total_checks": total,
            "passed": passed,
            "failed": status_counts.get("FAIL", 0),
            "skipped": status_counts.get("SKIP", 0),
            "severity_counts": severity_counts,
            "failed_rule_ids": failed_rules,
            "security_score": score,
            "grade": self._score_to_grade(score)
        }

    def _score_to_grade(self, score: int) -> str:
        if score >= 90:
            return "A"
        elif score >= 80:
            return "B"
        elif score >= 70:
            return "C"
        elif score >= 60:
            return "D"
        else:
            return "F"

六、报告生成器

分析完成后，生成人类可读的巡检报告：

6.1 Markdown 报告

# report_generator.py
from datetime import datetime
from typing import List, Dict

class ReportGenerator:
    """生成 Markdown 格式巡检报告"""

    SEVERITY_EMOJI = {
        "CRITICAL": "🔴",
        "HIGH": "🟠",
        "MEDIUM": "🟡",
        "LOW": "🔵",
        "INFO": "⚪"
    }

    STATUS_SYMBOL = {
        "PASS": "✅",
        "FAIL": "❌",
        "SKIP": "⏭️"
    }

    def generate_report(self, analyses: List[Dict]) -> str:
        """生成完整巡检报告"""
        lines = [
            f"# 🔍 VPS 配置智能巡检报告",
            f"",
            f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
            f"**巡检模式**: 安全基线检查（CIS Benchmark + 自定义规则）",
            f"",
            f"---",
            f""
        ]

        # 全局摘要
        total_score = sum(a["summary"]["security_score"] for a in analyses if "summary" in a)
        avg_score = total_score / len(analyses) if analyses else 0
        total_fail = sum(a["summary"]["failed"] for a in analyses if "summary" in a)

        lines.append("## 📊 全局摘要")
        lines.append("")
        lines.append(f"| 指标 | 数值 |")
        lines.append(f"|------|------|")
        lines.append(f"| 巡检服务器数 | {len(analyses)} |")
        lines.append(f"| 平均安全评分 | {avg_score:.0f}/100 |")
        lines.append(f"| 总违规项数 | {total_fail} |")
        lines.append(f"| 全局评级 | {self._score_to_grade(avg_score)} |")
        lines.append("")

        # 每台服务器的详细报告
        for analysis in analyses:
            hostname = analysis.get("hostname", "unknown")
            summary = analysis.get("summary", {})
            results = analysis.get("results", [])
            error = analysis.get("error")

            lines.append(f"---")
            lines.append(f"")
            lines.append(f"## 🖥️ {hostname}")
            lines.append(f"**采集时间**: {analysis.get('collected_at', 'N/A')}")
            lines.append(f"")

            if error:
                lines.append(f"⚠️ **采集或分析出错**: {error}")
                lines.append("")
                continue

            # 服务器评分卡片
            score = summary.get("security_score", 0)
            grade = summary.get("grade", "F")
            lines.append(f"> **安全评分**: {score}/100  **评级**: {grade}")
            lines.append(f">")
            lines.append(f"> 通过: {summary.get('passed', 0)} | 违规: {summary.get('failed', 0)} | "
                        f"跳过: {summary.get('skipped', 0)} | 总计: {summary.get('total_checks', 0)}")
            lines.append(f"")

            # 严重度分布
            sev_counts = summary.get("severity_counts", {})
            if any(v > 0 for v in sev_counts.values()):
                lines.append("**风险分布**: ")
                parts = []
                if sev_counts.get("CRITICAL", 0) > 0:
                    parts.append(f"🔴 严重 {sev_counts['CRITICAL']}")
                if sev_counts.get("HIGH", 0) > 0:
                    parts.append(f"🟠 高危 {sev_counts['HIGH']}")
                if sev_counts.get("MEDIUM", 0) > 0:
                    parts.append(f"🟡 中危 {sev_counts['MEDIUM']}")
                if sev_counts.get("LOW", 0) > 0:
                    parts.append(f"🔵 低危 {sev_counts['LOW']}")
                lines.append(" | ".join(parts))
                lines.append("")

            # 违规详情
            failed = [r for r in results if r.get("status") == "FAIL"]
            if failed:
                lines.append(f"### ❌ 违规项 ({len(failed)})")
                lines.append("")
                for r in failed:
                    emoji = self.SEVERITY_EMOJI.get(r.get("severity", "INFO"), "⚪")
                    lines.append(f"#### {emoji} {r['rule_id']} — {r.get('severity', 'INFO')}")
                    lines.append(f"- **问题**: {r.get('description', 'N/A')}")
                    if r.get("cis_ref"):
                        lines.append(f"- **CIS 引用**: {r['cis_ref']}")
                    fix = r.get("fix_command", "")
                    if fix:
                        lines.append(f"- **修复命令**:")
                        lines.append(f"  ```bash")
                        lines.append(f"  {fix}")
                        lines.append(f"  ```")
                    lines.append("")

            # 通过项（折叠）
            passed = [r for r in results if r.get("status") == "PASS"]
            if passed:
                lines.append(f"<details>")
                lines.append(f"<summary>✅ 合规项 ({len(passed)}) — 点击展开</summary>")
                lines.append(f"")
                for r in passed[:10]:  # 只显示前 10 条
                    lines.append(f"- ✅ {r['rule_id']}: {r.get('description', '合规')}")
                if len(passed) > 10:
                    lines.append(f"- ... 还有 {len(passed) - 10} 项合规")
                lines.append(f"</details>")
                lines.append("")

        # 整改优先级列表
        lines.append("---")
        lines.append("")
        lines.append("## 🎯 整改优先级建议")
        lines.append("")
        lines.append("请按以下顺序处理违规项（从高危到低危）：")
        lines.append("")

        all_failures = []
        for analysis in analyses:
            for r in analysis.get("results", []):
                if r.get("status") == "FAIL":
                    all_failures.append({
                        **r,
                        "hostname": analysis.get("hostname")
                    })

        # 按严重度排序
        severity_order = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3, "INFO": 4}
        all_failures.sort(key=lambda x: severity_order.get(x.get("severity", "INFO"), 4))

        for i, f in enumerate(all_failures, 1):
            emoji = self.SEVERITY_EMOJI.get(f.get("severity", "INFO"), "⚪")
            lines.append(f"{i}. **[{f['severity']}]** {emoji} **{f['hostname']}** — {f['rule_id']}: {f.get('description', '')}")
            if f.get("fix_command"):
                lines.append(f"   ```bash")
                lines.append(f"   # {f['hostname']}")
                lines.append(f"   {f['fix_command']}")
                lines.append(f"   ```")

        lines.append("")
        lines.append("---")
        lines.append("")
        lines.append(f"*报告由 AI 配置智能巡检系统自动生成 | 模型: Ollama Qwen2.5:7b*")

        return "\n".join(lines)

    def _score_to_grade(self, score: float) -> str:
        if score >= 90: return "A"
        elif score >= 80: return "B"
        elif score >= 70: return "C"
        elif score >= 60: return "D"
        else: return "F"

七、通知推送

巡检完成后，自动推送报告摘要：

# notifier.py
import os
import json
import requests
from pathlib import Path

class Notifier:
    """多通道通知推送"""

    def __init__(self, telegram_token: str = None, telegram_chat_id: str = None):
        self.tg_token = telegram_token or os.environ.get("TELEGRAM_BOT_TOKEN")
        self.tg_chat_id = telegram_chat_id or os.environ.get("TELEGRAM_CHAT_ID")

    def send_telegram(self, report_md: str, summary: dict):
        """通过 Telegram Bot 推送巡检摘要"""
        if not self.tg_token or not self.tg_chat_id:
            print("⚠️ Telegram 未配置，跳过通知")
            return

        # 构建摘要消息
        servers = summary.get("servers", [])
        avg_score = summary.get("average_score", 0)
        total_fail = summary.get("total_failures", 0)

        msg = f"🔍 **VPS 配置巡检完成**\n\n"
        msg += f"📅 {summary.get('generated_at', '')}\n"
        msg += f"🖥️  巡检服务器: {len(servers)} 台\n"
        msg += f"⭐ 平均安全评分: **{avg_score:.0f}/100**\n"
        msg += f"❌ 总违规项: {total_fail}\n\n"

        # 每台服务器状态
        for s in servers:
            emoji = "✅" if s["score"] >= 80 else "⚠️" if s["score"] >= 60 else "🔴"
            msg += f"{emoji} **{s['hostname']}** — 评分 {s['score']} ({s['grade']})，"
            msg += f"违规 {s['failed']} 项\n"

        msg += "\n📄 完整报告已保存至本地。"

        # 限制 Telegram 消息长度
        if len(msg) > 4000:
            msg = msg[:3997] + "..."

        url = f"https://api.telegram.org/bot{self.tg_token}/sendMessage"
        requests.post(url, json={
            "chat_id": self.tg_chat_id,
            "text": msg,
            "parse_mode": "Markdown"
        }, timeout=10)
        print("✅ Telegram 通知已发送")

    def save_report(self, report_md: str, filename: str = None):
        """保存报告到本地文件"""
        output_dir = Path("reports")
        output_dir.mkdir(exist_ok=True)

        if not filename:
            filename = f"inspection-report-{datetime.now().strftime('%Y%m%d-%H%M%S')}.md"

        filepath = output_dir / filename
        filepath.write_text(report_md, encoding="utf-8")
        print(f"📄 报告已保存: {filepath}")
        return filepath

八、主入口与定时任务

# inspector.py
#!/usr/bin/env python3
"""
AI VPS 配置智能巡检系统
用法: python inspector.py [--remote] [--profile strict|basic]
"""
import json
import argparse
from pathlib import Path
from datetime import datetime

from config_collector import ConfigCollector
from llm_analyzer import ConfigAnalyzer
from report_generator import ReportGenerator
from notifier import Notifier

def main():
    parser = argparse.ArgumentParser(description="AI VPS 配置智能巡检系统")
    parser.add_argument("--remote", action="store_true", help="采集远程服务器（需要 SSH 密钥）")
    parser.add_argument("--profile", choices=["strict", "basic"], default="strict",
                        help="巡检严格程度 (default: strict)")
    parser.add_argument("--config", default="config.json", help="配置文件路径")
    parser.add_argument("--save-report", action="store_true", help="保存报告文件")
    parser.add_argument("--notify", action="store_true", help="发送 Telegram 通知")
    args = parser.parse_args()

    # 加载配置
    config_path = Path(args.config)
    if not config_path.exists():
        print(f"❌ 配置文件不存在: {args.config}")
        return 1

    with open(config_path) as f:
        config = json.load(f)

    collector = ConfigCollector()
    analyzer = ConfigAnalyzer()
    reporter = ReportGenerator()
    notifier = Notifier()

    print(f"🔍 开始 AI 配置智能巡检 ({datetime.now().strftime('%Y-%m-%d %H:%M:%S')})")
    print(f"📋 巡检模式: {args.profile}")
    print()

    analyses = []
    servers = config.get("servers", [])

    for server in servers:
        name = server.get("name", "unknown")
        print(f"  🖥️  正在检查: {name} ({server.get('host', '')}) ... ", end="", flush=True)

        if args.remote:
            snapshot = collector.collect_remote(
                host=server["host"],
                port=server.get("port", 22),
                key_path=config.get("ssh_key_path")
            )
        else:
            snapshot = collector.collect_local()
            snapshot["hostname"] = name

        result = analyzer.analyze_snapshot(snapshot)
        analyses.append(result)
        score = result.get("summary", {}).get("security_score", 0)
        grade = result.get("summary", {}).get("grade", "?")
        print(f"✅ 评分 {score}/100 ({grade})")

    # 生成报告
    report_md = reporter.generate_report(analyses)

    if args.save_report:
        notifier.save_report(report_md)

    # 构建摘要用于通知
    summary = {
        "generated_at": datetime.now().isoformat(),
        "servers": [
            {
                "hostname": a.get("hostname", "unknown"),
                "score": a.get("summary", {}).get("security_score", 0),
                "grade": a.get("summary", {}).get("grade", "?"),
                "failed": a.get("summary", {}).get("failed", 0)
            }
            for a in analyses
        ],
        "average_score": sum(a.get("summary", {}).get("security_score", 0) for a in analyses) / len(analyses) if analyses else 0,
        "total_failures": sum(a.get("summary", {}).get("failed", 0) for a in analyses)
    }

    if args.notify:
        notifier.send_telegram(report_md, summary)

    # 打印摘要到终端
    print()
    print("=" * 60)
    print("📊 巡检结果摘要")
    print("=" * 60)
    for a in analyses:
        s = a.get("summary", {})
        print(f"  {a.get('hostname', '?'):20s} | 评分: {s.get('security_score', 0):3d}/100 | "
              f"评级: {s.get('grade', '?')} | 违规: {s.get('failed', 0)} 项")
    print("=" * 60)

    return 0

if __name__ == "__main__":
    exit(main())

设置定时巡检

# 每天凌晨 2 点执行巡检，结果推送到 Telegram
crontab -e

# 添加以下行：
0 2 * * * cd ~/ai-config-inspector && source venv/bin/activate && \
  python inspector.py --remote --save-report --notify >> /var/log/config-inspector.log 2>&1

# 每周一早上 9 点执行严格模式深度巡检
0 9 * * 1 cd ~/ai-config-inspector && source venv/bin/activate && \
  python inspector.py --remote --profile strict --save-report --notify \
  >> /var/log/config-inspector-weekly.log 2>&1

九、实战演示

场景 1：日常巡检

cd ~/ai-config-inspector
source venv/bin/activate
python inspector.py --remote --save-report --notify

输出示例：

🔍 开始 AI 配置智能巡检 (2026-09-21 02:00:01)
📋 巡检模式: strict

  🖥️  正在检查: prod-web-01 (10.0.1.10) ... ✅ 评分 72/100 (C)
  🖥️  正在检查: prod-db-01 (10.0.1.20) ... ✅ 评分 85/100 (B)
  🖥️  正在检查: staging-01 (10.0.2.10) ... ✅ 评分 68/100 (C)
  🖥️  正在检查: dev-01 (10.0.3.10) ...    ✅ 评分 91/100 (A)

============================================================
📊 巡检结果摘要
============================================================
  prod-web-01          | 评分:  72/100 | 评级: C | 违规: 5 项
  prod-db-01           | 评分:  85/100 | 评级: B | 违规: 2 项
  staging-01           | 评分:  68/100 | 评级: C | 违规: 7 项
  dev-01               | 评分:  91/100 | 评级: A | 违规: 0 项
============================================================
📄 报告已保存: reports/inspection-report-20260921-020001.md
✅ Telegram 通知已发送

场景 2：历史趋势对比

系统会自动保存每次巡检的历史数据（JSON 格式），支持对比分析：

# history.py — 历史趋势追踪
import json
from pathlib import Path
from datetime import datetime, timedelta

HISTORY_FILE = Path("history/inspection_history.json")

def load_history():
    if HISTORY_FILE.exists():
        return json.loads(HISTORY_FILE.read_text())
    return {"records": []}

def append_record(hostname: str, score: int, failed: int, grade: str):
    history = load_history()
    record = {
        "hostname": hostname,
        "timestamp": datetime.now().isoformat(),
        "score": score,
        "failed": failed,
        "grade": grade
    }
    history["records"].append(record)
    # 只保留最近 90 天
    cutoff = datetime.now() - timedelta(days=90)
    history["records"] = [
        r for r in history["records"]
        if datetime.fromisoformat(r["timestamp"]) > cutoff
    ]
    HISTORY_FILE.parent.mkdir(parents=True, exist_ok=True)
    HISTORY_FILE.write_text(json.dumps(history, ensure_ascii=False, indent=2))
    return record

def get_trend(hostname: str, days: int = 30) -> list:
    """获取指定服务器的分数趋势"""
    history = load_history()
    cutoff = datetime.now() - timedelta(days=days)
    records = [
        r for r in history["records"]
        if r["hostname"] == hostname
        and datetime.fromisoformat(r["timestamp"]) > cutoff
    ]
    return sorted(records, key=lambda x: x["timestamp"])

趋势数据可用于生成评分趋势图（配合 Grafana 或直接输出 ASCII 图表）：

prod-web-01 安全评分趋势（近 30 天）
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Sep  1  ████████████████████░░░░  78
Sep  3  ████████████████████░░░░  78
Sep  5  █████████████████████░░░  82  ← 修复了 SSH 根登录
Sep  8  ██████████████████████░░  85  ← 启用了 fail2ban
Sep 10  ██████████████████████░░  85
Sep 12  ███████████████████████░  88  ← 关闭了 IP 转发
Sep 15  ███████████████████████░  88
Sep 18  ███████████████████████░  88
Sep 21  ████████████████████████  91  ← 新增安全规则通过

十、扩展：自定义规则与动态基线

10.1 添加自定义规则

在 safety_rules.py 中追加你的业务规则：

CUSTOM_RULES = [
    {
        "id": "CUSTOM-001",
        "category": "application",
        "check": "app_health_check",
        "expected": "healthy",
        "severity": "HIGH",
        "description": "核心业务服务 health endpoint 应返回 200",
        "fix_command": "# 检查 /healthz 端点响应",
        "cis_ref": None
    }
]

10.2 动态基线（自适应规则）

对于不同环境的服务器，可以设置不同的基线标准：

ENVIRONMENTS = {
    "production": {"profile": "strict", "min_score": 85},
    "staging":    {"profile": "standard", "min_score": 70},
    "development":{"profile": "basic",    "min_score": 50}
}

系统在巡检时会根据服务器角色自动选择对应的基线标准，并在报告中突出显示低于环境基线的服务器。

总结

本文介绍了如何构建一套 AI 配置智能巡检系统，核心思路是：

自动化采集 — 从 SSH/内核/网络/Docker 等多源采集配置快照，告别手动逐项检查
LLM 智能分析 — 用本地 Qwen 模型对照 CIS Benchmark 规则库进行风险评估，准确率远超人工
结构化报告 — 生成带安全评分、分级整改建议和修复命令的 Markdown 报告
闭环推送 — 通过 Telegram 推送摘要，支持定时任务实现无人值守巡检

与传统巡检相比，这套系统的优势：

效率提升 10x+：10 台服务器的巡检从 2 小时缩短到 5 分钟
标准统一：所有服务器使用相同的安全规则库，消除人为判断差异
持续演进：规则库可随时扩展，适应新的安全要求
成本极低：本地 Ollama 运行，无 API 调用费用

每周一次智能巡检，让你的 VPS 集群始终保持在最佳安全状态。