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

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 集中存储的 OCSFOpen Cybersecurity Schema Framework格式安全数据并交由 Wazuh 规则引擎分析。读完本文你将掌握在ossec.conf中配置subscriber typesecurity_lake与bucket typesecurity_lake两种接入方式、为 IAM 角色配置最小权限、验证集成是否生效并理解该功能在仓库中的底层实现原理。背景Security Lake 与 Wazuh 的对接方式Amazon Security Lake 会自动将 AWS 服务、SaaS 提供商和第三方来源的安全数据集中到一个专门构建的、存储在 S3 中的数据湖中并使用 OCSF 开放标准统一数据模型。Wazuh 通过以下链路接入这一数据源Wazuh 订阅一个 SQS 队列该队列在 Security Lake 有新数据可用时收到通知Wazuh 依据通知中携带的对象键object key从 S3 拉取对应的日志文件日志事件经过解析后进入 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 nameaws-s3 disabledno/disabled interval5m/interval run_on_startyes/run_on_start skip_on_erroryes/skip_on_error subscriber typesecurity_lake sqs_namewazuh-security-lake-queue/sqs_name aws_profiledefault/aws_profile iam_role_arnarn: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 nameaws-s3 disabledno/disabled interval10m/interval run_on_startyes/run_on_start bucket typesecurity_lake nameaws-security-data-lake-bucket/name aws_profiledefault/aws_profile iam_role_arnarn:aws:iam::123456789012:role/WazuhSecurityLakeRole/iam_role_arn regionsus-east-1/regions /bucket /wodle该方式使用bucket typesecurity_lake通过name指定桶名、regions限定扫描区域。适用于已明确 Security Lake 数据桶位置、希望按固定周期全量扫描的场景interval可适当放宽如上例为 10 分钟以控制 S3 请求频率。Subscriber 配置选项详解下表汇总了subscriber场景下可用的配置项基于关联文档整理并对照 aws_s3.py 与 aws_tools.py 中的参数校验逻辑OptionRequiredDefaultDescriptiondisabledNono设为yes时禁用 AWS 模块。intervalNo5s两次 SQS 轮询之间的时间间隔。run_on_startNoyes模块启动时立即轮询队列。skip_on_errorNoyes出错时继续处理而非停止。subscriber typeYes—置为security_lake以启用 Security Lake 集成。sqs_nameYes—用于接收 Security Lake 通知的 SQS 队列名。aws_profileNo—用于认证的 AWS CLI Profile 名称。iam_role_arnNo—需要代入assume的 IAM 角色 ARN。iam_role_durationNo—代入 IAM 角色会话的时长秒。external_idNo—跨账户角色代入时使用的 External ID。discard_regexNo—用于过滤匹配事件的正则表达式需配合field属性指定 JSON 字段名格式为discard_regex fieldfieldNameregex/discard_regex。注意Security Lake 订阅者不支持该选项。sts_endpointNo—自定义 AWS STS 端点 URL适用于 VPC 端点场景。service_endpointNo—自定义 AWS 服务端点 URL。需要特别说明的两点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在 AWS 管理控制台进入Amazon Security Lake启用 Security Lake选择要收集数据的 AWS 区域与日志来源配置一个带 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 agentsystemctl restart wazuh-agent查看模块日志确认运行状态grep aws-s3 /var/ossec/logs/ossec.logSecurity 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考