AI辅助合规审计自动检查数据库配置是否满足等保和GDPR要求一、体系化起点合规审计为什么不能只靠手工检查数据库合规审计长期以来依赖手工方式DBA 对照等保 2.0 和 GDPR 的条文逐台检查数据库的配置参数、权限设置、日志策略。这种方式的根本问题有三个覆盖不全。一个中型互联网公司有 200 数据库实例每个实例 300 配置参数手工检查不可能面面俱到。DBA 通常只检查显眼的项——比如sql_mode、max_connections而忽略了log_bin_trust_function_creators、local_infile这类安全开关。判断不一致。不同 DBA 对同一条合规要求的理解可能不同。一份MyISAM 表是否允许存在的判断有的 DBA 认为业务既然在用就没事有的认为必须强制迁移到 InnoDB导致审计结果因人而异。无法持续监控。等保要求持续监测而非一次性检查但手工审计通常以月或季度为单位配置可能在检查后因运维操作而漂移形成审计当天合规第二天就违规的尴尬。AI 辅助审计的核心思路是将合规条文转化为机器可执行的规则引擎利用大模型理解自然语言条文自动生成检查规则并通过定时巡检实现持续合规。flowchart TB A[合规文档br/等保2.0 / GDPR / 行业规范] -- B[LLM 条文解析] B -- C[规则结构化输出br/JSON Schema] C -- D[规则引擎br/自动巡检调度] D -- E[数据库 1] D -- F[数据库 2] D -- G[数据库 N] E -- H[合规报告生成] F -- H G -- H H -- I[整改建议 跟踪闭环] style B fill:#c8e6c9 style D fill:#e3f2fd style I fill:#fff9c4图中从合规文档到整改建议的整个闭环核心难点在于LLM 如何理解合规条文并转化为可执行代码。这也是本章要拆解的重点。二、条文解析引擎LLM 如何将自然语言合规要求转化为检查规则我们面对的合规文档如 GB/T 22239-2019《信息安全技术 网络安全等级保护基本要求》是用自然语言书写的表述方式如应禁止数据库使用默认口令和弱口令应启用数据库审计功能审计记录应包括事件的日期、时间、类型、主体标识和客体标识应采用密码技术保证重要数据在传输和存储过程中的完整性这类表述对 DBA 来说是明确的但对机器来说需要转化为结构化的检查规则。我们设计了一个两步转换流程第一步条文提取。LLM 阅读合规文档抽取出每条要求对应的检查点。例如上面的三条分别对应密码强度检查、审计日志配置检查、传输/存储加密检查。第二步规则生成。为每个检查点生成 SQL 查询或配置检查脚本。以下是核心实现import json import pymysql import re from typing import List, Dict, Optional, Any from dataclasses import dataclass, field from datetime import datetime import hashlib dataclass class ComplianceRule: 单条合规检查规则 rule_id: str category: str # 等保/ GDPR / 行业 level: str # L1/L2/L3 description: str check_type: str # config / query / permission / log check_method: str # check的SQL或配置键 expected_value: Any severity: str # critical / high / medium / low def to_dict(self) - Dict: return { rule_id: self.rule_id, category: self.category, level: self.level, description: self.description, check_type: self.check_type, severity: self.severity } class ComplianceRuleEngine: 合规规则引擎管理规则库并执行巡检 def __init__(self): self.rules: List[ComplianceRule] [] self._init_builtin_rules() def _init_builtin_rules(self): 初始化内置合规规则库 self.rules [ # 等保 2.0 三级身份鉴别 (8.1.3.2) ComplianceRule( rule_idMLPS-AUTH-001, category等保2.0, levelL3, description禁止空口令用户存在, check_typequery, check_methodSELECT COUNT(*) AS cnt FROM mysql.user WHERE authentication_string OR authentication_string IS NULL, expected_value0, severitycritical ), ComplianceRule( rule_idMLPS-AUTH-002, category等保2.0, levelL3, description密码复杂度策略已启用, check_typeconfig, check_methodvalidate_password.policy, expected_valuelambda v: v in (MEDIUM, STRONG), severityhigh ), # 等保 2.0 三级访问控制 (8.1.3.3) ComplianceRule( rule_idMLPS-ACL-001, category等保2.0, levelL3, description禁止匿名用户存在, check_typequery, check_methodSELECT COUNT(*) FROM mysql.user WHERE User , expected_value0, severitycritical ), ComplianceRule( rule_idMLPS-ACL-002, category等保2.0, levelL3, description不允许 root 远程登录, check_typequery, check_methodSELECT COUNT(*) FROM mysql.user WHERE User root AND Host NOT IN (localhost, 127.0.0.1, ::1), expected_value0, severityhigh ), ComplianceRule( rule_idMLPS-ACL-003, category等保2.0, levelL3, description禁止给用户授予 SUPER 和 FILE 权限, check_typequery, check_method SELECT COUNT(*) AS cnt FROM ( SELECT User, Host FROM mysql.user WHERE Super_priv Y OR File_priv Y AND User NOT IN (mysql.session, mysql.sys, mysql.infoschema) ) t , expected_value0, severitycritical ), # 等保 2.0 三级安全审计 (8.1.3.4) ComplianceRule( rule_idMLPS-AUDIT-001, category等保2.0, levelL3, description通用查询日志未开启生产环境应关闭 general_log, check_typeconfig, check_methodgeneral_log, expected_valueOFF, severitymedium ), ComplianceRule( rule_idMLPS-AUDIT-002, category等保2.0, levelL3, description错误日志已启用, check_typeconfig, check_methodlog_error, expected_valuelambda v: v ! and v is not None, severityhigh ), ComplianceRule( rule_idMLPS-AUDIT-003, category等保2.0, levelL3, descriptionBinlog 已启用用于审计追溯, check_typeconfig, check_methodlog_bin, expected_valueON, severityhigh ), # 等保 2.0 三级数据完整性 (8.1.3.6) ComplianceRule( rule_idMLPS-DATA-001, category等保2.0, levelL3, descriptionSSL 加密已启用, check_typeconfig, check_methodhave_ssl, expected_valueYES, severityhigh ), ComplianceRule( rule_idMLPS-DATA-002, category等保2.0, levelL3, description表空间加密已启用, check_typeconfig, check_methodinnodb_encrypt_tables, expected_valueON, severityhigh ), # GDPR 相关 ComplianceRule( rule_idGDPR-DATA-001, categoryGDPR, levelL3, description数据库连接超时已配置防止会话劫持, check_typeconfig, check_methodwait_timeout, expected_valuelambda v: int(v) 28800, # ≤ 8小时 severitymedium ), ComplianceRule( rule_idGDPR-DATA-002, categoryGDPR, levelL3, descriptionMyISAM 表应迁移到 InnoDBMyISAM 不支持事务无法保证数据一致性, check_typequery, check_method SELECT COUNT(*) AS cnt FROM information_schema.TABLES WHERE ENGINE MyISAM AND TABLE_SCHEMA NOT IN (mysql, information_schema, performance_schema, sys) , expected_value0, severitymedium ), ComplianceRule( rule_idGDPR-DATA-003, categoryGDPR, levelL3, descriptionlocal_infile 必须关闭防止 LOAD DATA LOCAL 数据泄露, check_typeconfig, check_methodlocal_infile, expected_valueOFF, severitycritical ), ] def add_rule(self, rule: ComplianceRule): 动态添加规则 self.rules.append(rule) def get_rules_by_category(self, category: str) - List[ComplianceRule]: 按类别筛选规则 return [r for r in self.rules if r.category category] def get_rules_by_level(self, level: str) - List[ComplianceRule]: 按保护等级筛选规则 return [r for r in self.rules if r.level level] class ComplianceAuditor: 合规审计执行引擎 def __init__(self, conn_params: Dict[str, Any], rule_engine: ComplianceRuleEngine): self.conn_params conn_params self.rule_engine rule_engine def audit_config(self, config_name: str, expected: Any) - Dict[str, Any]: 检查单个配置项 conn None try: conn pymysql.connect(**self.conn_params) with conn.cursor() as cursor: cursor.execute(fSHOW VARIABLES LIKE {config_name}) result cursor.fetchone() if result is None: return { status: unknown, actual: None, message: f配置项 {config_name} 未找到 } actual result[1] # 支持 lambda 和直接值比较 if callable(expected): passed expected(actual) expected_desc lambda_condition else: passed str(actual).upper() str(expected).upper() expected_desc str(expected) return { status: pass if passed else fail, actual: str(actual), expected: expected_desc, message: 通过 if passed else f不通过: 实际值 {actual} ! 期望值 {expected_desc} } except pymysql.Error as e: return { status: error, actual: None, expected: None, message: f检查执行出错: {str(e)} } finally: if conn: conn.close() def audit_query(self, query: str, expected: Any) - Dict[str, Any]: 检查 SQL 查询结果 conn None try: conn pymysql.connect(**self.conn_params) with conn.cursor() as cursor: cursor.execute(query) result cursor.fetchone() if result is None: return { status: error, actual: None, message: 查询未返回结果 } actual result[0] expected_val expected() if callable(expected) else expected passed actual expected_val return { status: pass if passed else fail, actual: actual, expected: expected_val, message: 通过 if passed else f不通过: 实际值 {actual} ! 期望值 {expected_val} } except pymysql.Error as e: return { status: error, actual: None, message: f查询执行出错: {str(e)} } finally: if conn: conn.close() def run_audit(self, categories: Optional[List[str]] None) - Dict[str, Any]: 执行完整合规审计 rules_to_check self.rule_engine.rules if categories: rules_to_check [r for r in rules_to_check if r.category in categories] results [] summary {total: 0, pass: 0, fail: 0, error: 0, critical_failures: []} for rule in rules_to_check: summary[total] 1 if rule.check_type config: check_result self.audit_config(rule.check_method, rule.expected_value) elif rule.check_type query: check_result self.audit_query(rule.check_method, rule.expected_value) else: check_result { status: error, message: f不支持的检查类型: {rule.check_type} } audit_item { **rule.to_dict(), **check_result, audit_time: datetime.now().isoformat() } results.append(audit_item) if check_result[status] pass: summary[pass] 1 elif check_result[status] fail: summary[fail] 1 if rule.severity critical: summary[critical_failures].append({ rule_id: rule.rule_id, description: rule.description }) else: summary[error] 1 summary[compliance_rate] ( summary[pass] / summary[total] * 100 if summary[total] 0 else 0 ) return { host: self.conn_params.get(host), audit_time: datetime.now().isoformat(), summary: summary, details: results } def generate_report(self, audit_result: Dict[str, Any]) - str: 生成可读的合规审计报告 lines [] lines.append( * 60) lines.append(数据库合规审计报告) lines.append( * 60) lines.append(f审计目标: {audit_result[host]}) lines.append(f审计时间: {audit_result[audit_time]}) lines.append(- * 60) s audit_result[summary] lines.append(f检查项总数: {s[total]}) lines.append(f通过: {s[pass]} | 不通过: {s[fail]} | 错误: {s[error]}) lines.append(f合规率: {s[compliance_rate]:.1f}%) lines.append(- * 60) # 严重违规项优先展示 if s[critical_failures]: lines.append(\n 严重违规项:) for failure in s[critical_failures]: lines.append(f [{failure[rule_id]}] {failure[description]}) # 按类别分组展示 lines.append(\n 详细检查结果:) by_category {} for item in audit_result[details]: cat item[category] by_category.setdefault(cat, []).append(item) for cat, items in by_category.items(): lines.append(f\n## {cat}) for item in items: icon ✅ if item[status] pass else (❌ if item[status] fail else ⚠️) lines.append(f {icon} [{item[rule_id]}] {item[description]}) if item[status] fail: lines.append(f 详情: {item.get(message, )}) lines.append(\n * 60) return \n.join(lines) # LLM 规则生成辅助 class LLMRuleGenerator: 利用 LLM 从自然语言合规条文中自动生成检查规则 SYSTEM_PROMPT 你是一个数据库合规专家。请将给定的合规要求转化为结构化的检查规则。 输出格式必须严格遵循以下 JSON Schema { rule_id: 合规要求编号, category: 等保2.0 | GDPR | 行业规范, level: L1 | L2 | L3, description: 简短描述, check_type: config | query | permission | log, check_method: MySQL 变量名 或 SQL 查询语句, expected_value: 期望的值或条件, severity: critical | high | medium | low } 注意 1. check_method 如果是 config 类型填写 MySQL 变量名如 local_infile 2. check_method 如果是 query 类型填写完整的 SQL 查询 3. expected_value 填写期望值对于配置项填写字符串对于查询填写数值 staticmethod def parse_compliance_text(compliance_text: str) - List[Dict]: 解析合规条文实际使用时替换为 LLM API 调用 这里展示的是结构化输出示例 # 实际实现中这里调用 LLM API # response openai.ChatCompletion.create( # modelgpt-4, # messages[ # {role: system, content: LLMRuleGenerator.SYSTEM_PROMPT}, # {role: user, content: compliance_text} # ], # response_format{type: json_object} # ) # 返回模拟的解析结果 return [{ rule_id: MLPS-AUTO-001, category: 等保2.0, level: L3, description: compliance_text[:50] ..., check_type: config, check_method: 示例检查项, expected_value: ON, severity: high }] if __name__ __main__: conn_params { host: 127.0.0.1, port: 3306, user: audit_user, password: ***, database: mysql, connect_timeout: 5 } engine ComplianceRuleEngine() auditor ComplianceAuditor(conn_params, engine) # 执行等保和 GDPR 审计 result auditor.run_audit(categories[等保2.0, GDPR]) # 生成报告 report auditor.generate_report(result) print(report) # 保存到文件 with open(/var/log/db_compliance_audit.json, w) as f: json.dump(result, f, ensure_asciiFalse, indent2, defaultstr)三、从静态检查到持续合规巡检调度的工程化落地单次审计只解决当前是否合规的问题而等保和 GDPR 要求的是持续合规。这意味着审计必须定期自动化执行变更后自动触发并且合规状态可追踪。我们采用 定时巡检 事件触发 的混合调度模式定时巡检每日凌晨 3 点对所有数据库实例执行全量合规审计事件触发配置变更如修改my.cnf、用户权限变更、DDL 操作触发增量审计状态追踪每次审计结果存入时序数据库可视化合规趋势flowchart LR A[Cron 定时触发br/每日 3:00] -- C[合规审计引擎] B[配置变更事件br/my.cnf / GRANT] -- C C -- D{审计结果} D --|合规率 100%| E[记录日志br/状态正常] D --|存在违规| F[分级告警] F -- G[严重: 即时通知br/钉钉/企微/PagerDuty] F -- H[一般: 生成待办br/Jira/禅道] E -- I[InfluxDB 时序记录] H -- J[整改闭环跟踪] J -- C style C fill:#e3f2fd style G fill:#ffcdd2 style H fill:#fff9c4其中最关键的是整改闭环跟踪。我们发现很多团队的审计流程在发现问题后就结束了缺少修复 → 验证 → 关闭的闭环。因此引入了 TTLTime-To-Resolution追踪每个违规项分配修复责任人超时自动升级。四、LLM 规则生成的边界什么该交给模型什么该硬编码LLM 在合规审计中的最大价值是加速规则构建而不是替代规则执行。我们需要明确划分职责边界适合 LLM 做的事从大段合规文档中自动提取检查要点将自然语言要求转化为 Python/SQL 检查代码骨架对检查结果做智能解释为什么不通过、如何修复必须硬编码的事检查规则的核心逻辑安全红线不能由 LLM 自由发挥严重违规项的判定标准密钥管理、认证凭据等敏感操作安全底线原则所有由 LLM 生成的规则在部署前必须经过 DBA/安全工程师的人工审核。这是防止 LLM 幻觉生成不安全或不正确的规则的关键防线。五、总结AI 辅助合规审计的本质是让 LLM 理解自然语言合规条文产出结构化检查规则然后由规则引擎以工程化的方式执行持续巡检。这不是用 AI 替代 DBA而是把 DBA 从逐台手工检查的重复劳动中解放出来让他们专注于解读审计结果、制定修复方案这些真正需要经验和判断力的工作。从工程价值来看AI 辅助合规审计至少带来了三个层次的收益消除手工巡检的遗漏风险、统一审计判断标准、实现持续合规监控。在等保测评和 GDPR 审计的压力下这是每个数据库团队都值得投入的工程能力。
AI辅助合规审计：自动检查数据库配置是否满足等保和GDPR要求
发布时间：2026/7/11 19:12:00
AI辅助合规审计自动检查数据库配置是否满足等保和GDPR要求一、体系化起点合规审计为什么不能只靠手工检查数据库合规审计长期以来依赖手工方式DBA 对照等保 2.0 和 GDPR 的条文逐台检查数据库的配置参数、权限设置、日志策略。这种方式的根本问题有三个覆盖不全。一个中型互联网公司有 200 数据库实例每个实例 300 配置参数手工检查不可能面面俱到。DBA 通常只检查显眼的项——比如sql_mode、max_connections而忽略了log_bin_trust_function_creators、local_infile这类安全开关。判断不一致。不同 DBA 对同一条合规要求的理解可能不同。一份MyISAM 表是否允许存在的判断有的 DBA 认为业务既然在用就没事有的认为必须强制迁移到 InnoDB导致审计结果因人而异。无法持续监控。等保要求持续监测而非一次性检查但手工审计通常以月或季度为单位配置可能在检查后因运维操作而漂移形成审计当天合规第二天就违规的尴尬。AI 辅助审计的核心思路是将合规条文转化为机器可执行的规则引擎利用大模型理解自然语言条文自动生成检查规则并通过定时巡检实现持续合规。flowchart TB A[合规文档br/等保2.0 / GDPR / 行业规范] -- B[LLM 条文解析] B -- C[规则结构化输出br/JSON Schema] C -- D[规则引擎br/自动巡检调度] D -- E[数据库 1] D -- F[数据库 2] D -- G[数据库 N] E -- H[合规报告生成] F -- H G -- H H -- I[整改建议 跟踪闭环] style B fill:#c8e6c9 style D fill:#e3f2fd style I fill:#fff9c4图中从合规文档到整改建议的整个闭环核心难点在于LLM 如何理解合规条文并转化为可执行代码。这也是本章要拆解的重点。二、条文解析引擎LLM 如何将自然语言合规要求转化为检查规则我们面对的合规文档如 GB/T 22239-2019《信息安全技术 网络安全等级保护基本要求》是用自然语言书写的表述方式如应禁止数据库使用默认口令和弱口令应启用数据库审计功能审计记录应包括事件的日期、时间、类型、主体标识和客体标识应采用密码技术保证重要数据在传输和存储过程中的完整性这类表述对 DBA 来说是明确的但对机器来说需要转化为结构化的检查规则。我们设计了一个两步转换流程第一步条文提取。LLM 阅读合规文档抽取出每条要求对应的检查点。例如上面的三条分别对应密码强度检查、审计日志配置检查、传输/存储加密检查。第二步规则生成。为每个检查点生成 SQL 查询或配置检查脚本。以下是核心实现import json import pymysql import re from typing import List, Dict, Optional, Any from dataclasses import dataclass, field from datetime import datetime import hashlib dataclass class ComplianceRule: 单条合规检查规则 rule_id: str category: str # 等保/ GDPR / 行业 level: str # L1/L2/L3 description: str check_type: str # config / query / permission / log check_method: str # check的SQL或配置键 expected_value: Any severity: str # critical / high / medium / low def to_dict(self) - Dict: return { rule_id: self.rule_id, category: self.category, level: self.level, description: self.description, check_type: self.check_type, severity: self.severity } class ComplianceRuleEngine: 合规规则引擎管理规则库并执行巡检 def __init__(self): self.rules: List[ComplianceRule] [] self._init_builtin_rules() def _init_builtin_rules(self): 初始化内置合规规则库 self.rules [ # 等保 2.0 三级身份鉴别 (8.1.3.2) ComplianceRule( rule_idMLPS-AUTH-001, category等保2.0, levelL3, description禁止空口令用户存在, check_typequery, check_methodSELECT COUNT(*) AS cnt FROM mysql.user WHERE authentication_string OR authentication_string IS NULL, expected_value0, severitycritical ), ComplianceRule( rule_idMLPS-AUTH-002, category等保2.0, levelL3, description密码复杂度策略已启用, check_typeconfig, check_methodvalidate_password.policy, expected_valuelambda v: v in (MEDIUM, STRONG), severityhigh ), # 等保 2.0 三级访问控制 (8.1.3.3) ComplianceRule( rule_idMLPS-ACL-001, category等保2.0, levelL3, description禁止匿名用户存在, check_typequery, check_methodSELECT COUNT(*) FROM mysql.user WHERE User , expected_value0, severitycritical ), ComplianceRule( rule_idMLPS-ACL-002, category等保2.0, levelL3, description不允许 root 远程登录, check_typequery, check_methodSELECT COUNT(*) FROM mysql.user WHERE User root AND Host NOT IN (localhost, 127.0.0.1, ::1), expected_value0, severityhigh ), ComplianceRule( rule_idMLPS-ACL-003, category等保2.0, levelL3, description禁止给用户授予 SUPER 和 FILE 权限, check_typequery, check_method SELECT COUNT(*) AS cnt FROM ( SELECT User, Host FROM mysql.user WHERE Super_priv Y OR File_priv Y AND User NOT IN (mysql.session, mysql.sys, mysql.infoschema) ) t , expected_value0, severitycritical ), # 等保 2.0 三级安全审计 (8.1.3.4) ComplianceRule( rule_idMLPS-AUDIT-001, category等保2.0, levelL3, description通用查询日志未开启生产环境应关闭 general_log, check_typeconfig, check_methodgeneral_log, expected_valueOFF, severitymedium ), ComplianceRule( rule_idMLPS-AUDIT-002, category等保2.0, levelL3, description错误日志已启用, check_typeconfig, check_methodlog_error, expected_valuelambda v: v ! and v is not None, severityhigh ), ComplianceRule( rule_idMLPS-AUDIT-003, category等保2.0, levelL3, descriptionBinlog 已启用用于审计追溯, check_typeconfig, check_methodlog_bin, expected_valueON, severityhigh ), # 等保 2.0 三级数据完整性 (8.1.3.6) ComplianceRule( rule_idMLPS-DATA-001, category等保2.0, levelL3, descriptionSSL 加密已启用, check_typeconfig, check_methodhave_ssl, expected_valueYES, severityhigh ), ComplianceRule( rule_idMLPS-DATA-002, category等保2.0, levelL3, description表空间加密已启用, check_typeconfig, check_methodinnodb_encrypt_tables, expected_valueON, severityhigh ), # GDPR 相关 ComplianceRule( rule_idGDPR-DATA-001, categoryGDPR, levelL3, description数据库连接超时已配置防止会话劫持, check_typeconfig, check_methodwait_timeout, expected_valuelambda v: int(v) 28800, # ≤ 8小时 severitymedium ), ComplianceRule( rule_idGDPR-DATA-002, categoryGDPR, levelL3, descriptionMyISAM 表应迁移到 InnoDBMyISAM 不支持事务无法保证数据一致性, check_typequery, check_method SELECT COUNT(*) AS cnt FROM information_schema.TABLES WHERE ENGINE MyISAM AND TABLE_SCHEMA NOT IN (mysql, information_schema, performance_schema, sys) , expected_value0, severitymedium ), ComplianceRule( rule_idGDPR-DATA-003, categoryGDPR, levelL3, descriptionlocal_infile 必须关闭防止 LOAD DATA LOCAL 数据泄露, check_typeconfig, check_methodlocal_infile, expected_valueOFF, severitycritical ), ] def add_rule(self, rule: ComplianceRule): 动态添加规则 self.rules.append(rule) def get_rules_by_category(self, category: str) - List[ComplianceRule]: 按类别筛选规则 return [r for r in self.rules if r.category category] def get_rules_by_level(self, level: str) - List[ComplianceRule]: 按保护等级筛选规则 return [r for r in self.rules if r.level level] class ComplianceAuditor: 合规审计执行引擎 def __init__(self, conn_params: Dict[str, Any], rule_engine: ComplianceRuleEngine): self.conn_params conn_params self.rule_engine rule_engine def audit_config(self, config_name: str, expected: Any) - Dict[str, Any]: 检查单个配置项 conn None try: conn pymysql.connect(**self.conn_params) with conn.cursor() as cursor: cursor.execute(fSHOW VARIABLES LIKE {config_name}) result cursor.fetchone() if result is None: return { status: unknown, actual: None, message: f配置项 {config_name} 未找到 } actual result[1] # 支持 lambda 和直接值比较 if callable(expected): passed expected(actual) expected_desc lambda_condition else: passed str(actual).upper() str(expected).upper() expected_desc str(expected) return { status: pass if passed else fail, actual: str(actual), expected: expected_desc, message: 通过 if passed else f不通过: 实际值 {actual} ! 期望值 {expected_desc} } except pymysql.Error as e: return { status: error, actual: None, expected: None, message: f检查执行出错: {str(e)} } finally: if conn: conn.close() def audit_query(self, query: str, expected: Any) - Dict[str, Any]: 检查 SQL 查询结果 conn None try: conn pymysql.connect(**self.conn_params) with conn.cursor() as cursor: cursor.execute(query) result cursor.fetchone() if result is None: return { status: error, actual: None, message: 查询未返回结果 } actual result[0] expected_val expected() if callable(expected) else expected passed actual expected_val return { status: pass if passed else fail, actual: actual, expected: expected_val, message: 通过 if passed else f不通过: 实际值 {actual} ! 期望值 {expected_val} } except pymysql.Error as e: return { status: error, actual: None, message: f查询执行出错: {str(e)} } finally: if conn: conn.close() def run_audit(self, categories: Optional[List[str]] None) - Dict[str, Any]: 执行完整合规审计 rules_to_check self.rule_engine.rules if categories: rules_to_check [r for r in rules_to_check if r.category in categories] results [] summary {total: 0, pass: 0, fail: 0, error: 0, critical_failures: []} for rule in rules_to_check: summary[total] 1 if rule.check_type config: check_result self.audit_config(rule.check_method, rule.expected_value) elif rule.check_type query: check_result self.audit_query(rule.check_method, rule.expected_value) else: check_result { status: error, message: f不支持的检查类型: {rule.check_type} } audit_item { **rule.to_dict(), **check_result, audit_time: datetime.now().isoformat() } results.append(audit_item) if check_result[status] pass: summary[pass] 1 elif check_result[status] fail: summary[fail] 1 if rule.severity critical: summary[critical_failures].append({ rule_id: rule.rule_id, description: rule.description }) else: summary[error] 1 summary[compliance_rate] ( summary[pass] / summary[total] * 100 if summary[total] 0 else 0 ) return { host: self.conn_params.get(host), audit_time: datetime.now().isoformat(), summary: summary, details: results } def generate_report(self, audit_result: Dict[str, Any]) - str: 生成可读的合规审计报告 lines [] lines.append( * 60) lines.append(数据库合规审计报告) lines.append( * 60) lines.append(f审计目标: {audit_result[host]}) lines.append(f审计时间: {audit_result[audit_time]}) lines.append(- * 60) s audit_result[summary] lines.append(f检查项总数: {s[total]}) lines.append(f通过: {s[pass]} | 不通过: {s[fail]} | 错误: {s[error]}) lines.append(f合规率: {s[compliance_rate]:.1f}%) lines.append(- * 60) # 严重违规项优先展示 if s[critical_failures]: lines.append(\n 严重违规项:) for failure in s[critical_failures]: lines.append(f [{failure[rule_id]}] {failure[description]}) # 按类别分组展示 lines.append(\n 详细检查结果:) by_category {} for item in audit_result[details]: cat item[category] by_category.setdefault(cat, []).append(item) for cat, items in by_category.items(): lines.append(f\n## {cat}) for item in items: icon ✅ if item[status] pass else (❌ if item[status] fail else ⚠️) lines.append(f {icon} [{item[rule_id]}] {item[description]}) if item[status] fail: lines.append(f 详情: {item.get(message, )}) lines.append(\n * 60) return \n.join(lines) # LLM 规则生成辅助 class LLMRuleGenerator: 利用 LLM 从自然语言合规条文中自动生成检查规则 SYSTEM_PROMPT 你是一个数据库合规专家。请将给定的合规要求转化为结构化的检查规则。 输出格式必须严格遵循以下 JSON Schema { rule_id: 合规要求编号, category: 等保2.0 | GDPR | 行业规范, level: L1 | L2 | L3, description: 简短描述, check_type: config | query | permission | log, check_method: MySQL 变量名 或 SQL 查询语句, expected_value: 期望的值或条件, severity: critical | high | medium | low } 注意 1. check_method 如果是 config 类型填写 MySQL 变量名如 local_infile 2. check_method 如果是 query 类型填写完整的 SQL 查询 3. expected_value 填写期望值对于配置项填写字符串对于查询填写数值 staticmethod def parse_compliance_text(compliance_text: str) - List[Dict]: 解析合规条文实际使用时替换为 LLM API 调用 这里展示的是结构化输出示例 # 实际实现中这里调用 LLM API # response openai.ChatCompletion.create( # modelgpt-4, # messages[ # {role: system, content: LLMRuleGenerator.SYSTEM_PROMPT}, # {role: user, content: compliance_text} # ], # response_format{type: json_object} # ) # 返回模拟的解析结果 return [{ rule_id: MLPS-AUTO-001, category: 等保2.0, level: L3, description: compliance_text[:50] ..., check_type: config, check_method: 示例检查项, expected_value: ON, severity: high }] if __name__ __main__: conn_params { host: 127.0.0.1, port: 3306, user: audit_user, password: ***, database: mysql, connect_timeout: 5 } engine ComplianceRuleEngine() auditor ComplianceAuditor(conn_params, engine) # 执行等保和 GDPR 审计 result auditor.run_audit(categories[等保2.0, GDPR]) # 生成报告 report auditor.generate_report(result) print(report) # 保存到文件 with open(/var/log/db_compliance_audit.json, w) as f: json.dump(result, f, ensure_asciiFalse, indent2, defaultstr)三、从静态检查到持续合规巡检调度的工程化落地单次审计只解决当前是否合规的问题而等保和 GDPR 要求的是持续合规。这意味着审计必须定期自动化执行变更后自动触发并且合规状态可追踪。我们采用 定时巡检 事件触发 的混合调度模式定时巡检每日凌晨 3 点对所有数据库实例执行全量合规审计事件触发配置变更如修改my.cnf、用户权限变更、DDL 操作触发增量审计状态追踪每次审计结果存入时序数据库可视化合规趋势flowchart LR A[Cron 定时触发br/每日 3:00] -- C[合规审计引擎] B[配置变更事件br/my.cnf / GRANT] -- C C -- D{审计结果} D --|合规率 100%| E[记录日志br/状态正常] D --|存在违规| F[分级告警] F -- G[严重: 即时通知br/钉钉/企微/PagerDuty] F -- H[一般: 生成待办br/Jira/禅道] E -- I[InfluxDB 时序记录] H -- J[整改闭环跟踪] J -- C style C fill:#e3f2fd style G fill:#ffcdd2 style H fill:#fff9c4其中最关键的是整改闭环跟踪。我们发现很多团队的审计流程在发现问题后就结束了缺少修复 → 验证 → 关闭的闭环。因此引入了 TTLTime-To-Resolution追踪每个违规项分配修复责任人超时自动升级。四、LLM 规则生成的边界什么该交给模型什么该硬编码LLM 在合规审计中的最大价值是加速规则构建而不是替代规则执行。我们需要明确划分职责边界适合 LLM 做的事从大段合规文档中自动提取检查要点将自然语言要求转化为 Python/SQL 检查代码骨架对检查结果做智能解释为什么不通过、如何修复必须硬编码的事检查规则的核心逻辑安全红线不能由 LLM 自由发挥严重违规项的判定标准密钥管理、认证凭据等敏感操作安全底线原则所有由 LLM 生成的规则在部署前必须经过 DBA/安全工程师的人工审核。这是防止 LLM 幻觉生成不安全或不正确的规则的关键防线。五、总结AI 辅助合规审计的本质是让 LLM 理解自然语言合规条文产出结构化检查规则然后由规则引擎以工程化的方式执行持续巡检。这不是用 AI 替代 DBA而是把 DBA 从逐台手工检查的重复劳动中解放出来让他们专注于解读审计结果、制定修复方案这些真正需要经验和判断力的工作。从工程价值来看AI 辅助合规审计至少带来了三个层次的收益消除手工巡检的遗漏风险、统一审计判断标准、实现持续合规监控。在等保测评和 GDPR 审计的压力下这是每个数据库团队都值得投入的工程能力。
