最近做的企微二开运维要求能在企微里远程控制服务器——发条消息重启 nginx机器人执行命令并回结果。和之前聊的自然语言控制和快捷指令不同那些是控制业务系统这篇是远程控制程序执行——消息变指令、指令变命令、命令执行、结果回传。把完整链路怎么搭记下来。Eyun 平台开放的企微 API统一 POSTJSON鉴权用 App Token 加 appidAuthorization: Bearer eyk_xxxx路径{BASE_URL}/wx-api/api/模块/动作响应封套{code, data, detail, message, time}code 为 0 成功。指令接收Webhook 收消息运维在企微发重启 nginxWebhook 回调到后端from flask import Flask, request app Flask(__name__) app.route(/wx-api/webhook/, methods[POST]) def webhook(): event request.headers.get(X-Eyun-Event) if event ! message: return ok payload request.json[data] appid payload[appid] from_uin payload[fromUin] content payload[content] # 权限校验只有运维组能发控制指令 if not is_admin(from_uin): return ok # 解析指令 cmd parse_command(content) if cmd: execute_and_reply(appid, from_uin, cmd) return ok权限校验是第一道——只有授权人员能发控制指令普通员工发的消息不处理。指令解析消息变命令重启 nginx 要解析成具体命令。用指令映射表不直接执行用户输入COMMAND_MAP { 重启nginx: {cmd: systemctl restart nginx, risk: medium}, 重启redis: {cmd: systemctl restart redis, risk: medium}, 查看日志: {cmd: tail -100 /var/log/app.log, risk: low}, 清理缓存: {cmd: redis-cli flushall, risk: high}, 查看状态: {cmd: systemctl status nginx, risk: low} } def parse_command(content): content content.strip() for key, cmd_def in COMMAND_MAP.items(): if key in content: return cmd_def return None指令映射表是核心安全机制——不直接执行用户输入的字符串只执行预定义命令。用户说rm -rf /不会被执行不在映射表里就忽略。命令执行安全沙箱解析出命令后执行。高风险命令要二次确认import subprocess def execute_command(cmd_def, from_uin): if cmd_def[risk] high: # 高风险要二次确认 return {need_confirm: True, cmd: cmd_def[cmd]} try: result subprocess.run( cmd_def[cmd], shellTrue, capture_outputTrue, textTrue, timeout30 ) return { stdout: result.stdout[-2000:], # 截断防止太长 stderr: result.stderr[-500:], returncode: result.returncode } except subprocess.TimeoutExpired: return {error: 命令执行超时}timeout30防止命令卡死。输出截断防止消息太长发不出去。高风险命令如清理缓存返回需要确认不直接执行。关于消息接收和回复接口在Eyun 企业微信 API 平台有说明。结果回传调消息接口执行完调企微接口把结果发回去import requests BASE https://api.eyun.com HEADERS {Authorization: Bearer eyk_xxxx, Content-Type: application/json} def reply_result(appid, to_uin, result): if result.get(need_confirm): text f高风险操作{result[cmd]}\n回复确认执行其他忽略 elif result.get(error): text f执行失败{result[error]} else: stdout result.get(stdout, ).strip() text f执行完成返回码 {result[returncode]}\n输出\n{stdout} requests.post( f{BASE}/wx-api/api/message/sendText, headersHEADERS, json{appid: appid, to: to_uin, content: text} )结果格式化成文本发回去。输出太长截断关键字段返回码、错误保留。高风险操作发确认提示运维回复确认才执行。二次确认机制高风险命令的二次确认流程pending_confirms {} # from_uin - {cmd, timestamp} def handle_confirm(appid, from_uin, content): if content.strip() 确认 and from_uin in pending_confirms: pending pending_confirms[from_uin] if time.time() - pending[timestamp] 60: # 60秒内有效 result execute_command({cmd: pending[cmd], risk: low}, from_uin) reply_result(appid, from_uin, result) del pending_confirms[from_uin] def execute_and_reply(appid, from_uin, cmd_def): if cmd_def[risk] high: pending_confirms[from_uin] { cmd: cmd_def[cmd], timestamp: time.time() } reply_result(appid, from_uin, {need_confirm: True, cmd: cmd_def[cmd]}) else: result execute_command(cmd_def, from_uin) reply_result(appid, from_uin, result)确认有效期 60 秒超时要重新发指令。不做二次确认运维手滑发个清理缓存数据全没了。安全边界远程控制机器人安全是第一位的几条硬规矩指令映射表只执行预定义命令不执行任意输入权限校验只有授权人员能发指令风险分级高风险要二次确认执行超时防止命令卡死审计日志谁、什么时候、执行了什么、结果audit_log 表: from_uin command result timestamp ip审计日志事后可追溯。不做审计出了事故不知道谁干的。写在最后远程控制机器人这套东西本质是把企微消息接口和命令执行串起来——Webhook 收指令、映射表解析、subprocess 执行、sendText 回结果。安全靠指令映射表不执行任意输入、权限校验、风险分级、二次确认、审计日志。把安全做严运维在企微里发条消息就能管服务器——但每一步都可控可审计。