企业官网建设流程全解析

Wazuh 集成 Amazon Security Lake：通过 SQS 订阅与 S3 接入 OCSF 安全数据

【免费下载链接】wazuhWazuh - The Open Source Security Platform. Unified XDR and SIEM protection for endpoints and cloud workloads.项目地址: https://gitcode.com/GitHub_Trending/wa/wazuh

导读

本文介绍 Wazuh AWS 模块如何接入 Amazon Security Lake：通过订阅 SQS 通知队列，拉取 S3 中由 Security Lake 集中存储的 OCSF（Open Cybersecurity Schema Framework）格式安全数据，并交由 Wazuh 规则引擎分析。读完本文，你将掌握在ossec.conf中配置subscriber type="security_lake"与bucket type="security_lake"两种接入方式、为 IAM 角色配置最小权限、验证集成是否生效，并理解该功能在仓库中的底层实现原理。

背景：Security Lake 与 Wazuh 的对接方式

Amazon Security Lake 会自动将 AWS 服务、SaaS 提供商和第三方来源的安全数据集中到一个专门构建的、存储在 S3 中的数据湖中，并使用 OCSF 开放标准统一数据模型。Wazuh 通过以下链路接入这一数据源：

1. Wazuh 订阅一个 SQS 队列，该队列在 Security Lake 有新数据可用时收到通知；
2. Wazuh 依据通知中携带的对象键（object key）从 S3 拉取对应的日志文件；
3. 日志事件经过解析后进入 Wazuh 规则引擎，生成带aws数据字段（含 OCSF 格式事件数据）的告警。

这一模式的优势在于：Wazuh 无需直接轮询 Security Lake 的存储布局，而是借助 SQS 通知按需拉取，降低 API 调用与数据扫描开销。

从仓库源码看，这一流程被封装在 wodles/aws/subscribers 目录中：sqs_queue.py负责 SQS 队列的连接与消息消费，sqs_message_processor.py负责解析通知负载，s3_log_handler.py负责从 S3 读取并解析日志文件。

前置条件

在开始配置前，请确认以下条件已满足：

• 一个已启用 Amazon Security Lake 的 AWS 账户；
• 一个已配置为 Security Lake 订阅源的 SQS 队列；
• 具备读取 SQS 队列与 Security Lake S3 桶权限的 AWS 凭证或 IAM 角色；
• Wazuh agent 上已安装 Python 3 与boto3库（Security Lake 场景还需pyarrow，详见下文实现解析）。

配置方式一：SQS 订阅（推荐）

在 Wazuh agent 的ossec.conf中，通过subscriber元素并将type设为security_lake来启用集成：

<wodle name="aws-s3"> <disabled>no</disabled> <interval>5m</interval> <run_on_start>yes</run_on_start> <skip_on_error>yes</skip_on_error> <subscriber type="security_lake"> <sqs_name>wazuh-security-lake-queue</sqs_name> <aws_profile>default</aws_profile> <iam_role_arn>arn:aws:iam::123456789012:role/WazuhSecurityLakeRole</iam_role_arn> </subscriber> </wodle>

配置要点：

• disabled置为no启用模块，interval控制两次轮询 SQS 之间的时间间隔（此处为 5 分钟）；
• run_on_start为yes时模块启动即立即轮询一次队列；
• sqs_name必须是真实存在的队列名。在 sqs_queue.py 的_get_sqs_url()中，模块启动时会调用get_queue_url解析队列 URL，若队列不存在会报错并以退出码 20 终止；
• aws_profile指定 AWS CLI 配置文件中使用的凭证 Profile，缺省时使用default。

配置方式二：直接读取 S3 桶

如果不想依赖 SQS 通知，也可以通过bucket元素直接访问 Security Lake 的存储桶：

<wodle name="aws-s3"> <disabled>no</disabled> <interval>10m</interval> <run_on_start>yes</run_on_start> <bucket type="security_lake"> <name>aws-security-data-lake-bucket</name> <aws_profile>default</aws_profile> <iam_role_arn>arn:aws:iam::123456789012:role/WazuhSecurityLakeRole</iam_role_arn> <regions>us-east-1</regions> </bucket> </wodle>

该方式使用bucket type="security_lake"，通过<name>指定桶名、<regions>限定扫描区域。适用于已明确 Security Lake 数据桶位置、希望按固定周期全量扫描的场景；interval可适当放宽（如上例为 10 分钟）以控制 S3 请求频率。

Subscriber 配置选项详解

下表汇总了subscriber场景下可用的配置项（基于关联文档整理，并对照 aws_s3.py 与 aws_tools.py 中的参数校验逻辑）：

需要特别说明的两点：

1. 认证参数的强校验。在 aws_tools.py 的arg_validate_security_lake_auth_params()中，Security Lake 订阅场景对iam_role_arn、queue（对应sqs_name）和external_id做了强制要求——三者任一缺失都会以退出码 21 终止运行。这一点与普通buckets订阅不同（普通订阅允许仅凭aws_profile运行）。因此生产环境中若使用订阅方式，务必在 AWS 侧为角色配置 External ID，并在配置中显式填写。

2. 队列名校验。aws_tools.py 的args_valid_sqs_name()要求sqs_name长度为 1~80 个字符，且仅允许字母、数字、连字符（-）与下划线（_）。

AWS 侧配置

启用 Amazon Security Lake

1. 在 AWS 管理控制台进入Amazon Security Lake；
2. 启用 Security Lake，选择要收集数据的 AWS 区域与日志来源；
3. 配置一个带 SQS 队列的订阅者（subscriber），用于接收数据可用通知。

IAM 权限

运行 Wazuh AWS 模块的 IAM 用户或角色需要以下最小权限（Policy JSON 可直接使用）：

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:GetQueueUrl" ], "Resource": "arn:aws:sqs:*:*:wazuh-security-lake-queue" }, { "Effect": "Allow", "Action": [ "s3:GetObject", "s3:ListBucket" ], "Resource": [ "arn:aws:s3:::aws-security-data-lake-*", "arn:aws:s3:::aws-security-data-lake-*/*" ] } ] }

权限设计要点：

• SQS 侧只需ReceiveMessage、DeleteMessage与GetQueueUrl三项——对应源码中fetch_messages()（调用receive_message）、delete_message()（调用delete_message）与_get_sqs_url()（调用get_queue_url）三个操作；
• S3 侧GetObject用于拉取数据文件，ListBucket用于桶内容遍历（直接桶接入场景必需）；
• 若采用跨账户角色代入方式，还需在角色信任策略中配置sts:AssumeRole权限，并按上文所述设置 External ID。

验证集成

修改配置后重启 Wazuh agent：

systemctl restart wazuh-agent

查看模块日志确认运行状态：

grep "aws-s3" /var/ossec/logs/ossec.log

Security Lake 事件将生成包含aws数据字段的告警，其中携带 OCSF 格式的事件数据。

源码级实现解析

消息解析：SQS 通知 → 对象键

Security Lake 的 SQS 通知负载结构不同于普通 S3 事件通知。在 sqs_message_processor.py 中，AWSSSecLakeMessageProcessor.parse_message()从message["detail"]["object"]["key"]与message["detail"]["bucket"]["name"]提取日志文件路径与桶名（普通 S3 订阅则从message["Records"][0]["s3"]["object"]["key"]提取）。字段位置不匹配时会返回raw_message，由上层跳过该消息。

事件消费循环

sqs_queue.py 的sync_events()是核心循环：通过get_messages()批量拉取（每次最多 10 条，长轮询等待 20 秒）→ 逐条交由bucket_handler.process_file()处理 → 处理成功后调用delete_message()删除消息 → 继续拉取下一批，直至队列为空。skip_on_error语义也由此体现：单条消息处理异常不会中断整体循环。

Parquet 文件处理

Security Lake 的数据文件以 Apache Parquet 格式存储，这与普通 S3 订阅的 JSON/CSV/纯文本处理路径截然不同。s3_log_handler.py 中的AWSSLSubscriberBucket.obtain_logs()使用pyarrow.parquet读取文件：先通过 S3 客户端get_object拉取对象字节流，再以ParquetFile.iter_batches()分块迭代，将每条记录序列化为 JSON 后直接交给send_msg()发送。若机器缺少pyarrow，模块会在导入阶段即以退出码 10 报错（见 s3_log_handler.py）——因此 Security Lake 场景除boto3外还需安装pyarrow。

参数校验的测试佐证

在 test_aws_s3.py 中可以看到官方测试用例如下：

['main', '--subscriber', 'security_lake', '--iam_role_arn', ..., '--external_id', ..., '--queue', ...]

该用例同时断言实例化subscribers.sqs_queue.AWSSQSQueue并调用sync_events()，从测试侧印证了：Security Lake 订阅模式要求同时提供--iam_role_arn、--external_id与--queue三个参数，且最终执行路径与上述实现完全一致。

小结

通过 SQS 订阅方式，Wazuh 可以近乎实时地消费 Amazon Security Lake 提供的 OCSF 安全数据，并复用既有的规则引擎与分析管道；直接桶接入则提供了不依赖通知队列的兜底方案。配置时请特别留意：订阅模式必须配置iam_role_arn、external_id与sqs_name（源码强制校验），并在 agent 上安装boto3与pyarrow。相关实现均位于仓库 wodles/aws 目录，可进一步阅读 aws_s3.py、subscribers/sqs_queue.py、subscribers/sqs_message_processor.py 与 subscribers/s3_log_handler.py 获取完整细节。

【免费下载链接】wazuhWazuh - The Open Source Security Platform. Unified XDR and SIEM protection for endpoints and cloud workloads.项目地址: https://gitcode.com/GitHub_Trending/wa/wazuh