Skip to content
BranchesTags

Repository files navigation

wecom-sdk · 企业微信消息网关

一个独立、自包含的企业微信消息网关项目，从业务项目中抽取通用能力而来。 企业微信凭据只在本项目集中配置一次，其他项目无需安装 SDK、无需配置任何企业微信信息， 只要调用本网关的 HTTP 接口即可发消息、收回调。

flowchart LR
    subgraph 其他项目
        P1[项目A]
        P2[项目B]
        P3[项目C]
    end
    G[企业微信消息网关<br/>集中配置凭据 + API Key 鉴权]
    W[企业微信 API]
    P1 -->|HTTP + API Key| G
    P2 -->|HTTP + API Key| G
    P3 -->|HTTP + API Key| G
    G -->|发送消息/菜单/用户| W
    W -->|回调消息| G
    G -->|解密后明文| P1
Loading

特性

  • ✅ 集中配置：WECOM_CORP_ID / WECOM_AGENT_ID / WECOM_CORP_SECRET 只在网关 apps.json 配一次
  • ✅ 多组应用：支持配置多组企业微信应用，其他项目携带 group(code/uuid) 选择使用哪组
  • ✅ 零业务耦合：其他项目几行代码即可调用，无需任何企业微信知识
  • ✅ 回调免写加解密：网关代收企业微信回调，验签解密后把明文 JSON 转发给各项目 webhook
  • ✅ 安全：API Key 鉴权 + 每 Key 组权限隔离 + 审计日志（详见安全）
  • ✅ access_token 自动管理：线程安全缓存、失效自动重试、可选跨进程文件缓存

目录

快速开始

cd c:\project\wecom-sdk
pip install -r requirements.txt

# 1. 配置企业微信凭据（只在这里配一次）→ 编辑 apps.json
# 2. 配置调用方 Key 与监听端口 → 编辑 .env
python app.py

启动后控制台会打印可用应用组与所有端点。

也可以直接 pip 安装依赖后运行：python -m pip install -r requirements.txt

多组应用配置

企业微信凭据统一放在 apps.json（模板见 apps.json.example）。每组应用用 code 或 uuid 标识：

{
  "apps": [
    {
      "code": "wecom",
      "uuid": "0f8fad5b-d9cb-469f-a165-70867728950e",
      "name": "账本应用",
      "corp_id": "ww1234567890abcdef",
      "agent_id": "1000002",
      "corp_secret": "your-secret",
      "agent_url": "",
      "default_to_user": "@all",
      "token": "callback-token",
      "encoding_aes_key": "callback-aes-key",
      "webhook_url": "http://my-service:8000/wecom/webhook"
    }
  ]
}
字段 必填 说明
code ✅ 应用组标识，其他项目调用时用它（如 "wecom"）
uuid 可选 机器标识，缺省自动生成；其他项目也可用它路由
corp_id / agent_id / corp_secret ✅ 企业微信凭据
agent_url 可选 代理转发 URL（规避 IP 白名单），配置后优先
default_to_user 可选 默认推送用户，默认 @all
token / encoding_aes_key 仅回调 回调验证与解密
webhook_url 仅回调 回调转发地址（网关代收后转发明文）

也可用环境变量 WECOM_APPS_JSON 提供同样 JSON（Docker 场景）。 apps.json 含真实密钥，已加入 .gitignore，请提交 apps.json.example。

其他项目如何调用

其他项目无需安装本 SDK、无需配置任何企业微信信息，用任意语言发 HTTP 请求即可。

# 使用 code 指定应用组
curl -X POST http://<网关IP>:5002/api/message/text \
     -H "X-API-Key: key-for-project-a" \
     -H "Content-Type: application/json" \
     -d '{"group": "wecom", "content": "你好，世界"}'

Python 示例（可直接拷到其他项目，见 examples/call_gateway.py）：

import requests

GATEWAY_URL = 'http://127.0.0.1:5002'
API_KEY = 'key-for-project-a'   # 你项目分配到的 Key
GROUP = 'wecom'                 # 网关里的 code 或 uuid

resp = requests.post(f'{GATEWAY_URL}/api/message/text',
                     headers={'X-API-Key': API_KEY},
                     json={'group': GROUP, 'content': '你好'}, timeout=10)
print(resp.json())
# {'success': True, 'errcode': 0, 'errmsg': 'ok',
#  'group': {'code': 'wecom', 'uuid': '...', 'name': '账本应用'}, 'result': {...}}

端点一览

端点 说明
POST /api/message/text 发送文本 {content, to_user}
POST /api/message/textcard 发送文本卡片 {title, description, url, btntxt}
POST /api/message/markdown 发送 Markdown {content}
POST /api/message/news 发送图文 {articles}
POST /api/message/send 通用发送 {msgtype, payload}（含自定义类型）
GET/POST /api/menu 查询 / 创建菜单
GET /health 健康检查

所有发送端点都支持在请求体加 group（或 code/uuid）选择应用组，缺省用第一组；推送目标支持 to_user（@all/userid）、to_party、to_tag。

接收回调（无需再写加解密）

其他项目完全不用再实现企业微信那套加解密——网关代收：

  1. 企业微信后台「接收消息」回调地址填： http://<网关地址>:5002/qywx/callback/<code或uuid>
  2. 网关完成 URL 验证、签名校验、消息解密
  3. 解密后的明文 JSON 自动 POST 到该组配置的 webhook_url：
{
  "group": "wecom",
  "uuid": "0f8fad5b-...",
  "message": {
    "MsgType": "text",
    "Content": "买奶茶 15",
    "FromUserName": "zhangsan"
  }
}
  1. 各项目 webhook 只需处理明文 JSON；如需被动回复，返回 {"reply": "..."}，网关自动加密回复。
# 其他项目侧 webhook（Flask 示例）—— 零加解密代码
@app.route('/wecom/webhook', methods=['POST'])
def webhook():
    data = request.get_json()
    msg = data['message']
    if msg['MsgType'] == 'text':
        handle_text(msg['Content'], msg['FromUserName'])
    return {'reply': '收到'}

Docker 部署

仓库内已提供 Dockerfile、docker-compose.sample.yml、entrypoint.sh、.dockerignore， 以及一键构建脚本 build.cmd（Windows）/ build.sh（Linux）。

# 方式一：一键构建脚本（build 并打 latest 标签）
./build.sh         # Linux/macOS
build.cmd          # Windows

# 方式二：直接 docker build
docker build -t wecom-gateway:latest .

# 方式三：compose 本地构建 + 启动（compose 里已带 build: .）
cp docker-compose.sample.yml docker-compose.yml
docker compose up -d --build

密钥不进镜像（.dockerignore 已排除）：

  • apps.json（多组企业微信凭据）用只读卷挂载：/path/to/wecom-sdk/apps.json:/app/apps.json:ro
  • 或用环境变量 WECOM_APPS_JSON 注入（Docker 场景推荐）
  • 网关配置、API Key、传输安全开关全部走 environment:，见 docker-compose.sample.yml

生产建议：镜像只监听内网 5002，TLS 由前面的 nginx/Caddy 反代统一终结（网关侧可开 WECOM_REQUIRE_HTTPS=true 校验 X-Forwarded-Proto）。

安全

其他项目 ↔ 网关之间的数据安全，从以下层面保障：

层面 措施
传输加密 网关保持 HTTP，TLS/证书由项目外反代统一终结（nginx/Caddy）；网关侧用 WECOM_REQUIRE_HTTPS 强制校验 X-Forwarded-Proto 为 https，杜绝明文传输
身份认证 每个项目分配一把 API Key（WECOM_API_KEYS），请求头 X-API-Key 携带，缺失/错误返回 401
数据隔离 WECOM_API_KEY_GROUPS 限定每把 Key 可访问的应用组（如 {"keyA":["wecom"],"keyB":["other"]}），防止项目间互相发送；越权返回 403
最小化网络暴露 默认监听 0.0.0.0，生产建议只绑定内网网卡、防火墙放行、不暴露公网；反代层可加 IP 白名单
凭据保护 企业微信 Secret 只在网关 apps.json/.env，永不下发给其他项目；access_token 不外泄（日志已脱敏记录 Key）
审计 配置 WECOM_LOG_FILE 记录每次发送的 Key、应用组、内容与结果，可追溯"谁在什么时间发了什么"
回调安全 消息签名必须校验通过才处理；webhook 转发只使用预配置的 webhook_url，杜绝 SSRF
密钥管理 apps.json、.env 均已 gitignore，切勿提交；Key 与 Secret 建议通过环境变量/密钥管理服务注入

推荐的最小生产部署：

公网 → [nginx(HTTPS, 可选 IP/域名白名单)] → 网关(:5002, 内网) → 企业微信 API

传输安全：TLS 外置 + 策略校验

TLS/证书统一由项目外反向代理（nginx/Caddy）处理，网关本身只跑 HTTP 与策略校验，不持有证书。针对"其他项目 ↔ 网关"的数据传输，可在网关侧逐层开启：

层级 配置 说明
① 强制 HTTPS（开关） WECOM_REQUIRE_HTTPS=true 开关：true=强制校验 X-Forwarded-Proto 必须 https（反代场景），否则 403——防明文传输、防降级；初始值读环境变量，运行时可 set_require_https(True/False) 热切换，无需重启
② 域名白名单 WECOM_ALLOWED_DOMAINS=wecom.example.com 校验请求 Host 头必须匹配，防 DNS rebinding / Host 头注入 / 借用网关服务
③ 请求签名（开关） WECOM_SIGNATURE_REQUIRED=true 开关：调用方携带 X-Timestamp/X-Nonce/X-Signature，网关校验 HMAC-SHA256 签名与时间窗（WECOM_SIGNATURE_TTL，默认 300 秒），防篡改 + 防重放；初始值读环境变量，运行时可 set_signature_required(True/False) 热切换
④ IP 白名单（开关） WECOM_API_IP_WHITELIST=127.0.0.1,192.168.1.0/24 开关：只允许指定来源 IP / CIDR 调用；初始值读环境变量，运行时可 set_ip_whitelist([...]) 热切换，传空列表即关闭
日志脱敏 WECOM_LOG_MASK_CONTENT=true 审计日志只记录消息长度，不落明文

签名算法：以该项目的 API Key 为密钥，对 timestamp + "\n" + nonce + "\n" + 原始请求体 做 HMAC-SHA256。 调用方示例（可直接照抄 examples/call_gateway.py）：

import hashlib, hmac, secrets, time

def sign(key: str, raw_body: bytes):
    ts = str(int(time.time())); nonce = secrets.token_hex(8)
    msg = ts.encode() + b'\n' + nonce.encode() + b'\n' + raw_body
    return ts, nonce, hmac.new(key.encode(), msg, hashlib.sha256).hexdigest()
curl -X POST http://网关:5002/api/message/text \
     -H "X-API-Key: key-a" -H "Content-Type: application/json" \
     -H "X-Timestamp: <ts>" -H "X-Nonce: <nonce>" -H "X-Signature: <sig>" \
     -d '{"group":"wecom","content":"你好"}'

API 一览

  • 网关 HTTP 端点见上表；企业微信 API 返回结构：errcode==0 表示成功。
  • 底层库 wecom_sdk（网关内部引擎）也提供完整 Python API，见 wecom_sdk/ 源码与 docs 注释。

底层 SDK 库（可选）

本项目的 wecom_sdk/ 是独立、可 pip 安装的库（pyproject.toml），网关内部即用它发消息。 个别项目若想脱离网关直接使用（如自行处理回调加解密），可：

pip install -e c:\project\wecom-sdk     # 或拷贝 wecom_sdk/ 目录
from wecom_sdk import WeComClient
client = WeComClient(corp_id="ww...", agent_id="1000002", corp_secret="...")
client.send_text_message("你好", to_user="@all")

from wecom_sdk.crypto import WXBizMsgCrypt, parse_xml
crypt = WXBizMsgCrypt(token, aes_key, corp_id)   # 回调加解密（不依赖 Web 框架）

依赖仅 requests、python-dotenv、pycryptodome，无 Flask/数据库耦合。

常见问题

Q：为什么不做成"每个项目装个库、自己配 Secret"？ 那样每个项目仍要配企业微信凭据，等于直接调企业微信接口，没有复用价值。本网关把凭据集中配置一次， 其他项目零配置 HTTP 调用即可。

Q：其他项目还要写企业微信加解密吗？ 发送完全不用；接收回调也不用——网关代为验签解密并转发明文 webhook（见接收回调）。 仅当某个项目要脱离网关自己处理回调时，才可用 wecom_sdk.crypto（封装好的，不用自己实现）。

Q：多个项目共用一组应用，会被互相干扰吗？ 可给每组应用单独配置（apps.json 加一组），并用 WECOM_API_KEY_GROUPS 把每个项目的 Key 隔离到自己的组。

Q：如何规避 IP 白名单限制？ 在对应组的 agent_url 配反向代理地址（本机固定 IP 出网）即可。

Q：发送返回 errcode 40014 / 42001？ SDK 已内置 token 失效自动刷新重试；仍出现多为 corp_secret 与 agent 不匹配或代理未转发对应 API。

About

No description, website, or topics provided.

Resources

Readme
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages