通过预训练投毒实现的代理后门攻击

import os
import requests
from typing import Dict, Any

class AgentSecurityGateway:
    def __init__(self, api_key: str, endpoint: str = "https://api.n1n.ai/v1/chat/completions"):
        self.api_key = api_key
        self.endpoint = endpoint
        self.blocked_patterns = ["rm -rf", "DROP TABLE", "--bypass-auth"]

    def validate_tool_payload(self, tool_call_args: str) -> bool:
        for pattern in self.blocked_patterns:
            if pattern.lower() in tool_call_args.lower():
                print(f"[SECURITY ALERT] 检测到恶意模式: {pattern}")
                return False
        return True

    def execute_completion(self, model: str, messages: list) -> Dict[str, Any]:
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        payload = {
            "model": model,
            "messages": messages,
            "temperature": 0.1
        }
        
        response = requests.post(self.endpoint, json=payload, headers=headers)
        if response.status_code != 200:
            raise RuntimeError(f"API 错误: {response.text}")
            
        data = response.json()
        # 模拟对工具调用注入的检查
        completion_text = data["choices"][0]["message"].get("content", "")
        if not self.validate_tool_payload(completion_text):
            return {"error": "由于违反安全策略，执行已被中止。"}
            
        return data

# 使用 n1n.ai 的示例
gateway = AgentSecurityGateway(api_key=os.getenv("N1N_API_KEY"))
safe_response = gateway.execute_completion(
    model="claude-3-5-sonnet",
    messages=[{"role": "user", "content": "重构此数据库迁移脚本。"}]
})
print(safe_response)