### 问题解构与接入方案 基于 OpenClaw 框架使用 Python 接入钉钉，核心在于正确配置回调地址并实现符合钉钉协议的加解密逻辑。钉钉开放平台要求回调服务必须支持 HTTPS、能够处理 POST 请求，并使用 AES 加密算法对消息进行加解密 [ref_1]。以下将分步骤详解从环境准备到代码实现的完整流程。 ### 一、钉钉应用配置与回调地址设置 在钉钉开发者后台创建自建应用后，需配置回调 URL 以接收事件推送。回调地址必须是公网可访问的 HTTPS 地址，若本地开发需配合内网穿透工具 [ref_2]。 | 配置项 | 要求说明 | 示例值 | | :--- | :--- | :--- | | **回调协议** | 必须使用 HTTPS，HTTP 会被拒绝 [ref_4] | `https://superhermes.xyz/dingtalk/webhook` | | **请求方法** | 钉钉服务器强制使用 POST 发送数据 [ref_1] | `POST` | | **加密密钥** | 需在钉钉后台设置 AES 密钥，长度 32 位 [ref_5] | `dingtalk_aes_key_32chars` | | **Token** | 用于签名验证的随机字符串 [ref_2] | `random_token_string` | 配置完成后，钉钉会向该地址发送验证请求，服务端需正确解密并返回明文字符串才能通过验证 [ref_5]。 ### 二、Python 加解密核心实现代码 以下代码基于 `dingtalk-stream` SDK 及原生 `crypto` 库实现，涵盖签名验证、消息解密与响应加密全流程 [ref_6]。 ```python # -*- coding: utf-8 -*- """ 钉钉回调加解密模块 - 基于 OpenClaw 框架适配 [ref_1] """ import base64 import hashlib import json from Crypto.Cipher import AES from flask import Flask, request, jsonify app = Flask(__name__) # 从钉钉后台获取的配置参数 [ref_2] DINGTALK_TOKEN = "your_token_from_dingtalk" DINGTALK_AES_KEY = "your_32_char_aes_key" # 必须 32 字符 class DingTalkEncryptor: """钉钉消息加解密工具类 [ref_5]""" def __init__(self, token, aes_key): self.token = token self.aes_key = aes_key.encode('utf-8') def check_signature(self, signature, timestamp, nonce): """验证请求签名，确保来自钉钉服务器 [ref_4]""" # 按字典序排序参数后拼接 sorted_params = sorted([self.token, timestamp, nonce]) # 使用 SHA256 计算签名 sha256 = hashlib.sha256() sha256.update(''.join(sorted_params).encode('utf-8')) # 对比签名是否一致 return signature == sha256.hexdigest() def decrypt(self, encrypt_data): """解密钉钉推送的加密数据 [ref_1]""" # 解码 Base64 密文 decode_data = base64.b64decode(encrypt_data) # 创建 AES 解密器，模式为 CBC cipher = AES.new(self.aes_key, AES.MODE_CBC, self.aes_key[:16]) # 解密并去除 PKCS7 填充 decrypted = cipher.decrypt(decode_data) # 移除填充字节 pad_len = decrypted[-1] return decrypted[:-pad_len].decode('utf-8') def encrypt(self, plain_text): """加密响应数据返回给钉钉 [ref_5]""" # 添加 PKCS7 填充 pad_len = 16 - len(plain_text) % 16 padded_text = plain_text + chr(pad_len) * pad_len # 创建 AES 加密器 cipher = AES.new(self.aes_key, AES.MODE_CBC, self.aes_key[:16]) # 加密后 Base64 编码 encrypted = cipher.encrypt(padded_text.encode('utf-8')) return base64.b64encode(encrypted).decode('utf-8') # 初始化工具类实例 encryptor = DingTalkEncryptor(DINGTALK_TOKEN, DINGTALK_AES_KEY) @app.route('/dingtalk/webhook', methods=['POST']) def receive_callback(): """ 接收钉钉回调请求入口 必须限定为 POST 方法，否则钉钉会返回 405 错误 [ref_6] """ try: # 获取 URL 参数 signature = request.args.get('signature') timestamp = request.args.get('timestamp') nonce = request.args.get('nonce') # 获取请求体中的加密数据 encrypt_data = request.json.get('encrypt') if request.json else None # 验证签名，防止伪造请求 [ref_4] if not encryptor.check_signature(signature, timestamp, nonce): return jsonify({"error": "签名验证失败"}), 403 # 解密业务数据 decrypt_data = encryptor.decrypt(encrypt_data) business_data = json.loads(decrypt_data) # 处理业务逻辑（如解析事件类型、调用 OpenClaw 技能）[ref_2] process_event(business_data) # 返回加密的成功标识，钉钉要求必须返回加密的"success"[ref_1] return encryptor.encrypt("success") except Exception as e: # 捕获异常避免 Connection reset，记录日志后返回错误 [ref_6] print(f"回调处理异常：{str(e)}") return "error", 500 def process_event(data): """业务事件处理函数，可对接 OpenClaw 技能引擎 [ref_2]""" event_type = data.get('eventType') print(f"收到事件类型：{event_type}") # 此处可调用 OpenClaw 的 Skill 执行逻辑 if __name__ == '__main__': # 生产环境需使用 Gunicorn+Nginx 部署并配置 HTTPS[ref_3] app.run(host='0.0.0.0', port=5000, ssl_context='adhoc') ``` ### 三、部署与验证流程 完成代码开发后，需按以下步骤部署并验证回调功能： | 步骤 | 操作内容 | 验证标准 | | :--- | :--- | :--- | | **1. 内网穿透** | 使用 ngrok 或 frp 将本地服务暴露为 HTTPS 公网地址 [ref_2] | 浏览器可访问 `https://xxx.ngrok.io/dingtalk/webhook` | | **2. 后台配置** | 在钉钉开发者后台填入回调 URL、Token 和 AES 密钥 [ref_5] | 保存时显示"验证成功" | | **3. 日志监控** | 服务端打印接收到的事件数据和解密结果 [ref_6] | 日志中可见完整 JSON 业务数据 | | **4. 响应测试** | 触发钉钉事件（如审批通过）观察服务端响应 [ref_1] | 钉钉后台显示"推送成功"且无重试 | ### 四、常见故障排查 针对之前遇到的 `Connection reset` 和 `400 Bad Request` 错误，需重点检查以下环节： 1. **HTTPS 证书问题**：自签名证书可能导致钉钉拒绝连接，建议使用 Let's Encrypt 或云服务商提供的正式证书 [ref_4]。 2. **响应格式错误**：必须返回加密字符串而非纯 JSON，否则钉钉会判定为失败并重试 [ref_5]。 3. **超时设置**：服务端处理时间超过 5 秒会导致钉钉断开连接，建议异步处理业务逻辑 [ref_6]。 4. **防火墙限制**：确保服务器安全组放行 443 端口，且未拦截钉钉服务器 IP 段 [ref_2]。 通过上述步骤，可完成 Python 环境下 OpenClaw 与钉钉的完整对接，确保回调地址配置正确且加解密逻辑符合平台规范 [ref_1]。