服务器中了挖矿病毒?我让 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 日志分析、自动扩容这些场景。