收集檔案掃描架構記錄
剖析器版本:3.0
本文說明如何使用 Google Cloud Storage V2,將檔案掃描架構記錄檔擷取至 Google Security Operations。
檔案掃描架構 (FSF) 是 Emerson Electric Co. 開發的開放原始碼模組化遞迴檔案掃描解決方案。FSF 採用用戶端/伺服器架構分析檔案,並產生詳細的 JSON 掃描結果,包括檔案中繼資料、YARA 簽章比對結果、擷取的子物件,以及模組專屬中繼資料。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 已部署的 FSF 伺服器執行個體,具備記錄目錄的寫入權限
- FSF 伺服器主機的 root 或 sudo 存取權
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 fsf-logs-secops)。位置類型 請根據需求選擇 (區域、雙區域、多區域)。 位置 選取位置 (例如 us-central1)。儲存空間級別 標準 (建議用於經常存取的記錄)。 存取控管 一致 (建議)。 保護工具 選用:啟用物件版本管理或保留政策。 點選「建立」。
設定 FSF 記錄輸出目錄
FSF 會將 JSON 掃描結果寫入可設定的記錄目錄。為 Google SecOps 擷取作業設定專屬目錄。
- 透過 SSH 連線至 FSF 伺服器主機。
開啟 FSF 伺服器設定檔:
sudo nano /opt/fsf/fsf-server/conf/config.py找出
SCANNER_CONFIG字典。將
LOG_PATH參數更新為專屬目錄:SCANNER_CONFIG = { 'LOG_PATH': '/var/log/fsf', 'YARA_PATH': '/opt/fsf/fsf-server/yara/rules.yara', 'PID_PATH': '/tmp/scanner.pid', 'EXPORT_PATH': '/tmp', 'TIMEOUT': 60, 'MAX_DEPTH': 10 }儲存並關閉檔案。
建立具有適當權限的記錄目錄:
sudo mkdir -p /var/log/fsf sudo chown -R fsf:fsf /var/log/fsf sudo chmod 755 /var/log/fsf重新啟動 FSF 伺服器以套用變更:
sudo systemctl restart fsf確認 FSF 是否正在將記錄寫入新目錄:
ls -lh /var/log/fsf/
安裝及設定 Fluentd
Fluentd 會追蹤 FSF 記錄檔,並將其傳送至 Google Cloud Storage。
安裝 Fluentd
在 FSF 伺服器主機上安裝 Fluentd (td-agent):
curl -fsSL https://toolbelt.treasuredata.com/sh/install-ubuntu-jammy-td-agent4.sh | sh安裝 GCS 輸出外掛程式:
sudo td-agent-gem install fluent-plugin-gcs驗證外掛程式安裝作業:
td-agent-gem list | grep fluent-plugin-gcs
為 Fluentd 建立 GCP 服務帳戶
- 在 GCP 控制台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
fsf-fluentd-shipper。 - 服務帳戶說明:輸入
Service account for Fluentd to ship FSF logs to GCS。
- 服務帳戶名稱:輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分:
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 按一下「繼續」。
- 按一下 [完成]。
建立服務帳戶金鑰
- 在「服務帳戶」清單中,按一下服務帳戶 (
fsf-fluentd-shipper)。 - 前往「金鑰」分頁標籤。
- 依序點選「新增金鑰」>「建立新的金鑰」。
- 選取「JSON」做為金鑰類型。
- 點選「建立」。
- JSON 金鑰檔案會下載至您的電腦。
將金鑰檔案轉移到 FSF 伺服器主機:
scp /path/to/downloaded-key.json user@fsf-server:/etc/td-agent/gcp-key.json設定金鑰檔案的適當權限:
sudo chown td-agent:td-agent /etc/td-agent/gcp-key.json sudo chmod 600 /etc/td-agent/gcp-key.json
授予 GCS 值區的 IAM 權限
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (
fsf-logs-secops)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (例如
fsf-fluentd-shipper@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色:選取「Storage 物件管理員」。
- 新增主體:輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
設定 Fluentd
在 FSF 伺服器主機上,建立 Fluentd 設定檔:
sudo nano /etc/td-agent/td-agent.conf新增下列設定:
# Tail FSF JSON logs <source> @type tail path /var/log/fsf/*.log pos_file /var/log/td-agent/fsf.log.pos tag fsf.scan read_from_head true <parse> @type json time_key timestamp time_format %Y-%m-%dT%H:%M:%S.%L%z </parse> </source> # Ship to Google Cloud Storage <match fsf.scan> @type gcs project YOUR_GCP_PROJECT_ID keyfile /etc/td-agent/gcp-key.json bucket fsf-logs-secops object_key_format %{path}%{time_slice}_%{index}.%{file_extension} path fsf-logs/ <buffer tag,time> @type file path /var/log/td-agent/buffer/gcs timekey 3600 timekey_wait 10m timekey_use_utc true chunk_limit_size 10MB </buffer> <format> @type json </format> store_as json auto_create_bucket false </match>將
YOUR_GCP_PROJECT_ID替換為實際的 GCP 專案 ID。儲存並關閉檔案。
建立緩衝區目錄:
sudo mkdir -p /var/log/td-agent/buffer/gcs sudo chown -R td-agent:td-agent /var/log/td-agent/buffer重新啟動 Fluentd 以套用設定:
sudo systemctl restart td-agent啟用 Fluentd,在開機時啟動:
sudo systemctl enable td-agent確認 Fluentd 正在執行:
sudo systemctl status td-agent
確認記錄檔傳送情形
檢查 Fluentd 記錄是否有錯誤:
sudo tail -f /var/log/td-agent/td-agent.log觸發測試 FSF 掃描,產生記錄:
echo "test content" > /tmp/test.txt /opt/fsf/fsf-client/fsf_client.py /tmp/test.txt --suppress-report等待 1 到 2 分鐘,讓 Fluentd 處理及傳送記錄。
在 GCP 控制台中,依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱 (
fsf-logs-secops)。前往
fsf-logs/前置字元。確認系統是否正在建立含有時間戳記的 JSON 檔案。
下載並檢查檔案,確認其中包含 JSON 格式的 FSF 掃描結果。
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
FSF File Scanning Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「檔案掃描架構」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址,例如:
secops-12345678@secops-gcp-prod.iam.gserviceaccount.com複製電子郵件地址,以便在下一步中使用。
點選「下一步」。
指定下列輸入參數的值:
儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:
gs://fsf-logs-secops/fsf-logs/來源刪除選項:根據偏好設定選取刪除選項:
- 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。
檔案存在時間上限:包含在過去天數內修改的檔案 (預設為 180 天)。
資產命名空間:資產命名空間。
擷取標籤:要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (
fsf-logs-secops)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色:選取「Storage 物件檢視者」。
- 按一下 [儲存]。
驗證擷取作業
- 等待 10 到 15 分鐘,讓系統完成初始擷取作業。
- 在 Google SecOps 中,依序前往「SIEM 設定」>「動態消息」。
- 找出動態消息 (
FSF File Scanning Logs)。 - 確認「狀態」顯示為「有效」。
- 按一下動態饋給名稱,即可查看擷取指標。
- 確認「Events ingested」(擷取的事件) 計數正在增加。
- 前往 Google SecOps 中的「搜尋」。
執行搜尋查詢,確認系統是否正在擷取 FSF 記錄:
metadata.log_type = "FILE_SCANNING_FRAMEWORK"確認搜尋結果中會顯示 FSF 掃描結果。
疑難排解
GCS 未顯示任何記錄
確認 FSF 是否將記錄寫入
/var/log/fsf/:ls -lh /var/log/fsf/ tail -f /var/log/fsf/*.log檢查 Fluentd 記錄是否有錯誤:
sudo tail -f /var/log/td-agent/td-agent.log確認 GCP 服務帳戶金鑰有效,且具備正確權限。
確認 Fluentd 設定中的 bucket 名稱與實際 bucket 名稱相符。
Fluentd 權限錯誤
- 確認服務帳戶 (
fsf-fluentd-shipper) 在 bucket 中具備「Storage 物件管理員」角色。 - 確認 Fluentd 設定中的金鑰路徑是否正確。
確認金鑰檔案的擁有權和權限是否正確:
ls -l /etc/td-agent/gcp-key.json
Google SecOps 未擷取記錄
- 確認 Google SecOps 服務帳戶在 bucket 中具有「Storage 物件檢視者」角色。
- 確認動態消息設定中的 bucket URI 正確無誤,且包含尾端斜線。
- 確認檔案位於指定前置字串路徑的 GCS bucket 中。
- 在「SIEM 設定」>「動態饋給」中檢查動態饋給狀態,查看是否有錯誤訊息。
FSF 記錄格式不符
- 確認 FSF 已設定為寫入 JSON 輸出內容 (預設行為)。
- 確認 Fluentd
<parse>區段已設定@type json。 手動檢查記錄檔,確認其中包含有效的 JSON:
head -n 1 /var/log/fsf/*.log | jq .
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| CompressType_label、compressed_parents | about.labels | 從 CompressType_label (鍵為「Compress Type」,如果訊息包含「Compress Type」,則值來自 Object.EXTRACT_ZIP.Object_0.Compress Type) 和 compressed_parents (鍵為「Compressed Parent Files」,串連自 Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.additional_info.compressed_parents) 合併 |
| Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.MD5、Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.MD5、Object.EXTRACT_SWF.META_BASIC_INFO.MD5、Object.EXTRACT_GZIP.META_BASIC_INFO.MD5、Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.MD5 | intermediary.file.md5 | 如果存在 EXTRACT_EMBEDDED,則為 Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.MD5;如果存在 EXTRACT_ZIP,則為 Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.MD5;如果存在 EXTRACT_SWF,則為 Object.EXTRACT_SWF.META_BASIC_INFO.MD5;如果存在 EXTRACT_GZIP,則為 Object.EXTRACT_GZIP.META_BASIC_INFO.MD5;如果存在 EXTRACT_CAB,則為 Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.MD5 |
| Object.EXTRACT_EMBEDDED.Object_0.Description | intermediary.file.mime_type | 直接複製值 |
| Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA1、Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA1、Object.EXTRACT_SWF.META_BASIC_INFO.SHA1、Object.EXTRACT_GZIP.META_BASIC_INFO.SHA1、Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA1 | intermediary.file.sha1 | 如果存在 EXTRACT_EMBEDDED,則為 Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA1 的值;如果存在 EXTRACT_ZIP,則為 Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA1 的值;如果存在 EXTRACT_SWF,則為 Object.EXTRACT_SWF.META_BASIC_INFO.SHA1 的值;如果存在 EXTRACT_GZIP,則為 Object.EXTRACT_GZIP.META_BASIC_INFO.SHA1 的值;否則為 Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA1 的值。 |
| Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA256、Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA256、Object.EXTRACT_SWF.META_BASIC_INFO.SHA256、Object.EXTRACT_GZIP.META_BASIC_INFO.SHA256、Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA256 | intermediary.file.sha256 | 如果存在 EXTRACT_EMBEDDED,則為 Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.SHA256;如果存在 EXTRACT_ZIP,則為 Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.SHA256;如果存在 EXTRACT_SWF,則為 Object.EXTRACT_SWF.META_BASIC_INFO.SHA256;如果存在 EXTRACT_GZIP,則為 Object.EXTRACT_GZIP.META_BASIC_INFO.SHA256;否則為 Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.SHA256 |
| Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.Size、Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.Size、Object.EXTRACT_SWF.META_BASIC_INFO.Size、Object.EXTRACT_GZIP.META_BASIC_INFO.Size、Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.Size | intermediary.file.size | Value from Object.EXTRACT_EMBEDDED.Object_0.META_BASIC_INFO.Size if EXTRACT_EMBEDDED present, else Object.EXTRACT_ZIP.Object_0.META_BASIC_INFO.Size if EXTRACT_ZIP present, else Object.EXTRACT_SWF.META_BASIC_INFO.Size if EXTRACT_SWF present, else Object.EXTRACT_GZIP.META_BASIC_INFO.Size if EXTRACT_GZIP present, else Object.EXTRACT_CAB.Object_0.META_BASIC_INFO.Size; stripped of trailing " .*" and converted to uinteger |
| Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.scan_id | intermediary.resource.id | 直接複製值 |
| Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.permalink | intermediary.url | 直接複製值 |
| Object.META_EMERSON_INFO.results | intermediary.user.email_addresses | 從結果陣列中的 matched_email 合併 |
| Summary.Observations | metadata.description | 以「, 」分隔符號從陣列串連,並移除開頭的半形逗號 |
| 掃描時間 | metadata.event_timestamp | 使用日期篩選器轉換,格式為 yyyy-MM-dd HH:mm:ss |
| 來源 | metadata.event_type | 如果來源不為空白,請設為「SCAN_FILE」,否則設為「GENERIC_EVENT」 |
| Object.META_VT_CACHE._id | metadata.product_log_id | 直接複製值 |
| result.ad_data.message | network.http.response_code | 使用 grok 模式 INT 從 result.ad_data.message 中擷取為整數 |
| 來源 | principal.hostname | 直接複製值 |
| Object.META_EMERSON_INFO.result_summary、Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.verbose_msg | security_result.summary | 如有 Object.META_EMERSON_INFO.result_summary,則設為該值,否則設為 Object.EXTRACT_ZIP.Object_0.META_VT_CACHE.vt_data.verbose_msg |
| 檔案名稱 | target.file.full_path | 直接複製值 |
| Object.META_BASIC_INFO.MD5 | target.file.md5 | 直接複製值 |
| Summary.Yara | target.file.mime_type | 從 Summary.Yara 的第一個索引擷取,並轉換為大寫,如果存在 Yara,則移除「FT_」,否則如果存在 EXTRACT_ZIP,則設為「ZIP」;如果存在 EXTRACT_SWF,則設為「SWF」;如果存在 EXTRACT_GZIP,則設為「GZIP」;如果存在 EXTRACT_CAB,則設為「CAB」 |
| Object.META_BASIC_INFO.SHA1、Object.META_VT_CACHE.SHA1 | target.file.sha1 | 如果 Object.META_BASIC_INFO.SHA1 不為空,則為該值,否則為 Object.META_VT_CACHE.SHA1 |
| Object.META_BASIC_INFO.SHA256 | target.file.sha256 | 直接複製值 |
| Object.META_BASIC_INFO.Size | target.file.size | 已移除結尾的「 .*」,並轉換為 uinteger |
| metadata.vendor_name | 設為「EMERSON」 | |
| metadata.product_name | 設為「FILE SCANNING FRAMEWORK」 |
還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求答案。