收集 ZeroFox 平台記錄
支援的國家/地區：
Google SecOps
SIEM
本文說明如何使用 Google Cloud Storage，將 ZeroFox Platform 記錄檔擷取至 Google Security Operations。ZeroFox 平台會監控及分析社群媒體、行動應用程式、雲端、電子郵件和其他數位管道的威脅，提供數位風險防護。
事前準備
請確認您已完成下列事前準備事項：
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS 值區的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- ZeroFox Platform 租戶的特殊存取權
建立 Google Cloud Storage 值區
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 zerofox-platform-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或保留政策 點選「建立」。
收集 ZeroFox 平台憑證
取得 ZeroFox 個人存取權杖
- 前往 https://cloud.zerofox.com 登入 ZeroFox 平台。
- 依序前往「設定」>「資料連線」>「API 資料動態饋給」。
- 直接網址 (登入後)：https://cloud.zerofox.com/data_connectors/api
- 注意：如果沒有看到這個選單項目，請與 ZeroFox 管理員聯絡以取得存取權。視租戶 UI 版本而定，選單也可能標示為「資料連接器」>「API 資料動態饋給」。
- 按一下「產生權杖」或「建立個人存取權杖」。
- 請提供下列設定詳細資料：
- 名稱：輸入描述性名稱 (例如
Google SecOps GCS Ingestion)。 - 到期：根據貴機構的安全政策選取輪替週期。
- 權限/動態消息：選取下列項目的讀取權限：
- 快訊
- CTI 摘要
- 要匯出的其他資料類型
- 名稱：輸入描述性名稱 (例如
- 按一下 [產生]。
- 複製並將產生的個人存取權杖儲存在安全位置 (您無法再次查看)。
儲存 ZEROFOX_BASE_URL：
https://api.zerofox.com(大多數租戶的預設值)。
驗證權限
如要確認帳戶是否具備必要權限，請按照下列步驟操作：
- 登入 ZeroFox Platform。
- 依序點選「設定」圖示 (⚙️) >「資料連結」>「API 資料動態饋給」。
- 如果能看到「API 資料動態饋給」部分並產生權杖，表示您擁有必要權限。
- 如果沒有看到這個選項，請聯絡管理員，請對方授予 API 存取權。
測試 API 存取權
請先測試憑證，再繼續進行整合：
# Replace with your actual credentials ZEROFOX_API_TOKEN="your-personal-access-token" ZEROFOX_BASE_URL="https://api.zerofox.com" # Test API access (example endpoint - adjust based on your data type) curl -v -H "Authorization: Bearer $ZEROFOX_API_TOKEN" \ -H "Accept: application/json" \ "$ZEROFOX_BASE_URL/v1/alerts?limit=1"
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶，並由 Pub/Sub 叫用。
建立服務帳戶
- 在 GCP 主控台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下 [Create Service Account] (建立服務帳戶)。
- 請提供下列設定詳細資料：
- 服務帳戶名稱：輸入
zerofox-logs-collector-sa。 - 服務帳戶說明：輸入
Service account for Cloud Run function to collect ZeroFox Platform logs。
- 服務帳戶名稱：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下 [完成]。
這些角色適用於：
- Storage 物件管理員：將記錄檔寫入 GCS 值區，並管理狀態檔案
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者：允許函式叫用
授予 GCS 值區的 IAM 權限
將 GCS bucket 的寫入權限授予服務帳戶 (zerofox-logs-collector-sa)：
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (例如
zerofox-platform-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如
zerofox-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色：選取「Storage 物件管理員」。
- 新增主體：輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料：
- 主題 ID：輸入
zerofox-logs-trigger。 - 其他設定保留預設值。
- 主題 ID：輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 ZeroFox Platform API 擷取記錄，並將記錄寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 zerofox-logs-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取「Python 3.12」以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區：
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中，選擇 Pub/Sub 主題 (
zerofox-logs-trigger)。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶 (
zerofox-logs-collector-sa)。
- 服務帳戶：選取服務帳戶 (
前往「容器」分頁：
- 按一下「變數與密鑰」。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 說明 GCS_BUCKETzerofox-platform-logsGCS bucket 名稱 GCS_PREFIXzerofox/platform記錄檔的前置字串 STATE_KEYzerofox/platform/state.json狀態檔案路徑 ZEROFOX_BASE_URLhttps://api.zerofox.comAPI 基礎網址 ZEROFOX_API_TOKENyour-zerofox-personal-access-token個人存取權杖 LOOKBACK_HOURS24初始回溯期 PAGE_SIZE200每頁記錄數 MAX_PAGES20每次執行的頁數上限 HTTP_TIMEOUT60HTTP 要求逾時時間 (以秒為單位) HTTP_RETRIES3HTTP 重試次數 URL_TEMPLATE(選填) 含有 {SINCE}、{PAGE_TOKEN}、{PAGE_SIZE}的自訂網址範本在「變數與密鑰」部分，向下捲動至「要求」：
- 要求逾時：輸入
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, timedelta import time import urllib.parse # 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() # Environment variables GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'zerofox/platform') STATE_KEY = os.environ.get('STATE_KEY', 'zerofox/platform/state.json') ZEROFOX_BASE_URL = os.environ.get('ZEROFOX_BASE_URL', 'https://api.zerofox.com') ZEROFOX_API_TOKEN = os.environ.get('ZEROFOX_API_TOKEN') LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '200')) MAX_PAGES = int(os.environ.get('MAX_PAGES', '20')) HTTP_TIMEOUT = int(os.environ.get('HTTP_TIMEOUT', '60')) HTTP_RETRIES = int(os.environ.get('HTTP_RETRIES', '3')) URL_TEMPLATE = os.environ.get('URL_TEMPLATE', '') def parse_datetime(value: str) -> datetime: """Parse ISO datetime string to datetime object.""" if value.endswith("Z"): value = value[:-1] + "+00:00" return datetime.fromisoformat(value) @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch ZeroFox Platform logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, ZEROFOX_BASE_URL, ZEROFOX_API_TOKEN]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(GCS_BUCKET) # Load state state = load_state(bucket, STATE_KEY) # Determine time window now = datetime.now(timezone.utc) last_time = None if isinstance(state, dict) and state.get("last_since"): try: last_time = parse_datetime(state["last_since"]) # Overlap by 2 minutes to catch any delayed events last_time = last_time - timedelta(minutes=2) except Exception as e: print(f"Warning: Could not parse last_since: {e}") if last_time is None: last_time = now - timedelta(hours=LOOKBACK_HOURS) since_iso = last_time.strftime('%Y-%m-%dT%H:%M:%SZ') print(f"Fetching logs since {since_iso}") # Fetch logs records, newest_since = fetch_logs( api_base=ZEROFOX_BASE_URL, api_token=ZEROFOX_API_TOKEN, since=since_iso, page_size=PAGE_SIZE, max_pages=MAX_PAGES, ) if not records: print("No new log records found.") save_state(bucket, STATE_KEY, since_iso) return # Write to GCS as NDJSON timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}") # Update state with newest timestamp if newest_since: save_state(bucket, STATE_KEY, newest_since) else: save_state(bucket, STATE_KEY, since_iso) print(f"Successfully processed {len(records)} records") except Exception as e: print(f'Error processing logs: {str(e)}') raise def load_state(bucket, key): """Load state from GCS.""" try: blob = bucket.blob(key) if blob.exists(): state_data = blob.download_as_text() return json.loads(state_data) except Exception as e: print(f"Warning: Could not load state: {e}") return {} def save_state(bucket, key, last_since: str): """Save the last since timestamp to GCS state file.""" try: state = {'last_since': last_since} blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: last_since={last_since}") except Exception as e: print(f"Warning: Could not save state: {e}") def fetch_logs(api_base: str, api_token: str, since: str, page_size: int, max_pages: int): """ Fetch logs from ZeroFox Platform API with pagination and rate limiting. Args: api_base: API base URL api_token: Personal access token since: ISO timestamp for filtering logs page_size: Number of records per page max_pages: Maximum pages to fetch Returns: Tuple of (records list, newest_since ISO string) """ # Use URL_TEMPLATE if provided, otherwise construct default alerts endpoint if URL_TEMPLATE: base_url = URL_TEMPLATE.replace("{SINCE}", urllib.parse.quote(since)) else: base_url = f"{api_base}/v1/alerts?since={urllib.parse.quote(since)}" headers = { 'Authorization': f'Bearer {api_token}', 'Accept': 'application/json', 'Content-Type': 'application/json', 'User-Agent': 'GoogleSecOps-ZeroFoxCollector/1.0' } records = [] newest_since = since page_num = 0 page_token = "" backoff = 1.0 while page_num < max_pages: page_num += 1 # Construct URL with pagination if URL_TEMPLATE: url = (base_url .replace("{PAGE_TOKEN}", urllib.parse.quote(page_token)) .replace("{PAGE_SIZE}", str(page_size))) else: url = f"{base_url}&limit={page_size}" if page_token: url += f"&page_token={urllib.parse.quote(page_token)}" attempt = 0 while attempt <= HTTP_RETRIES: try: response = http.request('GET', url, headers=headers, timeout=HTTP_TIMEOUT) # Handle rate limiting with exponential backoff if response.status == 429: retry_after = int(response.headers.get('Retry-After', str(int(backoff)))) print(f"Rate limited (429). Retrying after {retry_after}s...") time.sleep(retry_after) backoff = min(backoff * 2, 30.0) attempt += 1 continue backoff = 1.0 if response.status != 200: print(f"HTTP Error: {response.status}") response_text = response.data.decode('utf-8') print(f"Response body: {response_text}") return records, newest_since data = json.loads(response.data.decode('utf-8')) # Extract results (try multiple possible keys) page_results = [] for key in ('results', 'data', 'alerts', 'items', 'logs', 'events'): if isinstance(data.get(key), list): page_results = data[key] break if not page_results: print(f"No more results (empty page)") return records, newest_since print(f"Page {page_num}: Retrieved {len(page_results)} events") records.extend(page_results) # Track newest timestamp for event in page_results: try: # Try multiple possible timestamp fields event_time = (event.get('timestamp') or event.get('created_at') or event.get('last_modified') or event.get('event_time') or event.get('log_time') or event.get('updated_at')) if event_time and isinstance(event_time, str): if event_time > newest_since: newest_since = event_time except Exception as e: print(f"Warning: Could not parse event time: {e}") # Check for next page token next_token = (data.get('next') or data.get('next_token') or data.get('nextPageToken') or data.get('next_page_token')) if isinstance(next_token, dict): next_token = (next_token.get('token') or next_token.get('cursor') or next_token.get('value')) if not next_token: print("No more pages (no next token)") return records, newest_since page_token = str(next_token) break except urllib3.exceptions.HTTPError as e: if attempt < HTTP_RETRIES: print(f"HTTP error (attempt {attempt + 1}/{HTTP_RETRIES}): {e}") time.sleep(1 + attempt) attempt += 1 continue else: print(f"Error fetching logs after {HTTP_RETRIES} retries: {e}") return records, newest_since except Exception as e: print(f"Error fetching logs: {e}") return records, newest_since print(f"Retrieved {len(records)} total records from {page_num} pages") return records, newest_since- 第二個檔案：requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0點選「部署」來儲存並部署函式。
等待部署作業完成 (2 到 3 分鐘)。
建立 Cloud Scheduler 工作
Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題 (zerofox-logs-trigger)，藉此觸發 Cloud Run 函式。
- 前往 GCP 主控台的「Cloud Scheduler」。
- 點選「建立工作」。
請提供下列設定詳細資料：
設定 值 名稱 zerofox-logs-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取 Pub/Sub 主題 ( zerofox-logs-trigger)郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間要求選擇頻率：
頻率 Cron 運算式 用途 每 5 分鐘 */5 * * * *高容量、低延遲 每 15 分鐘檢查一次 */15 * * * *普通量 每小時 0 * * * *標準 (建議採用) 每 6 小時 0 */6 * * *少量、批次處理 每日 0 0 * * *歷來資料集合
測試整合項目
- 在 Cloud Scheduler 控制台中，找出您的作業 (
zerofox-logs-collector-hourly)。 - 按一下「強制執行」，手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「Services」。
- 按一下函式名稱 (
zerofox-logs-collector)。 - 按一下 [Logs] (記錄) 分頁標籤。
確認函式是否已順利執行。請找出以下項目：
Fetching logs since YYYY-MM-DDTHH:MM:SSZ Page 1: Retrieved X events Wrote X records to gs://zerofox-platform-logs/zerofox/platform/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X records依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱 (
zerofox-platform-logs)。前往前置字元資料夾 (
zerofox/platform/)。確認是否已建立含有目前時間戳記的新
.ndjson檔案。
如果在記錄中發現錯誤：
- HTTP 401：檢查環境變數中的 API 憑證。確認
ZEROFOX_API_TOKEN正確無誤且尚未過期。 - HTTP 403：確認 ZeroFox 帳戶是否具備快訊和 CTI 資訊動態饋給的必要權限。依序前往「設定」>「資料連線」>「API 資料動態饋給」，然後檢查權杖權限。
- HTTP 404：預設
/v1/alerts端點可能不適用於您的租戶。請根據 ZeroFox API 說明文件設定URL_TEMPLATE環境變數，或與 ZeroFox 支援團隊聯絡。 - HTTP 429：頻率限制 - 函式會自動重試，並採用指數輪詢間隔。
- 缺少環境變數：檢查 Cloud Run 函式設定中是否已設定所有必要變數。
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
ZeroFox Platform Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「ZeroFox Platform」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址，以便在下一步中使用。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (
zerofox-platform-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。
在 Google SecOps 中設定動態消息，擷取 ZeroFox Platform 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
ZeroFox Platform Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「ZeroFox Platform」做為「記錄類型」。
- 點選 [下一步]。
指定下列輸入參數的值：
儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
gs://zerofox-platform-logs/zerofox/platform/取代：
zerofox-platform-logs：您的 GCS bucket 名稱。zerofox/platform：儲存記錄的前置字元/資料夾路徑。
來源刪除選項：根據偏好設定選取刪除選項：
- 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
資產命名空間：資產命名空間。
擷取標籤：要套用至這個動態饋給事件的標籤。
點選 [下一步]。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。
需要其他協助嗎？向社群成員和 Google SecOps 專業人員尋求答案。