#!/home/hz1/miniconda3/envs/openclaw/bin/python3.12 """ Matrix Synapse 批量注册账号脚本 用法: python3 batch_register.py --config config.json --users users.csv """ import argparse import csv import hmac import hashlib import json import requests import sys import time from pathlib import Path def get_nonce(server_url: str, admin_token: str) -> str: """从 Synapse Admin API 获取 nonce""" url = f"{server_url}/_synapse/admin/v1/register" headers = {"Authorization": f"Bearer {admin_token}"} resp = requests.get(url, headers=headers, timeout=10) resp.raise_for_status() return resp.json()["nonce"] def generate_mac(nonce: str, username: str, password: str, admin: bool, secret: str) -> str: """生成 HMAC 签名""" # Synapse 要求的格式: nonce + username + password + admin mac_content = f"{nonce}\n{username}\n{password}\n{admin}" mac = hmac.new( secret.encode("utf-8"), mac_content.encode("utf-8"), hashlib.sha1 ).hexdigest() return mac def register_user( server_url: str, admin_token: str, shared_secret: str, username: str, password: str, admin: bool = False, display_name: str = None, ) -> dict: """注册单个用户""" # 获取 nonce nonce = get_nonce(server_url, admin_token) # 生成 MAC mac = generate_mac(nonce, username, password, admin, shared_secret) # 构建请求 url = f"{server_url}/_synapse/admin/v1/register" headers = { "Authorization": f"Bearer {admin_token}", "Content-Type": "application/json", } data = { "nonce": nonce, "username": username, "password": password, "admin": admin, "mac": mac, } # 发送注册请求 resp = requests.post(url, headers=headers, json=data, timeout=10) result = { "username": username, "success": resp.status_code == 200, "status_code": resp.status_code, } if resp.status_code == 200: result["user_id"] = resp.json().get("user_id") result["access_token"] = resp.json().get("access_token") # 设置显示名称（可选） if display_name: set_display_name(server_url, admin_token, result["user_id"], display_name) else: result["error"] = resp.json().get("error", resp.text) return result def set_display_name(server_url: str, admin_token: str, user_id: str, display_name: str): """设置用户显示名称""" url = f"{server_url}/_synapse/admin/v1/users/{user_id}" headers = { "Authorization": f"Bearer {admin_token}", "Content-Type": "application/json", } data = {"displayname": display_name} try: requests.put(url, headers=headers, json=data, timeout=10) except Exception: pass # 显示名称设置失败不影响注册结果 def batch_register(config: dict, users: list, dry_run: bool = False) -> list: """批量注册用户""" results = [] server_url = config["server_url"] admin_token = config["admin_token"] shared_secret = config["shared_secret"] for user in users: username = user["username"] password = user.get("password", config.get("default_password", "changeme123")) admin = user.get("admin", False) display_name = user.get("display_name") if dry_run: print(f"[DRY-RUN] 将注册: {username} (admin={admin})") results.append({"username": username, "success": None, "dry_run": True}) continue print(f"正在注册: {username}...") try: result = register_user( server_url, admin_token, shared_secret, username, password, admin, display_name ) results.append(result) if result["success"]: print(f" ✅ 成功: {result['user_id']}") else: print(f" ❌ 失败: {result.get('error', '未知错误')}") # 避免请求过快 time.sleep(0.5) except Exception as e: print(f" ❌ 异常: {e}") results.append({ "username": username, "success": False, "error": str(e), }) return results def load_users_from_csv(filepath: str) -> list: """从 CSV 文件加载用户列表""" users = [] with open(filepath, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: user = { "username": row["username"].strip(), "password": row.get("password", "").strip() or None, "display_name": row.get("display_name", "").strip() or None, "admin": row.get("admin", "").strip().lower() in ("true", "1", "yes"), } users.append(user) return users def load_users_from_json(filepath: str) -> list: """从 JSON 文件加载用户列表""" with open(filepath, "r", encoding="utf-8") as f: return json.load(f) def main(): parser = argparse.ArgumentParser( description="Matrix Synapse 批量注册账号", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: # 使用配置文件和 CSV 用户列表 %(prog)s --config config.json --users users.csv # 使用 JSON 用户列表 %(prog)s --config config.json --users users.json # 干运行（只打印不执行） %(prog)s --config config.json --users users.csv --dry-run # 单个用户注册 %(prog)s --config config.json --single --username user1 --password pwd123 CSV 格式示例: username,password,display_name,admin user1,pass1,用户一,false user2,pass2,用户二,false admin1,pass3,管理员,true config.json 格式: { "server_url": "https://matrix.example.com", "admin_token": "syt_xxx_yyy", "shared_secret": "your_registration_shared_secret", "default_password": "changeme123" } """ ) parser.add_argument("--config", required=True, help="配置文件路径 (JSON)") parser.add_argument("--users", help="用户列表文件 (CSV 或 JSON)") parser.add_argument("--single", action="store_true", help="注册单个用户") parser.add_argument("--username", help="单个用户名") parser.add_argument("--password", help="单个用户密码") parser.add_argument("--display-name", help="显示名称") parser.add_argument("--admin", action="store_true", help="设为管理员") parser.add_argument("--dry-run", action="store_true", help="干运行，不实际注册") parser.add_argument("--output", help="结果输出文件 (JSON)") args = parser.parse_args() # 加载配置 with open(args.config, "r", encoding="utf-8") as f: config = json.load(f) # 加载用户列表 if args.single: if not args.username: print("错误: 单用户模式需要 --username", file=sys.stderr) sys.exit(1) users = [{ "username": args.username, "password": args.password, "display_name": args.display_name, "admin": args.admin, }] elif args.users: filepath = args.users if filepath.endswith(".csv"): users = load_users_from_csv(filepath) else: users = load_users_from_json(filepath) else: print("错误: 需要 --users 或 --single 模式", file=sys.stderr) sys.exit(1) print(f"准备注册 {len(users)} 个用户...") print(f"服务器: {config['server_url']}") print() # 执行批量注册 results = batch_register(config, users, args.dry_run) # 统计 success_count = sum(1 for r in results if r.get("success") is True) fail_count = sum(1 for r in results if r.get("success") is False) print() print("=" * 50) print(f"注册完成: 成功 {success_count}, 失败 {fail_count}") # 输出结果 if args.output: with open(args.output, "w", encoding="utf-8") as f: json.dump(results, f, indent=2, ensure_ascii=False) print(f"结果已保存: {args.output}") return 0 if fail_count == 0 else 1 if __name__ == "__main__": sys.exit(main())