From 89d4260eb8ec13458dde41e7a95b15fc19815924 Mon Sep 17 00:00:00 2001 From: hongz Date: Tue, 3 Feb 2026 00:54:48 +0800 Subject: [PATCH] init project --- .dockerignore | 38 ++++ .env.example | 17 ++ .gitignore | 30 +++ AGENTS.md | 57 ++++++ Dockerfile | 20 ++ README.md | 113 ++++++++++ docker-compose.yml | 33 +++ references/matrix-bot-chat-reference | 1 + requirements.txt | 13 ++ src/__init__.py | 1 + src/config.py | 81 ++++++++ src/main.py | 81 ++++++++ src/services/__init__.py | 1 + src/services/matrix_service.py | 270 ++++++++++++++++++++++++ src/services/token_manager.py | 294 +++++++++++++++++++++++++++ tests/__init__.py | 1 + tests/test_config.py | 76 +++++++ tests/test_token_manager.py | 277 +++++++++++++++++++++++++ 18 files changed, 1404 insertions(+) create mode 100644 .dockerignore create mode 100644 .env.example create mode 100644 .gitignore create mode 100644 AGENTS.md create mode 100644 Dockerfile create mode 100644 README.md create mode 100644 docker-compose.yml create mode 160000 references/matrix-bot-chat-reference create mode 100644 requirements.txt create mode 100644 src/__init__.py create mode 100644 src/config.py create mode 100644 src/main.py create mode 100644 src/services/__init__.py create mode 100644 src/services/matrix_service.py create mode 100644 src/services/token_manager.py create mode 100644 tests/__init__.py create mode 100644 tests/test_config.py create mode 100644 tests/test_token_manager.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..20aa308 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,38 @@ +# Git +.git +.gitignore + +# Python +__pycache__ +*.pyc +*.pyo +*.pyd +.Python +*.egg-info +.eggs +dist +build +.venv +venv + +# IDE +.idea +.vscode +*.swp +*.swo + +# Environment +.env + +# References (不需要包含在鏡像中) +references/ + +# Tests (生產鏡像不需要) +tests/ + +# Documentation +*.md +docs/ + +# Store (運行時掛載) +store/ diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..34a4e2f --- /dev/null +++ b/.env.example @@ -0,0 +1,17 @@ +# PROJECT M Environment Variables +# Copy this file to .env and fill in your values + +# Matrix Server Configuration +MATRIX_SERVER=https://your-matrix-server.com +MATRIX_USER_ID=@your-bot:your-matrix-server.com +MATRIX_PASSWORD=your-secure-password + +# Device Configuration +MATRIX_DEVICE_ID=PROJECT_M_BOT +MATRIX_DEVICE_NAME=ProjectMBot + +# OpenClaw Configuration +OPENCLAW_JSON_PATH=~/.openclaw/openclaw.json + +# Bot Greeting Message +BOT_GREETING_MESSAGE=你好！我係 PROJECT M Bot，已成功加入房間。有咩需要可以搵管理員幫手！ diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e2a0cdb --- /dev/null +++ b/.gitignore @@ -0,0 +1,30 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# Virtual environments +.venv/ +venv/ +ENV/ + +# Environment variables +.env + +# Store folder (E2EE keys) +store/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo + +# Distribution +dist/ +build/ +*.egg-info/ + +# OS files +.DS_Store +Thumbs.db diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..758f6e1 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,57 @@ +# PROJECT M - OpenClaw Matrix Bot + +> AI Agent 專用文檔，描述項目結構和開發指南 + +## 項目概述 + +| 屬性 | 內容 | +|------|------| +| 代號 | PROJECT M | +| 類型 | Matrix Bot | +| 核心功能 | 自動登錄、Token 同步、邀請響應、問候發送 | + +## 目錄結構 + +``` +openclaw-matrix-uplife-bot/ +├── src/ # 源代碼 +│ ├── main.py # 主入口 +│ ├── config.py # 配置管理 +│ └── services/ +│ ├── matrix_service.py # Matrix 服務 +│ └── token_manager.py # Token 管理器 +├── tests/ # 單元測試 +├── references/ # 參考代碼（只讀） +│ └── matrix-bot-chat-reference/ +├── docker-compose.yml # Docker 配置 +└── README.md +``` + +## 關鍵配置 + +- **Token JSON Key**: `channels.matrix.accessToken` +- **OpenClaw 路徑**: `~/.openclaw/openclaw.json` + +## 開發命令 + +```bash +# 啟動 +docker compose up -d + +# 停止 +docker compose down + +# 測試 +pytest tests/ -v +``` + +## 安全協議 + +1. 更新 Token 前**必須**備份 +2. 使用沙箱測試 +3. 密碼僅通過環境變量傳入 + +## 參考資料 + +- `references/matrix-bot-chat-reference/` - 舊版 Matrix Bot 代碼 +- [matrix-nio 文檔](https://matrix-nio.readthedocs.io/) diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..71d2c2e --- /dev/null +++ b/Dockerfile @@ -0,0 +1,20 @@ +FROM python:3.11-slim + +# 設置工作目錄 +WORKDIR /app + +# 複製依賴文件 +COPY requirements.txt . + +# 安裝依賴 +RUN pip install --no-cache-dir -r requirements.txt + +# 複製源代碼 +COPY src/ ./src/ + +# 設置環境變量 +ENV PYTHONUNBUFFERED=1 +ENV PYTHONPATH=/app + +# 運行 Bot +CMD ["python", "-m", "src.main"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..0c0392c --- /dev/null +++ b/README.md @@ -0,0 +1,113 @@ +# PROJECT M - Matrix Bot + +[![Python](https://img.shields.io/badge/Python-3.11+-blue.svg)](https://www.python.org/) +[![Matrix](https://img.shields.io/badge/Protocol-Matrix-green.svg)](https://matrix.org/) +[![Docker](https://img.shields.io/badge/Docker-Compose-blue.svg)](https://docs.docker.com/compose/) + +> Matrix Bot for OpenClaw integration - Auto-join rooms and sync access tokens. + +## 功能特點 + +- 🔐 **自動登錄**：使用配置的帳號密碼登錄 Matrix Server +- 🔄 **Token 同步**：自動更新 `~/.openclaw/openclaw.json` 中的 access token +- 📬 **邀請響應**：自動接受房間邀請並加入 +- 👋 **問候消息**：加入房間後自動發送問候信息 +- 🐳 **Docker 部署**：支持 Docker Compose 一鍵啟動 + +## 快速開始 + +### 1. 配置環境變量 + +```bash +# 複製環境變量模板 +cp .env.example .env + +# 編輯配置 +nano .env +``` + +必要配置項： +```env +MATRIX_SERVER=https://your-matrix-server.com +MATRIX_USER_ID=@your-bot:your-matrix-server.com +MATRIX_PASSWORD=your-secure-password +``` + +### 2. 使用 Docker Compose 啟動 + +```bash +# 構建並啟動 +docker compose up -d + +# 查看日誌 +docker compose logs -f + +# 停止 +docker compose down +``` + +### 3. 本地開發運行 + +```bash +# 創建虛擬環境 +python -m venv .venv +source .venv/bin/activate + +# 安裝依賴 +pip install -r requirements.txt + +# 運行 +python -m src.main +``` + +## 項目結構 + +``` +openclaw-matrix-uplife-bot/ +├── src/ +│ ├── __init__.py +│ ├── main.py # 主入口 +│ ├── config.py # 配置管理 +│ └── services/ +│ ├── __init__.py +│ ├── matrix_service.py # Matrix 服務 +│ └── token_manager.py # Token 管理器 +├── tests/ # 測試 +├── store/ # E2EE 密鑰存儲 +├── Dockerfile +├── docker-compose.yml +├── requirements.txt +└── README.md +``` + +## 安全協議 + +### Token 更新安全 + +1. **強制備份**：更新前必須創建帶時間戳的備份 +2. **驗證機制**：確認備份文件存在後才進行更新 +3. **錯誤回滾**：更新失敗時可從備份恢復 + +### 環境變量 + +- 敏感信息（密碼）僅通過環境變量傳入 +- `.env` 文件已加入 `.gitignore` +- 禁止在代碼中硬編碼密碼 + +## 測試 + +```bash +# 運行所有測試 +pytest tests/ -v + +# 運行特定測試 +pytest tests/test_token_manager.py -v +``` + +## License + +MIT License + +--- + +**PROJECT M** - OpenClaw Matrix Integration Bot diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..f322561 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,33 @@ +version: '3.8' + +services: + project-m-bot: + build: . + container_name: project-m-bot + restart: unless-stopped + + # 環境變量（從 .env 文件讀取） + environment: + - MATRIX_SERVER=${MATRIX_SERVER} + - MATRIX_USER_ID=${MATRIX_USER_ID} + - MATRIX_PASSWORD=${MATRIX_PASSWORD} + - MATRIX_DEVICE_ID=${MATRIX_DEVICE_ID:-PROJECT_M_BOT} + - MATRIX_DEVICE_NAME=${MATRIX_DEVICE_NAME:-ProjectMBot} + - BOT_GREETING_MESSAGE=${BOT_GREETING_MESSAGE:-你好！我係 PROJECT M Bot，已成功加入房間。} + # 容器內路徑 + - OPENCLAW_JSON_PATH=/root/.openclaw/openclaw.json + + # 掛載卷 + volumes: + # 掛載 openclaw 配置目錄（用於更新 token） + - ~/.openclaw:/root/.openclaw + # 掛載 store 目錄（用於持久化 E2EE 密鑰） + - ./store:/app/store + + # 網絡配置 + networks: + - project-m-network + +networks: + project-m-network: + driver: bridge diff --git a/references/matrix-bot-chat-reference b/references/matrix-bot-chat-reference new file mode 160000 index 0000000..972a538 --- /dev/null +++ b/references/matrix-bot-chat-reference @@ -0,0 +1 @@ +Subproject commit 972a538356431b58e86be929d5de4c86eb420f57 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..6e35ec6 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,13 @@ +# PROJECT M Dependencies +# Matrix Protocol Client +matrix-nio[e2e]>=0.25.0 + +# Async File Operations +aiofiles>=23.0.0 + +# Environment Variables +python-dotenv>=1.0.0 + +# Testing +pytest>=7.0.0 +pytest-asyncio>=0.21.0 diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..a605c69 --- /dev/null +++ b/src/__init__.py @@ -0,0 +1 @@ +# PROJECT M - Matrix Bot diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..d6d718f --- /dev/null +++ b/src/config.py @@ -0,0 +1,81 @@ +# config.py - PROJECT M 配置管理模組 +""" +Configuration management for PROJECT M Matrix Bot. +Loads settings from environment variables with sensible defaults. +""" + +import os +from pathlib import Path +from dotenv import load_dotenv + +# 載入 .env 文件 +load_dotenv() + + +class Config: + """PROJECT M Bot 配置類""" + + # Matrix Server 配置 + MATRIX_SERVER: str = os.getenv("MATRIX_SERVER", "https://matrix.org") + MATRIX_USER_ID: str = os.getenv("MATRIX_USER_ID", "") + MATRIX_PASSWORD: str = os.getenv("MATRIX_PASSWORD", "") + + # 設備配置 + DEVICE_ID: str = os.getenv("MATRIX_DEVICE_ID", "PROJECT_M_BOT") + DEVICE_NAME: str = os.getenv("MATRIX_DEVICE_NAME", "ProjectMBot") + + # OpenClaw 配置 + OPENCLAW_JSON_PATH: str = os.getenv( + "OPENCLAW_JSON_PATH", + str(Path.home() / ".openclaw" / "openclaw.json") + ) + + # Access Token 的 JSON Key 路徑 + # 格式: channels.matrix.accessToken + TOKEN_JSON_KEY: str = "channels.matrix.accessToken" + + # Bot 問候信息 + GREETING_MESSAGE: str = os.getenv( + "BOT_GREETING_MESSAGE", + "你好！我係 PROJECT M Bot，已成功加入房間。有咩需要可以搵管理員幫手！" + ) + + # 存儲路徑（E2EE 密鑰） + STORE_PATH: str = os.getenv("STORE_PATH", "store") + + @classmethod + def validate(cls) -> tuple[bool, list[str]]: + """ + 驗證必要的配置是否已設置 + + Returns: + tuple[bool, list[str]]: (是否有效, 缺失的配置列表) + """ + missing = [] + + if not cls.MATRIX_SERVER: + missing.append("MATRIX_SERVER") + if not cls.MATRIX_USER_ID: + missing.append("MATRIX_USER_ID") + if not cls.MATRIX_PASSWORD: + missing.append("MATRIX_PASSWORD") + + return len(missing) == 0, missing + + @classmethod + def get_openclaw_path(cls) -> Path: + """ + 獲取 openclaw.json 的完整路徑 + + Returns: + Path: openclaw.json 的路徑對象 + """ + path_str = cls.OPENCLAW_JSON_PATH + # 展開 ~ 為用戶主目錄 + if path_str.startswith("~"): + return Path(path_str).expanduser() + return Path(path_str) + + +# 創建全局配置實例 +config = Config() diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..898eed9 --- /dev/null +++ b/src/main.py @@ -0,0 +1,81 @@ +# main.py - PROJECT M 主入口 +""" +PROJECT M Matrix Bot - Main Entry Point + +This bot: +1. Logs into Matrix server +2. Updates access token in ~/.openclaw/openclaw.json +3. Listens for room invitations +4. Auto-joins and sends greeting messages +""" + +import asyncio +import signal +import sys + +from src.config import config +from src.services.matrix_service import MatrixService + + +async def main(): + """主函數""" + print("=" * 50) + print("PROJECT M - Matrix Bot") + print("=" * 50) + + # 驗證配置 + is_valid, missing = config.validate() + if not is_valid: + print(f"[ERROR] 缺少必要配置: {', '.join(missing)}") + print("[ERROR] 請設置環境變量或創建 .env 文件") + sys.exit(1) + + print(f"[INFO] Matrix Server: {config.MATRIX_SERVER}") + print(f"[INFO] User ID: {config.MATRIX_USER_ID}") + print(f"[INFO] OpenClaw Path: {config.get_openclaw_path()}") + print("-" * 50) + + # 創建 Matrix 服務 + matrix_service = MatrixService() + + # 設置信號處理 + loop = asyncio.get_event_loop() + + async def shutdown(): + """優雅關閉""" + print("\n[INFO] 正在關閉...") + await matrix_service.close() + print("[INFO] 已關閉") + + def signal_handler(sig, frame): + """處理終止信號""" + asyncio.create_task(shutdown()) + sys.exit(0) + + signal.signal(signal.SIGINT, signal_handler) + signal.signal(signal.SIGTERM, signal_handler) + + try: + # 登錄 + success = await matrix_service.login() + if not success: + print("[ERROR] 登錄失敗，請檢查配置") + sys.exit(1) + + print("[INFO] 登錄成功！開始監聽邀請...") + print("[INFO] 按 Ctrl+C 退出") + print("-" * 50) + + # 開始同步循環 + await matrix_service.sync_forever() + + except KeyboardInterrupt: + print("\n[INFO] 收到中斷信號") + except Exception as e: + print(f"[ERROR] 發生錯誤: {e}") + finally: + await matrix_service.close() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/services/__init__.py b/src/services/__init__.py new file mode 100644 index 0000000..0557eb6 --- /dev/null +++ b/src/services/__init__.py @@ -0,0 +1 @@ +# Services module diff --git a/src/services/matrix_service.py b/src/services/matrix_service.py new file mode 100644 index 0000000..ec8fa53 --- /dev/null +++ b/src/services/matrix_service.py @@ -0,0 +1,270 @@ +# matrix_service.py - Matrix 連接服務 +""" +Matrix Bot Service for PROJECT M. +Handles login, sync, invite callbacks, and message sending. + +Based on reference implementation from: +references/matrix-bot-chat-reference/services/matrix_service.py +""" + +import asyncio +import os +from typing import Optional +from nio import ( + AsyncClient, + LoginResponse, + InviteMemberEvent, + ClientConfig, +) + +from src.config import config +from src.services.token_manager import TokenManager + + +class MatrixService: + """ + Matrix Bot 核心服務 + + 功能： + - 登錄 Matrix Server + - 同步消息 + - 處理房間邀請 + - 發送問候消息 + - 更新 openclaw.json 中的 access token + """ + + _instance = None + + def __new__(cls): + """單例模式""" + if cls._instance is None: + cls._instance = super(MatrixService, cls).__new__(cls) + cls._instance._initialized = False + return cls._instance + + def __init__(self): + """初始化 Matrix 服務""" + if self._initialized: + return + + # 確保存儲目錄存在 + if not os.path.exists(config.STORE_PATH): + os.makedirs(config.STORE_PATH) + + # 創建客戶端配置 + client_config = ClientConfig( + store_sync_tokens=True, + encryption_enabled=False # PROJECT M 不需要 E2EE + ) + + # 提取用戶名（從 @user:server.com 格式） + username = config.MATRIX_USER_ID + if username.startswith("@"): + username = username.split(":")[0][1:] + + # 初始化客戶端 + self.client = AsyncClient( + config.MATRIX_SERVER, + username, + device_id=config.DEVICE_ID, + store_path=config.STORE_PATH, + config=client_config, + ) + + self.is_logged_in = False + self.welcomed_rooms = set() + + # Token 管理器 + self.token_manager = TokenManager( + json_path=config.get_openclaw_path(), + token_key=config.TOKEN_JSON_KEY + ) + + # 添加邀請回調 + self.client.add_event_callback( + self.invite_callback, + InviteMemberEvent + ) + + self._initialized = True + print("[MatrixService] 初始化完成") + + async def login(self) -> bool: + """ + 登錄 Matrix Server 並更新 openclaw.json 中的 token + + Returns: + 是否登錄成功 + """ + if self.is_logged_in: + print("[MatrixService] 已經登錄") + return True + + try: + print(f"[MatrixService] 正在登錄 {config.MATRIX_SERVER}...") + + response = await self.client.login( + password=config.MATRIX_PASSWORD, + device_name=config.DEVICE_NAME + ) + + if isinstance(response, LoginResponse): + self.is_logged_in = True + print(f"[MatrixService] 登錄成功！User ID: {response.user_id}") + print(f"[MatrixService] Access Token: {response.access_token[:30]}...") + + # 更新 openclaw.json 中的 token + await self._update_openclaw_token(response.access_token) + + return True + else: + print(f"[MatrixService] 登錄失敗: {response}") + return False + + except Exception as e: + print(f"[MatrixService] 登錄過程出錯: {e}") + return False + + async def _update_openclaw_token(self, access_token: str) -> bool: + """ + 更新 openclaw.json 中的 access token + + Args: + access_token: 新的 access token + + Returns: + 是否更新成功 + """ + print("[MatrixService] 準備更新 openclaw.json...") + + # 檢查文件是否存在 + openclaw_path = config.get_openclaw_path() + if not openclaw_path.exists(): + print(f"[MatrixService] openclaw.json 不存在: {openclaw_path}") + print("[MatrixService] 跳過 token 更新（這在測試環境是正常的）") + return False + + # 使用 TokenManager 安全更新 + success = await self.token_manager.update_access_token(access_token) + + if success: + print("[MatrixService] openclaw.json token 更新成功！") + else: + print("[MatrixService] openclaw.json token 更新失敗") + + return success + + async def invite_callback(self, room, event): + """ + 處理房間邀請事件 + + 當 Bot 被邀請到房間時： + 1. 自動加入房間 + 2. 發送問候消息 + """ + try: + inviter = event.sender + room_id = room.room_id + + print(f"[MatrixService] 收到來自 {inviter} 的邀請，房間: {room_id}") + + # 自動加入房間 + join_response = await self.client.join(room_id) + + if join_response: + print(f"[MatrixService] 已加入房間: {room_id}") + + # 同步以獲取房間狀態 + await self.client.sync(timeout=5000) + + # 發送問候消息 + await self.send_greeting(room_id) + + # 記錄已發送問候的房間 + self.welcomed_rooms.add(room_id) + else: + print(f"[MatrixService] 加入房間失敗: {room_id}") + + except Exception as e: + print(f"[MatrixService] 處理邀請時出錯: {e}") + + async def send_greeting(self, room_id: str) -> bool: + """ + 發送問候消息到指定房間 + + Args: + room_id: 房間 ID + + Returns: + 是否發送成功 + """ + return await self.send_message(config.GREETING_MESSAGE, room_id) + + async def send_message(self, message: str, room_id: str) -> bool: + """ + 發送消息到指定房間 + + Args: + message: 消息內容 + room_id: 房間 ID + + Returns: + 是否發送成功 + """ + if not self.is_logged_in: + print("[MatrixService] 未登錄，無法發送消息") + return False + + content = { + "msgtype": "m.text", + "body": message + } + + try: + await self.client.room_send( + room_id=room_id, + message_type="m.room.message", + content=content + ) + print(f"[MatrixService] 消息已發送到 {room_id}") + return True + + except Exception as e: + print(f"[MatrixService] 發送消息失敗: {e}") + return False + + async def sync_forever(self): + """ + 開始無限同步循環 + + 這個方法會持續運行，監聽所有事件 + """ + if not self.is_logged_in: + print("[MatrixService] 未登錄，無法開始同步") + return + + print("[MatrixService] 開始同步循環...") + + try: + # 先進行一次同步 + await self.client.sync(timeout=10000) + print("[MatrixService] 初始同步完成") + + # 開始無限同步 + await self.client.sync_forever( + timeout=30000, + full_state=True + ) + except Exception as e: + print(f"[MatrixService] 同步出錯: {e}") + raise + + async def close(self): + """關閉連接""" + if self.client: + await self.client.close() + print("[MatrixService] 連接已關閉") + + +# 創建全局服務實例 +matrix_service = MatrixService() diff --git a/src/services/token_manager.py b/src/services/token_manager.py new file mode 100644 index 0000000..4002079 --- /dev/null +++ b/src/services/token_manager.py @@ -0,0 +1,294 @@ +# token_manager.py - OpenClaw Token 管理器 +""" +Manages the access token synchronization with openclaw.json. +Handles backup creation and safe updates. + +SAFETY PROTOCOLS: +1. Always create timestamped backup before any modification +2. Verify backup exists before updating +3. Never corrupt the original file +""" + +import json +import os +import shutil +from datetime import datetime +from pathlib import Path +from typing import Optional +import aiofiles + + +class TokenManager: + """ + 管理 openclaw.json 中的 Matrix access token + + 安全協議： + - 更新前必須創建備份 + - 驗證備份存在後才進行更新 + - 任何錯誤都不會破壞原文件 + - 備份輪轉：最多保留 MAX_BACKUPS 個備份 + """ + + # 最大備份文件數量 + MAX_BACKUPS: int = 5 + + def __init__(self, json_path: Path, token_key: str = "channels.matrix.accessToken"): + """ + 初始化 Token 管理器 + + Args: + json_path: openclaw.json 的路徑 + token_key: Token 在 JSON 中的 key 路徑（點分隔） + """ + self.json_path = json_path + self.token_key = token_key + self._last_backup_path: Optional[Path] = None + + def _generate_backup_filename(self) -> str: + """生成帶時間戳的備份文件名（毫秒級精度避免重複）""" + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") + return f"{self.json_path.name}.{timestamp}.bak" + + def _get_nested_value(self, data: dict, key_path: str) -> Optional[str]: + """ + 從嵌套字典中獲取值 + + Args: + data: 字典數據 + key_path: 點分隔的 key 路徑，如 "channels.matrix.accessToken" + + Returns: + 對應的值，如果不存在則返回 None + """ + keys = key_path.split(".") + current = data + + for key in keys: + if isinstance(current, dict) and key in current: + current = current[key] + else: + return None + + return current if isinstance(current, str) else None + + def _set_nested_value(self, data: dict, key_path: str, value: str) -> dict: + """ + 在嵌套字典中設置值（會自動創建中間層級） + + Args: + data: 字典數據 + key_path: 點分隔的 key 路徑 + value: 要設置的值 + + Returns: + 更新後的字典 + """ + keys = key_path.split(".") + current = data + + # 遍歷到最後一個 key 的父級 + for key in keys[:-1]: + if key not in current: + current[key] = {} + current = current[key] + + # 設置最後一個 key 的值 + current[keys[-1]] = value + return data + + def create_backup(self) -> Optional[Path]: + """ + 創建 openclaw.json 的備份 + + Returns: + 備份文件路徑，如果失敗則返回 None + """ + if not self.json_path.exists(): + print(f"[TokenManager] 源文件不存在: {self.json_path}") + return None + + backup_filename = self._generate_backup_filename() + backup_path = self.json_path.parent / backup_filename + + try: + shutil.copy2(self.json_path, backup_path) + self._last_backup_path = backup_path + print(f"[TokenManager] 備份已創建: {backup_path}") + + # 備份輪轉：刪除超出限制的舊備份 + self._rotate_backups() + + return backup_path + except Exception as e: + print(f"[TokenManager] 創建備份失敗: {e}") + return None + + def _get_backup_files(self) -> list[Path]: + """ + 獲取所有備份文件，按時間排序（最新到最舊） + + 使用文件名中的時間戳排序（格式：YYYYMMDD_HHMMSS_ffffff） + 這比 mtime 更可靠 + + Returns: + 備份文件列表 + """ + if not self.json_path.parent.exists(): + return [] + + # 匹配備份文件模式：openclaw.json.YYYYMMDD_HHMMSS_ffffff.bak + pattern = f"{self.json_path.name}.*.bak" + backup_files = list(self.json_path.parent.glob(pattern)) + + # 按文件名排序（時間戳在文件名中，字典序即時間序） + # 格式：name.YYYYMMDD_HHMMSS_ffffff.bak + # 因為時間戳格式固定，字典序 = 時間序 + backup_files.sort(key=lambda x: x.name, reverse=True) + + return backup_files + + def _rotate_backups(self) -> int: + """ + 輪轉備份文件，刪除超出 MAX_BACKUPS 的舊備份 + + Returns: + 刪除的備份數量 + """ + backup_files = self._get_backup_files() + deleted_count = 0 + + # 如果備份數量超過限制，刪除最舊的 + if len(backup_files) > self.MAX_BACKUPS: + files_to_delete = backup_files[self.MAX_BACKUPS:] + + for old_backup in files_to_delete: + try: + old_backup.unlink() + print(f"[TokenManager] 已刪除舊備份: {old_backup.name}") + deleted_count += 1 + except Exception as e: + print(f"[TokenManager] 刪除備份失敗: {old_backup.name} - {e}") + + return deleted_count + + def get_backup_count(self) -> int: + """ + 獲取當前備份文件數量 + + Returns: + 備份文件數量 + """ + return len(self._get_backup_files()) + + def verify_backup_exists(self, backup_path: Optional[Path] = None) -> bool: + """ + 驗證備份文件是否存在 + + Args: + backup_path: 備份路徑，如果為 None 則使用上次創建的備份 + + Returns: + 備份是否存在 + """ + path = backup_path or self._last_backup_path + if path is None: + return False + return path.exists() + + async def read_current_token(self) -> Optional[str]: + """ + 讀取當前的 access token + + Returns: + 當前 token，如果不存在則返回 None + """ + if not self.json_path.exists(): + return None + + try: + async with aiofiles.open(self.json_path, 'r', encoding='utf-8') as f: + content = await f.read() + data = json.loads(content) + return self._get_nested_value(data, self.token_key) + except Exception as e: + print(f"[TokenManager] 讀取 token 失敗: {e}") + return None + + async def update_access_token(self, new_token: str) -> bool: + """ + 更新 access token（安全模式） + + 安全協議： + 1. 創建備份 + 2. 驗證備份存在 + 3. 讀取原文件 + 4. 更新 token + 5. 寫入文件 + + Args: + new_token: 新的 access token + + Returns: + 是否更新成功 + """ + # Step 1: 創建備份 + backup_path = self.create_backup() + if backup_path is None: + print("[TokenManager] 無法創建備份，中止更新操作") + return False + + # Step 2: 驗證備份存在 + if not self.verify_backup_exists(backup_path): + print("[TokenManager] 備份驗證失敗，中止更新操作") + return False + + print(f"[TokenManager] 備份驗證通過: {backup_path}") + + try: + # Step 3: 讀取原文件 + async with aiofiles.open(self.json_path, 'r', encoding='utf-8') as f: + content = await f.read() + data = json.loads(content) + + # Step 4: 更新 token + old_token = self._get_nested_value(data, self.token_key) + if old_token: + print(f"[TokenManager] 舊 token: {old_token[:20]}...") + + data = self._set_nested_value(data, self.token_key, new_token) + print(f"[TokenManager] 新 token: {new_token[:20]}...") + + # Step 5: 寫入文件（保持格式美觀） + async with aiofiles.open(self.json_path, 'w', encoding='utf-8') as f: + await f.write(json.dumps(data, indent=2, ensure_ascii=False)) + + print("[TokenManager] Token 更新成功！") + return True + + except Exception as e: + print(f"[TokenManager] 更新失敗: {e}") + print(f"[TokenManager] 可以從備份恢復: {backup_path}") + return False + + def restore_from_backup(self, backup_path: Optional[Path] = None) -> bool: + """ + 從備份恢復原文件 + + Args: + backup_path: 備份路徑，如果為 None 則使用上次創建的備份 + + Returns: + 是否恢復成功 + """ + path = backup_path or self._last_backup_path + if path is None or not path.exists(): + print("[TokenManager] 沒有可用的備份") + return False + + try: + shutil.copy2(path, self.json_path) + print(f"[TokenManager] 已從備份恢復: {path}") + return True + except Exception as e: + print(f"[TokenManager] 恢復失敗: {e}") + return False diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..11754ee --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +# Tests module diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..ca16467 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,76 @@ +# test_config.py - 配置模組測試 +""" +Tests for Config class. +""" + +import os +import pytest +from pathlib import Path + +import sys +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from src.config import Config + + +class TestConfig: + """Config 測試套件""" + + def test_default_values(self): + """測試默認值""" + assert Config.DEVICE_ID == os.getenv("MATRIX_DEVICE_ID", "PROJECT_M_BOT") + assert Config.STORE_PATH == os.getenv("STORE_PATH", "store") + + def test_token_json_key(self): + """測試 Token JSON Key 路徑""" + assert Config.TOKEN_JSON_KEY == "channels.matrix.accessToken" + + def test_validate_missing_config(self): + """測試配置驗證（缺少配置時）""" + # 臨時清空環境變量 + original_server = os.environ.get("MATRIX_SERVER") + original_user = os.environ.get("MATRIX_USER_ID") + original_password = os.environ.get("MATRIX_PASSWORD") + + try: + os.environ["MATRIX_SERVER"] = "" + os.environ["MATRIX_USER_ID"] = "" + os.environ["MATRIX_PASSWORD"] = "" + + # 重新載入配置類的值 + Config.MATRIX_SERVER = "" + Config.MATRIX_USER_ID = "" + Config.MATRIX_PASSWORD = "" + + is_valid, missing = Config.validate() + + assert is_valid is False + assert "MATRIX_SERVER" in missing + assert "MATRIX_USER_ID" in missing + assert "MATRIX_PASSWORD" in missing + finally: + # 恢復原值 + if original_server: + os.environ["MATRIX_SERVER"] = original_server + if original_user: + os.environ["MATRIX_USER_ID"] = original_user + if original_password: + os.environ["MATRIX_PASSWORD"] = original_password + + def test_get_openclaw_path_with_tilde(self): + """測試 ~ 路徑展開""" + Config.OPENCLAW_JSON_PATH = "~/.openclaw/openclaw.json" + + path = Config.get_openclaw_path() + + assert isinstance(path, Path) + assert str(path).startswith(str(Path.home())) + assert "~" not in str(path) + + def test_get_openclaw_path_absolute(self): + """測試絕對路徑""" + Config.OPENCLAW_JSON_PATH = "/tmp/test/openclaw.json" + + path = Config.get_openclaw_path() + + assert path == Path("/tmp/test/openclaw.json") diff --git a/tests/test_token_manager.py b/tests/test_token_manager.py new file mode 100644 index 0000000..115436a --- /dev/null +++ b/tests/test_token_manager.py @@ -0,0 +1,277 @@ +# test_token_manager.py - Token 管理器測試 +""" +Tests for TokenManager class. +Uses sandbox testing to avoid modifying real files. +""" + +import json +import pytest +import tempfile +from pathlib import Path +from datetime import datetime + +# 需要確保可以導入 +import sys +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from src.services.token_manager import TokenManager + + +class TestTokenManager: + """TokenManager 測試套件""" + + @pytest.fixture + def temp_json_file(self, tmp_path): + """創建臨時 JSON 文件用於沙箱測試""" + json_file = tmp_path / "openclaw.json" + test_data = { + "channels": { + "matrix": { + "accessToken": "old_token_12345", + "homeserver": "https://matrix.org" + }, + "telegram": { + "botToken": "telegram_token" + } + }, + "settings": { + "debug": True + } + } + json_file.write_text(json.dumps(test_data, indent=2)) + return json_file + + @pytest.fixture + def token_manager(self, temp_json_file): + """創建 TokenManager 實例""" + return TokenManager( + json_path=temp_json_file, + token_key="channels.matrix.accessToken" + ) + + def test_create_backup(self, token_manager, temp_json_file): + """測試備份創建""" + backup_path = token_manager.create_backup() + + assert backup_path is not None + assert backup_path.exists() + assert temp_json_file.name in backup_path.name + assert ".bak" in backup_path.name + + # 驗證備份內容與原文件相同 + original = json.loads(temp_json_file.read_text()) + backup = json.loads(backup_path.read_text()) + assert original == backup + + def test_verify_backup_exists(self, token_manager): + """測試備份驗證""" + # 未創建備份時應該返回 False + assert token_manager.verify_backup_exists() is False + + # 創建備份後應該返回 True + token_manager.create_backup() + assert token_manager.verify_backup_exists() is True + + def test_get_nested_value(self, token_manager): + """測試嵌套值讀取""" + data = { + "level1": { + "level2": { + "level3": "value" + } + } + } + + assert token_manager._get_nested_value(data, "level1.level2.level3") == "value" + assert token_manager._get_nested_value(data, "level1.level2") is None # 不是字符串 + assert token_manager._get_nested_value(data, "nonexistent") is None + + def test_set_nested_value(self, token_manager): + """測試嵌套值設置""" + data = {} + + result = token_manager._set_nested_value(data, "a.b.c", "new_value") + + assert result["a"]["b"]["c"] == "new_value" + + @pytest.mark.asyncio + async def test_read_current_token(self, token_manager): + """測試讀取當前 token""" + token = await token_manager.read_current_token() + + assert token == "old_token_12345" + + @pytest.mark.asyncio + async def test_update_access_token(self, token_manager, temp_json_file): + """測試更新 access token（核心功能）""" + new_token = "new_token_67890" + + # 執行更新 + success = await token_manager.update_access_token(new_token) + + # 驗證成功 + assert success is True + + # 驗證備份存在 + assert token_manager.verify_backup_exists() is True + + # 驗證新 token 已寫入 + updated_data = json.loads(temp_json_file.read_text()) + assert updated_data["channels"]["matrix"]["accessToken"] == new_token + + # 驗證其他數據未被破壞 + assert updated_data["channels"]["telegram"]["botToken"] == "telegram_token" + assert updated_data["settings"]["debug"] is True + + def test_restore_from_backup(self, token_manager, temp_json_file): + """測試從備份恢復""" + # 創建備份 + backup_path = token_manager.create_backup() + + # 修改原文件 + temp_json_file.write_text('{"corrupted": true}') + + # 恢復 + success = token_manager.restore_from_backup() + + assert success is True + + # 驗證恢復結果 + restored_data = json.loads(temp_json_file.read_text()) + assert restored_data["channels"]["matrix"]["accessToken"] == "old_token_12345" + + +class TestTokenManagerEdgeCases: + """TokenManager 邊緣情況測試""" + + def test_backup_nonexistent_file(self, tmp_path): + """測試備份不存在的文件""" + manager = TokenManager( + json_path=tmp_path / "nonexistent.json", + token_key="test" + ) + + backup_path = manager.create_backup() + assert backup_path is None + + @pytest.mark.asyncio + async def test_update_nonexistent_file(self, tmp_path): + """測試更新不存在的文件""" + manager = TokenManager( + json_path=tmp_path / "nonexistent.json", + token_key="test" + ) + + success = await manager.update_access_token("new_token") + assert success is False + + +class TestBackupRotation: + """備份輪轉功能測試（沙箱測試）""" + + @pytest.fixture + def temp_json_file(self, tmp_path): + """創建臨時 JSON 文件""" + json_file = tmp_path / "openclaw.json" + test_data = {"channels": {"matrix": {"accessToken": "test_token"}}} + json_file.write_text(json.dumps(test_data)) + return json_file + + @pytest.fixture + def token_manager(self, temp_json_file): + """創建 TokenManager 實例""" + return TokenManager( + json_path=temp_json_file, + token_key="channels.matrix.accessToken" + ) + + def test_max_backups_constant(self): + """測試最大備份數量常量""" + assert TokenManager.MAX_BACKUPS == 5 + + def test_get_backup_files_empty(self, token_manager): + """測試沒有備份時返回空列表""" + backups = token_manager._get_backup_files() + assert backups == [] + + def test_get_backup_files_with_backups(self, token_manager, tmp_path): + """測試獲取備份文件列表""" + # 創建幾個備份 + for i in range(3): + token_manager.create_backup() + import time + time.sleep(0.01) # 確保時間戳不同 + + backups = token_manager._get_backup_files() + assert len(backups) == 3 + + # 確認按文件名排序（時間戳在名稱中，字典序=時間序，最新在前） + for i in range(len(backups) - 1): + assert backups[i].name > backups[i + 1].name + + def test_rotation_under_limit(self, token_manager): + """測試備份數量未超過限制時不刪除""" + # 創建 3 個備份（低於限制） + for i in range(3): + token_manager.create_backup() + import time + time.sleep(0.01) + + assert token_manager.get_backup_count() == 3 + + def test_rotation_at_limit(self, token_manager): + """測試備份數量剛好達到限制""" + # 創建 5 個備份（剛好等於限制） + for i in range(5): + token_manager.create_backup() + import time + time.sleep(0.01) + + assert token_manager.get_backup_count() == 5 + + def test_rotation_over_limit(self, token_manager): + """測試備份數量超過限制時自動刪除最舊的""" + # 創建 7 個備份（超過限制） + for i in range(7): + token_manager.create_backup() + import time + time.sleep(0.01) + + # 應該只保留 5 個 + assert token_manager.get_backup_count() == 5 + + def test_rotation_preserves_newest(self, token_manager, temp_json_file): + """測試輪轉保留最新的備份""" + backup_paths = [] + + # 創建 7 個備份 + for i in range(7): + path = token_manager.create_backup() + if path: + backup_paths.append(path) + import time + time.sleep(0.01) + + # 檢查最新的 5 個備份仍然存在 + remaining_backups = token_manager._get_backup_files() + assert len(remaining_backups) == 5 + + # 最新的 5 個應該是最後創建的 5 個 + newest_5 = backup_paths[-5:] + for path in newest_5: + assert path.exists(), f"{path} should exist" + + # 最舊的 2 個應該被刪除 + oldest_2 = backup_paths[:2] + for path in oldest_2: + assert not path.exists(), f"{path} should be deleted" + + def test_get_backup_count(self, token_manager): + """測試獲取備份數量""" + assert token_manager.get_backup_count() == 0 + + token_manager.create_backup() + assert token_manager.get_backup_count() == 1 + + token_manager.create_backup() + assert token_manager.get_backup_count() == 2