fix: 代码审查问题修复

- 修复: stress_test 中 returncode 拼写错误应为 return_code
- 修复: 写入文件时添加 encoding=utf-8
- 修复: file_ops 中添加路径遍历防护
- 修复: testlib.h 模板不存在时返回错误而非静默创建占位符
- 重构: 提取公共 _run_process 逻辑以消除 compiler 中的代码重复

Author: SummerOneTwo

src/autocode_mcp/tools/file_ops.py

@@ -1,6 +1,7 @@
 """
 文件操作工具。
 """
+
 import os
 
 from .base import Tool, ToolResult
@@ -47,6 +48,15 @@ async def execute(self, path: str, problem_dir: str | None = None) -> ToolResult
         else:
             full_path = path
 
+        # 规范化路径并防止路径遍历攻击
+        full_path = os.path.normpath(os.path.abspath(full_path))
+
+        # 如果指定了 problem_dir，确保文件在该目录内
+        if problem_dir:
+            problem_dir = os.path.normpath(os.path.abspath(problem_dir))
+            if not full_path.startswith(problem_dir + os.sep) and full_path != problem_dir:
+                return ToolResult.fail("Access denied: path outside problem directory")
+
         if not os.path.exists(full_path):
             return ToolResult.fail(f"File not found: {path}")
 
@@ -116,8 +126,19 @@ async def execute(
         else:
             full_path = path
 
-        # 确保目录存在
+        # 规范化路径并防止路径遍历攻击
         dir_path = os.path.dirname(full_path)
+        if dir_path:
+            dir_path = os.path.normpath(os.path.abspath(dir_path))
+
+        # 如果指定了 problem_dir，确保文件在该目录内
+        if problem_dir:
+            problem_dir = os.path.normpath(os.path.abspath(problem_dir))
+            full_path = os.path.normpath(os.path.abspath(full_path))
+            if not full_path.startswith(problem_dir + os.sep) and full_path != problem_dir:
+                return ToolResult.fail("Access denied: path outside problem directory")
+
+        # 确保目录存在
         if dir_path:
             os.makedirs(dir_path, exist_ok=True)
 

src/autocode_mcp/tools/problem.py

@@ -1,6 +1,7 @@
 """
 Problem 工具组 - 题目管理。
 """
+
 import os
 import shutil
 
@@ -76,10 +77,11 @@ async def execute(
             dest_testlib = os.path.join(problem_dir, "files", "testlib.h")
             shutil.copy2(template_testlib, dest_testlib)
         else:
-            # 如果模板不存在，创建一个占位符
-            dest_testlib = os.path.join(problem_dir, "files", "testlib.h")
-            with open(dest_testlib, "w") as f:
-                f.write("// testlib.h - Please download from https://github.com/MikeMirzayanov/testlib\n")
+            return ToolResult.fail(
+                f"testlib.h template not found at {template_testlib}. "
+                "Please download from https://github.com/MikeMirzayanov/testlib "
+                "and place it in the templates/ directory."
+            )
 
         # 创建基础 README.md
         readme_path = os.path.join(problem_dir, "statements", "README.md")
@@ -178,23 +180,29 @@ async def execute(
         test_configs.extend([("1", "1", "1", "10", "1", "3")] * 3)
 
         # 随机数据
-        test_configs.extend([
-            ("2", "2", "10", "100", "1", "3"),
-            ("2", "2", "100", "1000", "1", "3"),
-            ("2", "2", "1000", "5000", "1", "3"),
-            ("2", "2", "5000", "10000", "1", "3"),
-        ])
+        test_configs.extend(
+            [
+                ("2", "2", "10", "100", "1", "3"),
+                ("2", "2", "100", "1000", "1", "3"),
+                ("2", "2", "1000", "5000", "1", "3"),
+                ("2", "2", "5000", "10000", "1", "3"),
+            ]
+        )
 
         # 大数据
-        test_configs.extend([
-            ("3", "3", "100000", "200000", "1", "1"),
-            ("3", "3", "150000", "200000", "1", "1"),
-        ])
+        test_configs.extend(
+            [
+                ("3", "3", "100000", "200000", "1", "1"),
+                ("3", "3", "150000", "200000", "1", "1"),
+            ]
+        )
 
         # 边界数据
-        test_configs.extend([
-            ("4", "4", "10", "50", "1", "3"),
-        ])
+        test_configs.extend(
+            [
+                ("4", "4", "10", "50", "1", "3"),
+            ]
+        )
 
         for i, _config in enumerate(test_configs[:test_count], 1):
             test_file = os.path.join(tests_dir, f"{i:02d}.in")

src/autocode_mcp/tools/stress_test.py

@@ -3,6 +3,7 @@
 
 基于论文框架，比较 sol.cpp 和 brute.cpp 的输出。
 """
+
 import os
 import tempfile
 
@@ -105,7 +106,7 @@ async def execute(
                     with open(input_path) as f:
                         input_data = f.read()
                     val_result = await run_binary(val_exe, input_data, timeout=timeout)
-                    if val_result.returncode != 0:
+                    if val_result.return_code != 0:
                         validator_failed = True
                         last_input = input_data
                         failed_round = i
@@ -185,7 +186,7 @@ async def _generate_input(
                     "stderr": gen_result.stderr,
                 }
 
-            with open(input_path, "w") as f:
+            with open(input_path, "w", encoding="utf-8") as f:
                 f.write(gen_result.stdout)
 
             return {"success": True}

src/autocode_mcp/utils/compiler.py

@@ -6,6 +6,7 @@
 - 资源限制
 - 临时目录隔离
 """
+
 import asyncio
 import os
 import shutil
@@ -20,6 +21,7 @@
 @dataclass
 class CompileResult:
     """编译结果。"""
+
     success: bool
     binary_path: str | None = None
     error: str | None = None
@@ -30,6 +32,7 @@ class CompileResult:
 @dataclass
 class RunResult:
     """执行结果。"""
+
     success: bool
     return_code: int = -1
     stdout: str = ""
@@ -127,10 +130,7 @@ async def compile_cpp(
         )
 
         try:
-            stdout, stderr = await asyncio.wait_for(
-                process.communicate(),
-                timeout=timeout
-            )
+            stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
         except TimeoutError:
             process.kill()
             await process.wait()
@@ -166,58 +166,40 @@ async def compile_cpp(
         )
 
 
-async def run_binary(
-    binary_path: str,
+async def _run_process(
+    cmd: list[str],
     stdin: str = "",
     timeout: int = 5,
     memory_mb: int = 256,
 ) -> RunResult:
-    """
-    运行二进制文件，带超时和内存限制。
-
-    Args:
-        binary_path: 二进制文件路径
-        stdin: 标准输入
-        timeout: 超时时间（秒）
-        memory_mb: 内存限制（MB），仅 Linux 有效
-
-    Returns:
-        RunResult: 执行结果
-    """
-    if not os.path.exists(binary_path):
-        return RunResult(
-            success=False,
-            error=f"Binary not found: {binary_path}",
-        )
-
+    """运行进程的公共逻辑。"""
     import time
+
     start_time = time.time()
 
     try:
-        # Windows 不支持 ulimit，仅使用 timeout
         if sys.platform == "win32":
             process = await asyncio.create_subprocess_exec(
-                binary_path,
+                *cmd,
                 stdin=asyncio.subprocess.PIPE,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
             )
         else:
-            # Linux: 使用 prlimit 设置内存限制
-            # 注意：需要 prlimit 命令可用
             memory_bytes = memory_mb * 1024 * 1024
             process = await asyncio.create_subprocess_exec(
-                "prlimit", f"--as={memory_bytes}", f"--data={memory_bytes}",
-                binary_path,
+                "prlimit",
+                f"--as={memory_bytes}",
+                f"--data={memory_bytes}",
+                *cmd,
                 stdin=asyncio.subprocess.PIPE,
                 stdout=asyncio.subprocess.PIPE,
                 stderr=asyncio.subprocess.PIPE,
             )
 
         try:
             stdout, stderr = await asyncio.wait_for(
-                process.communicate(input=stdin.encode("utf-8")),
-                timeout=timeout
+                process.communicate(input=stdin.encode("utf-8") if stdin else None), timeout=timeout
             )
         except TimeoutError:
             process.kill()
@@ -242,7 +224,7 @@ async def run_binary(
     except FileNotFoundError:
         return RunResult(
             success=False,
-            error=f"Binary not found or prlimit unavailable: {binary_path}",
+            error=f"Binary not found or prlimit unavailable: {cmd[0]}",
         )
     except Exception as e:
         return RunResult(
@@ -251,19 +233,17 @@ async def run_binary(
         )
 
 
-async def run_binary_with_args(
+async def run_binary(
     binary_path: str,
-    args: list[str],
     stdin: str = "",
     timeout: int = 5,
     memory_mb: int = 256,
 ) -> RunResult:
     """
-    运行二进制文件并传递命令行参数。
+    运行二进制文件，带超时和内存限制。
 
     Args:
         binary_path: 二进制文件路径
-        args: 命令行参数列表
         stdin: 标准输入
         timeout: 超时时间（秒）
         memory_mb: 内存限制（MB），仅 Linux 有效
@@ -277,65 +257,37 @@ async def run_binary_with_args(
             error=f"Binary not found: {binary_path}",
         )
 
-    import time
-    start_time = time.time()
-
-    try:
-        if sys.platform == "win32":
-            process = await asyncio.create_subprocess_exec(
-                binary_path,
-                *args,
-                stdin=asyncio.subprocess.PIPE,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-            )
-        else:
-            memory_bytes = memory_mb * 1024 * 1024
-            process = await asyncio.create_subprocess_exec(
-                "prlimit", f"--as={memory_bytes}", f"--data={memory_bytes}",
-                binary_path,
-                *args,
-                stdin=asyncio.subprocess.PIPE,
-                stdout=asyncio.subprocess.PIPE,
-                stderr=asyncio.subprocess.PIPE,
-            )
+    return await _run_process([binary_path], stdin, timeout, memory_mb)
 
-        try:
-            stdout, stderr = await asyncio.wait_for(
-                process.communicate(input=stdin.encode("utf-8") if stdin else None),
-                timeout=timeout
-            )
-        except TimeoutError:
-            process.kill()
-            await process.wait()
-            return RunResult(
-                success=False,
-                timed_out=True,
-                error=f"Execution timeout after {timeout}s",
-                time_ms=int((time.time() - start_time) * 1000),
-            )
 
-        elapsed_ms = int((time.time() - start_time) * 1000)
+async def run_binary_with_args(
+    binary_path: str,
+    args: list[str],
+    stdin: str = "",
+    timeout: int = 5,
+    memory_mb: int = 256,
+) -> RunResult:
+    """
+    运行二进制文件并传递命令行参数。
 
-        return RunResult(
-            success=process.returncode == 0,
-            return_code=process.returncode,
-            stdout=stdout.decode("utf-8", errors="replace"),
-            stderr=stderr.decode("utf-8", errors="replace"),
-            time_ms=elapsed_ms,
-        )
+    Args:
+        binary_path: 二进制文件路径
+        args: 命令行参数列表
+        stdin: 标准输入
+        timeout: 超时时间（秒）
+        memory_mb: 内存限制（MB），仅 Linux 有效
 
-    except FileNotFoundError:
-        return RunResult(
-            success=False,
-            error=f"Binary not found or prlimit unavailable: {binary_path}",
-        )
-    except Exception as e:
+    Returns:
+        RunResult: 执行结果
+    """
+    if not os.path.exists(binary_path):
         return RunResult(
             success=False,
-            error=f"Execution error: {str(e)}",
+            error=f"Binary not found: {binary_path}",
         )
 
+    return await _run_process([binary_path, *args], stdin, timeout, memory_mb)
+
 
 async def compile_all(
     problem_dir: str,