第319篇：ChatOps 聊天机器人运维

关键词

ChatOps、聊天机器人、运维自动化、Slack Bot、企业微信、钉钉、WeLink、即席命令

一、ChatOps 概述

1.1 什么是 ChatOps

ChatOps 是将运维操作 集成到聊天工具 中的实践，让工程师在聊天窗口即可执行网络操作：

ChatOps 的理念：

  传统方式：
  ┌─ 打开终端 → SSH 登录设备 → 执行命令 → 阅读回显
  ├─ 耗时：30 秒 ~ 2 分钟
  ├─ 需要：终端工具 + 凭证 + 网络可达
  └─ 不能协作（只有操作者看到输出）

  ChatOps 方式：
  ┌─ 在聊天窗口输入：/ping 192.168.1.1
  ├─ 机器人执行，结果直接发到聊天群
  ├─ 耗时：2 秒
  ├─ 所有人可见
  └─ 操作可审计

  ChatOps 三大优势：
  ┌─ 低门槛：聊天工具人人会
  ├─ 协作透明：群内操作全员可见
  └─ 安全可控：机器人统一管控权限

1.2 架构设计

ChatOps 系统架构：

二、命令行定义

2.1 命令集设计

ChatOps 网络运维命令：

  ┌─ 查询类
  │  /ping <ip>                      Ping 测试
  │  /show <device> <cmd>           执行展示命令
  │  /status <device>               设备健康状态
  │  /int <device> [interface]      接口状态
  │  /bgp <device> [peer]           BGP 状态
  │  /route <device> <prefix>       路由查询
  │
  ├─ 操作类
  │  /backup [device]               配置备份
  │  /deploy <config_file>          部署配置
  │  /rollback <device>             回滚配置
  │  /save <device>                 保存配置
  │
  ├─ 信息类
  │  /version <device>              设备版本
  │  /log <device> [lines]          日志查看
  │  /mac <vlan>                    MAC 地址查询
  │  /arp <ip>                      ARP 查询
  │
  └─ 管理类
       /help                        帮助信息
       /auth                        权限认证
       /whoami                      当前身份
       /history                     我的操作记录

2.2 命令解析引擎

#!/usr/bin/env python3
# chatops_engine.py — ChatOps 命令解析引擎

import re
from typing import Callable, Dict, Optional

class Command:
    """命令定义"""
    def __init__(self, name: str, handler: Callable,
                 description: str, permission: str = "user"):
        self.name = name
        self.handler = handler
        self.description = description
        self.permission = permission
        self.pattern = re.compile(
            f"^/{name}(\\s+(?P<args>.*))?$", re.IGNORECASE
        )

class ChatOpsEngine:
    """ChatOps 命令引擎"""

    def __init__(self):
        self.commands: Dict[str, Command] = {}
        self.users = {}  # user_id → permissions

    def register(self, cmd: Command):
        """注册命令"""
        self.commands[cmd.name] = cmd

    def parse(self, text: str) -> Optional[Dict]:
        """解析用户输入"""
        text = text.strip()

        for cmd in self.commands.values():
            match = cmd.pattern.match(text)
            if match:
                args_str = match.group("args") or ""
                args = args_str.split() if args_str.strip() else []
                return {
                    "command": cmd.name,
                    "handler": cmd.handler,
                    "args": args,
                    "raw_args": args_str,
                }

        return None

    def execute(self, user_id: str, text: str) -> str:
        """执行命令"""
        parsed = self.parse(text)

        if not parsed:
            return self._format_error(
                f"未知命令: {text}\n输入 /help 查看可用命令"
            )

        # 权限检查
        cmd_name = parsed["command"]
        cmd = self.commands[cmd_name]
        user_permission = self.users.get(user_id, "user")

        if cmd.permission == "admin" and user_permission != "admin":
            return self._format_error("权限不足，需要管理员权限")

        # 执行
        try:
            result = parsed["handler"](*parsed["args"])
            return self._format_success(cmd_name, result)
        except Exception as e:
            return self._format_error(str(e))

    def _format_success(self, cmd: str, result: str) -> str:
        return f"✅ *{cmd}* 执行成功\n```\n{result}\n```"

    def _format_error(self, msg: str) -> str:
        return f"❌ {msg}"


# ===== 命令处理器 =====

def cmd_ping(target: str) -> str:
    """Ping 测试"""
    # 实际调用 Netmiko 或系统 ping
    import subprocess
    result = subprocess.run(
        ["ping", "-n", "2", target],
        capture_output=True, text=True, timeout=10,
    )
    return result.stdout or result.stderr

def cmd_status(device: str) -> str:
    """设备状态"""
    from lib.connector import DeviceConnector
    connector = DeviceConnector()
    dev = connector.get_device(device)
    if not dev:
        return f"未知设备: {device}"

    conn = connector.connect_netmiko(dev)
    if not conn:
        return f"{device}: 连接失败"

    try:
        cpu = conn.send_command("display cpu-usage")
        mem = conn.send_command("display memory-usage")
        return f"设备: {device}\nCPU: {cpu[:100]}\n内存: {mem[:100]}"
    finally:
        connector.close(conn)

def cmd_backup(device: Optional[str] = None) -> str:
    """配置备份"""
    from lib.connector import DeviceConnector
    connector = DeviceConnector()
    devices = [connector.get_device(device)] if device else connector.devices
    results = []
    for dev in devices:
        if dev:
            conn = connector.connect_netmiko(dev)
            if conn:
                config = connector.backup_config(conn)
                connector.close(conn)
                results.append(f"✓ {dev['name']}: {len(config)} 字符")
            else:
                results.append(f"✗ {dev['name']}: 连接失败")
    return "\n".join(results) if results else "无设备"

def cmd_help() -> str:
    """帮助信息"""
    return """*可用命令：*
`/ping <ip>` — Ping 测试
`/show <device> <cmd>` — 执行展示命令
`/status <device>` — 设备状态
`/backup [device]` — 配置备份
`/int <device> [intf]` — 接口信息
`/version <device>` — 设备版本
`/help` — 本帮助"""

# ===== 初始化 =====
def create_engine() -> ChatOpsEngine:
    engine = ChatOpsEngine()

    engine.register(Command("ping", cmd_ping, "Ping 测试"))
    engine.register(Command("status", cmd_status, "设备状态"))
    engine.register(Command("backup", cmd_backup, "配置备份", "admin"))
    engine.register(Command("help", cmd_help, "帮助信息"))

    # 添加用户权限
    engine.users = {
        "zhangsan": "admin",
        "lisi": "user",
    }

    return engine

# ===== 测试 =====
if __name__ == "__main__":
    engine = create_engine()

    test_commands = [
        "/help",
        "/ping 8.8.8.8",
        "/status CORE-SW01",
        "/backup",
    ]

    for cmd in test_commands:
        print(f"\n用户: zhangsan 输入: {cmd}")
        print(engine.execute("zhangsan", cmd))

三、企业微信 Bot 集成

3.1 Webhook 接收

#!/usr/bin/env python3
# wework_chatops.py — 企业微信 ChatOps Bot

from flask import Flask, request, jsonify
import hashlib
import xml.etree.ElementTree as ET
import time
import requests

app = Flask(__name__)

# 配置
WEWORK_TOKEN = "your_wechat_token"
WEWORK_ENCODING_AES_KEY = "your_aes_key"
WEWORK_CORP_ID = "your_corp_id"
WEWORK_AGENT_ID = 1000001
WEWORK_SECRET = "your_secret"

# ChatOps 引擎
from chatops_engine import create_engine
engine = create_engine()

@app.route("/wechat", methods=["GET", "POST"])
def wechat_callback():
    """企业微信回调"""
    if request.method == "GET":
        # URL 验证
        msg_signature = request.args.get("msg_signature")
        timestamp = request.args.get("timestamp")
        nonce = request.args.get("nonce")
        echostr = request.args.get("echostr")
        return echostr

    # 处理消息
    xml_data = request.data
    root = ET.fromstring(xml_data)
    content = root.find("Content").text
    from_user = root.find("FromUserName").text

    # 执行命令
    response = engine.execute(from_user, content)

    # 发送回复
    send_wechat_message(from_user, response)
    return "ok"

def send_wechat_message(user_id: str, content: str):
    """发送企业微信消息"""
    url = f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={get_token()}"
    data = {
        "touser": user_id,
        "msgtype": "markdown",
        "agentid": WEWORK_AGENT_ID,
        "markdown": {"content": content},
    }
    requests.post(url, json=data)

def get_token() -> str:
    """获取 access_token"""
    url = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={WEWORK_CORP_ID}&corpsecret={WEWORK_SECRET}"
    resp = requests.get(url)
    return resp.json().get("access_token", "")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

3.2 Slack Bot 集成

#!/usr/bin/env python3
# slack_chatops.py — Slack ChatOps Bot

from slack_bolt import App
from slack_bolt.adapter.flask import SlackRequestHandler
from chatops_engine import create_engine

app = App(token="xoxb-your-token")
engine = create_engine()

@app.command("/network")
def handle_network_command(ack, command, respond):
    """处理 /network 命令"""
    ack()
    user_id = command["user_id"]
    text = command["text"]
    result = engine.execute(user_id, f"/{text}")
    respond(result)

@app.command("/ping")
def handle_ping(ack, command, respond):
    """快速 Ping"""
    ack()
    target = command["text"]
    result = engine.execute(command["user_id"], f"/ping {target}")
    respond(result)

@app.event("app_mention")
def handle_mention(event, say):
    """处理 @bot 消息"""
    text = event["text"]
    user = event["user"]
    # 去除 @bot 前缀
    text = text.split(">", 1)[-1].strip()
    result = engine.execute(user, text)
    say(result)

from flask import Flask
flask_app = Flask(__name__)
handler = SlackRequestHandler(app)

@flask_app.route("/slack/events", methods=["POST"])
def slack_events():
    return handler.handle(request)

if __name__ == "__main__":
    flask_app.run(port=5000)

四、WebSocket 实时推送

#!/usr/bin/env python3
# chatops_server.py — ChatOps 服务端

import asyncio
import websockets
import json
from chatops_engine import create_engine

engine = create_engine()

async def handle_websocket(websocket, path):
    """处理 WebSocket 连接"""
    async for message in websocket:
        data = json.loads(message)
        user_id = data.get("user_id", "anonymous")
        text = data.get("text", "")

        # 执行命令
        result = engine.execute(user_id, text)

        # 发送结果
        await websocket.send(json.dumps({
            "type": "result",
            "data": result,
        }))

async def main():
    async with websockets.serve(handle_websocket, "0.0.0.0", 8765):
        print("ChatOps WebSocket 服务启动: ws://0.0.0.0:8765")
        await asyncio.Future()

if __name__ == "__main__":
    asyncio.run(main())

五、审计与安全

5.1 操作审计

#!/usr/bin/env python3
# chatops_audit.py — ChatOps 审计日志

import json
from datetime import datetime

class AuditLogger:
    """ChatOps 审计日志"""

    def __init__(self, log_file: str = "chatops_audit.log"):
        self.log_file = log_file

    def log(self, user: str, command: str, args: list,
            result: str, status: str = "success"):
        """记录操作日志"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "user": user,
            "command": command,
            "args": args,
            "result_preview": result[:200] if result else "",
            "status": status,
        }

        with open(self.log_file, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")

    def query(self, user: str = None, command: str = None,
              limit: int = 50) -> list:
        """查询审计日志"""
        results = []
        with open(self.log_file, "r", encoding="utf-8") as f:
            for line in f:
                entry = json.loads(line)
                if user and entry["user"] != user:
                    continue
                if command and entry["command"] != command:
                    continue
                results.append(entry)
                if len(results) >= limit:
                    break
        return results

audit = AuditLogger()
# 使用：audit.log("zhangsan", "ping", ["8.8.8.8"], "Reply from 8.8.8.8")

六、最佳实践

ChatOps 实施建议：

  1. 命令设计
  ┌─ 简单直观（/ping、/show、/backup）
  ├─ 有自动补全提示
  ├─ 结果格式化（markdown/代码块）
  └─ 长操作异步返回

  2. 权限控制
  ┌─ 查询类命令开放给所有人
  ├─ 配置变更限制管理员
  ├─ 敏感命令二次确认
  └─ 操作前自动备份

  3. 安全
  ┌─ 操作日志完整记录
  ├─ 命令注入防护
  ├─ 敏感信息脱敏
  └─ 频率限制防滥用

  4. 用户体验
  ┌─ 命令帮助可发现
  ├─ 结果可视化（表格/图表）
  ├─ 错误提示清晰
  └─ 与现有工具集成

下篇预告：第320篇 — LLM 辅助网络排障，将介绍如何利用大语言模型辅助网络故障排查和分析。