homelab

/

stacks

/

wechat-bot

/

or Cancel

Edit File Preview Preview Changes

Spaces Tabs

2 4 8

Wrap No wrap

#!/usr/bin/env python3 # /// script # dependencies = [ # "cryptography", # "requests", # ] # /// """ Mac mini 家庭 All-in-One 微信双向控制与通知网关 (WeCom <-> Personal WeChat) 监听端口: 3005 (由 Azure https://wx.961504.xyz:443 通过 Tailscale 专线转发) """ import base64 import hashlib import json import os import socket import struct import subprocess import threading import time import urllib.parse import xml.etree.ElementTree as ET from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import requests from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes BASE_DIR = Path(__file__).resolve().parent CONFIG_PATH = BASE_DIR / "config.env" def load_config() -> dict: cfg = { "WECOM_CORP_ID": "", "WECOM_AGENT_ID": "", "WECOM_SECRET": "", "WECOM_TOKEN": "", "WECOM_AES_KEY": "", "WECOM_TO_USER": "@all", } if CONFIG_PATH.exists(): for line in CONFIG_PATH.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue k, v = line.split("=", 1) cfg[k.strip()] = v.strip().strip('"').strip("'") return cfg class WXBizMsgCrypt: """企业微信官方消息加解密算法实现 (AES-256-CBC + PKCS#7 32字节填充)""" def __init__(self, token: str, encoding_aes_key: str, receive_id: str): self.token = token self.receive_id = receive_id self.key = base64.b64decode(encoding_aes_key + "=") def verify_signature(self, msg_signature: str, timestamp: str, nonce: str, encrypt_msg: str) -> bool: items = sorted([self.token, timestamp, nonce, encrypt_msg]) sha1 = hashlib.sha1("".join(items).encode("utf-8")).hexdigest() return sha1 == msg_signature def decrypt(self, encrypt_msg: str) -> str: cipher_bytes = base64.b64decode(encrypt_msg) cipher = Cipher(algorithms.AES(self.key), modes.CBC(self.key[:16]), backend=default_backend()) decryptor = cipher.decryptor() plain = decryptor.update(cipher_bytes) + decryptor.finalize() pad_len = plain[-1] content = plain[:-pad_len] msg_len = struct.unpack("!I", content[16:20])[0] raw_msg = content[20 : 20 + msg_len].decode("utf-8") return raw_msg TOKEN_CACHE = {"token": "", "expires_at": 0.0} def send_wecom_text(content: str, to_user: str = None) -> dict: cfg = load_config() corp_id = cfg["WECOM_CORP_ID"] secret = cfg["WECOM_SECRET"] agent_id = cfg["WECOM_AGENT_ID"] if not (corp_id and secret and agent_id): return {"errcode": -1, "errmsg": "WeCom credentials not configured yet"} now = time.time() if not TOKEN_CACHE["token"] or now >= TOKEN_CACHE["expires_at"]: resp = requests.get( "https://qyapi.weixin.qq.com/cgi-bin/gettoken", params={"corpid": corp_id, "corpsecret": secret}, timeout=10, ).json() if resp.get("errcode") != 0: return resp TOKEN_CACHE["token"] = resp["access_token"] TOKEN_CACHE["expires_at"] = now + resp.get("expires_in", 7200) - 200 # 企业微信单条文本上限 2048 字节，超长自动截断 if len(content.encode("utf-8")) > 1900: content = content[:600] + "\n...(内容已截断)" payload = { "touser": to_user or cfg.get("WECOM_TO_USER") or "@all", "msgtype": "text", "agentid": int(agent_id), "text": {"content": content}, } return requests.post( f"https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token={TOKEN_CACHE['token']}", json=payload, timeout=10, ).json() def handle_user_command(user_id: str, text: str): text = text.strip() try: if text in ("状态", "/status", "status", "体检"): cmd = ( "export PATH=/opt/homebrew/bin:$PATH; " "echo '🖥️ Mac mini (M6) 状态报告'; " "echo \"⏱️ 运行时间: $(uptime | sed 's/.*up //; s/, [0-9]* users.*//')\"; " "echo \"💾 磁盘剩余: $(df -h / | awk 'NR==2{print $4 \" (已用 \" $5 \")\"}')\"; " "echo \"🌐 家里公网: $(cat ~/homelab/backups/last_wan_ip.txt 2>/dev/null || echo 未知)\"; " "echo \"🐳 运行容器: $(docker ps --format '{{.Names}}' | tr '\\n' ' ')\"" ) out = subprocess.check_output(cmd, shell=True, text=True, timeout=15) send_wecom_text(out.strip(), to_user=user_id) elif text.startswith("/sh "): shell_cmd = text[4:].strip() out = subprocess.check_output( f"export PATH=/opt/homebrew/bin:$HOME/.local/bin:$PATH; {shell_cmd}", shell=True, stderr=subprocess.STDOUT, text=True, timeout=60, ) send_wecom_text(f"✅ 执行完成:\n{out.strip() or '(无输出)'}", to_user=user_id) else: send_wecom_text( f"🤖 Mac mini 已收到指令：\n「{text}」\n\n" "可用快捷指令：\n" "• 发送「状态」：查看 Mac mini 实时健康度与公网 IP\n" "• 发送「/sh <命令>」：远程在 Mac mini 上执行终端命令", to_user=user_id, ) except Exception as e: send_wecom_text(f"⚠️ 执行异常: {e}", to_user=user_id) class GatewayHandler(BaseHTTPRequestHandler): def do_GET(self): parsed = urllib.parse.urlparse(self.path) if parsed.path == "/wechat/callback": qs = urllib.parse.parse_qs(parsed.query) msg_sig = qs.get("msg_signature", [""])[0] timestamp = qs.get("timestamp", [""])[0] nonce = qs.get("nonce", [""])[0] echostr = qs.get("echostr", [""])[0] cfg = load_config() if not (cfg["WECOM_TOKEN"] and cfg["WECOM_AES_KEY"] and cfg["WECOM_CORP_ID"]): self.send_response(400) self.end_headers() self.wfile.write(b"WeCom callback parameters not configured yet") return crypt = WXBizMsgCrypt(cfg["WECOM_TOKEN"], cfg["WECOM_AES_KEY"], cfg["WECOM_CORP_ID"]) if not crypt.verify_signature(msg_sig, timestamp, nonce, echostr): self.send_response(403) self.end_headers() self.wfile.write(b"Signature verification failed") return plain_echo = crypt.decrypt(echostr) self.send_response(200) self.send_header("Content-Type", "text/plain; charset=utf-8") self.end_headers() self.wfile.write(plain_echo.encode("utf-8")) return cfg = load_config() configured = bool(cfg["WECOM_CORP_ID"] and cfg["WECOM_SECRET"]) self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write( json.dumps( { "service": "macmini-wechat-gateway", "status": "ok", "wecom_configured": configured, "callback_url": "https://wx.961504.xyz/wechat/callback", "notify_webhook": "https://wx.961504.xyz/notify", }, ensure_ascii=False, ).encode("utf-8") ) def do_POST(self): parsed = urllib.parse.urlparse(self.path) length = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(length).decode("utf-8") if parsed.path == "/wechat/callback": qs = urllib.parse.parse_qs(parsed.query) msg_sig = qs.get("msg_signature", [""])[0] timestamp = qs.get("timestamp", [""])[0] nonce = qs.get("nonce", [""])[0] cfg = load_config() crypt = WXBizMsgCrypt(cfg["WECOM_TOKEN"], cfg["WECOM_AES_KEY"], cfg["WECOM_CORP_ID"]) root = ET.fromstring(body) encrypt_node = root.find("Encrypt") if encrypt_node is not None and crypt.verify_signature(msg_sig, timestamp, nonce, encrypt_node.text): xml_plain = crypt.decrypt(encrypt_node.text) msg_root = ET.fromstring(xml_plain) msg_type = msg_root.findtext("MsgType", "") from_user = msg_root.findtext("FromUserName", "@all") if msg_type == "text": content = msg_root.findtext("Content", "") threading.Thread(target=handle_user_command, args=(from_user, content), daemon=True).start() self.send_response(200) self.end_headers() self.wfile.write(b"success") return if parsed.path == "/notify": try: data = json.loads(body) if body else {} msg = data.get("msg") or data.get("text") or body except Exception: msg = body res = send_wecom_text(f"🔔 【家庭服务器通知】\n{msg}") self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps(res, ensure_ascii=False).encode("utf-8")) return self.send_response(404) self.end_headers() if __name__ == "__main__": server = ThreadingHTTPServer(("0.0.0.0", 3005), GatewayHandler) print("WeChat Gateway listening on 0.0.0.0:3005...") server.serve_forever()

Commit Changes

Add a Signed-off-by trailer by the committer at the end of the commit log message.

Commit directly to the main branch.

Create a new branch for this commit and start a pull request.

Commit Changes Cancel