收集 CrowdStrike Identity Protection (IDP) 服務記錄
剖析器版本：1.0
支援的國家/地區：
Google SecOps
SIEM
本文說明如何使用 Google Cloud Storage，將 CrowdStrike Identity Protection (IDP) 服務記錄檔擷取至 Google Security Operations。這項整合功能會使用 CrowdStrike Unified Alerts API 收集 Identity Protection 事件，並以 NDJSON 格式儲存，以便內建的 CS_IDP 剖析器處理。
事前準備
請確認您已完成下列事前準備事項：
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 GCS bucket IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- CrowdStrike Falcon 控制台和 API 金鑰管理員的特殊權限
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 crowdstrike-idp-logs-bucket)位置類型 請根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或保留政策 點選「建立」。
取得 CrowdStrike Identity Protection 的必要條件
- 登入 CrowdStrike Falcon 控制台。
- 依序前往「支援與資源」 > 「API 用戶端和金鑰」。
- 按一下「新增 API 用戶端」。
- 請提供下列設定詳細資料：
- 「Client Name」(用戶端名稱)：輸入
Google SecOps IDP Integration。 - 說明：輸入
API client for Google SecOps integration。 - 範圍：選取「Alerts: READ」(快訊：讀取) 範圍 (alerts:read)，包括 Identity Protection 快訊。
- 「Client Name」(用戶端名稱)：輸入
- 按一下「新增」。
- 複製下列詳細資料，並儲存至安全位置：
- 用戶端 ID
- 用戶端密鑰 (只會顯示一次)
- 基準網址 (例如：美國 1 為
api.crowdstrike.com、美國 2 為api.us-2.crowdstrike.com、歐洲 1 為api.eu-1.crowdstrike.com)
建立 Cloud Run 函式的服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶。
建立服務帳戶
- 在 GCP 控制台中，依序前往「IAM 與管理」>「服務帳戶」。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料：
- 「Service account name」(服務帳戶名稱)：輸入
crowdstrike-idp-collector-sa。 - 服務帳戶說明：輸入
Service account for Cloud Run function to collect CrowdStrike IDP logs。
- 「Service account name」(服務帳戶名稱)：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分：
- 按一下「請選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下「完成」。
這些角色適用於：
- Storage 物件管理員：將記錄檔寫入 GCS bucket，並管理狀態檔案
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者：允許叫用函式
授予 GCS bucket 的 IAM 權限
授予服務帳戶 GCS bucket 的寫入權限：
- 依序前往「Cloud Storage」 > 「Bucket」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如
crowdstrike-idp-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色：選取「Storage 物件管理員」。
- 新增主體：輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題，Cloud Scheduler 會發布至該主題，Cloud Run 函式則會訂閱該主題。
- 在 GCP 控制台中，前往「Pub/Sub」>「主題」。
- 按一下「建立主題」。
- 請提供下列設定詳細資料：
- 主題 ID：輸入
crowdstrike-idp-trigger。 - 其他設定保留預設值。
- 主題 ID：輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 CrowdStrike Identity Protection API 擷取記錄，並將記錄寫入 GCS。
- 在 GCP 控制台中，前往「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 crowdstrike-idp-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「觸發條件 (選用)」部分：
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中，選擇主題
crowdstrike-idp-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「容器、網路、安全性」。
前往「安全性」分頁標籤：
- 服務帳戶：選取服務帳戶
crowdstrike-idp-collector-sa。
- 服務帳戶：選取服務帳戶
前往「容器」分頁：
- 按一下「變數與密鑰」。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 GCS_BUCKETcrowdstrike-idp-logs-bucketGCS_PREFIXcrowdstrike-idp/STATE_KEYcrowdstrike-idp/state.jsonCROWDSTRIKE_CLIENT_IDyour-client-idCROWDSTRIKE_CLIENT_SECRETyour-client-secretAPI_BASEapi.crowdstrike.com(US-1)、api.us-2.crowdstrike.com(US-2)、api.eu-1.crowdstrike.com(EU-1)ALERTS_LIMIT1000(選用，每頁最多 10000 個)在「變數與密鑰」分頁中，向下捲動至「要求」：
- 要求逾時：輸入
600秒 (10 分鐘)。
- 要求逾時：輸入
前往「容器」的「設定」分頁：
- 在「Resources」(資源) 部分：
- 記憶體：選取 512 MiB 以上。
- CPU：選取 1。
- 按一下「完成」。
- 在「Resources」(資源) 部分：
捲動至「執行環境」：
- 選取「預設」 (建議)。
在「Revision scaling」(修訂版本資源調度) 部分：
- 「執行個體數量下限」：輸入
0。 - 「執行個體數量上限」：輸入
100(或依預期負載調整)。
- 「執行個體數量下限」：輸入
點選「建立」。
等待服務建立完成 (1 至 2 分鐘)。
服務建立完成後，系統會自動開啟內嵌程式碼編輯器。
新增函式程式碼
- 在「函式進入點」中輸入 main
在內嵌程式碼編輯器中，建立兩個檔案：
- 第一個檔案：main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone from urllib.parse import urlencode # Initialize HTTP client http = urllib3.PoolManager() # Initialize Storage client storage_client = storage.Client() @functions_framework.cloud_event def main(cloud_event): """ Fetch CrowdStrike Identity Protection alerts (Unified Alerts API) and store RAW JSON (NDJSON) to GCS for the CS_IDP parser. No transformation is performed. """ # Get environment variables bucket_name = os.environ.get('GCS_BUCKET') prefix = os.environ.get('GCS_PREFIX', 'crowdstrike-idp/') state_key = os.environ.get('STATE_KEY', 'crowdstrike-idp/state.json') client_id = os.environ.get('CROWDSTRIKE_CLIENT_ID') client_secret = os.environ.get('CROWDSTRIKE_CLIENT_SECRET') api_base = os.environ.get('API_BASE') if not all([bucket_name, client_id, client_secret, api_base]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(bucket_name) # Get OAuth token token = get_token(client_id, client_secret, api_base) # Load last processed timestamp last_ts = get_last_timestamp(bucket, state_key) # FQL filter for Identity Protection alerts only, newer than checkpoint fql_filter = f"product:'idp' + updated_timestamp:>'{last_ts}'" sort = 'updated_timestamp.asc' # Step 1: Get list of alert IDs all_ids = [] per_page = int(os.environ.get('ALERTS_LIMIT', '1000')) offset = 0 while True: page_ids = query_alert_ids(api_base, token, fql_filter, sort, per_page, offset) if not page_ids: break all_ids.extend(page_ids) if len(page_ids) < per_page: break offset += per_page if not all_ids: print('No new Identity Protection alerts.') return # Step 2: Get alert details in batches (max 1000 IDs per request) details = [] max_batch = 1000 for i in range(0, len(all_ids), max_batch): batch = all_ids[i:i + max_batch] details.extend(fetch_alert_details(api_base, token, batch)) if details: # Sort by updated_timestamp details.sort(key=lambda d: d.get('updated_timestamp', d.get('created_timestamp', ''))) latest = details[-1].get('updated_timestamp') or details[-1].get('created_timestamp') # Write to GCS timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') blob_name = f"{prefix}cs_idp_{timestamp}.json" blob = bucket.blob(blob_name) # NDJSON format log_lines = '\n'.join([json.dumps(d, separators=(',', ':')) for d in details]) blob.upload_from_string(log_lines, content_type='application/x-ndjson') print(f'Wrote {len(details)} alerts to {blob_name}') # Update state update_state(bucket, state_key, latest) except Exception as e: print(f'Error processing alerts: {str(e)}') raise def get_token(client_id, client_secret, api_base): """Get OAuth2 token from CrowdStrike API""" url = f"https://{api_base}/oauth2/token" data = f"client_id={client_id}&client_secret={client_secret}&grant_type=client_credentials" headers = {'Content-Type': 'application/x-www-form-urlencoded'} r = http.request('POST', url, body=data, headers=headers) if r.status != 200: raise Exception(f'Auth failed: {r.status} {r.data}') return json.loads(r.data.decode('utf-8'))['access_token'] def query_alert_ids(api_base, token, fql_filter, sort, limit, offset): """Query alert IDs using filters""" url = f"https://{api_base}/alerts/queries/alerts/v2" params = { 'filter': fql_filter, 'sort': sort, 'limit': str(limit), 'offset': str(offset) } qs = urlencode(params) r = http.request('GET', f"{url}?{qs}", headers={'Authorization': f'Bearer {token}'}) if r.status != 200: raise Exception(f'Query alerts failed: {r.status} {r.data}') resp = json.loads(r.data.decode('utf-8')) return resp.get('resources', []) def fetch_alert_details(api_base, token, composite_ids): """Fetch detailed alert data by composite IDs""" url = f"https://{api_base}/alerts/entities/alerts/v2" body = {'composite_ids': composite_ids} headers = { 'Authorization': f'Bearer {token}', 'Content-Type': 'application/json' } r = http.request('POST', url, body=json.dumps(body).encode('utf-8'), headers=headers) if r.status != 200: raise Exception(f'Fetch alert details failed: {r.status} {r.data}') resp = json.loads(r.data.decode('utf-8')) return resp.get('resources', []) def get_last_timestamp(bucket, key, default='2023-01-01T00:00:00Z'): """Get last processed timestamp from GCS state file""" try: blob = bucket.blob(key) if blob.exists(): state_data = blob.download_as_text() state = json.loads(state_data) return state.get('last_timestamp', default) except Exception as e: print(f'Warning: Could not load state: {str(e)}') return default def update_state(bucket, key, ts): """Update last processed timestamp in GCS state file""" state = { 'last_timestamp': ts, 'updated': datetime.now(timezone.utc).isoformat() } blob = bucket.blob(key) blob.upload_from_string(json.dumps(state), content_type='application/json')- 第二個檔案：requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0按一下「部署」，即可儲存並部署函式。
等待部署作業完成 (2 到 3 分鐘)。
建立 Cloud Scheduler 工作
Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題，觸發 Cloud Run 函式。
- 前往 GCP 控制台中的「Cloud Scheduler」。
- 點選「建立工作」。
提供下列設定詳細資料：
設定 值 名稱 crowdstrike-idp-collector-15m區域 選取與 Cloud Run 函式相同的區域 頻率 */15 * * * *(每 15 分鐘)時區 選取時區 (建議選取世界標準時間) 目標類型 Pub/Sub 主題 選取主題「 crowdstrike-idp-trigger」郵件內文 {}(空白 JSON 物件)點選「建立」。
測試排程器工作
- 在 Cloud Scheduler 控制台中，找出您的工作。
- 按一下「強制執行」即可手動觸發。
- 等待幾秒鐘，然後依序前往「Cloud Run」 >「服務」 >「crowdstrike-idp-collector」 >「記錄」。
- 確認函式是否執行成功。
- 檢查 GCS bucket，確認記錄已寫入。
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往 「SIEM 設定」 > 「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中，輸入動態饋給名稱 (例如
CrowdStrike Identity Protection Services logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「Crowdstrike Identity Protection Services」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址，以便在下一步中使用。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱。
- 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。
在 Google SecOps 中設定動態饋給，擷取 CrowdStrike Identity Protection Services 服務記錄
- 依序前往 「SIEM 設定」 > 「動態饋給」。
- 按一下「新增動態饋給」。
- 按一下「設定單一動態饋給」。
- 在「Feed name」(動態饋給名稱) 欄位中，輸入動態饋給名稱 (例如
CrowdStrike Identity Protection Services logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 在「記錄類型」部分，選取「Crowdstrike Identity Protection Services」。
- 點選「下一步」。
指定下列輸入參數的值：
儲存空間 bucket 網址：輸入 GCS bucket URI，並加上前置路徑：
gs://crowdstrike-idp-logs-bucket/crowdstrike-idp/更改項目：
crowdstrike-idp-logs-bucket：GCS bucket 名稱。crowdstrike-idp/：儲存記錄的前置字元/資料夾路徑。
來源刪除選項：根據偏好設定選取刪除選項：
- Never：轉移後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案：在成功轉移後刪除檔案。
刪除已轉移的檔案和空白目錄：成功轉移檔案後，刪除檔案和空白目錄。
檔案存在時間上限：包含過去幾天內修改過的檔案。預設值為 180 天。
Asset namespace：資產命名空間。
擷取標籤：要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。
變更記錄
還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求解答。