MySQL数据库审计最佳实践：三层方案满足等保2.0合规要求

大家好，我是数据库小学妹 👋

去年年底，安全团队收到告警：一批用户手机号和身份证号出现在了外部数据交易群里。源头还没确认，初步怀疑是从数据库侧泄露的。领导说"查一下最近谁在批量导出用户数据"，我打开数据库，愣住了——general_log没开。binlog开了，但只记录了DML操作，没有记录执行SQL的用户来源IP。performance_schema有连接记录，只记录了谁连过，没记录谁查了什么。我什么日志都有，但没有一条能告诉我：哪个账号、在什么时间、用哪条SQL、导出了多少用户数据。后来通过应用层的操作日志反查，花了3天才锁定范围。3天，足够数据被转发无数遍了。

那次之后我花了两周时间重新搭了一套数据库审计方案。不涉及厂商产品推荐，只讲技术实现：MySQL原生能力能做什么，不能做什么，怎么用最少的成本搭一套能用的审计系统。

审计到底要记录什么

很多人一提到数据库审计，第一反应是开general_log，这是最大的误区。general_log确实记录了所有SQL，但问题很大：它是同步写入，每条SQL都要写磁盘，我在测试环境做过对比，开启后TPS下降约40%，生产环境根本不敢开。日志体积也爆炸，中等规模系统每天几百万条SQL，一天轻松超过10GB，存储成本和检索效率都是问题。

最关键的是，它无法区分正常查询和敏感操作，把所有SQL一视同仁记下来，从几百万条日志里找"谁导出了用户数据"，等于大海捞针。

审计不是把所有东西都记下来，是把关键的东西精确记下来。我最终的方案记录了四类信息：谁执行了SQL（账号和来源IP），精确到毫秒的时间戳，SQL语句类型和操作的表，影响行数和执行时间。出事的时候需要回答三个问题——数据被谁导出的、导出了多少、什么时候导出的——这四类信息组合起来才能回答。

MySQL原生的审计能力拆解

我研究了一圈MySQL自带的日志机制，每种都试用过，全有坑。

general_log是最简单的审计手段，开启后所有SQL都会被记录：

SET GLOBAL general_log = 'ON';
SET GLOBAL log_output = 'TABLE';  -- 写入 mysql.general_log 表

写入TABLE比写入FILE稍好，可以建索引做查询，但性能损耗一样大。1000 TPS的场景下，开启general_log后TPS降到600左右，CPU开销增加约35%。适合短期排查问题，不适合长期审计。

binlog记录了所有数据变更操作，binlog_format是ROW的话还记录了每行变更的前后值。但binlog不记录SELECT，只记录DML和DDL，而数据泄露往往是通过SELECT完成的。binlog也不记录来源IP，只记录了线程ID，ROW格式还是二进制文件，需要mysqlbinlog工具解析。binlog作为审计补充有用，但作为唯一手段覆盖不了SELECT操作，毕竟binlog的主要用途是复制和恢复，不是审计。

performance_schema从MySQL 5.7开始提供了events_statements_history和events_statements_current表：

SELECT THREAD_ID, EVENT_ID, SQL_TEXT, TIMER_WAIT
FROM performance_schema.events_statements_current
WHERE SQL_TEXT IS NOT NULL;

但默认只保留最近10条SQL（per thread），超过就会被覆盖，MySQL重启后数据全部丢失，因为它存在内存中。适合做实时监控，不适合做审计追溯。

MySQL企业版自带审计插件，功能最完善，可以记录所有SQL，支持过滤规则，输出格式规范。但这是企业版功能，社区版用不了。

我需要的方案是能长期记录关键操作、性能可控、社区版能用。原生能力各有局限，我的思路是用MySQL原生能力做数据采集，外部系统做存储和检索。

我的方案：社区版的审计实现

第一步，用init_connect记录连接信息。init_connect是每个客户端连接时自动执行的一条SQL，我用它在审计日志表里记录连接信息：

CREATE TABLE audit_db.connection_log (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    conn_id INT NOT NULL,
    user_name VARCHAR(128) NOT NULL,
    source_ip VARCHAR(45),
    connect_time DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3),
    db_name VARCHAR(64),
    INDEX idx_conn_time (connect_time),
    INDEX idx_user (user_name)
) ENGINE=InnoDB;

SET GLOBAL init_connect = "
    INSERT INTO audit_db.connection_log(conn_id, user_name, source_ip, db_name)
    VALUES(CONNECTION_ID(), CURRENT_USER(), SUBSTRING_INDEX(USER(), '@', -1), DATABASE())
";

init_connect对root用户不生效，而且如果审计日志表用的是InnoDB，init_connect里的INSERT失败会导致普通用户无法连接。我的做法是用专门的审计账号管理审计日志表，init_connect里的INSERT失败不影响主业务。

第二步，用触发器记录敏感表的变更。对于核心敏感表（用户信息、订单、支付），在表上加触发器，记录每次UPDATE和DELETE：

CREATE TABLE audit_db.user_data_change_log (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    table_name VARCHAR(64) NOT NULL,
    operation VARCHAR(10) NOT NULL,
    old_values JSON,
    new_values JSON,
    changed_by VARCHAR(128),
    change_time DATETIME(3) DEFAULT CURRENT_TIMESTAMP(3),
    source_ip VARCHAR(45),
    INDEX idx_table_time (table_name, change_time)
) ENGINE=InnoDB;

DELIMITER //
CREATE TRIGGER trg_users_after_update
AFTER UPDATE ON app_db.users
FOR EACH ROW
BEGIN
    INSERT INTO audit_db.user_data_change_log
        (table_name, operation, old_values, new_values, changed_by, source_ip)
    VALUES
        ('users', 'UPDATE',
         JSON_OBJECT('user_id', OLD.user_id, 'phone', OLD.phone, 'id_card', OLD.id_card),
         JSON_OBJECT('user_id', NEW.user_id, 'phone', NEW.phone, 'id_card', NEW.id_card),
         CURRENT_USER(),
         (SELECT source_ip FROM audit_db.connection_log
          WHERE conn_id = CONNECTION_ID() ORDER BY connect_time DESC LIMIT 1));
END //
DELIMITER ;

触发器的优点是可以精确记录数据变更的前后值，缺点是只对DML有效，SELECT操作无法捕获，而且本身有性能开销。只在最核心的1-2张敏感表上加，不是所有表都加。

第三步，应用层埋点补上SELECT的审计空白。binlog不记录SELECT，触发器也捕获不到SELECT，对于"谁查了什么数据"这个需求，只能在应用代码里记录审计日志：

@Aspect
public class AuditAspect {
    @Around("execution(* com.example.dao.*.*(..))")
    public Object audit(ProceedingJoinPoint pjp) throws Throwable {
        String sql = extractSql(pjp);
        String user = SecurityContextHolder.getCurrentUser();
        String ip = RequestContextHolder.getRequestIP();
        auditLogDao.insert(new AuditLog(user, ip, sql, System.currentTimeMillis()));
        return pjp.proceed();
    }
}

应用层审计的优势是记录完整上下文：哪个用户、哪个请求、哪条SQL、执行结果。缺点是必须改应用代码，而且绕过应用直接连数据库的场景（比如DBA用命令行）就审计不到了。

最终方案是三层叠加：init_connect记录连接，触发器记录变更，应用层记录查询。三层互相补充，出问题时能快速定位到人、定位到数据、定位到时间。

信创合规下的审计留存要求

国内做数据库审计，绕不开《数据安全法》、《个人信息保护法》、等保2.0。等保2.0三级要求审计记录包括事件的日期和时间、用户、事件类型、事件是否成功，要防止未授权的删除、修改或覆盖，留存时间不少于6个月。

我的方案是：审计日志写入独立的audit_db库，每天把当天的审计日志导出到只读存储（比如OSS归档存储桶），导出后从MySQL中删除超过30天的记录：

#!/bin/bash
# 每天凌晨2点执行
DATE=$(date -d "yesterday" +%Y%m%d)
mysqldump --single-transaction audit_db connection_log \
  --where="connect_time >= '${DATE} 00:00:00' AND connect_time < '${DATE} 23:59:59'" \
  | gzip > /audit_backup/connection_log_${DATE}.sql.gz

# 删除30天前的本地记录
mysql -e "DELETE FROM audit_db.connection_log WHERE connect_time < DATE_SUB(NOW(), INTERVAL 30 DAY)"

6个月的留存靠OSS归档存储满足，MySQL里只保留最近30天的热数据用于快速查询。

避坑清单

init_connect里的SQL不能出错。普通用户连接时会先执行init_connect，如果SQL报错（审计日志表不存在、权限不够），用户连接会直接失败。我之前测试时就遇到过：审计日志表的权限没给对，所有应用连接全部报错，排查了半天才发现是init_connect搞的鬼。先在测试环境验证init_connect的SQL，确认无误再上生产。

触发器会影响写入性能。每张加了触发器的表，每次INSERT/UPDATE/DELETE都要多写一次审计日志表。我实测过，单表触发器写入延迟增加约15%-20%。只在最核心的表上加。

审计日志表的索引设计很重要。审计日志是写多读少场景，平时没人查，出事时要秒级定位。索引必须覆盖最常用的查询条件：时间范围加用户名加表名。我建了(table_name, change_time)的联合索引和(user_name, change_time)的联合索引。索引不是越多越好，会影响写入性能，只保留核心查询的索引。

审计日志本身也要做权限隔离。别把审计日志放在业务库的同一个账号下。独立的数据库、独立的账号，应用账号只有INSERT权限。能删审计日志的人应该是系统权限最高的——我见过有人把审计日志表和业务表放同一个账号，结果开发误操作直接truncate了，审计记录全没了。

总结

经历过那次数据泄露排查，我最大的感触是：审计不是"有了更好"的东西，"有没有"直接决定事后排查是几小时还是几天。MySQL社区版虽然没有开箱即用的审计插件，但init_connect抓连接、触发器抓变更、应用层抓查询，这三层叠在一起，完全能搭出一套满足合规要求的系统。少了哪一层都有盲区，三层互补才是这套方案的关键。

朋友，你的数据库有审计方案吗？出了事能在一小时内定位到谁导出了什么数据吗？来评论区聊聊。

我是数据库小学妹，咱们下篇见 👋