收集 Fivetran 記錄

剖析器版本：1.0

支援的國家/地區：

Google SecOps SIEM

本文說明如何設定 Fivetran，透過 Webhook 將記錄檔推送至 Google Security Operations。

Fivetran 是一個資料整合平台，可自動將各種來源的資料管道傳輸至資料倉儲。Fivetran 會產生作業事件，包括連接器同步事件、轉換事件和連線狀態變更。這些事件可透過外送 Webhook 傳送至外部端點，用於監控、快訊和安全性分析。

事前準備

請確認您已完成下列事前準備事項：

Google SecOps 執行個體。
具備管理員或帳戶層級權限的 Fivetran 帳戶。
存取 Google Cloud 控制台 (用於建立 API 金鑰)。
已啟用 REST API 存取權的 Fivetran 帳戶。

在 Google SecOps 中建立 Webhook 動態饋給

建立動態饋給

依序前往「SIEM 設定」>「動態消息」。
按一下「+ 新增」。
選取「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入名稱 (例如 Fivetran Events)。
選取「Webhook」做為「來源類型」。
選取「Fivetran」做為「記錄類型」。
點選「下一步」。
指定下列輸入參數的值：
- 分割分隔符 (選填)：留空。
- 資產命名空間：資產命名空間。
- 擷取標籤：要套用至這個動態饋給事件的標籤。
點選「下一步」。
檢查新的動態饋給設定，然後按一下「提交」。

產生並儲存密鑰

在動態饋給詳細資料頁面中，按一下「產生密鑰」。
對話方塊會顯示密鑰。
複製並妥善儲存密鑰。

取得動態消息端點網址

前往動態消息的「詳細資料」分頁。
在「端點資訊」部分，複製「動態消息端點網址」。
請儲存這個網址，以供後續步驟使用。
按一下 [完成]。

建立 Google Cloud API 金鑰

Google SecOps 需要 API 金鑰才能進行驗證。

建立 API 金鑰

前往 Google Cloud 控制台的「憑證」頁面。
選取專案。
依序按一下「建立憑證」>「API 金鑰」。
按一下「編輯 API 金鑰」即可限制金鑰。

限制 API 金鑰

在「API 金鑰」設定中：
- 名稱：輸入描述性名稱 (例如 SecOps Webhook API Key)。
在「API 限制」下方：
1. 選取「Restrict key」(限制金鑰)。
2. 在下拉式選單中，搜尋並選取「Google SecOps API」 (或「Chronicle API」)。
按一下 [儲存]。
複製 API 金鑰值並妥善保存。

設定 Fivetran Webhook

建構 Webhook 網址

合併端點網址和 API 金鑰：
```
<ENDPOINT_URL>?key=<API_KEY>
```

使用 Fivetran REST API 建立 Webhook

取得 Fivetran API 憑證

登入 Fivetran 帳戶。
依序前往「帳戶設定」>「API 設定」。
如果沒有 API 金鑰，請按一下「產生 API 金鑰」。
複製 API 金鑰 和 API 密鑰。

建立帳戶層級 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 詳細資料

可用的 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」DATABASE。
不適用	`metadata.product_name`	設為「FIVETRAN」FIVETRAN。
不適用	`metadata.vendor_name`	設為「FIVETRAN」FIVETRAN。

變更記錄

查看這個剖析器的變更記錄

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。