收集 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 設定」>「動態饋給」。
- 按一下「+ 新增」。
- 選取「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中輸入名稱 (例如
Fivetran Events)。 - 「Source type」(來源類型) 請選取「Webhook」。
- 選取「Fivetran」做為「記錄類型」。
- 點選「下一步」。
- 指定下列輸入參數的值:
- 分割分隔符號 (選用):留空。
- 資產命名空間:資產命名空間。
- 擷取標籤:要套用至這個動態饋給事件的標籤。
- 點選「下一步」。
- 檢查新的新聞動態消息設定,然後按一下「提交」。
產生並儲存密鑰
- 在動態饋給詳細資料頁面中,按一下「產生密鑰」。
- 對話方塊會顯示密鑰。
- 複製並妥善儲存密鑰。
取得動態饋給端點網址
- 前往動態饋給的「詳細資料」分頁。
- 在「Endpoint Information」部分,複製「Feed endpoint URL」。
- 請儲存這個網址,以便進行後續步驟。
- 按一下「完成」。
建立 Google Cloud API 金鑰
Google SecOps 需要 API 金鑰才能進行驗證。
建立 API 金鑰
- 前往 Google Cloud 控制台憑證頁面。
- 選取專案。
- 依序點選「建立憑證」>「API 金鑰」。
- 按一下「編輯 API 金鑰」即可限制金鑰。
限制 API 金鑰
- 在「API 金鑰」設定中:
- 名稱:輸入描述性名稱 (例如
SecOps Webhook API Key)。
- 名稱:輸入描述性名稱 (例如
- 在「API 限制」下方:
- 選取「Restrict key」(限制金鑰)。
- 在下拉式選單中,搜尋並選取「Google SecOps API」 (或「Chronicle API」)。
- 按一下 [儲存]。
- 複製 API 金鑰值並妥善保存。
設定 Fivetran Webhook
建構 Webhook 網址
合併端點網址和 API 金鑰:
<ENDPOINT_URL>?key=<API_KEY>
使用 Fivetran REST API 建立 Webhook
取得 Fivetran API 憑證
- 登入 Fivetran 帳戶。
- 依序前往「Account Settings」>「API Config」。
- 如果沒有 API 金鑰,請按一下「產生 API 金鑰」。
- 複製 API 金鑰和 API 密鑰。
建立帳戶層級 Webhook
使用這個方法,即可接收帳戶中所有連接器的事件。
開啟終端機並執行下列指令:
curl -X POST https://api.fivetran.com/v1/webhooks/account \ -u "API_KEY:API_SECRET" \ -H "Content-Type: application/json" \ -H "Accept: application/json" \ -d '{ "url": "https://{region}-chronicle.googleapis.com/v1/{parent}/logs:import?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" }'- 其中
{region}是 Google SecOps 執行個體的區域 (例如us、europe或asia-southeast1),{parent}是資源路徑,格式為projects/<PROJECT_ID>/locations/<LOCATION>/instances/<INSTANCE_ID>/logTypes/<LOG_TYPE>。
- 其中
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 專業人員尋求解答。