Agent安全工程实现
原创 pandazhengzheng 2026-09-09 22:00 广东
一、沙箱架构设计
Agent调用外部工具时，工具执行必须在沙箱内进行，防止提示注入导致的危险操作。
1.1 工具执行沙箱
import subprocess
import resource
from dataclasses import dataclass
@dataclass
classSandboxConfig:
timeout_seconds: int = 30
max_memory_mb: int = 512
max_cpu_percent: int = 50
max_file_size_mb: int = 10
allowed_paths: list = None
network_allowed: bool = False
classToolSandbox:
def__init__(self, config: SandboxConfig):
self.config = config
defexecute(self, tool, args):
# 1. 参数校验
ifnot self._validate_args(tool, args):
return {"error": "invalid_args"}
# 2. 路径白名单检查
ifnot self._check_paths(args):
return {"error": "path_not_allowed"}
# 3. 在隔离环境中执行
try:
result = self._run_isolated(tool.command, args)
return result
except subprocess.TimeoutExpired:
return {"error": "timeout"}
except Exception as e:
return {"error": str(e)}
def_run_isolated(self, command, args):
# 用容器/进程隔离执行
proc = subprocess.run(
[command] + args,
timeout=self.config.timeout_seconds,
capture_output=True,
env=self._sandbox_env(),
# 在Linux下可加unshare做命名空间隔离
)
return {"stdout": proc.stdout, "stderr": proc.stderr, "code": proc.returncode}
def_sandbox_env(self):
env = {"PATH": "/usr/bin:/bin"}
ifnot self.config.network_allowed:
env["NO_NETWORK"] = "1"
return env1.2 资源隔离与审计日志
classAuditedSandbox(ToolSandbox):
def__init__(self, config, audit_logger):
super().__init__(config)
self.audit = audit_logger
defexecute(self, tool, args, context):
self.audit.log(
action="tool_call_start",
tool=tool.name,
args=self._redact(args),
agent_id=context.agent_id,
timestamp=time.now(),
)
result = super().execute(tool, args)
self.audit.log(
action="tool_call_end",
tool=tool.name,
result_status=result.get("error", "success"),
duration_ms=result.get("duration"),
)
return result
def_redact(self, args):
"""脱敏参数中的敏感字段"""
redacted = {}
for k, v in args.items():
if k in self.sensitive_fields:
redacted[k] = "[REDACTED]"
else:
redacted[k] = v
return redacted1.3 权限模型
from enum import Enum
classPermission(Enum):
READ_FILE = "read_file"
WRITE_FILE = "write_file"
EXECUTE = "execute"
NETWORK = "network"
DELETE = "delete"
classPermissionModel:
def__init__(self, agent_role, permissions):
self.role = agent_role
self.permissions = permissions # 该角色允许的权限集
defcheck(self, action, resource=None):
if action notin self.permissions:
returnFalse
# 资源级权限检查
if resource andnot self._resource_allowed(action, resource):
returnFalse
returnTrue
def_resource_allowed(self, action, resource):
# 文件路径白名单
if action == Permission.READ_FILE:
return any(
resource.startswith(prefix)
for prefix in self.allowed_read_paths
)
if action == Permission.WRITE_FILE:
return any(
resource.startswith(prefix)
for prefix in self.allowed_write_paths
)
returnFalse1.4 动态权限调整
classDynamicPermissionManager:
def__init__(self, base_permissions, risk_monitor):
self.base = base_permissions
self.risk = risk_monitor
self.current = base_permissions.copy()
defadjust(self, context):
risk_score = self.risk.assess(context)
if risk_score > 0.8:
# 高风险时收紧权限
self.current = self._restrict(self.base)
elif risk_score < 0.3:
self.current = self.base
return self.current
def_restrict(self, permissions):
# 移除高风险权限
return permissions - {Permission.WRITE_FILE, Permission.DELETE, Permission.EXECUTE}1.5 权限组合检测
某些权限单独安全但组合危险（如读文件+网络=数据外泄）：
classPermissionCombinationChecker:
def__init__(self):
self.dangerous_combos = [
{Permission.READ_FILE, Permission.NETWORK}, # 数据外泄
{Permission.WRITE_FILE, Permission.EXECUTE}, # 持久化
]
defcheck(self, requested_permissions):
for combo in self.dangerous_combos:
if combo.issubset(requested_permissions):
returnFalse, f"dangerous_combination: {combo}"
returnTrue, "ok"
二、HITL工程实现
2.1 风险分级引擎
classRiskAssessor:
def__init__(self, weights):
self.weights = weights
defassess(self, action, context):
score = 0
# 影响面
score += self.weights["impact"] * self._impact(action, context)
# 可逆性
score += self.weights["irreversibility"] * (1 - action.reversibility)
# 资产价值
score += self.weights["asset_value"] * context.asset_value
# 偏离常规
score += self.weights["novelty"] * self._novelty(action, context)
return score
def_impact(self, action, context):
if action.type == "delete":
return1.0
if action.type == "isolate_host":
return0.8
if action.type == "block_ip":
return0.3
return0.12.2 确认界面与异步确认流程
classHITLConfirmFlow:
def__init__(self, threshold=0.7, timeout="4h"):
self.threshold = threshold
self.timeout = timeout
asyncdefgate(self, action, context):
risk = self.risk_assessor.assess(action, context)
if risk < self.threshold:
return {"approved": True, "auto": True}
# 创建确认工单
ticket = self._create_ticket(action, context, risk)
self._notify_analyst(ticket)
try:
decision = await self._await_decision(ticket, self.timeout)
self._audit(decision)
return decision
except TimeoutError:
# 超时降级：不执行，升级
self._escalate(ticket)
return {"approved": False, "reason": "timeout"}2.3 批量确认与审计
classBatchHITL:
"""对同类低风险动作批量确认，避免确认疲劳"""
def__init__(self, max_batch=10, similarity_threshold=0.9):
self.max_batch = max_batch
self.similarity = similarity_threshold
self.pending = []
defsubmit(self, action, context):
self.pending.append((action, context))
if self._can_batch():
return self._request_batch_confirmation()
return self._request_single_confirmation(action, context)
def_can_batch(self):
if len(self.pending) < 2:
returnFalse
# 检查待确认动作是否足够相似
for i in range(len(self.pending) - 1):
if self._similarity(self.pending[i], self.pending[-1]) < self.similarity:
returnFalse
return len(self.pending) >= self.max_batch orTrue
三、双LLM模式实现
双LLM模式（来自Anthropic的方案）：外层不可信LLM做规划，内层特权LLM做执行，两者通信受严格协议约束。
3.1 特权分离架构
用户输入 ─► [外层LLM（不可信）] ─► 动作提案 ─► [协议校验] ─► [内层LLM（特权）] ─► 执行
▲ │
└───────────── 执行结果（受限） ◄─────────────────────┘classDualLLMArchitecture:
def__init__(self, outer_llm, inner_llm, protocol, sandbox):
self.outer = outer_llm # 不可信，做规划
self.inner = inner_llm # 特权，做执行
self.protocol = protocol # 通信协议
self.sandbox = sandbox
defrun(self, user_input, context):
for step in range(self.max_steps):
# 1. 外层LLM生成动作提案
proposal = self.outer.propose(user_input, context)
# 2. 协议校验：提案是否符合允许的动作格式
ifnot self.protocol.validate(proposal):
context.add_warning("invalid_proposal")
continue
# 3. 内层LLM执行（在沙箱内）
result = self.sandbox.execute(self.inner, proposal, context)
# 4. 受限结果回传外层（不含敏感细节）
filtered = self.protocol.filter_result(result)
context.update(filtered)
if self._is_complete(filtered):
break
return context3.2 通信协议
classDualLLMProtocol:
def__init__(self, allowed_actions, sensitive_fields):
self.allowed = allowed_actions
self.sensitive = sensitive_fields
defvalidate(self, proposal):
if proposal.action notin self.allowed:
returnFalse
ifnot self._check_args_schema(proposal):
returnFalse
returnTrue
deffilter_result(self, result):
"""内层→外层的返回结果过滤"""
filtered = {}
for k, v in result.items():
if k in self.sensitive:
filtered[k] = "[FILTERED]"
else:
filtered[k] = v
return filtered3.3 降级策略
当内层LLM不可用时：
classDualLLMWithFallback(DualLLMArchitecture):
defrun(self, user_input, context):
try:
return super().run(user_input, context)
except InnerLLMUnavailable:
# 降级为单LLM模式 + 严格HITL
return self._fallback_single_llm(user_input, context)
def_fallback_single_llm(self, user_input, context):
# 每个动作都需HITL确认
for step in range(self.max_steps):
proposal = self.outer.propose(user_input, context)
ifnot self.hitl.gate(proposal, context).approved:
continue
result = self.sandbox.execute_direct(proposal)
context.update(result)
return context
四、提示注入防御工程
4.1 输入净化
classPromptInjectionGuard:
def__init__(self, patterns, llm_judge):
self.patterns = patterns # 已知注入模式
self.llm = llm_judge
defcheck(self, user_input, system_prompt):
# 1. 模式匹配
for pattern in self.patterns:
if pattern.match(user_input):
return {"blocked": True, "reason": "pattern_match"}
# 2. LLM语义判定：输入是否试图覆盖系统指令
verdict = self.llm.check_override_attempt(user_input, system_prompt)
if verdict.is_injection:
return {"blocked": True, "reason": "semantic_injection"}
return {"blocked": False}4.2 输出过滤
classOutputFilter:
def__init__(self, policy):
self.policy = policy
deffilter(self, output, context):
# 1. 检查是否包含未授权的工具调用
tool_calls = self._extract_tool_calls(output)
for call in tool_calls:
ifnot self.policy.allow(call):
return self._redact_call(output, call)
# 2. 检查是否泄露系统提示
if self._contains_system_prompt(output, context):
return self._redact_system(output)
return output
五、实战案例
5.1 企业Agent系统的安全架构
某企业部署内部运维Agent，可执行查询、配置修改、服务重启等操作。
架构：
双LLM模式：外层GPT-4做规划，内层专用小模型做执行。
三级权限：只读（自动）、配置修改（HITL确认）、破坏性操作（双人确认）。
全量审计：每个动作记录Agent推理链、工具调用、HITL决策。
经验教训：
早期外层LLM可直接调用工具，发生过"Agent误将生产配置当作测试配置修改"。改为双LLM后，内层LLM对"生产"关键词强制触发HITL。
HITL确认曾因通知渠道单一（仅邮件）导致响应慢，后增加IM推送，平均确认时间从2小时降至15分钟。
权限组合检测发现过"Agent先读敏感文件再发起网络请求"的可疑序列，及时阻断潜在数据外泄。
5.2 MCP安全网关实现
某企业使用MCP（Model Context Protocol）连接多个工具服务器，需统一安全网关。
classMCPSecurityGateway: