文章总结： 本文介绍agent安全工程实现，重点包括沙箱架构设计（工具执行沙箱、资源隔离与审计日志、权限模型及动态调整、权限组合检测）以及HITL工程实现（风险分级引擎、确认流程与批量确认）。核心思想是通过沙箱隔离、权限最小化、风险分级和人工审批来保障agent调用外部工具的安全性，防止提示注入和危险操作。 综合评分： 82 文章分类： 安全开发,安全建设,应用安全
Agent安全工程实现
原创
pandazhengzheng pandazhengzheng
安全分析与研究
2026年9月9日 22:00 广东
在小说阅读器读本章
去阅读
在公众号小说中沉浸阅读
一、沙箱架构设计
Agent调用外部工具时，工具执行必须在沙箱内进行，防止提示注入导致的危险操作。
1.1 工具执行沙箱
import subprocess
import resource
from dataclasses import dataclass
@dataclass
class SandboxConfig:
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
class ToolSandbox:
def __init__(self, config: SandboxConfig):
self.config = config
def execute(self, tool, args):
# 1. 参数校验
if not self._validate_args(tool, args):
return {"error": "invalid_args"}
# 2. 路径白名单检查
if not 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"}
if not self.config.network_allowed:
env["NO_NETWORK"] = "1"
return env
1.2 资源隔离与审计日志
class AuditedSandbox(ToolSandbox):
def __init__(self, config, audit_logger):
super().__init__(config)
self.audit = audit_logger
def execute(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 redacted
1.3 权限模型
from enum import Enum
class Permission(Enum):
READ_FILE = "read_file"
WRITE_FILE = "write_file"
EXECUTE = "execute"
NETWORK = "network"
DELETE = "delete"
class PermissionModel:
def __init__(self, agent_role, permissions):
self.role = agent_role
self.permissions = permissions # 该角色允许的权限集
def check(self, action, resource=None):
if action not in self.permissions:
return False
# 资源级权限检查
if resource and not self._resource_allowed(action, resource):
return False
return True
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
)
return False
1.4 动态权限调整
class DynamicPermissionManager:
def __init__(self, base_permissions, risk_monitor):
self.base = base_permissions
self.risk = risk_monitor
self.current = base_permissions.copy()
def adjust(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 权限组合检测
某些权限单独安全但组合危险（如读文件+网络=数据外泄）：
class PermissionCombinationChecker:
def __init__(self):
self.dangerous_combos = [
{Permission.READ_FILE, Permission.NETWORK}, # 数据外泄
{Permission.WRITE_FILE, Permission.EXECUTE}, # 持久化
]
def check(self, requested_permissions):
for combo in self.dangerous_combos:
if combo.issubset(requested_permissions):
return False, f"dangerous_combination: {combo}"
return True, "ok"
二、HITL工程实现
2.1 风险分级引擎
class RiskAssessor:
def __init__(self, weights):
self.weights = weights
def assess(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":
return 1.0
if action.type == "isolate_host":
return 0.8
if action.type == "block_ip":
return 0.3
return 0.1
2.2 确认界面与异步确认流程
class HITLConfirmFlow:
def __init__(self, threshold=0.7, timeout="4h"):
self.threshold = threshold
self.timeout = timeout
async def gate(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 批量确认与审计
class BatchHITL:
"""对同类低风险动作批量确认，避免确认疲劳"""
def __init__(self, max_batch=10, similarity_threshold=0.9):
self.max_batch = max_batch
self.similarity = similarity_threshold
self.pending = []
def submit(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:
return False
# 检查待确认动作是否足够相似
for i in range(len(self.pending) - 1):
if self._similarity(self.pending[i], self.pending[-1]) < self.similarity:
return False
return len(self.pending) >= self.max_batch or True
三、双LLM模式实现
双LLM模式（来自Anthropic的方案）：外层不可信LLM做规划，内层特权LLM做执行，两者通信受严格协议约束。
3.1 特权分离架构
用户输入 ─► [外层LLM（不可信）] ─► 动作提案 ─► [协议校验] ─► [内层LLM（特权）] ─► 执行
▲ │
└───────────── 执行结果（受限） ◄─────────────────────┘
class DualLLMArchitecture:
def __init__(self, outer_llm, inner_llm, protocol, sandbox):
self.outer = outer_llm # 不可信，做规划
self.inner = inner_llm # 特权，做执行
self.protocol = protocol # 通信协议
self.sandbox = sandbox
def run(self, user_input, context):
for step in range(self.max_steps):
# 1. 外层LLM生成动作提案
proposal = self.outer.propose(user_input, context)
# 2. 协议校验：提案是否符合允许的动作格式
if not 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 context
3.2 通信协议
class DualLLMProtocol:
def __init__(self, allowed_actions, sensitive_fields):
self.allowed = allowed_actions
self.sensitive = sensitive_fields
def validate(self, proposal):
if proposal.action not in self.allowed:
return False
if not self._check_args_schema(proposal):
return False
return True
def filter_result(self, result):
"""内层→外层的返回结果过滤"""
filtered = {}
for k, v in result.items():
if k in self.sensitive:
filtered[k] = "[FILTERED]"
else:
filtered[k] = v
return filtered
3.3 降级策略
当内层LLM不可用时：
class DualLLMWithFallback(DualLLMArchitecture):
def run(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)
if not self.hitl.gate(proposal, context).approved:
continue
result = self.sandbox.execute_direct(proposal)
context.update(result)
return context
四、提示注入防御工程
4.1 输入净化
class PromptInjectionGuard:
def __init__(self, patterns, llm_judge):
self.patterns = patterns # 已知注入模式
self.llm = llm_judge
def check(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 输出过滤
class OutputFilter:
def __init__(self, policy):
self.policy = policy
def filter(self, output, context):
# 1. 检查是否包含未授权的工具调用
tool_calls = self._extract_tool_calls(output)
for call in tool_calls:
if not 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）连接多个工具服务器，需统一安全网关。
class MCPSecurityGateway:
`
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：安全分析与研究 pandazhengzheng pandazhengzheng《Agent安全工程实现》
版权声明
本站仅做备份收录，仅供研究与教学参考之用。
读者将信息用于其他用途的，全部法律及连带责任由读者自行承担，本站不承担任何责任。
评论