收集 Proofpoint Emerging Threats Pro 入侵指標記錄

支援的國家/地區：

本文說明如何使用 Google Cloud Storage，將 Proofpoint Emerging Threats Pro IOC 記錄檔擷取至 Google Security Operations。Emerging Threats Intelligence 會每小時發布 IP 和網域的信譽清單 (CSV 格式)，其中包含威脅情報資料，包括類別、分數和時間資訊。剖析器程式碼會處理 CSV 格式的 ET_PRO 威脅情報資料。這項功能會擷取 IP 位址、網域、類別、分數和其他相關資訊，並將這些資訊對應至標準化 IOC 格式和 Chronicle UDM 架構，以便在 Google SecOps 中進一步分析及使用。

事前準備

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

  • Google SecOps 執行個體
  • 已啟用 Cloud Storage API 的 GCP 專案
  • 建立及管理 GCS 值區的權限
  • 管理 Google Cloud Storage 值區 IAM 政策的權限
  • 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
  • 訂閱 Proofpoint ET Intelligence，並具備信譽清單的存取權
  • 從 https://etadmin.proofpoint.com/api-access 取得 ET Intelligence API 金鑰

收集新興威脅專業認證的必要條件

  1. 前往 https://etadmin.proofpoint.com 登入 ET Intelligence 管理入口網站。
  2. 前往「API 存取權」。
  3. 複製並儲存 API 金鑰。
  4. 請與 Proofpoint 代表聯絡，取得下列資訊：
    • 詳細 IP 信譽清單網址
    • 詳細網域信譽清單網址

ET Intelligence 會提供 IP 和網域信譽清單的個別 CSV 檔案，並每小時更新。使用「詳細」格式，其中包含下列各欄：

  • 網域清單：網域名稱、類別、分數、首次出現時間、上次出現時間、連接埠
  • IP 清單：IP 位址、類別、分數、首次出現時間、上次出現時間、通訊埠

詳細格式網址通常遵循以下模式：

  • IP 清單：https://rules.emergingthreatspro.com/[your-code]/reputation/detailed-iprepdata.txt
  • 網域清單：https://rules.emergingthreatspro.com/[your-code]/reputation/detailed-domainrepdata.txt

Authorization 標頭必須包含原始 API 金鑰值，且不得有 Bearer 前置字元，以符合 Emerging Threats API 的預期。

建立 Google Cloud Storage 值區

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

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

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

  6. 點選「建立」。

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

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

這些角色適用於：

  • Storage 物件管理員：將記錄檔寫入 GCS 值區，並管理狀態檔案
  • Cloud Run 叫用者：允許 Pub/Sub 叫用函式
  • Cloud Functions 叫用者：允許函式叫用

授予 GCS 值區的 IAM 權限

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

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

建立 Pub/Sub 主題

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

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

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 Proofpoint ET Intelligence API 擷取記錄，並將記錄寫入 GCS。

  1. 前往 GCP Console 的「Cloud Run」。
  2. 按一下「Create service」(建立服務)。
  3. 選取「函式」 (使用內嵌編輯器建立函式)。

  4. 在「設定」部分，提供下列設定詳細資料：

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

  5. 在「Trigger (optional)」(觸發條件 (選用)) 專區：

    1. 按一下「+ 新增觸發條件」。
    2. 選取「Cloud Pub/Sub」。
    3. 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中，選擇 et-pro-ioc-trigger。
    4. 按一下 [儲存]。

  6. 在「Authentication」(驗證) 部分：

    1. 選取「需要驗證」。
    2. 檢查 Identity and Access Management (IAM)。

  7. 向下捲動並展開「Containers, Networking, Security」。

  8. 前往「安全性」分頁：

    • 服務帳戶：選取 et-pro-ioc-fetcher-sa。

  9. 前往「容器」分頁：

    1. 按一下「變數與密鑰」。
    2. 針對每個環境變數，按一下「+ 新增變數」：
    變數名稱 範例值 說明
    GCS_BUCKET et-pro-ioc-bucket GCS bucket 名稱
    GCS_PREFIX et-pro-ioc 記錄檔的前置字串
    STATE_KEY et-pro-ioc/state.json 狀態檔案路徑
    ET_API_KEY your-et-api-key ET Intelligence API 金鑰
    ET_IP_LIST_URL https://rules.emergingthreatspro.com/[your-code]/reputation/detailed-iprepdata.txt 詳細 IP 信譽清單網址
    ET_DOMAIN_LIST_URL https://rules.emergingthreatspro.com/[your-code]/reputation/detailed-domainrepdata.txt 詳細網域信譽清單網址
    TIMEOUT 120 HTTP 要求逾時時間 (以秒為單位)

  10. 在「變數與密鑰」分頁中向下捲動至「要求」：

    • 要求逾時：輸入 600 秒 (10 分鐘)。

  11. 前往「容器」中的「設定」分頁：

    • 在「資源」部分：
      • 記憶體：選取 512 MiB 以上。
      • CPU：選取 1。
    • 按一下 [完成]。

  12. 捲動至「執行環境」：

    • 選取「預設」 (建議選項)。

  13. 在「修訂版本資源調度」部分：

    • 執行個體數量下限：輸入 0。
    • 「Maximum number of instances」(執行個體數量上限)：輸入 100 (或根據預期負載調整)。

  14. 點選「建立」。

  15. 等待服務建立完成 (1 到 2 分鐘)。

  16. 服務建立完成後，系統會自動開啟內嵌程式碼編輯器。

新增函式程式碼

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

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

    • 第一個檔案：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 ET Pro IOC reputation lists 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', 'et-pro-ioc').strip('/')
    state_key = os.environ.get('STATE_KEY', f'{prefix}/state.json')
    et_api_key = os.environ.get('ET_API_KEY')
    et_ip_list_url = os.environ.get('ET_IP_LIST_URL')
    et_domain_list_url = os.environ.get('ET_DOMAIN_LIST_URL')
    timeout = int(os.environ.get('TIMEOUT', '120'))

    if not all([bucket_name, et_api_key, et_ip_list_url, et_domain_list_url]):
        print('Error: Missing required environment variables')
        return

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

        # Generate timestamp for file naming
        now = datetime.now(timezone.utc)
        timestamp = now.strftime('%Y/%m/%d/%H%M%S')

        results = []
        errors = []

        # Fetch IP reputation list
        try:
            print('Fetching IP reputation list...')
            ip_data = fetch_with_retry(et_ip_list_url, et_api_key, timeout)
            ip_key = f'{prefix}/ip/{timestamp}.csv'
            save_to_gcs(bucket, ip_key, ip_data)
            results.append({'type': 'ip', 'key': ip_key, 'size': len(ip_data)})
            print(f'Successfully fetched IP list: {len(ip_data)} bytes')
        except Exception as e:
            error_msg = f'Failed to fetch IP list: {str(e)}'
            print(error_msg)
            errors.append(error_msg)

        # Fetch Domain reputation list
        try:
            print('Fetching Domain reputation list...')
            domain_data = fetch_with_retry(et_domain_list_url, et_api_key, timeout)
            domain_key = f'{prefix}/domain/{timestamp}.csv'
            save_to_gcs(bucket, domain_key, domain_data)
            results.append({'type': 'domain', 'key': domain_key, 'size': len(domain_data)})
            print(f'Successfully fetched Domain list: {len(domain_data)} bytes')
        except Exception as e:
            error_msg = f'Failed to fetch Domain list: {str(e)}'
            print(error_msg)
            errors.append(error_msg)

        # Save state
        state = {
            'last_fetch': now.isoformat(),
            'results': results,
            'errors': errors
        }
        save_state(bucket, state_key, state)

        if errors:
            print(f'Completed with {len(errors)} error(s)')
        else:
            print('Successfully completed all fetches')

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

def fetch_with_retry(url, api_key, timeout, max_retries=3):
    """Fetch URL with retry logic for rate limits."""
    if not url.lower().startswith('https://'):
        raise ValueError('Only HTTPS URLs are allowed')

    headers = {'Authorization': api_key}

    for attempt in range(max_retries):
        try:
            response = http.request('GET', url, headers=headers, timeout=timeout)

            if response.status == 200:
                return response.data
            elif response.status == 429:
                # Rate limited, wait and retry
                wait_time = min(30 * (2 ** attempt), 300)
                print(f'Rate limited, waiting {wait_time}s...')
                time.sleep(wait_time)
            else:
                raise Exception(f'HTTP {response.status}: {response.reason}')
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            time.sleep(5 * (attempt + 1))

    raise Exception(f'Failed to fetch {url} after {max_retries} attempts')

def save_to_gcs(bucket, key, content):
    """Save content to GCS with appropriate content type."""
    blob = bucket.blob(key)
    blob.upload_from_string(content, content_type='text/csv')
    print(f'Saved {len(content)} bytes to gs://{bucket.name}/{key}')

def save_state(bucket, key, state):
    """Save state to GCS."""
    try:
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json'
        )
    except Exception as e:
        print(f'Warning: Could not save state: {str(e)}')
    • 第二個檔案：requirements.txt:
    functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

  3. 點選「部署」來儲存並部署函式。

  4. 等待部署作業完成 (2 到 3 分鐘)。

建立 Cloud Scheduler 工作

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

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

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

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

  4. 點選「建立」。

排程頻率選項

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

    頻率 Cron 運算式 用途
    每小時 0 * * * * 標準 (建議用於 ET Pro IOC)
    每 2 小時 0 */2 * * * 調低音頻
    每 6 小時 0 */6 * * * 最少更新次數

測試整合項目

  1. 在 Cloud Scheduler 控制台中找出您的工作。
  2. 按一下「強制執行」，手動觸發工作。
  3. 稍等幾秒鐘。
  4. 前往「Cloud Run」>「Services」。
  5. 按一下函式名稱 (et-pro-ioc-fetcher)。
  6. 按一下 [Logs] (記錄) 分頁標籤。

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

    Fetching IP reputation list...
Successfully fetched IP list: X bytes
Fetching Domain reputation list...
Successfully fetched Domain list: X bytes
Successfully completed all fetches

  8. 依序前往「Cloud Storage」>「Buckets」。

  9. 按一下 bucket 名稱。

  10. 前往前置字元資料夾 (et-pro-ioc/ip/ 和 et-pro-ioc/domain/)。

  11. 確認是否已建立含有目前時間戳記的新 .csv 檔案。

如果在記錄中發現錯誤：

  • HTTP 401：檢查環境變數中的 ET_API_KEY
  • HTTP 403：確認 API 金鑰是否具備必要權限
  • HTTP 429：頻率限制 - 函式會自動重試並延遲
  • 缺少環境變數：檢查是否已設定所有必要變數

擷取 Google SecOps 服務帳戶

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

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

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 ET Pro IOC - IP Reputation)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「新興威脅專業版」做為「記錄類型」。

  7. 按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址，例如：

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com

  8. 複製這個電子郵件地址，以便在下一步中使用。

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

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

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

  6. 按一下 [儲存]。

在 Google SecOps 中設定資訊提供，擷取 Proofpoint Emerging Threats Pro IOC 記錄

您需要建立兩個不同的動態饋給，分別用於 IP 信譽和網域信譽。

建立 IP 信譽動態饋給

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入 ET Pro IOC - IP Reputation。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「新興威脅專業版」做為「記錄類型」。
  7. 點選 [下一步]。

  8. 指定下列輸入參數的值：

    • 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：

      gs://et-pro-ioc-bucket/et-pro-ioc/ip/
      • 請將 et-pro-ioc-bucket 替換成實際的 GCS 值區名稱。

    • 來源刪除選項：根據偏好設定選取刪除選項：

      • 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
      • 刪除已轉移的檔案：成功轉移檔案後刪除檔案。

      • 刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。

    • 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。

    • 資產命名空間：資產命名空間。

    • 擷取標籤：套用至這個動態饋給事件的標籤。

  9. 點選 [下一步]。

  10. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

建立網域信譽動態饋給

重複動態饋給建立程序：

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態消息」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入 ET Pro IOC - Domain Reputation。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「新興威脅專業版」做為「記錄類型」。
  7. 點選 [下一步]。

  8. 指定下列輸入參數的值：

    • 儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
    gs://et-pro-ioc-bucket/et-pro-ioc/domain/

    請將 et-pro-ioc-bucket 替換成實際的 GCS 值區名稱。

    • 來源刪除選項：根據偏好設定選取。
    • 檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
    • 資產命名空間：資產命名空間。
    • 擷取標籤：套用至這個動態饋給事件的標籤。

  9. 點選 [下一步]。

  10. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。

UDM 對應表

記錄欄位 UDM 對應 邏輯
category 這個欄位會用於剖析器邏輯，但不會直接對應至 UDM。 系統會透過查閱表判斷 event.ioc.categorization 的值。
collection_time.nanos event.idm.entity.metadata.collected_timestamp.nanos 直接從原始記錄對應。
collection_time.seconds event.idm.entity.metadata.collected_timestamp.seconds 直接從原始記錄對應。
資料 系統會根據這個欄位的內容，剖析成多個 UDM 欄位。
first_seen event.idm.entity.metadata.interval.start_time 剖析為日期並對應至 UDM。
first_seen event.ioc.active_timerange.start 剖析為日期並對應至 UDM。
ip_or_domain event.idm.entity.entity.hostname 如果 grok 模式從欄位中擷取主機，則會對應至 UDM。
ip_or_domain event.idm.entity.entity.ip 如果 grok 模式未從欄位擷取主機，則會對應至 UDM。
ip_or_domain event.ioc.domain_and_ports.domain 如果 grok 模式從欄位中擷取主機，則會對應至 UDM。
ip_or_domain event.ioc.ip_and_ports.ip_address 如果 grok 模式未從欄位擷取主機，則會對應至 UDM。
last_seen event.idm.entity.metadata.interval.end_time 剖析為日期並對應至 UDM。
last_seen event.ioc.active_timerange.end 剖析為日期並對應至 UDM。
ports event.idm.entity.entity.labels.value 如果有多個通訊埠，系統會剖析這些通訊埠，以半形逗號分隔符號加入，並對應至 UDM。
ports event.idm.entity.entity.port 如果只有一個通訊埠，則會剖析並對應至 UDM。
ports event.ioc.domain_and_ports.ports 如果 grok 模式從欄位擷取主機，系統會剖析並對應至 UDM。
ports event.ioc.ip_and_ports.ports 如果 grok 模式未從欄位擷取主機，則會剖析並對應至 UDM。
分數 event.ioc.confidence_score 直接從原始記錄對應。
event.idm.entity.entity.labels.key 如果有多個通訊埠，請設為「ports」。
event.idm.entity.metadata.entity_type 如果 grok 模式從 ip_or_domain 欄位擷取主機，請設為「DOMAIN_NAME」，否則請設為「IP_ADDRESS」。
event.idm.entity.metadata.threat.category 設為「SOFTWARE_MALICIOUS」。
event.idm.entity.metadata.threat.category_details 使用對照表從類別欄位衍生而來。
event.idm.entity.metadata.threat.threat_name 設為「ET Intelligence Rep List」。
event.idm.entity.metadata.vendor_name 設為「ET_PRO_IOC」。
event.ioc.feed_name 設為「ET Intelligence Rep List」。
event.ioc.raw_severity 設為「惡意」。
timestamp.nanos 從 collection_time.nanos 複製。
timestamp.seconds 從 collection_time.seconds 複製。

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