MCP是手腳，Skill是靈魂· 第四篇:別讓AI瞎操作

#!/usr/bin/env python3
"""sandbox_skill.py - MCP安全防火牆沙箱Skill基類：用於隔離AI執行危險Shell/文件系統操作的權限圍欄"""
import os
import re
import time
import shlex
import logging
import subprocess
import resource
from pathlib import Path
from typing import List, Optional
from dataclasses import dataclass, field

audit_logger = logging.getLogger("sandbox.audit")

@dataclass
class SandboxPolicy:
 """沙箱策略配置"""
 allowed_commands: List[str] = field(default_factory=list)
 allowed_paths: List[str] = field(default_factory=list)
 forbidden_patterns: List[str] = field(default_factory=lambda: [
 r'rm\s+-rf',
 r'\.\.\/',
 r';\s*rm',
 r'\|\s*sh',
 r'\$\(.*\)',
 r'`.*`',
 r'>\s*/dev/',
 r'chmod\s+777',
 r'sudo',
 r'/etc/passwd',
 r'/etc/shadow',
 ])
 timeout_seconds: int = 30
 max_output_bytes: int = 1024 * 1024
 max_memory_bytes: int = 256 * 1024 * 1024

@dataclass
class ExecutionResult:
 """命令執行結果"""
 success: bool
 stdout: str = ""
 stderr: str = ""
 return_code: int = -1
 blocked: bool = False
 block_reason: str = ""
 execution_time_ms: float = 0.0

class SandboxSkill:
 """
 沙箱Skill基類
 核心原則：
 1. 白名單優先：未明確允許的，一律拒絕
 2. 路徑歸一化：先解析真實路徑，再做邊界檢查
 3. 確定性安全：安全檢查用代碼邏輯，不依賴AI判斷
 4. 全程審計：每一步操作都留下記錄
 """
 SKILL_NAME: str = "base_sandbox"
 POLICY: SandboxPolicy = SandboxPolicy()

 def __init__(self, session_id: str, user_id: str):
 self.session_id = session_id
 self.user_id = user_id

 def _check_forbidden_patterns(self, command: str) -> Optional[str]:
 for pattern in self.POLICY.forbidden_patterns:
 if re.search(pattern, command, re.IGNORECASE):
 return f"命令包含禁止模式: {pattern}"
 return None

 def _check_command_whitelist(self, command: str) -> Optional[str]:
 try:
 parts = shlex.split(command)
 except ValueError as e:
 return f"命令解析失敗（可能存在注入）: {e}"
 if not parts:
 return "空命令被拒絕"
 cmd_name = os.path.basename(parts[0])
 if cmd_name not in self.POLICY.allowed_commands:
 return (
 f"命令 '{cmd_name}' 不在白名單中。"
 f"允許的命令: {', '.join(self.POLICY.allowed_commands)}"
 )
 return None

 def _check_path_boundary(self, path: str) -> Optional[str]:
 if not self.POLICY.allowed_paths:
 return "沒有配置允許的路徑，操作被拒絕"
 real_path = os.path.realpath(path)
 for allowed_root in self.POLICY.allowed_paths:
 real_root = os.path.realpath(allowed_root)
 try:
 common = os.path.commonpath([real_path, real_root])
 if common == real_root:
 return None
 except ValueError:
 continue
 return (
 f"路徑 '{path}'（真實路徑: '{real_path}'）"
 f"超出允許範圍: {self.POLICY.allowed_paths}"
 )

 def _set_resource_limits(self):
 resource.setrlimit(
 resource.RLIMIT_AS,
 (self.POLICY.max_memory_bytes, self.POLICY.max_memory_bytes)
 )
 resource.setrlimit(resource.RLIMIT_CORE, (0, 0))

 def safe_execute(self, command: str, cwd: Optional[str] = None) -> ExecutionResult:
 start_time = time.monotonic()
 checks = [
 self._check_forbidden_patterns(command),
 self._check_command_whitelist(command),
 self._check_path_boundary(cwd) if cwd else None,
 ]
 for reason in checks:
 if reason:
 result = ExecutionResult(
 success=False, blocked=True, block_reason=reason
 )
 audit_logger.warning(
 f"[{self.SKILL_NAME}] BLOCKED | {reason} | user={self.user_id}"
 )
 return result
 safe_env = {"PATH": "/usr/local/bin:/usr/bin:/bin", "HOME": "/tmp"}
 try:
 proc = subprocess.run(
 shlex.split(command),
 shell=False,
 capture_output=True,
 text=True,
 timeout=self.POLICY.timeout_seconds,
 cwd=cwd,
 env=safe_env,
 preexec_fn=self._set_resource_limits,
 )
 elapsed = (time.monotonic() - start_time) * 1000
 stdout = proc.stdout
 if len(stdout.encode()) > self.POLICY.max_output_bytes:
 stdout = stdout[:self.POLICY.max_output_bytes] + "\n[輸出已截斷]"
 result = ExecutionResult(
 success=proc.returncode == 0,
 stdout=stdout,
 stderr=proc.stderr[:4096],
 return_code=proc.returncode,
 execution_time_ms=elapsed
 )
 except subprocess.TimeoutExpired:
 result = ExecutionResult(
 success=False, blocked=True,
 block_reason=f"執行超時（限制: {self.POLICY.timeout_seconds}秒）",
 execution_time_ms=(time.monotonic() - start_time) * 1000
 )
 except Exception as e:
 result = ExecutionResult(success=False, stderr=str(e))
 audit_logger.info(
 f"[{self.SKILL_NAME}] EXECUTE | "
 f"blocked={result.blocked} | user={self.user_id} | "
 f"time={result.execution_time_ms:.1f}ms"
 )
 return result

 def safe_read_file(self, path: str, max_size_bytes: int = 102400) -> ExecutionResult:
 block_reason = self._check_path_boundary(path)
 if block_reason:
 result = ExecutionResult(success=False, blocked=True, block_reason=block_reason)
 audit_logger.warning(f"[{self.SKILL_NAME}] FILE_READ_BLOCKED | {block_reason}")
 return result
 try:
 file_path = Path(os.path.realpath(path))
 if not file_path.is_file():
 return ExecutionResult(success=False, stderr=f"路徑不是文件: {path}")
 size = file_path.stat().st_size
 if size > max_size_bytes:
 return ExecutionResult(
 success=False, blocked=True,
 block_reason=f"文件大小({size}字節)超過限制({max_size_bytes}字節)"
 )
 content = file_path.read_text(errors='replace')
 return ExecutionResult(success=True, stdout=content)
 except Exception as e:
 return ExecutionResult(success=False, stderr=str(e))