收集 Fivetran 日志

解析器版本：1.0

支持的平台：

Google SecOps SIEM

本文档介绍了如何配置 Fivetran 以使用 Webhook 将日志推送到 Google Security Operations。

Fivetran 是一个数据集成平台，可自动将各种来源的数据流水线传输到数据仓库。Fivetran 会生成运营事件，包括连接器同步事件、转换事件和连接状态更改。这些事件可以通过出站 Webhook 发送到外部端点，以进行监控、提醒和安全分析。

准备工作

请确保满足以下前提条件：

Google SecOps 实例。
具有管理员或账号级权限的 Fivetran 账号。
对 Google Cloud Console 的访问权限（用于创建 API 密钥）。
已启用 REST API 访问权限的 Fivetran 账号。

在 Google SecOps 中创建 Webhook Feed

创建 Feed

依次前往 SIEM 设置 > Feed。
点击 + 添加新对比项。
选择配置单个 Feed。
在 Feed 名称字段中，输入一个名称（例如 Fivetran Events）。
选择 Webhook 作为来源类型。
选择 Fivetran 作为日志类型。
点击下一步。
为以下输入参数指定值：
- 拆分分隔符（可选）：留空。
- 资产命名空间：资产命名空间。
- 注入标签：要应用于此 Feed 中事件的标签。
点击下一步。
检查新的 Feed 配置，然后点击提交。

生成并保存密钥

在 Feed 详情页面上，点击生成密钥。
系统会显示一个对话框，其中包含密钥。
复制并妥善保存此密钥。

获取 Feed 端点网址

前往相应 Feed 的详细信息标签页。
在端点信息部分，复制 Feed 端点网址。
保存此网址以供后续步骤使用。
点击完成。

创建 Google Cloud API 密钥

Google SecOps 需要使用 API 密钥进行身份验证。

创建 API 密钥

前往 Google Cloud 控制台的“凭据”页面。
选择您的项目。
依次点击创建凭据 > API 密钥。
点击修改 API 密钥以限制密钥。

限制 API 密钥

在 API 密钥设置中：
- 名称：输入一个描述性名称（例如 SecOps Webhook API Key）。
在 API 限制下：
1. 选择限制密钥。
2. 在下拉菜单中，搜索并选择 Google SecOps API（或 Chronicle API）。
点击保存。
复制 API 密钥值并妥善保存。

配置 Fivetran webhook

构建网络钩子网址

将端点网址和 API 密钥组合在一起：
```
<ENDPOINT_URL>?key=<API_KEY>
```

使用 Fivetran REST API 创建 Webhook

获取 Fivetran API 凭据

登录您的 Fivetran 账号。
依次前往账号设置 > API 配置。
如果您还没有 API 密钥，请点击生成 API 密钥。
复制 API 密钥和 API Secret。

创建账号级 webhook

使用此方法可接收账号中所有连接器的事件。

打开终端并运行以下命令：

curl -X POST [https://api.fivetran.com/v1/webhooks/account](https://api.fivetran.com/v1/webhooks/account) \
        -u "API_KEY:API_SECRET" \
        -H "Content-Type: application/json" \
        -H "Accept: application/json" \
        -d '{
                "url": "[https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate?key=YOUR_CHRONICLE_API_KEY](https://malachiteingestion-pa.googleapis.com/v2/unstructuredlogentries:batchCreate?key=YOUR_CHRONICLE_API_KEY)",
                "events": [
                        "sync_start",
                        "sync_end",
                        "transformation_start",
                        "transformation_succeeded",
                        "transformation_failed",
                        "connection_successful",
                        "connection_failure",
                        "create_connector",
                        "pause_connector",
                        "resume_connector",
                        "edit_connector",
                        "delete_connector",
                        "force_update_connector",
                        "resync_connector",
                        "resync_table"
                ],
                "active": true,
                "secret": "YOUR_CHRONICLE_SECRET_KEY"
        }'

网络钩子详细信息

可用的 Webhook 事件

事件	说明
`sync_start`	连接器同步已开始
`sync_end`	连接器同步已完成
`transformation_start`	已开始转换
`transformation_succeeded`	已成功完成转换
`transformation_failed`	转换失败
`connection_successful`	连接测试成功
`connection_failure`	连接测试失败
`create_connector`	已创建新连接器

Webhook 重试行为

Fivetran 会自动重试失败的 Webhook，最长重试时间为 24 小时，重试时间表如下：

重试	首次尝试后的时间
第 1 次重试	6 分钟
第 2 次重试	27 分钟
第 3 次重试	1 小时 45 分钟
第 4 次重试	6 小时 25 分钟
第 5 次重试	23 小时 13 分钟

UDM 映射表

日志字段	UDM 映射	逻辑
`jsonPayload.connector_id`	`additional.connector_id`	直接复制值。
`jsonPayload.connector_type`	`additional.connector_type`	直接复制值。
`jsonPayload.data.query`	`additional.query`	直接复制值。
不适用	`metadata.event_type`	根据正文和目标的存在情况设置。
`jsonPayload.event`	`metadata.product_event_type`	直接复制值。
`jsonPayload.sync_id`	`metadata.product_log_id`	直接复制值。
`jsonPayload.connector_name`	`principal.hostname`	直接复制值。
`resource.labels.email_id`	`principal.user.email_addresses`	如果格式为有效的电子邮件地址，则进行映射。
`resource.labels.unique_id`	`principal.user.userid`	直接复制值。
`severity`	`security_result.severity`	“INFO”映射到 INFORMATIONAL。
`logName`	`target.resource.name`	直接复制值。
不适用	`target.resource.type`	设置为 DATABASE。
不适用	`metadata.product_name`	设置为 FIVETRAN。
不适用	`metadata.vendor_name`	设置为 FIVETRAN。

更新日志

查看相应解析器的更改日志