添加 Matrix Synapse 批量注册账号脚本

2026-05-29 11:42:42 +08:00
5 changed files with 391 additions and 0 deletions
--- a/README.md
+++ b/README.md
@@ -9,6 +9,11 @@ snippets/
 ├── llama-inference/     # llama.cpp 服务启动 + 推理任务
 │   ├── llama_inference.sh
 │   └── README.md
+├── matrix-batch-register/ # Matrix Synapse 批量注册账号
+│   ├── batch_register.py
+│   ├── config.json.example
+│   ├── users.csv.example
+│   └── README.md
 ├── .../
 └── README.md            # 本文件
 ```
@@ -18,6 +23,7 @@ snippets/
 | 目录 | 说明 |
 |------|------|
 | llama-inference | llama.cpp 服务启动后自动执行推理任务 |
+| matrix-batch-register | Matrix Synapse 批量注册账号 |

 ---

--- a/matrix-batch-register/README.md
+++ b/matrix-batch-register/README.md
@@ -0,0 +1,106 @@
+# Matrix Synapse 批量注册账号
+
+自动化批量注册 Matrix (Synapse) 用户账号。
+
+## 文件
+
+| 文件 | 说明 |
+|------|------|
+| `batch_register.py` | 主脚本 |
+| `config.json.example` | 配置示例 |
+| `users.csv.example` | 用户列表示例 |
+
+## 前置准备
+
+### 1. 获取 Admin Token
+
+登录 Matrix，获取 access token：
+- Element Web: 设置 → 帮助与关于 → 访问令牌
+- 或 API: `POST /_matrix/client/v3/login`
+
+### 2. 获取 registration_shared_secret
+
+在 Synapse 的 `homeserver.yaml` 中：
+```yaml
+registration_shared_secret: "你的密钥"
+```
+
+如果没有配置，需要先设置一个。
+
+## 用法
+
+### 批量注册
+
+```bash
+# 从 CSV 注册
+python3 batch_register.py --config config.json --users users.csv
+
+# 从 JSON 注册
+python3 batch_register.py --config config.json --users users.json
+
+# 干运行（先测试）
+python3 batch_register.py --config config.json --users users.csv --dry-run
+
+# 保存结果
+python3 batch_register.py --config config.json --users users.csv --output results.json
+```
+
+### 单个用户注册
+
+```bash
+python3 batch_register.py --config config.json --single \
+  --username user1 --password pwd123 --display-name "用户一"
+```
+
+## 配置文件格式
+
+`config.json`:
+```json
+{
+  "server_url": "https://matrix.tphai.com",
+  "admin_token": "syt_xxx_yyy_zzz",
+  "shared_secret": "your_registration_shared_secret",
+  "default_password": "changeme123"
+}
+```
+
+| 字段 | 说明 |
+|------|------|
+| server_url | Matrix 服务器地址 |
+| admin_token | 管理员 access token |
+| shared_secret | homeserver.yaml 中的 registration_shared_secret |
+| default_password | 默认密码（用户未指定时使用） |
+
+## 用户列表格式
+
+### CSV 格式
+
+`users.csv`:
+```csv
+username,password,display_name,admin
+user1,pass1,用户一,false
+user2,pass2,用户二,false
+admin1,pass3,管理员,true
+```
+
+### JSON 格式
+
+`users.json`:
+```json
+[
+  {"username": "user1", "password": "pass1", "display_name": "用户一"},
+  {"username": "user2", "password": "pass2", "display_name": "用户二"},
+  {"username": "admin1", "password": "pass3", "display_name": "管理员", "admin": true}
+]
+```
+
+## 工作原理
+
+1. 从 Admin API 获取 nonce
+2. 用 shared_secret 生成 HMAC 签名
+3. 调用 `/_synapse/admin/v1/register` 注册用户
+4. 可选设置 display_name
+
+---
+
+*黄庄4号程序员*
--- a/matrix-batch-register/batch_register.py
+++ b/matrix-batch-register/batch_register.py
@@ -0,0 +1,269 @@
+#!/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())
--- a/matrix-batch-register/config.json.example
+++ b/matrix-batch-register/config.json.example
@@ -0,0 +1,6 @@
+{
+  "server_url": "https://matrix.tphai.com",
+  "admin_token": "syt_xxx_yyy_zzz",
+  "shared_secret": "your_registration_shared_secret_from_homeserver_yaml",
+  "default_password": "changeme123"
+}
--- a/matrix-batch-register/users.csv.example
+++ b/matrix-batch-register/users.csv.example
@@ -0,0 +1,4 @@
+username,password,display_name,admin
+user1,pass123,用户一,false
+user2,pass456,用户二,false
+admin1,pass789,管理员,true