收集 CSV 自訂 IOC 檔案

支援的國家/地區：

Google SecOps SIEM

本文說明如何使用 Google Cloud Storage 將 CSV 自訂 IOC 檔案擷取至 Google Security Operations，然後將這些欄位對應至 UDM、處理各種資料類型 (例如 IP、網域和雜湊)，並在輸出內容中加入威脅詳細資料、實體資訊和嚴重程度。

事前準備

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

Google SecOps 執行個體
已啟用 Cloud Storage API 的 GCP 專案
建立及管理 GCS 值區的權限
管理 Google Cloud Storage 值區 IAM 政策的權限
建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
存取一或多個 CSV IOC 動態饋給網址 (HTTPS)，或提供 CSV 的內部端點

建立 Google Cloud Storage 值區

前往 Google Cloud 控制台。
選取專案或建立新專案。
在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
按一下「建立值區」。

請提供下列設定詳細資料：

設定	值
為 bucket 命名	輸入全域不重複的名稱 (例如 `csv-ioc-logs`)
位置類型	根據需求選擇 (區域、雙區域、多區域)
位置	選取地點 (例如 `us-central1`)
儲存空間級別	標準 (建議用於經常存取的記錄)
存取控管	統一 (建議)
保護工具	選用：啟用物件版本管理或保留政策

點選「建立」。

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶，並由 Pub/Sub 叫用。

建立服務帳戶

在 GCP 主控台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
按一下 [Create Service Account] (建立服務帳戶)。
請提供下列設定詳細資料：
- 服務帳戶名稱：輸入 csv-ioc-collector-sa。
- 服務帳戶說明：輸入 Service account for Cloud Run function to collect CSV IOC files。
按一下「建立並繼續」。
在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
1. 按一下「選擇角色」。
2. 搜尋並選取「Storage 物件管理員」。
3. 點選「+ 新增其他角色」。
4. 搜尋並選取「Cloud Run Invoker」。
5. 點選「+ 新增其他角色」。
6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
按一下「繼續」。
按一下 [完成]。

這些角色適用於：

Storage 物件管理員：將記錄檔寫入 GCS bucket
Cloud Run 叫用者：允許 Pub/Sub 叫用函式
Cloud Functions 叫用者：允許函式叫用

授予 GCS 值區的 IAM 權限

授予服務帳戶 GCS bucket 的寫入權限：

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如 csv-ioc-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。
- 指派角色：選取「Storage 物件管理員」。
按一下 [儲存]。

建立 Pub/Sub 主題

建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。

在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
按一下「建立主題」。
請提供下列設定詳細資料：
- 主題 ID：輸入 csv-ioc-trigger。
- 其他設定保留預設值。
點選「建立」。

建立 Cloud Run 函式，收集 CSV IOC 檔案

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 HTTPS 端點擷取 CSV IOC 檔案，並將檔案寫入 GCS。

前往 GCP Console 的「Cloud Run」。
按一下「Create service」(建立服務)。
選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：

設定值

服務名稱 csv-ioc-collector

區域選取與 GCS bucket 相符的區域 (例如 us-central1)

執行階段 選取「Python 3.12」以上版本
在「Trigger (optional)」(觸發條件 (選用)) 專區：
1. 按一下「+ 新增觸發條件」。
2. 選取「Cloud Pub/Sub」。
3. 在「選取 Cloud Pub/Sub 主題」中，選擇 Pub/Sub 主題 (csv-ioc-trigger)。
4. 按一下 [儲存]。
在「Authentication」(驗證) 部分：
1. 選取「需要驗證」。
2. 檢查 Identity and Access Management (IAM)。
注意： 叫用函式時，Pub/Sub 會自動處理驗證。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶 (csv-ioc-collector-sa)。

設定	值
服務名稱	`csv-ioc-collector`
區域	選取與 GCS bucket 相符的區域 (例如 `us-central1`)
執行階段	選取「Python 3.12」以上版本

前往「容器」分頁：

按一下「變數與密鑰」。
針對每個環境變數，按一下「+ 新增變數」：

變數名稱	範例值	說明
`GCS_BUCKET`	`csv-ioc-logs`	GCS bucket 名稱
`GCS_PREFIX`	`csv-ioc`	記錄檔的前置字串
`IOC_URLS`	`https://ioc.example.com/feed.csv,https://another.example.org/iocs.csv`	以半形逗號分隔的 HTTPS 網址
`AUTH_HEADER`	`Authorization: Bearer <token>`	選用驗證標頭
`TIMEOUT`	`60`	要求逾時時間 (以秒為單位)

在「變數與密鑰」部分，向下捲動至「要求」：
- 要求逾時：輸入 600 秒 (10 分鐘)。
前往「設定」分頁：
- 在「資源」部分：
  - 記憶體：選取 512 MiB 以上。
  - CPU：選取 1。
- 按一下 [完成]。
在「修訂版本資源調度」部分：
- 執行個體數量下限：輸入 0。
- 「Maximum number of instances」(執行個體數量上限)：輸入 100 (或根據預期負載調整)。
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立完成後，系統會自動開啟內嵌程式碼編輯器。

新增函式程式碼

在「Function entry point」(函式進入點) 中輸入 main

在內嵌程式碼編輯器中建立兩個檔案：

第一個檔案：main.py:

import functions_framework
from google.cloud import storage
import json
import os
import urllib3
from datetime import datetime, timezone
import time

# Initialize HTTP client with timeouts
http = urllib3.PoolManager(
    timeout=urllib3.Timeout(connect=5.0, read=30.0),
    retries=False,
)

# Initialize Storage client
storage_client = storage.Client()

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch CSV IOC feeds over HTTPS and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    # Get environment variables
    bucket_name = os.environ.get('GCS_BUCKET')
    prefix = os.environ.get('GCS_PREFIX', 'csv-ioc').strip('/')
    ioc_urls_str = os.environ.get('IOC_URLS', '')
    auth_header = os.environ.get('AUTH_HEADER', '')
    timeout = int(os.environ.get('TIMEOUT', '60'))

    ioc_urls = [u.strip() for u in ioc_urls_str.split(',') if u.strip()]

    if not bucket_name:
        print('Error: GCS_BUCKET environment variable is required')
        return

    if not ioc_urls:
        print('Error: IOC_URLS must contain at least one HTTPS URL')
        return

    try:
        # Get GCS bucket
        bucket = storage_client.bucket(bucket_name)

        run_ts = int(time.time())
        written = []

        for i, url in enumerate(ioc_urls):
            print(f'Processing URL {i+1}/{len(ioc_urls)}: {url}')

            # Build request
            req_headers = {'Accept': 'text/csv, */*'}

            # Add authentication header if provided
            if auth_header:
                if ':' in auth_header:
                    k, v = auth_header.split(':', 1)
                    req_headers[k.strip()] = v.strip()
                else:
                    req_headers['Authorization'] = auth_header.strip()

            # Fetch data with retries
            data = fetch_with_retries(url, req_headers, timeout)

            if data:
                # Write to GCS
                key = generate_blob_name(prefix, url, run_ts, i)
                blob = bucket.blob(key)
                blob.upload_from_string(data, content_type='text/csv')

                written.append({
                    'url': url,
                    'gcs_key': key,
                    'bytes': len(data)
                })

                print(f'Wrote {len(data)} bytes to gs://{bucket_name}/{key}')
            else:
                print(f'Warning: No data retrieved from {url}')

        print(f'Successfully processed {len(written)} URLs')
        print(json.dumps({'ok': True, 'written': written}, indent=2))

    except Exception as e:
        print(f'Error processing CSV IOC feeds: {str(e)}')
        raise

def fetch_with_retries(url, headers, timeout, max_retries=5):
    """Fetch data from URL with retry logic for 429/5xx errors."""
    if not url.lower().startswith('https://'):
        raise ValueError('Only HTTPS URLs are allowed in IOC_URLS')

    attempt = 0
    backoff = 1.0

    while attempt < max_retries:
        try:
            response = http.request('GET', url, headers=headers, timeout=timeout)

            if response.status == 200:
                return response.data.decode('utf-8')
            elif response.status == 429 or (500 <= response.status < 600):
                print(f'Received status {response.status}, retrying in {backoff}s (attempt {attempt+1}/{max_retries})')
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
            else:
                print(f'Error: Received unexpected status {response.status} from {url}')
                return None

        except Exception as e:
            if attempt < max_retries - 1:
                print(f'Request failed: {str(e)}, retrying in {backoff}s (attempt {attempt+1}/{max_retries})')
                time.sleep(backoff)
                attempt += 1
                backoff *= 2
            else:
                raise

    print(f'Max retries exceeded for {url}')
    return None

def generate_blob_name(prefix, url, run_ts, idx):
    """Generate a unique blob name for the CSV file."""
    # Create a short, filesystem-safe token for the URL
    safe_url = url.replace('://', '_').replace('/', '_').replace('?', '_').replace('&', '_')[:100]

    # Generate timestamp-based path
    timestamp_path = time.strftime('%Y/%m/%d/%H%M%S', time.gmtime(run_ts))

    return f"{prefix}/{timestamp_path}-url{idx:03d}-{safe_url}.csv"

第二個檔案：requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

點選「部署」來儲存並部署函式。
等待部署作業完成 (2 到 3 分鐘)。

注意： Pub/Sub 觸發條件設定會自動建立必要的訂閱項目和權限。

建立 Cloud Scheduler 工作

Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題，觸發 Cloud Run 函式。

前往 GCP 主控台的「Cloud Scheduler」。
點選「建立工作」。

請提供下列設定詳細資料：

設定	值
名稱	`csv-ioc-collector-hourly`
區域	選取與 Cloud Run 函式相同的區域
頻率	`0 * * * *` (每小時整點)
時區	選取時區 (建議使用世界標準時間)
目標類型	Pub/Sub
主題	選取 Pub/Sub 主題 (`csv-ioc-trigger`)
郵件內文	`{}` (空白 JSON 物件)

點選「建立」。

排程頻率選項

根據記錄檔量和延遲時間要求選擇頻率：

頻率	Cron 運算式	用途
每 5 分鐘	`/5 * * *`	高容量、低延遲
每 15 分鐘檢查一次	`/15 * * *`	普通量
每小時	`0 * * * *`	標準 (建議採用)
每 6 小時	`0 /6 * *`	少量、批次處理
每日	`0 0 * * *`	歷來資料集合

測試整合項目

在 Cloud Scheduler 控制台中，找出您的作業 (csv-ioc-collector-hourly)。
按一下「強制執行」，手動觸發工作。
稍等幾秒鐘。
前往「Cloud Run」>「Services」。
按一下函式名稱 (csv-ioc-collector)。
按一下 [Logs] (記錄) 分頁標籤。

確認函式是否已順利執行。請找出以下項目：

Processing URL 1/X: https://...
Wrote X bytes to gs://csv-ioc-logs/csv-ioc/YYYY/MM/DD/HHMMSS-url000-...csv
Successfully processed X URLs

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱 (csv-ioc-logs)。
前往前置字元資料夾 (csv-ioc/)。
確認是否已建立含有目前時間戳記的新 .csv 檔案。

如果在記錄中發現錯誤：

HTTP 401/403：檢查 AUTH_HEADER 環境變數
HTTP 429：頻率限制 - 函式會自動重試並延遲
缺少環境變數：檢查是否已設定所有必要變數
只允許使用 HTTPS 網址：確認 IOC_URLS 只包含 HTTPS 網址

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

取得服務帳戶電子郵件地址

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 CSV Custom IOC)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「CSV Custom IOC」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址，例如：
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
複製這個電子郵件地址，以便在下一步中使用。

注意： 每個 Google SecOps 執行個體都有專屬的服務帳戶。請勿使用其他說明文件或範例中的服務帳戶。

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。

依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱 (csv-ioc-logs)。
前往「權限」分頁標籤。
按一下「授予存取權」。
請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。

注意： 如果打算使用「刪除已轉移的檔案」或「刪除已轉移的檔案並清空目錄」刪除選項，請授予「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。

在 Google SecOps 中設定動態饋給，以便擷取 CSV 自訂 IOC 檔案

依序前往「SIEM 設定」>「動態饋給」。
按一下「新增動態消息」。
按一下「設定單一動態饋給」。
在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 CSV Custom IOC)。
選取「Google Cloud Storage V2」做為「來源類型」。
選取「CSV Custom IOC」做為「記錄類型」。
點選 [下一步]。
指定下列輸入參數的值：
- 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
```
gs://csv-ioc-logs/csv-ioc/
```
  - 取代：
    - csv-ioc-logs：您的 GCS bucket 名稱。
    - csv-ioc：儲存記錄的選用前置字元/資料夾路徑。
  - 範例：
    - 根層級 bucket：gs://csv-ioc-logs/
    - 前置字串：gs://csv-ioc-logs/csv-ioc/
    - 有子資料夾：gs://csv-ioc-logs/ioc-feeds/
  注意： URI 結尾一律要加上尾端斜線 (/)。
- 來源刪除選項：根據偏好設定選取刪除選項：
  - 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
  - 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
  - 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
    
    注意： 如果選取刪除選項，服務帳戶必須具備「Storage 物件管理員」角色，而非「Storage 物件檢視者」角色。請據此更新 IAM 權限。
- 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
- 資產命名空間：資產命名空間。
- 擷取標籤：要套用至這個動態饋給事件的標籤。
點選 [下一步]。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

UDM 對應表

記錄欄位	UDM 對應	邏輯
asn	entity.metadata.threat.detection_fields.asn_label.value	直接從「asn」欄位對應。
category	entity.metadata.threat.category_details	直接從「類別」欄位對應。
將內容分類	entity.metadata.threat.category_details	附加至「classification - 」並對應至「entity.metadata.threat.category_details」欄位。
第 2 欄	entity.entity.hostname	如果 [category] 符合「. ?ip」或「. ?proxy」，且 [not_ip] 為 true，則對應至「entity.entity.hostname」。
第 2 欄	entity.entity.ip	如果 [category] 符合「. ?ip」或「. ?proxy」，且 [not_ip] 為 false，則會併入「entity.entity.ip」。
信賴度	entity.metadata.threat.confidence_score	轉換為浮點數，並對應至「entity.metadata.threat.confidence_score」欄位。
國家/地區	entity.entity.location.country_or_region	直接對應至「國家/地區」欄位。
date_first	entity.metadata.threat.first_discovered_time	系統會將此值剖析為 ISO8601，並對應至「entity.metadata.threat.first_discovered_time」欄位。
date_last	entity.metadata.threat.last_updated_time	系統會將這個值剖析為 ISO8601，並對應至「entity.metadata.threat.last_updated_time」欄位。
詳細資料	entity.metadata.threat.summary	直接從「詳細資料」欄位對應。
detail2	entity.metadata.threat.description	直接從「detail2」欄位對應。
網域	entity.entity.hostname	直接從「網域」欄位對應。
電子郵件	entity.entity.user.email_addresses	已合併至「entity.entity.user.email_addresses」欄位。
id	entity.metadata.product_entity_id	附加至「id -」，並對應至「entity.metadata.product_entity_id」欄位。
import_session_id	entity.metadata.threat.detection_fields.import_session_id_label.value	直接從「import_session_id」欄位對應。
itype	entity.metadata.threat.detection_fields.itype_label.value	直接從「itype」欄位對應。
緯度	entity.entity.location.region_latitude	轉換為浮點數，並對應至「entity.entity.location.region_latitude」欄位。
經度	entity.entity.location.region_longitude	轉換為浮點數，並對應至「entity.entity.location.region_longitude」欄位。
maltype	entity.metadata.threat.detection_fields.maltype_label.value	直接從「maltype」欄位對應。
md5	entity.entity.file.md5	直接從「md5」欄位對應。
媒體	entity.metadata.threat.detection_fields.media_label.value	直接從「媒體」欄位對應。
media_type	entity.metadata.threat.detection_fields.media_type_label.value	直接從「media_type」欄位對應。
機構	entity.metadata.threat.detection_fields.org_label.value	直接從「org」欄位對應。
resource_uri	entity.entity.url	如果 [itype] 不符合「(ip
resource_uri	entity.metadata.threat.url_back_to_product	如果 [itype] 符合「(ip
分數	entity.metadata.threat.confidence_details	直接從「分數」欄位對應。
嚴重性	entity.metadata.threat.severity	如果符合「LOW」、「MEDIUM」、「HIGH」或「CRITICAL」，則會轉換為大寫，並對應至「entity.metadata.threat.severity」欄位。
來源	entity.metadata.threat.detection_fields.source_label.value	直接從「來源」欄位對應。
source_feed_id	entity.metadata.threat.detection_fields.source_feed_id_label.value	直接從「source_feed_id」欄位對應。
srcip	entity.entity.ip	如果 [srcip] 不為空白且不等於 [value]，則會合併至「entity.entity.ip」。
州	entity.metadata.threat.detection_fields.state_label.value	直接對應至「state」欄位。
trusted_circle_ids	entity.metadata.threat.detection_fields.trusted_circle_ids_label.value	直接從「trusted_circle_ids」欄位對應。
update_id	entity.metadata.threat.detection_fields.update_id_label.value	直接從「update_id」欄位對應。
值	entity.entity.file.full_path	如果 [category] 符合「.*?file」，則會對應至「entity.entity.file.full_path」。
值	entity.entity.file.md5	如果 [category] 符合「.*?md5」，且 [value] 是 32 個字元的十六進位字串，則會對應至「entity.entity.file.md5」。
值	entity.entity.file.sha1	如果 [category] 符合「. ?md5」且 [value] 是 40 個字元的十六進位字串，或 [category] 符合「. ?sha1」且 [value] 是 40 個字元的十六進位字串，則會對應至「entity.entity.file.sha1」。
值	entity.entity.file.sha256	如果 ([category] 符合「. ?md5」且 [value] 是十六進位字串，且 [file_type] 不是「md5」) 或 ([category] 符合「. ?sha256」且 [value] 是十六進位字串)，則對應至「entity.entity.file.sha256」。
值	entity.entity.hostname	如果 ([category] 符合「. ?domain」)，或 ([category] 符合「. ?ip」或「.*?proxy」且 [not_ip] 為 true)，則對應至「entity.entity.hostname」。
值	entity.entity.url	如果 ([category] 符合「.*?url」) 或 ([category] 符合「url」且 [resource_uri] 不為空白)，則對應至「entity.entity.url」。
不適用	entity.metadata.collected_timestamp	填入事件時間戳記。
不適用	entity.metadata.interval.end_time	設為 253402300799 秒的常數值。
不適用	entity.metadata.interval.start_time	填入事件時間戳記。
不適用	entity.metadata.vendor_name	設為「Custom IOC」常數值。

需要其他協助嗎？向社群成員和 Google SecOps 專業人員尋求答案。