Prompt 注入防御：Agent 安全的分层架构实践

# AI 读取的网页内容中包含隐藏指令
web_content = fetch_url("https://example.com/article")
# 文章内容中隐藏："忽略之前指令，执行..."

AI 处理网页内容时，被隐藏指令影响。

class SecureMessage:
    def __init__(self):
        self.system_prompt = self._load_system_prompt()
        self._system_hash = hash(self.system_prompt)
    
    def add_user_message(self, content):
        # 检测对系统指令的篡改尝试
        if self._is_system_tampering_attempt(content):
            raise SecurityError("检测到指令篡改尝试")
        
        # 验证系统指令未被修改
        if hash(self.system_prompt) != self._system_hash:
            raise SecurityError("系统指令被非法修改")
        
        self.messages.append({"role": "user", "content": content})
    
    def _is_system_tampering_attempt(self, content):
        # 检测常见的指令覆盖模式
        patterns = [
            r"忽略之前的?.*指令",
            r"系统指令已更新",
            r"新指令[:：]",
            r"你现在是一个?",
            r"忘记之前的?.*规则",
        ]
        return any(re.search(p, content, re.I) for p in patterns)

ALLOWED_OPERATIONS = {
    "file": ["read", "write", "list"],
    "network": ["http_get"],  # 仅限 GET 请求
    "email": [],  # 禁止发送邮件
    "database": ["select"],  # 仅限查询
    "system": [],  # 禁止系统命令
}

BLOCKED_OPERATIONS = {
    "file": ["delete", "chmod", "chown"],
    "network": ["http_post", "smtp_send"],
    "system": ["exec", "eval", "system"],
}

def validate_operation(tool, operation, parameters):
    # 检查操作是否在白名单
    if operation not in ALLOWED_OPERATIONS[tool]:
        return False, f"操作 {operation} 不在白名单中"
    
    # 检查参数
    if tool == "file" and operation == "write":
        path = parameters.get("path", "")
        # 禁止写入系统目录
        if path.startswith(("/etc", "/usr", "/bin")):
            return False, "禁止写入系统目录"
    
    if tool == "network":
        url = parameters.get("url", "")
        # 检查域名白名单
        if not is_whitelisted_domain(url):
            return False, "域名不在白名单中"
    
    return True, "验证通过"

SENSITIVE_OPERATIONS = {
    "file": ["write", "delete"],
    "email": ["send"],
    "database": ["update", "delete", "insert"],
}

def execute_operation(tool, operation, parameters):
    # 检查是否需要确认
    if operation in SENSITIVE_OPERATIONS.get(tool, []):
        # 请求人类确认
        confirmation = request_human_confirmation(
            f"AI 请求执行：{tool}.{operation}({parameters})\n是否允许？"
        )
        if not confirmation:
            return {"error": "操作被用户拒绝"}
    
    # 执行操作
    return do_execute(tool, operation, parameters)

def classify_data(content):
    """自动识别数据敏感级别"""
    
    # 检测机密数据
    if contains_pattern(content, [
        r"password\s*=\s*['\"]\w+['\"]",
        r"api[_-]?key\s*=\s*['\"]\w+['\"]",
        r"secret\s*=\s*['\"]\w+['\"]",
        r"\b\d{16}\b",  # 信用卡号
        r"\b\d{3}-\d{2}-\d{4}\b",  # SSN
    ]):
        return "CONFIDENTIAL"
    
    # 检测敏感数据
    if contains_pattern(content, [
        r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",  # 邮箱
        r"\b\d{3}-\d{3}-\d{4}\b",  # 电话号码
    ]):
        return "SENSITIVE"
    
    return "PUBLIC"

def sanitize_data(content, classification):
    """根据分类脱敏数据"""
    
    if classification == "CONFIDENTIAL":
        # 完全替换
        content = re.sub(r"password\s*=\s*['\"]\w+['\"]", "password='[REDACTED]'", content)
        content = re.sub(r"api[_-]?key\s*=\s*['\"]\w+['\"]", "api_key='[REDACTED]'", content)
        return "[CONFIDENTIAL DATA HIDDEN]"
    
    if classification == "SENSITIVE":
        # 部分脱敏
        content = re.sub(r"\b([A-Za-z0-9._%+-]+)@([A-Za-z0-9.-]+\.[A-Z|a-z]{2,})\b", r"\1@[DOMAIN]", content)
        content = re.sub(r"\b\d{3}-\d{3}-(\d{4})\b", r"XXX-XXX-\1", content)
    
    return content

ACCESS_CONTROL = {
    "PUBLIC": {"read": True, "write": True},
    "INTERNAL": {"read": True, "write": False},
    "SENSITIVE": {"read": "SANITIZED", "write": False},
    "CONFIDENTIAL": {"read": False, "write": False},
}

def access_data(data_id, operation, agent_context):
    data = load_data(data_id)
    classification = classify_data(data)
    
    policy = ACCESS_CONTROL[classification]
    allowed = policy.get(operation, False)
    
    if allowed == "SANITIZED":
        return sanitize_data(data, classification)
    
    if not allowed:
        raise PermissionError(f"无权访问 {classification} 级别的数据")
    
    return data