跳到主要内容

ChatOps

什么是 ChatOps​

ChatOps 是将运维操作集成到聊天工具中的实践：通过在钉钉/Slack/飞书等群聊中发送命令，触发自动化运维操作。

核心价值​

传统运维ChatOps
SSH 到服务器操作群聊发命令
操作不可见全员可见、可追溯
依赖个人经验标准化命令
缺少审批流程群内审批后执行

钉钉 Bot 示例​

dingtalk_bot.py
from flask import Flask, request, jsonify
import subprocess
import hmac, hashlib, base64, time

app = Flask(__name__)
SECRET = "your-dingtalk-secret"

# 安全：只允许白名单命令
ALLOWED_COMMANDS = {
    "/status": "systemctl status nginx",
    "/disk": "df -h | head -10",
    "/deploy": None,  # 需要审批
}

@app.route("/webhook", methods=["POST"])
def webhook():
    data = request.json
    text = data.get("text", {}).get("content", "").strip()

    if text not in ALLOWED_COMMANDS:
        return jsonify({"msgtype": "text", "text": {"content": "未知命令"}})

    cmd = ALLOWED_COMMANDS[text]
    if cmd is None:
        return jsonify({"msgtype": "text",
            "text": {"content": f"⚠️ {text} 需要审批，请回复 /approve"}})

    # 执行命令并返回结果
    result = subprocess.run(cmd, shell=True, capture_output=True,
                           text=True, timeout=30)
    output = result.stdout[:2000] or result.stderr[:2000]

    return jsonify({"msgtype": "text", "text": {"content": output}})
危险

ChatOps Bot 必须严格限制可执行的命令（白名单），禁止任意命令注入。危险操作（部署、重启）应设置审批流程。

告警通知集成​

alert_notify.py
import requests
import json

def send_dingtalk(webhook_url, title, content):
    """发送钉钉 Markdown 消息"""
    data = {
        "msgtype": "markdown",
        "markdown": {
            "title": title,
            "text": content
        }
    }
    requests.post(webhook_url, json=data, timeout=10)

# 调用示例
send_dingtalk(
    "https://oapi.dingtalk.com/robot/send?access_token=xxx",
    "🔴 服务告警",
    "### CPU 使用率过高\n- **主机**: web-01\n- **CPU**: 95%\n- **时间**: 2024-01-15 10:30"
)

常见面试问题​

Q1: ChatOps 的安全风险有哪些？如何防范？​

答案：

  1. 命令注入：白名单命令 + 参数校验，禁止拼接用户输入到 shell 命令
  2. 权限控制：按用户/群组限制可执行命令，危险操作需双人审批
  3. 审计追踪：所有操作日志化，谁在什么时间执行了什么命令
  4. Webhook 安全：验证请求签名（钉钉/Slack 都支持），防止伪造请求
  5. 超时控制：命令执行设超时，防止阻塞

相关链接​