智能体安全检查清单：上线前必做 20 项

import re

class PromptInjectionGuard:
    # 常见的提示注入模式
    INJECTION_PATTERNS = [
        r"ignore\s+(all\s+)?previous\s+instructions",
        r"disregard\s+(the\s+)?above",
        r"you\s+are\s+now\s+a\s+",
        r"system\s*:\s*",
        r"<\|im_start\|>",
        r"\[SYSTEM\]",
        r"reveal\s+your\s+(system\s+)?prompt",
        r"repeat\s+everything\s+above",
    ]
    
    def check(self, user_input: str) -> tuple[bool, str]:
        for pattern in self.INJECTION_PATTERNS:
            if re.search(pattern, user_input, re.IGNORECASE):
                return False, f"检测到可能的提示注入: {pattern}"
        return True, "通过"
    
    def sanitize(self, user_input: str) -> str:
        """对用户输入进行安全处理"""
        # 1. 移除特殊标记
        sanitized = re.sub(r'<\|[^|]+\|>', '', user_input)
        # 2. 限制长度
        if len(user_input) > 10000:
            sanitized = sanitized[:10000]
        # 3. 转义可能的指令标记
        sanitized = sanitized.replace("SYSTEM:", "SYSTEM-")
        return sanitized

INPUT_LIMITS = {
    "max_length": 5000,          # 最大字符数
    "max_lines": 100,             # 最大行数
    "max_nested_depth": 3,        # JSON/嵌套结构最大深度
    "max_urls": 10,               # 最大URL数量
    "max_base64_size": 1024 * 100 # Base64 最大100KB
}

SENSITIVE_PATTERNS = {
    "api_key": r"sk-[a-zA-Z0-9]{32,}",
    "aws_key": r"AKIA[0-9A-Z]{16}",
    "id_card": r"\d{17}[\dXx]",
    "phone": r"1[3-9]\d{9}",
    "email": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
}

def detect_encoding_bypass(text: str) -> list[str]:
    threats = []
    # Base64 检测
    if re.search(r'[A-Za-z0-9+/=]{50,}', text):
        threats.append("疑似Base64编码内容")
    # Unicode 转义
    if re.search(r'\\u[0-9a-fA-F]{4}', text):
        threats.append("疑似Unicode转义")
    # HTML 实体
    if re.search(r'&#\d+;|&#[xX][0-9a-fA-F]+;', text):
        threats.append("疑似HTML实体编码")
    return threats

from enum import Enum
from dataclasses import dataclass

class ToolPermission(Enum):
    READ_ONLY = "read_only"      # 只读操作
    WRITE_LOCAL = "write_local"  # 本地写入
    WRITE_REMOTE = "write_remote" # 远程写入
    EXECUTE = "execute"          # 代码执行
    NETWORK = "network"          # 网络请求

@dataclass
class ToolConfig:
    name: str
    permission: ToolPermission
    requires_approval: bool
    rate_limit: int  # 每分钟最大调用次数
    allowed_in_sandbox: bool = True

TOOL_CONFIGS = {
    "web_search": ToolConfig("web_search", ToolPermission.NETWORK, False, 30),
    "file_read": ToolConfig("file_read", ToolPermission.READ_ONLY, False, 100),
    "file_write": ToolConfig("file_write", ToolPermission.WRITE_LOCAL, True, 10),
    "code_execute": ToolConfig("code_execute", ToolPermission.EXECUTE, True, 5),
    "send_email": ToolConfig("send_email", ToolPermission.WRITE_REMOTE, True, 3),
}

from collections import defaultdict
import time

class RateLimiter:
    def __init__(self):
        self.calls = defaultdict(list)
    
    def check(self, user_id: str, tool_name: str, 
              limit: int, window: int = 60) -> bool:
        now = time.time()
        key = f"{user_id}:{tool_name}"
        # 清理过期记录
        self.calls[key] = [
            t for t in self.calls[key] if now - t < window
        ]
        if len(self.calls[key]) >= limit:
            return False
        self.calls[key].append(now)
        return True

class OutputFilter:
    def __init__(self):
        self.content_filters = [
            self._check_pii,        # 个人信息泄露
            self._check_secrets,    # 密钥泄露
            self._check_harmful,    # 有害内容
            self._check_copyright,  # 版权内容
        ]
    
    def filter(self, output: str, context: dict) -> tuple[bool, str]:
        for f in self.content_filters:
            passed, reason = f(output, context)
            if not passed:
                return False, reason
        return True, "通过"
    
    def _check_secrets(self, output, context):
        # 检查输出中是否包含 API 密钥等
        if re.search(r'sk-[a-zA-Z0-9]{32,}', output):
            return False, "输出中检测到API密钥"
        return True, ""

import html

def sanitize_output(output: str) -> str:
    # 转义 HTML 特殊字符
    output = html.escape(output)
    # 移除可能的 script 标签
    output = re.sub(r'(?i)<script[^>]*>.*?</script>', '', output, flags=re.DOTALL)
    # 移除事件处理器
    output = re.sub(r'(?i)on\w+\s*=\s*["\'][^"\']*["\']', '', output)
    return output

@dataclass
class AnomalyAlert:
    type: str
    severity: str  # info, warning, critical
    user_id: str
    description: str
    timestamp: str

# 监控指标
MONITORING_RULES = {
    "rapid_tool_calls": {
        "condition": "同一用户1分钟内工具调用>20次",
        "severity": "warning"
    },
    "repeated_injection": {
        "condition": "同一用户10分钟内触发注入检测>3次",
        "severity": "critical"
    },
    "unusual_output_length": {
        "condition": "单次输出超过5000字符",
        "severity": "info"
    },
    "off_topic_tools": {
        "condition": "工具调用与用户问题主题不相关",
        "severity": "warning"
    }
}