第十二章：Harness 安全体系¶

# 特征向量 (简化)
features = {
    "tool_name": "bash",
    "arg_length": len(command),
    "contains_dangerous_pattern": has_rm_rf_pattern,
    "execution_count_today": 45,
    "user_approval_rate": 0.92,
    "similar_calls_approved": 23,
}

risk_score = risk_classifier.predict_proba(features)
# risk_score: 0.0 (低风险) -> 1.0 (高风险)

if risk_score > 0.8:
    mode = "manual_only"  # 高风险,完全人工
elif risk_score > 0.6:
    mode = "approve_always"  # 中高风险,每步审批
elif risk_score > 0.4:
    mode = "approve_once"  # 中等风险,任务开始审批
elif risk_score > 0.2:
    mode = "ask_first"  # 低中风险,事前询问
else:
    mode = "auto_notification"  # 低风险,自动+通知

class ToolPermission(Enum):
    """工具权限等级定义(6级信任模型)"""
    MANUAL_ONLY = "manual_only"              # 完全人工操作
    APPROVE_ALWAYS = "approve_always"        # 每步都需要审批
    APPROVE_ONCE = "approve_once"            # 整个流程开始前审批一次
    ASK_FIRST = "ask_first"                  # 首次询问,后续自动
    AUTO_WITH_NOTIFICATION = "auto_with_notification"  # 自动执行+通知
    FULL_TRUST = "full_trust"                # 完全信任

# 使用示例
@tool(permission=ToolPermission.ASK_FIRST)
def read_file(path: str) -> str:
    """读取文件"""
    return open(path).read()

# 执行流程
if tool.permission == ASK_FIRST:
    canonical_path = canonicalize_path(path)
    approval_scope = (user_id, tool.name, canonical_path, tool.risk_level)

    if approval_scope not in approved_cache:
        approval = await ask_user_permission(f"读取文件 {canonical_path}?")
        if approval:
            approved_cache.add(approval_scope)
        else:
            raise PermissionDenied()
execute_tool(tool, args)

class DecisionAuthority(Enum):
    """决策权归属：位于风险分级之上的类别化前置门，不可被风险分类器覆盖"""
    USER_ONLY = "user_only"                # 仅限用户本人决策
    USER_AUTHORIZED = "user_authorized"    # 需用户授权，且不得超出授权范围
    AGENT_AUTONOMOUS = "agent_autonomous"  # 可自主决策，用户保留知情权与最终决策权

def gate(call: "ToolCall", risk_score: float) -> PermissionLevel:
    """先按决策权归属做类别化裁决，再决定是否落入风险梯度"""
    authority = classify_authority(call)         # 只看操作类别，与 risk_score 无关
    if authority == DecisionAuthority.USER_ONLY:
        return PermissionLevel.MANUAL_ONLY       # 转账即使 risk_score≈0 也不放行
    if authority == DecisionAuthority.USER_AUTHORIZED and not within_user_grant(call):
        return PermissionLevel.MANUAL_ONLY       # 越权回落为用户本人决策
    return risk_to_level(risk_score)             # 仅自主类操作才走原有风险梯度

from enum import Enum
from typing import Optional
import logging

logger = logging.getLogger(__name__)

class PermissionDecision(Enum):
    ALLOW = "allow"
    DENY = "deny"
    ASK_USER = "ask_user"

class PermissionDecisionEngine:
    def __init__(self, policy_store: dict, audit_log: "AuditLog"):
        self.policies = policy_store  # tool_name -> ToolPermissionPolicy
        self.audit_log = audit_log
        self.user_approvals = {}  # (user_id, tool_name) -> set of approved_params

    async def decide(self,
                    tool_call: "ToolCall",
                    user_id: str,
                    context: "ExecutionContext") -> PermissionDecision:
        """
        决策流程：
        1. 参数验证
        2. 权限等级评估
        3. 用户历史记录查询
        4. 最终决策
        """

        # 步骤1: 参数Schema验证
        if not self._validate_schema(tool_call):
            logger.warning(f"Schema验证失败: {tool_call}")
            self.audit_log.log("schema_validation_failed", tool_call, user_id)
            return PermissionDecision.DENY

        # 步骤2: 获取工具权限策略
        policy = self.policies.get(tool_call.tool_name)
        if not policy:
            logger.error(f"未知工具: {tool_call.tool_name}")
            return PermissionDecision.DENY

        # 步骤3: 评估权限等级
        permission_level = policy.evaluate(tool_call, context)

        # 步骤4: 基于等级决策
        if permission_level == PermissionLevel.DENY:
            logger.warning(f"权限拒绝: {tool_call.tool_name}")
            self.audit_log.log("permission_denied", tool_call, user_id)
            return PermissionDecision.DENY

        elif permission_level == PermissionLevel.OVERRIDE:
            logger.info(f"管理员覆盖: {tool_call.tool_name}")
            self.audit_log.log("override_allowed", tool_call, user_id)
            return PermissionDecision.ALLOW

        elif permission_level == PermissionLevel.FULL_TRUST:
            logger.info(f"完全信任,自动批准: {tool_call.tool_name}")
            self.audit_log.log("full_trust_approved", tool_call, user_id)
            return PermissionDecision.ALLOW

        elif permission_level == PermissionLevel.AUTO_WITH_NOTIFICATION:
            logger.info(f"自动执行并通知: {tool_call.tool_name}")
            self.audit_log.log("auto_with_notification", tool_call, user_id)
            return PermissionDecision.ALLOW

        elif permission_level == PermissionLevel.ASK_FIRST:
            # 检查用户历史批准记录
            approval_key = (user_id, tool_call.tool_name)
            if self._is_approved_before(approval_key, tool_call.args):
                logger.info(f"自动批准 (历史): {tool_call.tool_name}")
                self.audit_log.log("ask_first_cached", tool_call, user_id)
                return PermissionDecision.ALLOW
            else:
                logger.info(f"首次调用需询问: {tool_call.tool_name}")
                return PermissionDecision.ASK_USER

        else:  # APPROVE_ONCE, APPROVE_ALWAYS, MANUAL_ONLY
            return PermissionDecision.ASK_USER

    def _validate_schema(self, tool_call: "ToolCall") -> bool:
        """验证工具调用的参数Schema"""
        # 实现略
        return True

    def _is_approved_before(self, key: tuple, args: dict) -> bool:
        """检查用户是否曾批准过类似调用"""
        # 简化实现：仅检查完全相同的调用
        if key not in self.user_approvals:
            return False
        return str(args) in self.user_approvals[key]

    def record_approval(self, user_id: str, tool_name: str, args: dict):
        """记录用户批准"""
        key = (user_id, tool_name)
        if key not in self.user_approvals:
            self.user_approvals[key] = set()
        self.user_approvals[key].add(str(args))

class AsyncPolicyEngine:
    """异步权限策略引擎"""

    async def evaluate(self, tool_call: "ToolCall", user_id: str) -> PermissionLevel:
        """异步评估权限等级(支持数据库查询、API调用)"""
        # 步骤1:异步查询用户权限缓存
        cached_level = await self._get_cached_permission(user_id, tool_call.tool_name)
        if cached_level:
            return cached_level

        # 步骤2:异步查询策略库(可能涉及数据库)
        policy = await self._fetch_policy(tool_call.tool_name)

        # 步骤3:评估并缓存结果
        level = policy.evaluate(tool_call, user_id)
        await self._cache_permission(user_id, tool_call.tool_name, level, ttl=300)
        return level

class SubAgentPermissionInheritance:
    """子智能体权限继承管理"""

    def create_sub_agent(self,
                        parent_context: "ExecutionContext",
                        sub_agent_config: dict) -> "Agent":
        """
        创建子智能体时,若子智能体需要权限提升,
        请求冒泡到父Agent或用户
        """
        sub_agent = Agent(config=sub_agent_config)
        sub_agent.parent_context = parent_context
        sub_agent.permission_mode = parent_context.permission_mode
        return sub_agent

    async def execute_with_inherited_permissions(self,
                                                sub_agent: "Agent",
                                                tool_call: "ToolCall") -> "ToolResult":
        """
        执行子智能体工具调用,通过父上下文继承权限
        """
        decision = await self.decide(tool_call, sub_agent.context)

        if decision == PermissionDecision.ASK_USER:
            # 子智能体没有独立提权，向父上下文或用户请求批准
            if sub_agent.permission_mode == sub_agent.parent_context.permission_mode:
                parent_context = sub_agent.parent_context
                # 向用户或父Agent请求批准
                approval = await self._request_parent_approval(
                    parent_context,
                    tool_call,
                    reason="Sub-agent permission requirement"
                )
                if not approval:
                    raise PermissionDenied(f"父级拒绝了权限提升")
            else:
                # 直接询问用户
                approval = await self._request_user_approval(tool_call)
                if not approval:
                    raise PermissionDenied("用户拒绝")

        return await sub_agent.execute_tool(tool_call)

import resource
import os

class ProcessSandbox:
    """进程级隔离：通过setrlimit限制资源"""

    def setup(self):
        # CPU时间限制(秒)
        resource.setrlimit(resource.RLIMIT_CPU, (30, 30))

        # 内存限制(字节)
        resource.setrlimit(resource.RLIMIT_AS, (512 * 1024 * 1024, 512 * 1024 * 1024))

        # 文件大小限制(字节)
        resource.setrlimit(resource.RLIMIT_FSIZE, (1024 * 1024 * 1024, 1024 * 1024 * 1024))

        # 打开文件数限制
        resource.setrlimit(resource.RLIMIT_NOFILE, (256, 256))

        # 子进程数限制
        resource.setrlimit(resource.RLIMIT_NPROC, (32, 32))

        # 禁止生成core dump
        resource.setrlimit(resource.RLIMIT_CORE, (0, 0))

def run_in_process_sandbox(tool_func, *args, **kwargs):
    """在沙箱内执行工具函数"""
    import json
    import subprocess

    code = """
import importlib
import json
import sys

try:
    request = json.loads(sys.stdin.read())
    module_name = request["module"]
    func_name = request["function"]
    args = request.get("args", [])
    kwargs = request.get("kwargs", {})

    if not module_name.startswith("trusted_tools."):
        raise ValueError("only trusted_tools modules are allowed")

    tool_func = getattr(importlib.import_module(module_name), func_name)
    output = tool_func(*args, **kwargs)
    print(json.dumps({"ok": True, "output": output}))
except Exception as exc:
    print(json.dumps({"ok": False, "error": str(exc)}), file=sys.stderr)
    sys.exit(1)
"""

    proc = subprocess.Popen(
        ["python", "-c", code],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        preexec_fn=ProcessSandbox().setup
    )

    stdout, stderr = proc.communicate(
        json.dumps({
            "module": tool_func.__module__,
            "function": tool_func.__name__,
            "args": args,
            "kwargs": kwargs,
        }).encode(),
        timeout=30
    )
    if proc.returncode != 0:
        message = json.loads(stderr.decode())
        raise RuntimeError(message["error"])
    return json.loads(stdout.decode())["output"]

import docker
import json

class ContainerSandbox:
    """容器级隔离：每次工具调用创建新容器"""

    def __init__(self, image: str = "python:3.11-slim"):
        self.client = docker.from_env()
        self.image = image

    def run_tool(self,
                tool_code: str,
                timeout_sec: int = 30,
                memory_limit: str = "512m") -> str:
        """在容器内运行工具"""

        # 一次性容器(auto_remove=True)
        # 受限资源(mem_limit, memswap_limit)
        # 只读文件系统除了工作目录

        container = self.client.containers.run(
            self.image,
            command=["python", "-c", tool_code],

            # 资源限制
            mem_limit=memory_limit,
            memswap_limit=memory_limit,
            cpu_quota=100000,           # CPU份额
            cpu_period=100000,

            # 文件系统隔离：镜像文件系统只读，/tmp 是容器内临时 tmpfs
            read_only=True,
            tmpfs={"/tmp": "rw,noexec,nosuid,size=64m"},

            # 安全隔离
            cap_drop=["ALL"],           # 删除所有capability
            cap_add=["NET_BIND_SERVICE"],
            security_opt=["no-new-privileges"],
            pids_limit=10,              # 进程数限制

            # 网络隔离
            network_mode="none",        # 禁用网络

            # 超时
            timeout=timeout_sec,

            # 自动清理
            auto_remove=True,
            detach=False,
        )

        # 容器自动清理
        return container.decode() if isinstance(container, bytes) else str(container)

# 使用
sandbox = ContainerSandbox()
result = sandbox.run_tool(
    "print('Hello from sandbox')",
    timeout_sec=10
)

class VMSandbox:
    """VM级隔离：最强隔离,但开销最大"""

    def __init__(self, vm_image: str):
        self.vm_image = vm_image
        # 初始化Firecracker或其他轻量级VM管理器

    def run_tool(self, tool_code: str) -> str:
        """在隔离的VM内运行工具"""
        # 1. 启动VM (通常500ms)
        # 2. 注入代码
        # 3. 执行并收集结果
        # 4. 销毁VM
        pass

class SecureToolExecutor:
    def __init__(self,
                 permission_engine: PermissionDecisionEngine,
                 sandbox: ContainerSandbox):
        self.permission_engine = permission_engine
        self.sandbox = sandbox

    async def execute(self,
                     tool_call: "ToolCall",
                     user_id: str,
                     context: "ExecutionContext") -> "ToolResult":

        # 第一层：权限决策
        decision = await self.permission_engine.decide(
            tool_call, user_id, context
        )

        if decision == PermissionDecision.DENY:
            raise PermissionDenied(f"Tool {tool_call.tool_name} denied")

        if decision == PermissionDecision.ASK_USER:
            approval = await self._ask_user(tool_call, user_id)
            if not approval:
                raise PermissionDenied("User rejected")

        # 第二层：沙箱执行
        result = self.sandbox.run_tool(
            tool_code=tool_call.code,
            timeout_sec=tool_call.timeout or 30,
            memory_limit="512m"
        )

        return ToolResult(output=result, success=True)

# 危险命令黑名单
DANGEROUS_PATTERNS = [
    "rm",           # 删除文件
    "dd",           # 低级磁盘操作
    "mkfs",         # 创建文件系统(格式化)
    "shred",        # 安全删除
    "sysctl",       # 修改内核参数
    "iptables",     # 防火墙规则修改
    "insmod",       # 加载内核模块
    "rmmod",        # 卸载内核模块
    "kill -9",      # 强制杀进程
    "reboot",       # 重启系统
    "shutdown",     # 关闭系统
    "chown",        # 修改文件所有者(权限提升)
    "chmod",        # 修改文件权限(权限提升)
    "sudo",         # 提升权限(if not in allowed_list)
    "passwd",       # 修改密码
    "useradd",      # 添加用户
    "userdel",      # 删除用户
    "groupadd",     # 添加组
    "crontab",      # 计划任务编辑(权限提升风险)
    "visudo",       # 编辑sudoers(权限提升风险)
    "at",           # 延迟任务执行(权限提升风险)
    "cryptsetup",   # 磁盘加密
    "gdisk",        # 分区表修改
    "mdadm",        # RAID配置
    "lvcreate",     # 逻辑卷创建
    "lvremove",     # 逻辑卷删除
    "zypper",       # 包管理器(权限提升风险)
    "apt",          # Debian包管理器
    "yum",          # RHEL包管理器
]

import shlex
import subprocess

class DangerousCommandDetector:
    """危险命令检测器：通过AST分析而非字符串匹配"""

    DANGEROUS_COMMANDS = {
        "rm", "dd", "mkfs", "shred", "sysctl", "iptables",
        "insmod", "rmmod", "reboot", "shutdown", "chown", "chmod",
        "sudo", "passwd", "useradd", "userdel", "cryptsetup", "gdisk",
        "crontab", "visudo", "at", "mount", "umount"
    }

    # 某些命令有安全的子命令
    SAFE_SUBCOMMANDS = {
        "apt": ["list", "search", "show"],  # 仅允许查询
        "yum": ["list", "search", "info"],
    }

    def detect(self, command: str) -> tuple[bool, str]:
        """
        检测危险操作
        返回： (is_dangerous, reason)
        """
        try:
            # 分词(处理引号、转义)
            tokens = shlex.split(command)
            if not tokens:
                return False, ""

            # 提取主命令(第一个非管道/重定向符号)
            main_cmd = self._extract_main_command(tokens)

            if not main_cmd:
                return False, ""

            # 检查黑名单
            if main_cmd in self.DANGEROUS_COMMANDS:
                return True, f"禁止的命令: {main_cmd}"

            # 检查子命令(如apt list是否安全)
            if main_cmd in self.SAFE_SUBCOMMANDS:
                if len(tokens) < 2:
                    return True, f"命令 {main_cmd} 必须有子命令"
                subcommand = tokens[1]
                if subcommand not in self.SAFE_SUBCOMMANDS[main_cmd]:
                    return True, f"不安全的子命令: {main_cmd} {subcommand}"

            # 检查管道链中的所有命令
            for cmd in self._extract_piped_commands(tokens):
                if cmd in self.DANGEROUS_COMMANDS:
                    return True, f"管道中包含危险命令: {cmd}"

            return False, ""

        except ValueError as e:
            # shlex.split失败(如引号不匹配)
            return True, f"命令解析失败: {e}"

    def _extract_main_command(self, tokens: list) -> str:
        """提取主命令(忽略路径)"""
        if not tokens:
            return ""
        cmd = tokens[0]
        # 移除路径： /bin/rm -> rm
        return cmd.split("/")[-1]

    def _extract_piped_commands(self, tokens: list) -> list:
        """提取管道链中的所有命令"""
        commands = []
        current_cmd = []

        for token in tokens:
            if token in ["|", "||", "&", "&&"]:
                if current_cmd:
                    commands.append(self._extract_main_command(current_cmd))
                current_cmd = []
            else:
                current_cmd.append(token)

        if current_cmd:
            commands.append(self._extract_main_command(current_cmd))

        return commands

# 危险命令检测使用示例
detector = DangerousCommandDetector()

test_cases = [
    "ls -la /tmp",                  # 安全
    "rm -rf /",                     # 危险
    "cat file.txt | grep pattern",  # 安全
    "ls | rm -rf /",                # 危险
    "apt list",                     # 安全(白名单子命令)
    "apt install curl",             # 危险(非白名单子命令)
]

for cmd in test_cases:
    is_dangerous, reason = detector.detect(cmd)
    print(f"{cmd}: {'危险' if is_dangerous else '安全'} - {reason}")

class ReadOnlyConstraint:
    """只读约束：防止write操作"""

    WRITE_PATTERNS = [
        r"^(rm|touch|mkdir|mv|cp|tee|sed -i|awk -i)",  # 直接修改文件
        r">",                                            # 输出重定向
        r"dd of=",                                       # 磁盘写
        r"write\(",                                      # API write调用
    ]

    def enforce(self, command: str, tool_name: str) -> bool:
        """检查命令是否违反只读约束"""
        if "read_only" not in tool_name:
            return True  # 无约束

        import re
        for pattern in self.WRITE_PATTERNS:
            if re.search(pattern, command):
                raise PermissionDenied(f"只读工具不允许: {pattern}")

        return True

# 使用
constraint = ReadOnlyConstraint()

# 这应该成功
constraint.enforce("cat /etc/passwd", tool_name="read_file_read_only")

# 这应该失败
try:
    constraint.enforce("echo malware > /etc/passwd", tool_name="read_file_read_only")
except PermissionDenied as e:
    print(f"约束违反: {e}")

from dataclasses import dataclass
from typing import Any, Callable

@dataclass
class ParameterConstraint:
    name: str
    validator: Callable[[Any], bool]
    error_message: str

class ParameterConstraintEnforcer:
    """参数范围约束"""

    def __init__(self):
        self.constraints = {
            "timeout_seconds": ParameterConstraint(
                name="timeout_seconds",
                validator=lambda x: 0 < x <= 300,  # 0-300秒
                error_message="timeout必须在0-300秒之间"
            ),
            "memory_mb": ParameterConstraint(
                name="memory_mb",
                validator=lambda x: 0 < x <= 2048,  # 0-2GB
                error_message="内存限制必须在0-2GB之间"
            ),
            "file_size_mb": ParameterConstraint(
                name="file_size_mb",
                validator=lambda x: 0 < x <= 1024,  # 0-1GB
                error_message="文件大小必须在0-1GB之间"
            ),
        }

    def validate(self, params: dict) -> None:
        """验证所有参数"""
        for param_name, constraint in self.constraints.items():
            if param_name in params:
                value = params[param_name]
                if not constraint.validator(value):
                    raise ValueError(constraint.error_message)

# 使用
enforcer = ParameterConstraintEnforcer()

# 正常参数
try:
    enforcer.validate({"timeout_seconds": 60, "memory_mb": 512})
    print("参数有效")
except ValueError as e:
    print(f"参数错误: {e}")

# 超出范围的参数
try:
    enforcer.validate({"timeout_seconds": 3600})  # 超过300秒
except ValueError as e:
    print(f"参数错误: {e}")

import signal
import subprocess
from contextlib import contextmanager

class TimeoutEnforcer:
    """超时强制：确保工具调用在规定时间内完成"""

    @contextmanager
    def enforce_timeout(self, timeout_seconds: int):
        """
        上下文管理器：强制执行超时
        """
        def timeout_handler(signum, frame):
            raise TimeoutError(f"执行超时({timeout_seconds}秒)")

        # 设置信号处理器
        old_handler = signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(timeout_seconds)

        try:
            yield
        finally:
            # 取消警报
            signal.alarm(0)
            signal.signal(signal.SIGALRM, old_handler)

    def run_with_timeout(self,
                        command: list[str],
                        timeout_seconds: int = 30) -> str:
        """运行命令并强制超时"""
        try:
            with self.enforce_timeout(timeout_seconds):
                result = subprocess.run(
                    command,
                    shell=False,
                    capture_output=True,
                    text=True,
                    timeout=timeout_seconds
                )
                return result.stdout
        except subprocess.TimeoutExpired:
            raise TimeoutError(f"命令在{timeout_seconds}秒内未完成")
        except TimeoutError as e:
            raise e

# 使用
enforcer = TimeoutEnforcer()

# 这会在30秒后被中断
try:
    result = enforcer.run_with_timeout(["sleep", "60"], timeout_seconds=5)
except TimeoutError as e:
    print(f"超时: {e}")

from typing import List, Dict, Any
import logging

logger = logging.getLogger(__name__)

class GuardrailFramework:
    """统一的护栏框架"""

    def __init__(self):
        self.dangerous_detector = DangerousCommandDetector()
        self.readonly_constraint = ReadOnlyConstraint()
        self.param_constraint = ParameterConstraintEnforcer()
        self.timeout_enforcer = TimeoutEnforcer()

    async def check_tool_call(self,
                             tool_call: "ToolCall",
                             user_id: str) -> "GuardrailCheckResult":
        """
        执行完整的护栏检查

        返回GuardrailCheckResult,包含：
        - is_safe: bool
        - violations: List[str]
        - recommendation: str (ALLOW / ASK / DENY)
        """

        violations = []

        # 1. 危险命令检测
        if hasattr(tool_call, 'command'):
            is_dangerous, reason = self.dangerous_detector.detect(tool_call.command)
            if is_dangerous:
                violations.append(f"危险命令: {reason}")

        # 2. 只读约束检查
        try:
            self.readonly_constraint.enforce(
                tool_call.command if hasattr(tool_call, 'command') else "",
                tool_call.tool_name
            )
        except PermissionDenied as e:
            violations.append(str(e))

        # 3. 参数范围检查
        try:
            self.param_constraint.validate(tool_call.args)
        except ValueError as e:
            violations.append(str(e))

        # 4. 审计记录
        logger.info(f"护栏检查: user={user_id}, tool={tool_call.tool_name}, "
                   f"violations={len(violations)}")

        # 5. 生成建议
        if violations:
            return GuardrailCheckResult(
                is_safe=False,
                violations=violations,
                recommendation="DENY"
            )
        else:
            return GuardrailCheckResult(
                is_safe=True,
                violations=[],
                recommendation="ALLOW"
            )

    async def execute_with_guardrails(self,
                                     tool_call: "ToolCall",
                                     user_id: str) -> Any:
        """带护栏的工具执行"""

        # 检查护栏
        check_result = await self.check_tool_call(tool_call, user_id)

        if not check_result.is_safe:
            logger.warning(f"护栏拒绝: {check_result.violations}")
            raise PermissionDenied(
                f"工具调用被护栏阻止: {check_result.violations[0]}"
            )

        # 执行(在超时保护下)
        try:
            result = self.timeout_enforcer.run_with_timeout(
                command=tool_call.command if hasattr(tool_call, 'command') else "",
                timeout_seconds=tool_call.args.get('timeout_seconds', 30)
            )
            logger.info(f"工具成功执行: {tool_call.tool_name}")
            return result
        except TimeoutError as e:
            logger.error(f"工具超时: {tool_call.tool_name}")
            raise

class GuardrailCheckResult:
    def __init__(self, is_safe: bool, violations: List[str], recommendation: str):
        self.is_safe = is_safe
        self.violations = violations
        self.recommendation = recommendation

class GuardrailSandwich:
    """三明治护栏模式"""

    async def execute_with_guardrails(self, action: ToolCall) -> ToolResult:
        # 第一层：输入清洗
        sanitized = await self.input_sanitizer.clean(action)

        # 第二层：有界推理
        if not self.boundary_checker.within_bounds(sanitized):
            raise BoundaryViolation(f"Action exceeds bounds: {sanitized}")

        result = await self.tool_executor.execute(sanitized)

        # 第三层：输出验证
        validated = await self.output_validator.validate(result)
        return validated

import os
import re
from pathlib import Path
from urllib.parse import unquote
import unicodedata

class PathValidator:
    """
    5层递进式路径校验(基于Claude Code pathValidation.ts)
    """

    def __init__(self, base_path: str = "/tmp/agent"):
        self.base_path = os.path.abspath(base_path)
        self.max_path_length = 4096

    def validate(self, user_path: str) -> str:
        """
        完整的路径校验流程
        返回规范化后的安全路径,或抛异常
        """

        # 第一层：长度检查
        self._check_length(user_path)

        # 第二层：URL解码(全部解码,迭代检查)
        decoded_path = self._decode_all_encodings(user_path)

        # 第三层：Unicode规范化
        normalized_path = self._normalize_unicode(decoded_path)

        # 第四层：平台特定规范化(处理反斜杠)
        platform_normalized = self._normalize_platform(normalized_path)

        # 第五层：符号链接解析与边界检查
        resolved_path = self._resolve_and_check_boundaries(platform_normalized)

        return resolved_path

    # ============ 第一层：长度检查 ============
    def _check_length(self, path: str) -> None:
        """防止过长路径导致的缓冲区溢出或DoS"""
        if len(path) > self.max_path_length:
            raise ValueError(f"路径过长: {len(path)} > {self.max_path_length}")

    # ============ 第二层：URL解码(完全)============
    def _decode_all_encodings(self, path: str) -> str:
        """
        迭代解码直到稳定,防止多重编码攻击

        例： ..%252f -> ..%2f -> ../
        """
        previous = None
        current = path

        # 迭代解码，直到输出不再变化（收敛）。使用迭代上限防止畸形输入。
        # 注意：不能用"% 数量不变"作为提前退出条件，因为 %25xx 解码为 %xx
        # 时 % 数量恰好保持不变，会导致双重编码的路径穿越绕过。
        for _ in range(20):
            if previous == current:
                break
            previous = current
            current = unquote(current)
        else:
            # 20 次仍未收敛，视为攻击输入
            raise ValueError(f"URL 解码未收敛，疑似编码攻击：{path!r}")

        return current

    # ============ 第三层：Unicode规范化 ============
    def _normalize_unicode(self, path: str) -> str:
        """
        Unicode规范化为NFC形式,防止利用不同Unicode形式的绕过

        例： é (U+00E9) 和 e (U+0065) + ◌́ (U+0301) 看起来相同
        都规范化为 NFC: é
        """
        # NFC: 标准分解形式,最常用于文件系统
        return unicodedata.normalize('NFC', path)

    # ============ 第四层：平台特定规范化 ============
    def _normalize_platform(self, path: str) -> str:
        """
        处理平台特定字符
        - Windows: 反斜杠 \ 等同于 /
        - 移除 ./ 当前目录引用
        """
        # 统一使用正斜杠
        path = path.replace('\\', '/')

        # 移除重复的 //
        while '//' in path:
            path = path.replace('//', '/')

        # 使用 os.path.normpath 安全地处理 ./ 和冗余分隔符
        # 注意：不能简单使用 str.replace('./', ''),因为会误伤合法路径
        # 如 "notes./backup" 中的 "./" 并非目录引用
        import posixpath
        path = posixpath.normpath(path)

        return path

    # ============ 第五层：符号链接解析与边界检查 ============
    def _resolve_and_check_boundaries(self, path: str) -> str:
        """
        1. 将相对路径转为绝对路径(相对于base_path)
        2. 解析所有 .. 和 符号链接
        3. 检查最终路径是否仍在base_path内
        """

        # 如果是相对路径,相对于base_path解析
        if not path.startswith('/'):
            full_path = os.path.join(self.base_path, path)
        else:
            full_path = path

        # 使用realpath解析符号链接和..
        # 这是关键步骤：会实际访问文件系统
        try:
            resolved = os.path.realpath(full_path)
        except OSError as e:
            # 如果路径不存在或无法访问,仍然进行规范化
            # 使用abspath代替realpath(不解析符号链接)
            resolved = os.path.abspath(full_path)

        # 规范化base_path
        base_resolved = os.path.realpath(self.base_path)

        # 关键检查：resolved 路径必须在 base_path 内。
        # 使用 commonpath 避免 "/tmp/agent-evil" 或 "..notes" 这类前缀误判。
        if os.path.commonpath([base_resolved, resolved]) != base_resolved:
            raise ValueError(
                f"路径逃逸: {resolved} 不在允许范围 {base_resolved} 内"
            )

        return resolved

    # 辅助方法：白名单检查
    def validate_with_whitelist(self,
                               user_path: str,
                               whitelist: list) -> str:
        """
        可选的白名单检查：仅允许特定目录
        """
        resolved = self.validate(user_path)

        for allowed_dir in whitelist:
            allowed_resolved = os.path.realpath(allowed_dir)
            if resolved.startswith(allowed_resolved + '/') or resolved == allowed_resolved:
                return resolved

        raise ValueError(f"路径 {resolved} 不在白名单内")

# 创建测试环境
import tempfile
import os

def test_path_validator():
    """测试路径校验器"""

    # 创建临时目录作为base_path
    with tempfile.TemporaryDirectory() as tmpdir:
        base = os.path.join(tmpdir, "agent")
        os.makedirs(base)

        # 创建一些文件用于测试
        safe_dir = os.path.join(base, "data")
        os.makedirs(safe_dir)
        with open(os.path.join(safe_dir, "safe.txt"), "w") as f:
            f.write("safe content")

        # 在base外创建一个要保护的文件
        protected = os.path.join(tmpdir, "protected.txt")
        with open(protected, "w") as f:
            f.write("protected content")

        # 创建符号链接(Linux)
        try:
            symlink = os.path.join(base, "link_to_protected")
            os.symlink(protected, symlink)
        except:
            symlink = None

        validator = PathValidator(base_path=base)

        # 测试用例
        test_cases = [
            # (输入, 应该成功, 描述)
            ("data/safe.txt", True, "合法的相对路径"),
            ("./data/safe.txt", True, "带./的合法路径"),
            ("data/../data/safe.txt", True, "包含..但仍在范围内"),
            ("../../../etc/passwd", False, "相对路径穿越"),
            ("..%2f..%2fetc%2fpasswd", False, "URL编码穿越"),
            ("..%252f..%252fetc%252fpasswd", False, "双重URL编码穿越"),
            ("data/../../../../../../etc/passwd", False, "多层相对穿越"),
        ]

        for user_path, should_succeed, description in test_cases:
            try:
                result = validator.validate(user_path)
                if should_succeed:
                    print(f"✓ {description}: {user_path} -> {result}")
                else:
                    print(f"✗ {description}: {user_path} 本应被阻止,但成功了")
            except ValueError as e:
                if not should_succeed:
                    print(f"✓ {description}: {user_path} 正确被阻止 ({e})")
                else:
                    print(f"✗ {description}: {user_path} 本应成功,但被拒绝了 ({e})")

# 运行测试
test_path_validator()