一、程序概述与核心定位一个面向Agent权限系统的快速审计工具主要用于自动化收集、统计、分析和导出系统授权行为的审计日志。其设计目标是为系统管理员和安全审计人员提供轻量化、可落地的审计能力覆盖风险监控、异常行为检测、合规性数据导出等核心场景。程序基于Python 3开发依赖自定义的AuditLogger类来自src.audit_service.logger模块实现日志全生命周期管理通过模块化设计将审计流程拆解为统计概览、高风险事件筛查、特定Agent行为追踪、日志导出、核心算法验证五大环节兼顾实时性与可追溯性。二、核心功能模块详解程序通过main函数串联五大功能模块各模块职责明确形成完整的审计闭环1. 审计环境初始化程序启动阶段首先完成环境配置与依赖加载路径处理通过sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))将当前脚本目录添加到Python路径确保能正确导入同目录下的src.audit_service.logger模块避免相对导入错误。日志服务实例化创建AuditLogger对象logger AuditLogger()该对象是整个程序的核心封装了日志存储、查询、统计、导出等底层能力推测其内部可能集成了文件I/O、数据库交互或内存缓存机制具体取决于src.audit_service.logger的实现。2. 统计概览模块全局态势感知stats logger.get_statistics() print(f总事件: {stats[total_events]} | 允许率: {stats[allow_rate]*100:.2f}% | 拒绝率: {stats[deny_rate]*100:.2f}%)功能快速输出系统授权行为的宏观统计数据帮助管理员掌握整体态势。核心指标total_events累计审计事件总数反映系统活跃度allow_rate/deny_rate授权通过/拒绝比例异常波动如拒绝率突增可能暗示攻击或配置错误。实现逻辑get_statistics()方法需遍历全量日志或预计算的统计缓存统计decision字段中ALLOW/DENY的数量计算占比。若日志量较大可能采用增量统计如定时更新统计结果至Redis以提升性能。3. 高风险事件筛查模块精准定位威胁high_risk logger.query_logs(risk_levelHIGH, limit10)功能聚焦高风险行为优先展示潜在安全威胁。筛选维度通过risk_levelHIGH过滤高风险事件limit10限制返回数量避免信息过载。输出内容打印高风险事件的audit_id唯一标识、timestamp时间戳、subject.agent_id触发事件的Agent ID、authorization.reason授权原因如“权限不足”“敏感操作”。业务价值高风险事件可能包括“跨权限访问”“未授权API调用”“敏感数据导出”等需人工介入核查。例如若某Agent频繁出现reason: attempt to access admin resource可能存在恶意攻击意图。4. 特定Agent行为追踪WebAgent越权监控web_deny logger.query_logs(agent_idweb-agent, decisionDENY, limit10)功能针对特定Agent此处为web-agent的拒绝事件专项分析常用于监控前端Agent的越权尝试。筛选逻辑组合agent_id主体标识和decisionDENY结果过滤精准定位web-agent的所有授权失败行为。场景意义WebAgent通常直接面向用户是越权攻击的高发入口。若该模块输出次数突增如单日超过10次可能暗示暴力破解、会话劫持或业务逻辑漏洞如未校验用户权限直接调用接口。5. 日志导出模块合规性数据落地logs_24h logger.query_logs( start_time(datetime.datetime.utcnow() - datetime.timedelta(days1)).isoformat(), limit10000 ) csv logger.export_logs(logs_24h, formatcsv) with open(latest_24h_audit.csv, w, encodingutf-8-sig) as f: f.write(csv)功能导出最近24小时的审计日志为CSV格式满足合规性留存要求如等保2.0要求日志留存≥6个月。时间范围控制通过start_time参数指定查询窗口utcnow() - 1天使用UTC时间避免时区偏移导致的日志遗漏。导出优化limit10000防止单次查询数据量过大导致内存溢出encodingutf-8-sig确保Windows Excel打开时中文正常显示解决BOM头问题。扩展性export_logs支持format参数推测可扩展JSON、Parquet等格式适配不同数据分析工具如ELK Stack、Spark。6. 审计ID生成验证模块数据完整性校验test_id logger._generate_audit_id(AUTHORIZATION_DECISION, ALLOW) print(f格式验证: {通过 if test_id.startswith(audit-AUTHORIZ-ALLOW-) else 失败})功能验证审计ID生成的规范性与唯一性确保日志可追溯。ID结构测试生成的ID需符合audit-{event_type}-{decision}-{uuid}格式如audit-AUTHORIZ-ALLOW-550e8400-e29b-41d4-a716-446655440000其中AUTHORIZ是AUTHORIZATION_DECISION的缩写ALLOW对应授权结果。必要性审计ID是日志关联的关键如通过ID追溯原始请求、关联上下游系统日志格式验证可提前发现ID生成逻辑的缺陷如缺少事件类型前缀导致无法分类。三、数据结构设计程序的核心数据流转围绕审计日志条目展开其结构设计直接影响查询效率与分析深度。结合代码中对日志字段的访问如log[subject][agent_id]可推断底层日志采用嵌套字典结构具体定义如下1. 审计日志条目Audit Log Entry一级字段二级字段类型说明audit_id-string唯一审计ID格式为audit-{event_type}-{decision}-{uuid}用于日志去重与关联timestamp-string事件时间戳ISO 8601格式如2024-05-20T12:34:56.789ZUTC时区event_type-string事件类型如AUTHORIZATION_DECISION授权决策、POLICY_UPDATE策略更新subjectagent_idstring发起请求的Agent唯一标识如web-agent、data-processoragent_typestringAgent类型如WEB、BATCH、APIip_addressstringAgent所在IP地址用于定位物理位置resourceresource_idstring访问的资源标识如/api/admin/users、db:user_tableresource_typestring资源类型如API、DATABASE、FILEactionaction_namestring执行的操作如READ、WRITE、DELETEauthorizationdecisionstring授权决策结果ALLOW允许、DENY拒绝、PENDING待审批reasonstring决策原因如ROLE_NOT_MATCH、RESOURCE_NOT_EXISTpolicy_idstring匹配的权限策略ID用于追溯权限配置risk_level-string风险等级LOW低、MEDIUM中、HIGH高contextrequest_idstring原始请求ID关联上游系统日志user_agentstring客户端User-AgentWebAgent专用2. 统计结果结构Statisticsget_statistics()返回的stats字典结构{ total_events: 15230, # 总事件数 allow_count: 14500, # 允许事件数 deny_count: 730, # 拒绝事件数 allow_rate: 0.951, # 允许率14500/15230 deny_rate: 0.049 # 拒绝率730/15230 }四、核心算法与实现逻辑程序的智能化能力依赖于AuditLogger类的底层算法以下是关键逻辑的推断与解析1. 审计ID生成算法_generate_audit_id目标生成全局唯一、语义清晰的审计ID便于日志分类与检索。实现逻辑def _generate_audit_id(self, event_type: str, decision: str) - str: # 事件类型缩写如AUTHORIZATION_DECISION → AUTHORIZ type_abbr self._abbreviate_event_type(event_type) # 生成UUIDv4随机UUID确保唯一性 uuid_part uuid.uuid4().hex # 拼接格式audit-{type_abbr}-{decision}-{uuid} return faudit-{type_abbr}-{decision}-{uuid_part}缩写规则通过映射表简化事件类型如POLICY_UPDATE→POLICY、ACCESS_DENIED→DENY减少ID长度UUID选择采用UUIDv4而非自增ID避免分布式部署时的ID冲突多节点同时生成日志格式验证通过startswith(audit-AUTHORIZ-ALLOW-)确保生成逻辑未被篡改如误写为auth-前缀。2. 日志查询算法query_logs目标高效过滤符合条件的日志条目支持多维度组合查询。参数设计支持risk_level风险等级、agent_id主体、decision决策结果、start_time时间范围、limit返回数量等参数推测内部采用链式过滤逻辑def query_logs(self, risk_levelNone, agent_idNone, decisionNone, start_timeNone, limit100): filtered_logs self.logs # 假设self.logs是存储全量日志的列表/数据库游标 # 按风险等级过滤 if risk_level: filtered_logs [log for log in filtered_logs if log[risk_level] risk_level] # 按Agent ID过滤 if agent_id: filtered_logs [log for log in filtered_logs if log[subject][agent_id] agent_id] # 按决策结果过滤 if decision: filtered_logs [log for log in filtered_logs if log[authorization][decision] decision] # 按时间范围过滤start_time ≤ timestamp if start_time: start_dt datetime.datetime.fromisoformat(start_time) filtered_logs [ log for log in filtered_logs if datetime.datetime.fromisoformat(log[timestamp]) start_dt ] # 限制返回数量 return filtered_logs[:limit]性能优化若日志量达百万级上述列表推导式会效率低下此时需引入索引机制如对timestamp、agent_id建立B树索引或基于数据库的查询如Elasticsearch的DSL查询。3. 风险等级判定算法隐含逻辑程序中high_risk的筛选依赖risk_levelHIGH但风险等级的判定并非程序本身实现而是由AuditLogger在日志记录阶段完成。典型的风险判定规则包括基于权限强度访问admin资源的操作自动标记为HIGH基于行为频率同一Agent 1分钟内发起10次以上DENY事件标记为HIGH基于资源敏感度操作涉及PII个人身份信息数据时风险等级提升。五、运行流程与时序程序执行时序如下初始化0-1s加载路径→导入AuditLogger→实例化日志服务统计概览1-2s调用get_statistics()→计算总事件数、允许/拒绝率→打印结果高风险筛查2-3s调用query_logs(risk_levelHIGH)→遍历日志过滤高风险事件→打印前5条WebAgent监控3-4s调用query_logs(agent_idweb-agent, decisionDENY)→统计越权次数→打印结果日志导出4-10s查询24小时内日志可能耗时较长→转换为CSV→写入文件ID验证10-11s生成测试审计ID→验证格式→输出结果。六、安全性与可靠性设计路径安全通过os.path.dirname(os.path.abspath(__file__))动态获取脚本目录避免硬编码路径导致的跨平台兼容性问题编码安全CSV导出使用utf-8-sig编码防止中文乱码输入校验虽未显式体现但AuditLogger需对agent_id、risk_level等参数做合法性校验如decision只能是ALLOW/DENY/PENDING避免无效查询数据完整性审计ID的唯一性确保日志不可篡改若ID重复可直接发现数据异常。七、局限性与优化方向性能瓶颈全量日志扫描get_statistics、query_logs在日志量超10万时效率低下需引入索引优化对高频查询字段建索引异步查询通过Celery后台执行导出任务避免阻塞主线程功能缺失缺乏告警机制如高风险事件实时通知、可视化界面需对接Grafana展示统计图表扩展性不足web-agent硬编码建议改为配置文件驱动支持动态添加监控对象。八、总结一个轻量化但功能完备的审计工具通过模块化设计实现了从数据采集到分析导出的全流程覆盖。其核心优势在于场景针对性强聚焦Agent权限系统、落地成本低无需复杂依赖可作为中小规模系统的审计解决方案。未来通过引入分布式存储如Elasticsearch、实时计算如Flink和可视化界面可进一步升级为企业级审计平台。源代码#!/usr/bin/env python3 import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from src.audit_service.logger import AuditLogger import datetime import json if __name__ __main__: logger AuditLogger() print(*70) print(Agent权限系统 快速审计工具) print(*70) # 1. 统计概览 stats logger.get_statistics() print(f总事件: {stats[total_events]} | 允许率: {stats[allow_rate]*100:.2f}% | 拒绝率: {stats[deny_rate]*100:.2f}%) # 2. 高风险事件 high_risk logger.query_logs(risk_levelHIGH, limit10) print(f\n[WARN] 高风险事件: {len(high_risk)}条) for log in high_risk[:5]: print(f {log[audit_id]} | {log[timestamp]} | {log[subject][agent_id]} | {log[authorization][reason]}) # 3. WebAgent越权事件 web_deny logger.query_logs(agent_idweb-agent, decisionDENY, limit10) print(f\n[DENY] WebAgent越权尝试: {len(web_deny)}次) # 4. 导出最近24小时日志 logs_24h logger.query_logs( start_time(datetime.datetime.utcnow() - datetime.timedelta(days1)).isoformat(), limit10000 ) csv logger.export_logs(logs_24h, formatcsv) with open(latest_24h_audit.csv, w, encodingutf-8-sig) as f: f.write(csv) print(f\n[INFO] 最近24小时日志已导出到 latest_24h_audit.csv) # 5. 验证审计ID生成 print(\n[TEST] 验证审计ID生成:) test_id logger._generate_audit_id(AUTHORIZATION_DECISION, ALLOW) print(f生成测试审计ID: {test_id}) print(f格式验证: {通过 if test_id.startswith(audit-AUTHORIZ-ALLOW-) else 失败})
面向Agent权限系统的快速审计工具
发布时间：2026/6/1 6:39:17
一、程序概述与核心定位一个面向Agent权限系统的快速审计工具主要用于自动化收集、统计、分析和导出系统授权行为的审计日志。其设计目标是为系统管理员和安全审计人员提供轻量化、可落地的审计能力覆盖风险监控、异常行为检测、合规性数据导出等核心场景。程序基于Python 3开发依赖自定义的AuditLogger类来自src.audit_service.logger模块实现日志全生命周期管理通过模块化设计将审计流程拆解为统计概览、高风险事件筛查、特定Agent行为追踪、日志导出、核心算法验证五大环节兼顾实时性与可追溯性。二、核心功能模块详解程序通过main函数串联五大功能模块各模块职责明确形成完整的审计闭环1. 审计环境初始化程序启动阶段首先完成环境配置与依赖加载路径处理通过sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))将当前脚本目录添加到Python路径确保能正确导入同目录下的src.audit_service.logger模块避免相对导入错误。日志服务实例化创建AuditLogger对象logger AuditLogger()该对象是整个程序的核心封装了日志存储、查询、统计、导出等底层能力推测其内部可能集成了文件I/O、数据库交互或内存缓存机制具体取决于src.audit_service.logger的实现。2. 统计概览模块全局态势感知stats logger.get_statistics() print(f总事件: {stats[total_events]} | 允许率: {stats[allow_rate]*100:.2f}% | 拒绝率: {stats[deny_rate]*100:.2f}%)功能快速输出系统授权行为的宏观统计数据帮助管理员掌握整体态势。核心指标total_events累计审计事件总数反映系统活跃度allow_rate/deny_rate授权通过/拒绝比例异常波动如拒绝率突增可能暗示攻击或配置错误。实现逻辑get_statistics()方法需遍历全量日志或预计算的统计缓存统计decision字段中ALLOW/DENY的数量计算占比。若日志量较大可能采用增量统计如定时更新统计结果至Redis以提升性能。3. 高风险事件筛查模块精准定位威胁high_risk logger.query_logs(risk_levelHIGH, limit10)功能聚焦高风险行为优先展示潜在安全威胁。筛选维度通过risk_levelHIGH过滤高风险事件limit10限制返回数量避免信息过载。输出内容打印高风险事件的audit_id唯一标识、timestamp时间戳、subject.agent_id触发事件的Agent ID、authorization.reason授权原因如“权限不足”“敏感操作”。业务价值高风险事件可能包括“跨权限访问”“未授权API调用”“敏感数据导出”等需人工介入核查。例如若某Agent频繁出现reason: attempt to access admin resource可能存在恶意攻击意图。4. 特定Agent行为追踪WebAgent越权监控web_deny logger.query_logs(agent_idweb-agent, decisionDENY, limit10)功能针对特定Agent此处为web-agent的拒绝事件专项分析常用于监控前端Agent的越权尝试。筛选逻辑组合agent_id主体标识和decisionDENY结果过滤精准定位web-agent的所有授权失败行为。场景意义WebAgent通常直接面向用户是越权攻击的高发入口。若该模块输出次数突增如单日超过10次可能暗示暴力破解、会话劫持或业务逻辑漏洞如未校验用户权限直接调用接口。5. 日志导出模块合规性数据落地logs_24h logger.query_logs( start_time(datetime.datetime.utcnow() - datetime.timedelta(days1)).isoformat(), limit10000 ) csv logger.export_logs(logs_24h, formatcsv) with open(latest_24h_audit.csv, w, encodingutf-8-sig) as f: f.write(csv)功能导出最近24小时的审计日志为CSV格式满足合规性留存要求如等保2.0要求日志留存≥6个月。时间范围控制通过start_time参数指定查询窗口utcnow() - 1天使用UTC时间避免时区偏移导致的日志遗漏。导出优化limit10000防止单次查询数据量过大导致内存溢出encodingutf-8-sig确保Windows Excel打开时中文正常显示解决BOM头问题。扩展性export_logs支持format参数推测可扩展JSON、Parquet等格式适配不同数据分析工具如ELK Stack、Spark。6. 审计ID生成验证模块数据完整性校验test_id logger._generate_audit_id(AUTHORIZATION_DECISION, ALLOW) print(f格式验证: {通过 if test_id.startswith(audit-AUTHORIZ-ALLOW-) else 失败})功能验证审计ID生成的规范性与唯一性确保日志可追溯。ID结构测试生成的ID需符合audit-{event_type}-{decision}-{uuid}格式如audit-AUTHORIZ-ALLOW-550e8400-e29b-41d4-a716-446655440000其中AUTHORIZ是AUTHORIZATION_DECISION的缩写ALLOW对应授权结果。必要性审计ID是日志关联的关键如通过ID追溯原始请求、关联上下游系统日志格式验证可提前发现ID生成逻辑的缺陷如缺少事件类型前缀导致无法分类。三、数据结构设计程序的核心数据流转围绕审计日志条目展开其结构设计直接影响查询效率与分析深度。结合代码中对日志字段的访问如log[subject][agent_id]可推断底层日志采用嵌套字典结构具体定义如下1. 审计日志条目Audit Log Entry一级字段二级字段类型说明audit_id-string唯一审计ID格式为audit-{event_type}-{decision}-{uuid}用于日志去重与关联timestamp-string事件时间戳ISO 8601格式如2024-05-20T12:34:56.789ZUTC时区event_type-string事件类型如AUTHORIZATION_DECISION授权决策、POLICY_UPDATE策略更新subjectagent_idstring发起请求的Agent唯一标识如web-agent、data-processoragent_typestringAgent类型如WEB、BATCH、APIip_addressstringAgent所在IP地址用于定位物理位置resourceresource_idstring访问的资源标识如/api/admin/users、db:user_tableresource_typestring资源类型如API、DATABASE、FILEactionaction_namestring执行的操作如READ、WRITE、DELETEauthorizationdecisionstring授权决策结果ALLOW允许、DENY拒绝、PENDING待审批reasonstring决策原因如ROLE_NOT_MATCH、RESOURCE_NOT_EXISTpolicy_idstring匹配的权限策略ID用于追溯权限配置risk_level-string风险等级LOW低、MEDIUM中、HIGH高contextrequest_idstring原始请求ID关联上游系统日志user_agentstring客户端User-AgentWebAgent专用2. 统计结果结构Statisticsget_statistics()返回的stats字典结构{ total_events: 15230, # 总事件数 allow_count: 14500, # 允许事件数 deny_count: 730, # 拒绝事件数 allow_rate: 0.951, # 允许率14500/15230 deny_rate: 0.049 # 拒绝率730/15230 }四、核心算法与实现逻辑程序的智能化能力依赖于AuditLogger类的底层算法以下是关键逻辑的推断与解析1. 审计ID生成算法_generate_audit_id目标生成全局唯一、语义清晰的审计ID便于日志分类与检索。实现逻辑def _generate_audit_id(self, event_type: str, decision: str) - str: # 事件类型缩写如AUTHORIZATION_DECISION → AUTHORIZ type_abbr self._abbreviate_event_type(event_type) # 生成UUIDv4随机UUID确保唯一性 uuid_part uuid.uuid4().hex # 拼接格式audit-{type_abbr}-{decision}-{uuid} return faudit-{type_abbr}-{decision}-{uuid_part}缩写规则通过映射表简化事件类型如POLICY_UPDATE→POLICY、ACCESS_DENIED→DENY减少ID长度UUID选择采用UUIDv4而非自增ID避免分布式部署时的ID冲突多节点同时生成日志格式验证通过startswith(audit-AUTHORIZ-ALLOW-)确保生成逻辑未被篡改如误写为auth-前缀。2. 日志查询算法query_logs目标高效过滤符合条件的日志条目支持多维度组合查询。参数设计支持risk_level风险等级、agent_id主体、decision决策结果、start_time时间范围、limit返回数量等参数推测内部采用链式过滤逻辑def query_logs(self, risk_levelNone, agent_idNone, decisionNone, start_timeNone, limit100): filtered_logs self.logs # 假设self.logs是存储全量日志的列表/数据库游标 # 按风险等级过滤 if risk_level: filtered_logs [log for log in filtered_logs if log[risk_level] risk_level] # 按Agent ID过滤 if agent_id: filtered_logs [log for log in filtered_logs if log[subject][agent_id] agent_id] # 按决策结果过滤 if decision: filtered_logs [log for log in filtered_logs if log[authorization][decision] decision] # 按时间范围过滤start_time ≤ timestamp if start_time: start_dt datetime.datetime.fromisoformat(start_time) filtered_logs [ log for log in filtered_logs if datetime.datetime.fromisoformat(log[timestamp]) start_dt ] # 限制返回数量 return filtered_logs[:limit]性能优化若日志量达百万级上述列表推导式会效率低下此时需引入索引机制如对timestamp、agent_id建立B树索引或基于数据库的查询如Elasticsearch的DSL查询。3. 风险等级判定算法隐含逻辑程序中high_risk的筛选依赖risk_levelHIGH但风险等级的判定并非程序本身实现而是由AuditLogger在日志记录阶段完成。典型的风险判定规则包括基于权限强度访问admin资源的操作自动标记为HIGH基于行为频率同一Agent 1分钟内发起10次以上DENY事件标记为HIGH基于资源敏感度操作涉及PII个人身份信息数据时风险等级提升。五、运行流程与时序程序执行时序如下初始化0-1s加载路径→导入AuditLogger→实例化日志服务统计概览1-2s调用get_statistics()→计算总事件数、允许/拒绝率→打印结果高风险筛查2-3s调用query_logs(risk_levelHIGH)→遍历日志过滤高风险事件→打印前5条WebAgent监控3-4s调用query_logs(agent_idweb-agent, decisionDENY)→统计越权次数→打印结果日志导出4-10s查询24小时内日志可能耗时较长→转换为CSV→写入文件ID验证10-11s生成测试审计ID→验证格式→输出结果。六、安全性与可靠性设计路径安全通过os.path.dirname(os.path.abspath(__file__))动态获取脚本目录避免硬编码路径导致的跨平台兼容性问题编码安全CSV导出使用utf-8-sig编码防止中文乱码输入校验虽未显式体现但AuditLogger需对agent_id、risk_level等参数做合法性校验如decision只能是ALLOW/DENY/PENDING避免无效查询数据完整性审计ID的唯一性确保日志不可篡改若ID重复可直接发现数据异常。七、局限性与优化方向性能瓶颈全量日志扫描get_statistics、query_logs在日志量超10万时效率低下需引入索引优化对高频查询字段建索引异步查询通过Celery后台执行导出任务避免阻塞主线程功能缺失缺乏告警机制如高风险事件实时通知、可视化界面需对接Grafana展示统计图表扩展性不足web-agent硬编码建议改为配置文件驱动支持动态添加监控对象。八、总结一个轻量化但功能完备的审计工具通过模块化设计实现了从数据采集到分析导出的全流程覆盖。其核心优势在于场景针对性强聚焦Agent权限系统、落地成本低无需复杂依赖可作为中小规模系统的审计解决方案。未来通过引入分布式存储如Elasticsearch、实时计算如Flink和可视化界面可进一步升级为企业级审计平台。源代码#!/usr/bin/env python3 import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from src.audit_service.logger import AuditLogger import datetime import json if __name__ __main__: logger AuditLogger() print(*70) print(Agent权限系统 快速审计工具) print(*70) # 1. 统计概览 stats logger.get_statistics() print(f总事件: {stats[total_events]} | 允许率: {stats[allow_rate]*100:.2f}% | 拒绝率: {stats[deny_rate]*100:.2f}%) # 2. 高风险事件 high_risk logger.query_logs(risk_levelHIGH, limit10) print(f\n[WARN] 高风险事件: {len(high_risk)}条) for log in high_risk[:5]: print(f {log[audit_id]} | {log[timestamp]} | {log[subject][agent_id]} | {log[authorization][reason]}) # 3. WebAgent越权事件 web_deny logger.query_logs(agent_idweb-agent, decisionDENY, limit10) print(f\n[DENY] WebAgent越权尝试: {len(web_deny)}次) # 4. 导出最近24小时日志 logs_24h logger.query_logs( start_time(datetime.datetime.utcnow() - datetime.timedelta(days1)).isoformat(), limit10000 ) csv logger.export_logs(logs_24h, formatcsv) with open(latest_24h_audit.csv, w, encodingutf-8-sig) as f: f.write(csv) print(f\n[INFO] 最近24小时日志已导出到 latest_24h_audit.csv) # 5. 验证审计ID生成 print(\n[TEST] 验证审计ID生成:) test_id logger._generate_audit_id(AUTHORIZATION_DECISION, ALLOW) print(f生成测试审计ID: {test_id}) print(f格式验证: {通过 if test_id.startswith(audit-AUTHORIZ-ALLOW-) else 失败})
