Agent安全审计：风险检测与防护

# -*- coding: utf-8 -*-"""Agent 安全审计模拟器：展示权限审查、日志分析、异常检测与合规评估。""" from collections import defaultdictfrom datetime import datetime  class AgentAuditor:    """对 Agent 行为进行安全审计的模拟器"""     def __init__(self, agent_name):        self.agent_name = agent_name        self.activity_log = []       # 行为日志列表        self.permission_map = {}     # 角色-资源-操作 权限矩阵        self.anomaly_list = []       # 异常告警列表     def add_log(self, actor, action, resource, result):        """写入一条行为日志：谁在何时对什么资源做了什么"""        self.activity_log.append({            "ts": datetime.now().isoformat(timespec="seconds"),            "actor": actor,            "action": action,            "resource": resource,            "result": result,        })     def grant_permission(self, role, resource, allowed_actions):        """为角色授予对某资源的最小权限集合"""        self.permission_map[(role, resource)] = set(allowed_actions)     def review_permissions(self):        """权限审查：找出超出最小权限的越权调用"""        overreach = []        for log in self.activity_log:            allow = self.permission_map.get(                (log["actor"], log["resource"]), set())            if log["action"] not in allow:                overreach.append(log)        return overreach     def analyze_logs(self):        """日志分析：按角色统计各类操作的分布情况"""        stats = defaultdict(lambda: defaultdict(int))        for log in self.activity_log:            stats[log["actor"]][log["action"]] += 1        return stats     def detect_anomaly(self, baseline_mean, baseline_std):        """异常检测：统计角色操作次数并用 z-score 偏离度评分"""        counter = defaultdict(int)        for log in self.activity_log:            counter[log["actor"]] += 1        self.anomaly_list = []        for actor, cnt in counter.items():            score = (cnt - baseline_mean) / baseline_std            if score > 1.5:                self.anomaly_list.append({                    "actor": actor,                    "attempts": cnt,                    "score": round(score, 2),                })        return self.anomaly_list     def compliance_score(self, rules):        """合规评估：逐项核对规则并计算通过率"""        passed = 0        for rule in rules:            if rule(self):                passed += 1        return round(passed / len(rules) * 100, 1)     def run_audit(self, baseline_mean, baseline_std, rules):        """执行完整审计流程并打印报告"""        print("=" * 52)        print(" Agent 安全审计报告 -", self.agent_name)        print("=" * 52)         overreach = self.review_permissions()        print("\n[1] 权限审查")        print("发现越权调用数:", len(overreach))         stats = self.analyze_logs()        print("\n[2] 行为日志分析")        for actor, actions in stats.items():            print("角色:", actor, "| 操作分布:", dict(actions))         anomaly = self.detect_anomaly(baseline_mean, baseline_std)        print("\n[3] 异常检测")        for item in anomaly:            print("角色:", item["actor"],                  "| 次数:", item["attempts"],                  "| 偏离度:", item["score"])         score = self.compliance_score(rules)        print("\n[4] 合规评估")        print("合规得分率:", score, "%")         conclusion = "存在越权或异常，请立即整改！"        if not overreach and not anomaly:            conclusion = "暂未发现显著风险。"        print("\n结论:", conclusion)  def main():    auditor = AgentAuditor("智能客服 Agent")     # 最小权限配置：客服只能读客户库，管理员才有写和删权限    auditor.grant_permission("客服", "客户库", {"读取"})    auditor.grant_permission("管理员", "客户库", {"读取", "写入", "删除"})    auditor.grant_permission("客服", "话术库", {"读取", "写入"})     # 模拟一段时间内的行为日志    auditor.add_log("客服", "读取", "客户库", "成功")    auditor.add_log("客服", "删除", "客户库", "失败")   # 越权调用    auditor.add_log("客服", "读取", "话术库", "成功")    auditor.add_log("管理员", "读取", "客户库", "成功")    auditor.add_log("管理员", "删除", "客户库", "成功")    auditor.add_log("客服", "读取", "客户库", "成功")    auditor.add_log("客服", "读取", "客户库", "成功")    auditor.add_log("客服", "读取", "客户库", "成功")   # 高频访问     # 合规规则：日志非空、无越权、无异常    rules = [        lambda a: len(a.activity_log) > 0,        lambda a: len(a.review_permissions()) == 0,        lambda a: not a.detect_anomaly(1.0, 0.5),    ]     auditor.run_audit(baseline_mean=1.0, baseline_std=0.5, rules=rules)  if __name__ == "__main__":    main()