在团队协作里，把聊天群变成操作入口已经不算新鲜事。可一旦操作对象是生产集群，单纯一个按钮就能删除命名空间或踢出节点，风险不言而喻。于是审批流成了 ChatOps 机器人落地时绕不开的一环。本文从实际工程角度拆解集成方案，先明确机器人和审批系统各自的职责，再给出状态机、权限隔离和审计的实现要点。
先别急着写代码，很多人把 ChatOps 机器人当成一个能直接操作集群的全能工具，这本身就埋下了安全炸弹。正确的边界应该是：机器人只负责两件事——接收用户意图和推动审批流程，真正的集群操作交给专门的服务去执行。这样即使机器人被恶意调用，它也没有任何集群权限，最多只能发起一堆待审批的请求。
一个典型的请求链路是这样的：用户在群里 @机器人 并输入类似 deploy production --version v1.2.3 的命令；机器人解析命令后创建一个审批单，把消息推送给对应的审批人；审批人在群里点击按钮或回复关键词；机器人收到审批结果后，通过内部 API 调用执行服务；执行服务此时才使用受控凭证去操作集群。整个过程中，机器人完全不接触 kubeconfig 或云厂商的 AccessKey。
机器人接入与命令解析
首要是让机器人能稳定接入你们的 IM 平台。以常见的 Slack、飞书或钉钉为例，它们都提供事件订阅机制。你可以用 Flask 起一个 HTTP 服务，接收平台推送的 mention 事件。下面这段代码展示了一个最小可用的命令入口，它不依赖特定 SDK，易于迁移到任意平台。
import re
from flask import Flask, request
app = Flask(__name__)
@app.route("/chatops", methods=["POST"])
def chatops_webhook():
payload = request.get_json()
text = payload.get("text", "")
user = payload.get("user_name", "unknown")
# 去掉 @机器人 前缀，取出真正的命令
cmd_match = re.search(r"@\w+\s+(.+)", text)
if not cmd_match:
return {"text": "用法：@bot <命令>"}
command_line = cmd_match.group(1)
# 命令格式：action target [参数]
parts = command_line.split()
action = parts[0]
target = parts[1] if len(parts) > 1 else ""
params = parts[2:]
# 这里只做分发，不执行任何集群操作
dispatch_command(action, target, params, user)
return {"text": f"已收到 {action} 请求，等待审批。"}
命令解析的关键不是写正则有多炫，而是要把意图标准化。建议定义一个命令白名单，比如 deploy、restart、scale、rollback 等。不要直接用 eval 或反射机制去执行任意方法，那等于给攻击者开了一扇后门。每个 action 对应一个处理函数，函数内部只负责创建审批单，不触碰集群 API。
另外，命令分发需要做参数校验。比如 deploy 必须带上环境名和版本号，scale 必须带上副本数且限制范围。这些校验越早做，审批单的质量越高，审批人也越容易判断风险。你可以在 dispatch_command 里集中定义规则字典，保持代码可读性。
审批流的状态机设计
审批流不是简单的“同意/拒绝”二元开关，它需要明确的状态迁移路径。一个实用的状态集合至少包含：pending（待审批）、approved（已通过）、rejected（已拒绝）、expired（已过期）、executing（执行中）、failed（执行失败）和 done（已完成）。状态机负责保证非法迁移不会发生，比如已拒绝的单子不能直接变成执行中。
下面这段代码用一个字典来定义合法的迁移关系，简单直观。持久化可以选用 PostgreSQL 或 MySQL，审批单表至少包含 id、action、target、params、requester、approver、status、created_at、updated_at 等字段。状态字段建议用枚举而不是字符串，避免拼写错误。
ALLOWED_TRANSITIONS = {
"pending": {"approved", "rejected", "expired"},
"approved": {"executing"},
"executing": {"done", "failed"},
"failed": {"executing"}, # 允许重试
}
def transition(current_state: str, new_state: str) -> bool:
return new_state in ALLOWED_TRANSITIONS.get(current_state, set())
# 使用示例
if not transition("pending", "approved"):
raise ValueError("非法状态迁移")
很多时候审批流需要多人会签或一人通过即可。这两种模式的差异在于：会签要求所有审批人都同意，而或签只需要任意一人同意。你可以在审批单上记录 approval_type 字段，并在状态迁移时检查已收集到的审批结果。例如会签模式下，只有当所有审批人都提交了 approved 时，才能将单子从 pending 迁到 approved。否则走到 rejected 或继续等待。
还有一个容易被忽略的点：超时自动过期。如果审批人一直不处理，单子会一直占用资源。你可以在创建审批单时设置 expires_at 字段，用一个定时任务每分钟扫描一次，把超过有效期的 pending 单子批量更新为 expired。注意不要用机器人本身的定时器，因为机器人可能重启或横向扩容，容易重复处理。独立的调度服务更可靠。
集群操作与权限隔离
真正执行集群操作的服务必须是一个独立进程，它通过内部 API 或消息队列接收已批准的审批单。这个服务运行在受限网络环境中，只有它能访问集群的认证信息。机器人只通过 HTTP 回调通知执行服务，传递审批单 ID 和操作指令，但绝不传递任何集群凭证。
执行服务在每次操作前需要重新校验审批单状态，防止出现“审批通过后被篡改”的情况。你可以使用数据库行级锁或版本号来避免并发问题。下面是一个典型的执行服务逻辑骨架，它从队列中取出任务，检查状态是否为 approved，然后调用集群客户端执行操作，最后更新状态。
def execute_task(task_id: str):
task = get_task_by_id(task_id)
if task.status != "approved":
return # 状态已变，放弃执行
# 使用短期凭证，而不是长期 token
client = create_kubernetes_client(
cluster=task.target,
token=get_short_lived_token(task.target)
)
if task.action == "deploy":
client.apps_v1.patch_namespaced_deployment(
name=task.params.get("name"),
namespace=task.params.get("namespace"),
body=task.params.get("patch")
)
mark_task_status(task_id, "done")
权限隔离的另一个关键是短期凭证。如果你的集群是 Kubernetes，建议使用 ServiceAccount 的 TokenRequest API 为每次执行生成一个有效期很短的 token，比如 15 分钟。对于云厂商 API，则可以使用临时安全凭证（STS）。执行服务在启动时不需要任何静态密钥，而是从密钥管理服务中动态获取。这样即使某次令牌泄露，影响范围也极其有限。
审批通过后，执行动作的审计也不能丢。每一条执行日志都应该包含：审批单 ID、操作人、查询参数、集群响应摘要、执行开始和结束时间。这些日志最好写入只追加的存储，比如 S3 或 Elasticsearch，确保事后可追踪。审计日志本身也要设置访问权限，不能让普通开发者随意删除。
常见故障与避坑指南
第一个坑是审批人离职或转岗。如果审批单的审批人字段写死了用户 ID，一旦这个人离开，审批单就会永远卡在 pending。解决办法是在组织架构服务里维护“角色到人员”的映射，审批单只记录角色，不记录具体人。当有人离职时，管理员只需更新角色对应的人，机器人推送消息时会动态解析角色下的当前人员。必要时设置一个“升级审批”机制：如果一级审批人 30 分钟未处理，自动转给其上级。
第二个坑是机器人回调超时和重试风暴。IM 平台可能因为网络抖动多发几次按钮点击事件，如果机器人不做幂等处理，同一个审批动作可能被执行多次。尤其对于“拒绝”操作，重复拒绝问题不大，但对于“同意”并触发执行，就必须保证执行服务具备幂等性。你可以在执行服务里用审批单 ID 加操作类型做唯一键，数据库插入前检查是否已存在执行记录。
第三个坑是命令的权限模型太粗糙。有些人给所有用户都开放了 deploy 命令，审批流程形同虚设。应该根据用户角色和集群环境进行细粒度控制：开发环境可以允许开发者直接触发，测试环境需要 leader 审批，生产环境则必须多人会签。机器人解析命令时先查询权限表，发现无权操作直接返回提示，根本不会创建审批单。
最后，别忘了给机器人加超时会话管理。有些用户会连续发送多条命令，机器人需要能区分这是同一条命令的重试还是新的操作。你可以在解析时引入短暂的会话 ID，比如把用户 ID 和最近一次命令时间绑定，10 秒内重复命令视为重试。否则集群操作可能因为重复触发而产生意外。
审批流和 ChatOps 机器人的结合，本质上是把“人的判断”嵌入到自动化流程里。它不会降低执行速度，反而让高风险操作有了可追溯的停靠点。只要把状态机、权限隔离和审计这三块做扎实，集成并不复杂。照着本文的思路，你可以从一个小型 Flask 服务开始，先跑通一条含审批的部署流程，再逐步扩展命令库和审批策略。