技术教程 · 阅读约 9 分钟

服务器中了挖矿病毒?我让 AI 远程 SSH 上去,30 分钟搞定了

服务器中了挖矿病毒?我让 AI 远程 SSH 上去,30 分钟搞定了

昨天 V2EX 有个帖子让我印象很深——

"服务器 CPU 100% 下不去,最后让 Codex 远程 SSH 上去硬刚"

我第一反应是:这操作有点猛。但仔细想想,让 AI agent 拿着 SSH 权限去处理突发事故,其实是个很值得深挖的方向。于是我照着这个思路,自己搭了一套可运行的方案,用的是 Claude + Python + paramiko,记录下来给大家参考。


为什么不自己上去排查?

不是不会,是慌的时候容易误操作。

CPU 100% 的时候,kill 错进程、rm 错文件,这种事不是没有。更别说挖矿木马往往会:

  • 改 crontab 定时复活
  • 伪装成系统进程名(kworker、sshd 等)
  • 在多个路径写入副本

人工排查要靠经验,还要沉住气。AI 不会慌,而且它的排查思路可以被你提前写死,不会乱来。


整体思路


你 → Python 脚本 → SSH 连上服务器 → 把诊断输出喂给 Claude → Claude 给出命令 → 脚本执行 → 循环

不是让 AI 直接拿 shell,而是每一步都经过你的 Python 层做过滤和确认,可控。


完整代码

依赖安装:


pip install paramiko openai


import paramiko

import openai

import json



# ---- 配置区 ----

SSH_HOST = "your.server.ip"

SSH_PORT = 22

SSH_USER = "root"

SSH_KEY  = "/path/to/your/id_rsa"



client = openai.OpenAI(

    api_key="sk-xxxx",                        # 换成你的 key

    base_url="https://api2everything.xyz/v1"  # 国内直连,无需代理

)



SYSTEM_PROMPT = """

你是一名 Linux 安全运维专家。

我会把服务器的诊断输出发给你,你负责:

1. 判断是否存在挖矿/恶意进程

2. 给出下一步要执行的 shell 命令(单条)

3. 如果排查完毕,回复 DONE



严格用 JSON 格式回复:

{"action": "command"|"done", "cmd": "...", "reason": "..."}



注意:绝对不要给出 rm -rf、格式化等破坏性命令。

""".strip()



# ---- SSH 工具 ----

def ssh_run(ssh, cmd):

    stdin, stdout, stderr = ssh.exec_command(cmd, timeout=15)

    out = stdout.read().decode(errors="replace")

    err = stderr.read().decode(errors="replace")

    return (out + err).strip()[:3000]  # 截断,避免 token 爆炸



# ---- 诊断初始快照 ----

INIT_COMMANDS = [

    "top -bn1 | head -30",

    "ps aux --sort=-%cpu | head -20",

    "crontab -l 2>/dev/null",

    "ls /tmp && ls /var/tmp",

    "last | head -10",

    "ss -tnp | grep ESTABLISHED"

]



def collect_snapshot(ssh):

    parts = []

    for cmd in INIT_COMMANDS:

        output = ssh_run(ssh, cmd)

        parts.append(f"$ {cmd}\n{output}")

    return "\n\n".join(parts)



# ---- 主循环 ----

def investigate(ssh, max_rounds=10):

    messages = [{"role": "system", "content": SYSTEM_PROMPT}]

    

    print("[*] 采集初始快照...")

    snapshot = collect_snapshot(ssh)

    messages.append({"role": "user", "content": f"服务器当前状态:\n\n{snapshot}"})



    for i in range(max_rounds):

        print(f"\n[Round {i+1}] 请求 Claude 分析...")

        resp = client.chat.completions.create(

            model="claude-sonnet-4-5",

            messages=messages,

            temperature=0.2

        )

        raw = resp.choices[0].message.content.strip()

        print(f"Claude: {raw}")



        try:

            data = json.loads(raw)

        except json.JSONDecodeError:

            print("[!] 解析失败,跳过本轮")

            break



        if data["action"] == "done":

            print("\n[✓] Claude 认为排查完毕:", data.get("reason", ""))

            break



        cmd = data["cmd"]

        reason = data.get("reason", "")

        print(f"[执行] {cmd}  # {reason}")



        # 安全检查:拦截危险命令

        BLOCKED = ["rm -rf /", "mkfs", "dd if=", "> /dev/sda", "shutdown", "reboot"]

        if any(b in cmd for b in BLOCKED):

            print("[!] 危险命令已拦截,终止")

            break



        # 可以在这里加 input("确认执行?") 做人工确认

        output = ssh_run(ssh, cmd)

        print(f"输出:\n{output[:500]}")



        messages.append({"role": "assistant", "content": raw})

        messages.append({"role": "user", "content": f"执行结果:\n{output}"})



    print("\n[*] 排查结束")



# ---- 入口 ----

if __name__ == "__main__":

    ssh = paramiko.SSHClient()

    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    ssh.connect(SSH_HOST, SSH_PORT, SSH_USER, key_filename=SSH_KEY)

    try:

        investigate(ssh)

    finally:

        ssh.close()


几个值得注意的细节

模型选 claude-sonnet-4-5 而不是 GPT-4o

排查这种任务需要理解上下文链路,Claude 在多轮推理上表现更稳,而且对"不要乱删文件"这类约束遵守得更好。

token 截断很重要

top 和 ps 的输出可能很长,每轮都全量传会让成本飙升。代码里截了 3000 字符,实际使用可以按需调整。

人工确认点

代码里有一行被注释掉的 input("确认执行?")。如果你不放心全自动,取消注释,每步都让你确认再执行——这是最稳的方式。

别给 root 权限

能用普通用户 + sudo 就不要直接给 root 的 SSH key。


跑起来的效果大概是这样


[*] 采集初始快照...



[Round 1] 请求 Claude 分析...

Claude: {"action": "command", "cmd": "ps aux --sort=-%cpu | awk 'NR>1 && $3>50 {print $2, $11}'", "reason": "定位 CPU 占用超过 50% 的进程 PID 和名称"}



[执行] ps aux --sort=-%cpu | awk 'NR>1 && $3>50 {print $2, $11}'

输出: 3821 /tmp/.x/kworker



[Round 2] 请求 Claude 分析...

Claude: {"action": "command", "cmd": "ls -la /tmp/.x/ && cat /proc/3821/cmdline | tr '\\0' ' '", "reason": "确认 /tmp/.x 目录内容,读取进程真实命令行"}

...

AI 会一步步把挖矿进程的路径、crontab 里的复活项、可疑的 /tmp 目录都翻出来。


关于 API 费用

跑一次完整排查大概 10-15 轮对话,claude-sonnet-4-5 的 token 消耗在 2 万左右,官方价格大概要 ¥1.5 左右一次。

我用的是 无量Api,同样的模型比官方便宜约 60%,国内直连不需要代理,OpenAI 格式兼容,就改一行 base_url 就行。注册还送 ¥1 余额,用来跑这种脚本测试足够了。

支持 Claude、GPT-4o、Gemini、DeepSeek 等 300+ 模型,切模型也就改个参数的事。


最后

这套方案不是让 AI 替代你,而是在你最慌的时候给你一个不会慌的助手。

代码拿走可以直接用,如果你的服务器不是用 key 登录而是密码登录,把 key_filename 换成 password 参数就行。

遇到问题欢迎评论区说,或者把你的诊断输出贴出来一起分析。

觉得有用的话点个赞,我后面还会写 AI agent 处理 Nginx 日志分析、自动扩容这些场景。