RDS MySQL 基于 audit log 构建数据库审计系统

背景介绍

在人们对数据越来越依赖的今天，数据安全性和合规性对于用户来说至关重要，数据库的监控与审计是安全与合规中必不可少的一环。作为存储核心数据的关系型数据库往往存储着大量敏感和关键数据，任何未经授权的访问或操作都可能导致数据泄露，数据损坏或者违规等严重后果。

Amazon RDS MySQL 数据库提供了审计日志功能，记录对数据库实例的访问活动，包括连接信息，执行的 SQL 语句等，为数据安全与合规审计提供了宝贵的数据来源。然而，原生的审计日志以文件形式集成在 RDS 中，难以集中管理和实时监控，给安全运维带来了挑战。

一种常见做法是将 RDS MySQL 审计日志与 CloudWatch Logs 集成，将审计日志实时流入 CloudWatch Logs。但对于大型数据库，由于会产生大量的审计日志数据，将这些日志全部发送到 CloudWatch logs 并基于合规要求长期保存会产生较高的成本。同时用户如果希望能基于审计日志和其他安全日志联合分析，CloudWatch Logs 也缺乏灵活定制分析和告警的能力。

因此，构建一个实时、可扩展、高性价比的 RDS MySQL 数据库审计日志监控和分析方案就显得尤为重要。它不仅有助于提高数据安全性，及时发现和响应安全威胁，还可以提高运维效率，同时控制总体成本。

解决方案架构

该解决方案利用了 AWS 的无服务器架构，通过以下组件的协同工作，实现了对 RDS MySQL 数据库审计日志的实时监控、分析和告警，同时落入 S3 桶长期保存。

1. Lambda 函数定期轮询 RDS MySQL 数据库审计日志文件，根据日志文件的最后写入时间，获取新的日志记录。
2. 所有原始审计日志数据都会被存储在 Amazon S3 桶中，同时可以使用 S3 的 lifecycle configuration 将日志分步归档， 利于长期保留和分析。
3. Lambda 函数可以过滤掉特定用户（如 rdsadmin）的日志，只将有价值的记录推送到 Kinesis Data Streams。
4. 在 Kinesis Data Streams 中，通过 EventBridge 的管道功能，筛选不同审计日志进入不同的 Lambda 函数，进一步检测判断。EventBridge 管道功能可以进行定制匹配，满足不同筛选的需要。
5. 下游 Lambda 函数进行按需检查，可以将触发告警的事件转换为告警消息，发送到告警系统。
6. Athena 可以直接查询存储在 S3 桶中的原始审计日志数据。

先决条件

• 本文以 RDS MySQL 集群作为展示用例，需要拥有已经运行中的 RDS 集群，并开启审计日志。如果审计日志没有开启，可以查看此处链接。
• 本文中相关服务的创建需要相应的 IAM 权限，此处不再赘述。

服务部署

1. 创建 S3 桶

首先创建一个 S3 桶来长期储存审计日志。可以按需配置加密选项，bucket policy 等，本文使用了默认配置创建。用户也可以按需配置 S3 桶的 lifecycle configuration， 按需归档，优化长期储存成本。关于 S3 桶的 lifecycle configuration， 可以查看此处文档。

2. 创建 Kinesis Data Streams

创建 Kinesis Data Streams 来接入业务相关的数据库审计日志。本文使用的按需模式，创建一个 data stream。用户可以根据自己环境选择按需或者预置模式，以及数据留存期等。本文使用了默认配置。

3. 创建 DynamoDB 表

创建一个 DynamoDB 表来储存元数据，Partition key 设置为 DatabaseInstanceName， 使用按需模式，其他配置用户可以按需调整。

表数据格式示例如下：

    {
        "DatabaseInstanceName": "small-mysql"
        "AuditLogLastWritten": 1713161934,
        "LogFileName": "audit/server_audit.log"
    }

4. 创建连接 RDS MySQL 数据库的 Lambda 函数以及定时器

Lambda 函数 Python 代码查看此处链接。

在全局变量中填写对应的资源与配置。

本文使用 Python 3.12 Runtime 部署。使用默认配置创建 Lambda 函数后，按需调整 Timeout 和内存。本文调整为 1 分钟 Timeout，1024 MB 内存。

Lambda 函数 Execution role，在基本权限外，需要额外加入 S3，RDS, DynamoDB, Kinesis Data Streams 的相应权限，RDS 需要读取审计日志的权限，S3 和 Kinesis Data Streams 需要写入的权限，DynamoDB table 需要读取权限，权限 JSON 代码查看此处链接。创建 IAM policy 可以查看此处文档。

部署后可以测试运行，观察 Lambda 函数是否有 Timeout 现象，以及是否成功写入 S3 桶与 Kinesis Data Streams。

Lambda 函数需要定时检查 RDS MySQL 的审计日志是否有更新，通过对审计日志的定时拉取，按照检查审计日志对应的最后写入时间来判断更新的部分。Lambda 函数的定时触发需要 EventBridge Scheduler 功能来实现，由于本文为测试环境，定时选择为每 5 分钟触发一次 Lambda 函数。

这里值得注意的是，RDS MySQL 审计日志的大小是通过 option group 中的 MARIADB_AUDIT_PLUGIN 来配置的。用户可以通过 SERVER_AUDIT_QUERY_LOG_LIMIT 配置一个记录的长度，通过 SERVER_AUDIT_FILE_ROTATE_SIZE 配置每个审计日志的文件大小，通过 SERVER_AUDIT_FILE_ROTATION 配置审计日志文件 rotation 的数值。具体配置可以查看此处文档。

本文测试均采用默认数值，SERVER_AUDIT_FILE_ROTATE_SIZE 为 1MB，SERVER_AUDIT_FILE_ROTATIONS 为 9。

注意，用户在配置定时触发 Lambda 函数的间隔时间不要超过整体审计日志全部 rotation 的时间长度，不然会出现审计日志文件覆盖的问题，此处需要用户根据自己数据库使用情况配置这几个参数，以及调整 Lambda 函数的触发间隔。用户可以通过观察审计日志最后写入的间隔时间来进行判断，同时考虑波峰流量的情况进行冗余配置。

5. 配置消费 EventBridge 管道下游 Lambda 函数

在配置 EventBridge 管道之前，我们需要配置两个 Lambda 函数来消费不同管道筛选出的事件。这个两个 Lambda 函数用来处理不同管道筛选过的审计日志。用户可以按照业务需求来创建不同的管道，接入不同的 Lambda 函数处理。用户可以在 Lambda 函数中编写合适的检查逻辑，并可以接入告警系统。

如果用户考虑使用电话告警的方式，可以参考此处链接进行配置，Lambda 函数将需要告警的信息接入 SQS 队列中，触发之后的电话告警。

6. 创建 EventBridge 管道

EventBridge 管道是允许您在不同的事件源和目标之间点对点的传输事件数据，并支持事件的筛选与富集。具体内容与配置可以查看此处文档。

在 Kinesis Data Streams 面板选择 EventBridge 管道，选择 Connect Kinesis Data Stream to Pipe。之后会跳转到 EventBridge 管道页面。

在 Source 中我们可以配置不同的起始位置和批次大小。在筛选中我们可以配置筛选的条件。本文中使用筛选数据库名称的方式来提取不同数据库的审计事件，跳过富集步骤，之后接入不同的目标 Lambda 函数中。本文配置了两个管道，分别对应过滤 demo 和 test 为首字的数据库。

筛选数据库名称开头为 demo

{
  "data": {
    "database": [{
      "prefix": "demo"
    }]
  }
}

筛选数据库名称开头为 test

{
  "data": {
    "database": [{
      "prefix": "test"
    }]
  }
}

管道配置如下图所示，依次配置源，筛选，目标即可。

测试

部署完成后，我们可以连接 RDS MySQL 写节点，创建不同的 demo 和 test 数据库来观察审计日志是否按照筛选条件正常流入到相应的下游 Lambda 函数中。测试脚本查看此处链接。脚本中会随机选择创建 demo 或者 test 开头的数据库并录入数据。

观察接入 RDS MySQL 数据库的 Lambda 函数的指标，确认没有 Error 或者 Throttles。如果有问题可以调整 Lambda 函数内存大小和 Timeout 时间。这里我们看到 Lambda 函数的执行时间为 1s 左右，测试的审计日志量在 1M-5M 不等，Lambda 函数的执行时间包括了审计日志上传 S3 桶的时间和写入 Kinesis Data Streams 的时间。

确认审计日志已经落入 S3 桶中。

用户可以通过 Athena 对 S3 桶中的审计日志进行查询。使用 Athena 与 Glue 数据目录配合，根据 RDS MySQL 的审计日志格式，在 Athena query editor 中创建表结构储存在 Glue 数据目录上，之后可以使用这个目录进行 select 请求。示例如下所示。

-- create datbase in glue metadata store
CREATE DATABASE rds_mysql_audit_log
-- create table
CREATE EXTERNAL TABLE rds_mysql_audit_log (
    timestamp STRING,
    serverhost STRING,
    username STRING,
    host STRING,
    connectionid STRING,
    queryid STRING,
    operation STRING,
    `database` STRING,
    object STRING,
    retcode STRING
)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
LOCATION 's3://rds-mysql-audit-log/small-mysql_demo/';
-- sample query
select * from rds_mysql_audit_log where database like 'demo%' limit 10;

之后我们可以检查 EventBridge 管道，确保从 Kinesis Data Streams 进入的数据可以正常调用下游的 Lambda 函数，注意是否存在调用数量不一致，以及错误的情况。之后检查下游对应的 Lambda 函数是否收到筛选正确的审计日志。

总结

本解决方案利用 AWS 云上的无服务器架构，通过 Lambda 函数，Kinesis Data Streams、S3，Athena 和 EventBridge 等服务，实现了对 Amazon RDS MySQL 数据库审计日志的实时采集、处理、存储、分析的功能。

其中，Lambda 函数负责提取 RDS MySQL 数据库审计日志，Kinesis Data Streams 确保了审计日志事件的实时传输，S3 保证了日志的长期储存，EventBridge 管道则支持基于特定条件触发下游系统。同时，Athena 使用 SQL 语法直接分析存储在 S3 的结构化日志数据，提供了强大的数据洞察能力。

通过将 RDS MySQL 审计日志无缝集成到数据分析和安全监控体系中，用户可以显著提高数据安全性和合规性，简化运维流程。该解决方案具有很强的灵活性和扩展性，可根据具体需求进行定制，并与现有的安全运维系统集成。

参考资料

• https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html
• https://aws.amazon.com/blogs/database/configuring-an-audit-log-to-capture-database-activities-for-amazon-rds-for-mysql-and-amazon-aurora-with-mysql-compatibility/
• https://github.com/personio/rds-audit-logs-s3

本篇作者