From 62b8fef18ab706bcd481eb8f29929df4b291b0e3 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Fri, 26 Jun 2026 12:10:41 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BB=A3=E7=A0=81=E5=AE=A1=E6=9F=A5?= =?UTF-8?q?=E9=97=AE=E9=A2=98=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - C1 (Critical): 添加 PyYAML 解析 frontmatter，支持多行值和列表 - M1 (Major): 添加 FTS5 查询验证，防止注入攻击 - M2/M3 (Major): 工具类使用显式 wiki_vault_path 参数 - M4 (Major): 集成 aiofiles 实现真正的异步文件 I/O - M5 (Major): 改用 logger.exception() 记录完整堆栈 Co-Authored-By: Claude Dev --- mcp_server/main.py | 11 ++++-- mcp_server/services/indexer.py | 15 ++++---- mcp_server/services/parser.py | 63 +++++++++++++++++++++++--------- mcp_server/storage/database.py | 63 ++++++++++++++++++++++++++++++-- mcp_server/tools/daily_update.py | 17 +++++---- mcp_server/tools/wiki_lint.py | 10 +++-- requirements.txt | 8 +++- 7 files changed, 142 insertions(+), 45 deletions(-) diff --git a/mcp_server/main.py b/mcp_server/main.py index d1bc380..92ed8bc 100644 --- a/mcp_server/main.py +++ b/mcp_server/main.py @@ -92,11 +92,11 @@ class MCPServer: wiki_query = WikiQueryTool(self.query_service) memory_bridge = MemoryBridgeTool(self.query_service) wiki_status = WikiStatusTool(self.query_service, self.indexer_service, self.graph_service) - wiki_lint = WikiLintTool(self.parser_service, self.query_service, self.graph_service) + wiki_lint = WikiLintTool(self.parser_service, self.query_service, self.graph_service, self.wiki_vault_path) cross_linker = CrossLinkerTool(self.graph_service) tag_taxonomy = TagTaxonomyTool(self.query_service) wiki_synthesize = WikiSynthesizeTool(self.query_service, self.graph_service) - daily_update = DailyUpdateTool(self.query_service, self.indexer_service) + daily_update = DailyUpdateTool(self.query_service, self.indexer_service, self.wiki_vault_path) # 注册工具 self.tools = { @@ -151,10 +151,13 @@ class MCPServer: result = await self.tools[name](**arguments) return [TextContent(type="text", text=json.dumps(result, ensure_ascii=False, indent=2))] except Exception as e: - logger.error(f"Error calling tool {name}: {e}") + # 记录完整堆栈用于调试 + logger.exception(f"Error calling tool {name} with arguments {arguments}") + # 返回用户友好的错误消息 return [TextContent(type="text", text=json.dumps({ "success": False, - "error": str(e) + "error": str(e), + "tool": name }, ensure_ascii=False))] async def run(self) -> None: diff --git a/mcp_server/services/indexer.py b/mcp_server/services/indexer.py index 631fbde..d703b7a 100644 --- a/mcp_server/services/indexer.py +++ b/mcp_server/services/indexer.py @@ -11,6 +11,7 @@ import asyncio from pathlib import Path from typing import List, Set, Optional from datetime import datetime +import aiofiles from ..storage import Database, WikiPage, compute_content_hash from .parser import ParserService @@ -31,9 +32,9 @@ class IndexerService: try: full_path = self.wiki_vault_path / page_path - # 读取文件内容 - with open(full_path, 'r', encoding='utf-8') as f: - content = f.read() + # 读取文件内容（使用 aiofiles 实现真正的异步 I/O） + async with aiofiles.open(full_path, 'r', encoding='utf-8') as f: + content = await f.read() # 解析页面 parsed = self.parser.parse_wiki_page(page_path, content) @@ -120,8 +121,8 @@ class IndexerService: for page_path in all_pages: full_path = self.wiki_vault_path / page_path try: - with open(full_path, 'r', encoding='utf-8') as f: - content = f.read() + async with aiofiles.open(full_path, 'r', encoding='utf-8') as f: + content = await f.read() current_hash = compute_content_hash(content) stored_hash = await self.db.get_page_hash(page_path) if stored_hash != current_hash: @@ -143,8 +144,8 @@ class IndexerService: for page_path in all_pages: try: full_path = self.wiki_vault_path / page_path - with open(full_path, 'r', encoding='utf-8') as f: - content = f.read() + async with aiofiles.open(full_path, 'r', encoding='utf-8') as f: + content = await f.read() current_hash = compute_content_hash(content) stored_hash = await self.db.get_page_hash(page_path) diff --git a/mcp_server/services/parser.py b/mcp_server/services/parser.py index e59f3dd..cf3f456 100644 --- a/mcp_server/services/parser.py +++ b/mcp_server/services/parser.py @@ -13,6 +13,14 @@ from typing import List, Dict, Optional, Tuple from datetime import datetime from pathlib import Path +try: + import yaml + YAML_AVAILABLE = True +except ImportError: + YAML_AVAILABLE = False + logger = logging.getLogger(__name__) + logger.warning("PyYAML not available, using simple parser") + logger = logging.getLogger(__name__) @@ -30,7 +38,7 @@ class ParserService: def parse_frontmatter(self, content: str) -> Dict: """ - 解析 frontmatter + 解析 frontmatter（使用 PyYAML） 返回格式： { @@ -45,14 +53,28 @@ class ParserService: if match: yaml_content = match.group(1) - # 简单解析（对于复杂情况，应该使用 PyYAML） - for line in yaml_content.split('\n'): - if ':' in line: - key, value = line.split(':', 1) - key = key.strip() - value = value.strip() - frontmatter[key] = value + # 使用 PyYAML 进行完整解析 + if YAML_AVAILABLE: + try: + frontmatter = yaml.safe_load(yaml_content) or {} + except yaml.YAMLError as e: + logger.warning(f"Failed to parse YAML: {e}") + frontmatter = self._simple_parse(yaml_content) + else: + frontmatter = self._simple_parse(yaml_content) + + return frontmatter + + def _simple_parse(self, yaml_content: str) -> Dict: + """简单 YAML 解析（fallback）""" + frontmatter = {} + for line in yaml_content.split('\n'): + if ':' in line: + key, value = line.split(':', 1) + key = key.strip() + value = value.strip() + frontmatter[key] = value return frontmatter def extract_links(self, content: str) -> List[str]: @@ -71,20 +93,25 @@ class ParserService: 支持： - tags: ["tag1", "tag2"] - tags: [tag1, tag2] + - YAML 列表格式 """ tags = [] - match = self.TAG_PATTERN.search(content) + frontmatter = self.parse_frontmatter(content) - if match: - tag_content = match.group(1) - # 尝试 JSON 解析 - try: - tags = json.loads(f"[{tag_content}]") - except: - # 简单逗号分隔 - tags = [t.strip().strip('"\'') for t in tag_content.split(',')] + if "tags" in frontmatter: + tags_value = frontmatter["tags"] - return tags + if isinstance(tags_value, list): + tags = tags_value + elif isinstance(tags_value, str): + # 尝试解析字符串格式的标签 + try: + tags = json.loads(tags_value) if tags_value.startswith('[') else tags_value.split(',') + except: + tags = [tags_value] + + # 确保 tags 是字符串列表 + return [str(tag).strip() for tag in tags if tag] def extract_title(self, content: str) -> str: """ diff --git a/mcp_server/storage/database.py b/mcp_server/storage/database.py index b465492..74b5dea 100644 --- a/mcp_server/storage/database.py +++ b/mcp_server/storage/database.py @@ -272,13 +272,68 @@ class Database: # === FTS5 搜索 === + def _validate_fts_query(self, query: str) -> str: + """ + 验证和清理 FTS5 查询字符串，防止注入攻击 + + FTS5 支持的特殊字符： + - 双引号：短语查询 + - *：前缀查询 + - AND, OR, NOT：布尔运算符 + + 验证规则： + - 移除不安全的控制字符 + - 限制查询长度（防止 DoS） + - 转义双引号防止短语注入 + """ + if not query: + raise ValueError("Query cannot be empty") + + # 1. 限制查询长度 + max_query_length = 500 + if len(query) > max_query_length: + logger.warning(f"Query too long ({len(query)} chars), truncating to {max_query_length}") + query = query[:max_query_length] + + # 2. 移除控制字符（除了换行、制表符） + import re + query = re.sub(r'[\x00-\x08\x0b-\x0c\x0e-\x1f\x7f]', '', query) + + # 3. 转义未闭合的双引号（防止短语查询注入） + # 计算引号数量，如果是奇数则转义最后一个 + quote_count = query.count('"') + if quote_count % 2 != 0: + # 找到最后一个引号并转义 + last_quote_idx = query.rfind('"') + query = query[:last_quote_idx] + '\\"' + query[last_quote_idx+1:] + + # 4. 防止布尔运算符注入（移除前后空格的运算符） + # 这是为了防止类似 "term AND DROP TABLE" 的攻击 + # FTS5 会在查询语法错误时返回空结果，但我们需要额外保护 + query = re.sub(r'\s+(AND|OR|NOT)\s+', ' ', query, flags=re.IGNORECASE) + + return query.strip() + async def fts_search(self, query: str, limit: int = 10) -> List[WikiPage]: """FTS5 全文搜索""" + # 验证和清理查询 + try: + safe_query = self._validate_fts_query(query) + except ValueError as e: + logger.warning(f"Invalid FTS query: {e}") + return [] + # 先从 FTS5 获取匹配的路径 - fts_rows = await self.fetch_all( - "SELECT path FROM wiki_fts WHERE wiki_fts MATCH ? ORDER BY rank LIMIT ?", - (query, limit) - ) + try: + fts_rows = await self.fetch_all( + "SELECT path FROM wiki_fts WHERE wiki_fts MATCH ? ORDER BY rank LIMIT ?", + (safe_query, limit) + ) + except aiosqlite.OperationalError as e: + # FTS5 语法错误时返回空结果（不应该崩溃） + logger.warning(f"FTS5 query failed: {e}") + return [] + paths = [row[0] for row in fts_rows] if not paths: diff --git a/mcp_server/tools/daily_update.py b/mcp_server/tools/daily_update.py index 5dc0073..733a9d2 100644 --- a/mcp_server/tools/daily_update.py +++ b/mcp_server/tools/daily_update.py @@ -7,9 +7,11 @@ Tool Layer - daily_update 工具 """ import logging +import os from typing import Dict, Any from datetime import datetime from pathlib import Path +import aiofiles from ..services import QueryService, IndexerService logger = logging.getLogger(__name__) @@ -18,9 +20,11 @@ logger = logging.getLogger(__name__) class DailyUpdateTool: """daily_update 工具实现""" - def __init__(self, query_service: QueryService, indexer_service: IndexerService): + def __init__(self, query_service: QueryService, indexer_service: IndexerService, wiki_vault_path: str = None): self.query_service = query_service self.indexer_service = indexer_service + # 优先使用传入的路径，否则使用环境变量 + self.wiki_vault_path = Path(wiki_vault_path or os.environ.get("WIKI_VAULT_PATH", "/Volumes/KnowledgeBase/wiki-vault")) async def handle(self) -> Dict[str, Any]: """ @@ -61,9 +65,8 @@ class DailyUpdateTool: async def _generate_hot_md(self, recent_pages, new_tags, orphans) -> bool: """生成热点文件""" try: - # 确定 hot.md 保存路径 - vault_path = Path(self.query_service.db.path).parent.parent / "wiki-vault" - hot_path = vault_path / "hot.md" + # 使用配置的 wiki vault 路径 + hot_path = self.wiki_vault_path / "hot.md" # 格式化内容 content = f"""# Wiki Hot - {datetime.now().strftime('%Y-%m-%d')} @@ -82,9 +85,9 @@ class DailyUpdateTool: for orphan in sorted(list(orphans))[:10]: content += f"- [[{orphan}]]\n" - # 写入文件 - with open(hot_path, 'w', encoding='utf-8') as f: - f.write(content) + # 写入文件（使用 aiofiles 实现异步 I/O） + async with aiofiles.open(hot_path, 'w', encoding='utf-8') as f: + await f.write(content) logger.info(f"Generated hot.md: {hot_path}") return True diff --git a/mcp_server/tools/wiki_lint.py b/mcp_server/tools/wiki_lint.py index 4ad26bc..24ca06a 100644 --- a/mcp_server/tools/wiki_lint.py +++ b/mcp_server/tools/wiki_lint.py @@ -7,7 +7,9 @@ Tool Layer - wiki_lint 工具 """ import logging +import os from typing import Dict, Any, List +from pathlib import Path from ..services import ParserService, QueryService, GraphService logger = logging.getLogger(__name__) @@ -16,10 +18,12 @@ logger = logging.getLogger(__name__) class WikiLintTool: """wiki_lint 工具实现""" - def __init__(self, parser: ParserService, query_service: QueryService, graph_service: GraphService): + def __init__(self, parser: ParserService, query_service: QueryService, graph_service: GraphService, wiki_vault_path: str = None): self.parser = parser self.query_service = query_service self.graph_service = graph_service + # 优先使用传入的路径，否则使用环境变量 + self.wiki_vault_path = Path(wiki_vault_path or os.environ.get("WIKI_VAULT_PATH", "/Volumes/KnowledgeBase/wiki-vault")) async def handle(self, path: str = "", level: str = "basic") -> Dict[str, Any]: """ @@ -68,9 +72,7 @@ class WikiLintTool: for page in pages: # 读取内容 try: - from pathlib import Path - vault_path = Path(self.query_service.db.path).parent.parent / "wiki-vault" - full_path = vault_path / page.path + full_path = self.wiki_vault_path / page.path with open(full_path, 'r', encoding='utf-8') as f: content = f.read() except Exception as e: diff --git a/requirements.txt b/requirements.txt index 9acafec..8a734e4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,9 +4,15 @@ mcp>=0.1.0 # 数据库 aiosqlite>=0.19.0 -# 配置和日志 +# YAML 解析 pyyaml>=6.0 +# 异步文件 I/O +aiofiles>=23.0.0 + +# 配置和日志 +# 已包含在 pyyaml 中 + # 开发依赖（可选） # pytest>=7.0.0 # pytest-asyncio>=0.21.0