fix: 代码审查问题修复

- 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 <noreply@anthropic.com>

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:

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
 
-            # 读取文件内容
+            # 读取文件内容（使用 aiofiles 实现真正的异步 I/O）
-            with open(full_path, 'r', encoding='utf-8') as f:
+            async with aiofiles.open(full_path, 'r', encoding='utf-8') as f:
-                content = f.read()
+                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:
+                async with aiofiles.open(full_path, 'r', encoding='utf-8') as f:
-                    content = f.read()
+                    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:
+                    async with aiofiles.open(full_path, 'r', encoding='utf-8') as f:
-                        content = f.read()
+                        content = await f.read()
                     current_hash = compute_content_hash(content)
                     stored_hash = await self.db.get_page_hash(page_path)
 

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:
+        if "tags" in frontmatter:
-            tag_content = match.group(1)
+            tags_value = frontmatter["tags"]
-            # 尝试 JSON 解析
-            try:
-                tags = json.loads(f"[{tag_content}]")
-            except:
-                # 简单逗号分隔
-                tags = [t.strip().strip('"\'') for t in tag_content.split(',')]
 
-        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:
         """

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(
+        try:
-            "SELECT path FROM wiki_fts WHERE wiki_fts MATCH ? ORDER BY rank LIMIT ?",
+            fts_rows = await self.fetch_all(
-            (query, limit)
+                "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:

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 保存路径
+            # 使用配置的 wiki vault 路径
-            vault_path = Path(self.query_service.db.path).parent.parent / "wiki-vault"
+            hot_path = self.wiki_vault_path / "hot.md"
-            hot_path = 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"
 
-            # 写入文件
+            # 写入文件（使用 aiofiles 实现异步 I/O）
-            with open(hot_path, 'w', encoding='utf-8') as f:
+            async with aiofiles.open(hot_path, 'w', encoding='utf-8') as f:
-                f.write(content)
+                await f.write(content)
 
             logger.info(f"Generated hot.md: {hot_path}")
             return True

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
+                full_path = self.wiki_vault_path / page.path
-                vault_path = Path(self.query_service.db.path).parent.parent / "wiki-vault"
-                full_path = vault_path / page.path
                 with open(full_path, 'r', encoding='utf-8') as f:
                     content = f.read()
             except Exception as e:

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