收集 CrowdStrike FileVantage 記錄
支援的國家/地區：
Google SecOps
SIEM
本文說明如何使用 Google Cloud Storage，將 CrowdStrike FileVantage 記錄檔擷取至 Google Security Operations。CrowdStrike FileVantage 是一種檔案完整性監控解決方案，可追蹤環境中重要檔案和目錄的變更。
事前準備
請確認您已完成下列事前準備事項：
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS 值區的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
CrowdStrike Falcon 控制台的特殊權限
建立 Google Cloud Storage 值區
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中，依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料：
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 crowdstrike-filevantage-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用：啟用物件版本管理或保留政策 點選「建立」。
收集 CrowdStrike FileVantage API 憑證
- 登入 CrowdStrike Falcon Console。
- 依序前往「支援與資源」>「API 用戶端和金鑰」。
- 按一下「新增 API 用戶端」。
- 請提供下列設定詳細資料：
- 用戶端名稱：輸入描述性名稱 (例如
Google SecOps FileVantage Integration)。 - 說明：簡要說明整合作業的用途。
- API 範圍：選取「Falcon FileVantage:read」。
- 用戶端名稱：輸入描述性名稱 (例如
- 按一下「新增」即可完成程序。
複製下列詳細資料並存放在安全的地方：
- 用戶端 ID
- 用戶端密碼
- 基準網址 (決定雲端區域)
驗證權限
如要確認帳戶是否具備必要權限，請按照下列步驟操作：
- 登入 CrowdStrike Falcon Console。
- 依序前往「支援與資源」>「API 用戶端和金鑰」。
- 如果可以查看「API clients and keys」(API 用戶端和金鑰) 頁面並建立新的 API 用戶端，表示您擁有必要權限。
- 如果無法存取這個頁面，請與 CrowdStrike 管理員聯絡，請對方授予 Falcon 管理員角色。
測試 API 存取權
請先測試憑證，再繼續進行整合：
# Replace with your actual credentials FALCON_CLIENT_ID="your-client-id" FALCON_CLIENT_SECRET="your-client-secret" FALCON_BASE_URL="https://api.crowdstrike.com" # Get OAuth token TOKEN=$(curl -s -X POST "${FALCON_BASE_URL}/oauth2/token" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "client_id=${FALCON_CLIENT_ID}&client_secret=${FALCON_CLIENT_SECRET}&grant_type=client_credentials" \ | grep -o '"access_token":"[^"]*' | cut -d'"' -f4) # Test FileVantage API access curl -v -H "Authorization: Bearer ${TOKEN}" \ "${FALCON_BASE_URL}/filevantage/queries/changes/v3?limit=1"
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶，並由 Pub/Sub 叫用。
建立服務帳戶
- 在 GCP 主控台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下 [Create Service Account] (建立服務帳戶)。
- 請提供下列設定詳細資料：
- 服務帳戶名稱：輸入
crowdstrike-filevantage-sa。 - 服務帳戶說明：輸入
Service account for Cloud Run function to collect CrowdStrike FileVantage logs。
- 服務帳戶名稱：輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下 [完成]。
這些角色適用於：
- Storage 物件管理員：將記錄檔寫入 GCS 值區，並管理狀態檔案
- Cloud Run 叫用者：允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者：允許函式叫用
授予 GCS 值區的 IAM 權限
授予服務帳戶 GCS bucket 的寫入權限：
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (例如
crowdstrike-filevantage-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：輸入服務帳戶電子郵件地址 (例如
crowdstrike-filevantage-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色：選取「Storage 物件管理員」。
- 新增主體：輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題，Cloud Scheduler 會將訊息發布至該主題，而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中，前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料：
- 主題 ID：輸入
crowdstrike-filevantage-trigger。 - 其他設定保留預設值。
- 主題 ID：輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，從 CrowdStrike FileVantage API 擷取記錄，並將記錄寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分，提供下列設定詳細資料：
設定 值 服務名稱 crowdstrike-filevantage-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取「Python 3.12」以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區：
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中，選擇 Pub/Sub 主題 (
crowdstrike-filevantage-trigger)。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分：
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「Containers, Networking, Security」。
前往「安全性」分頁：
- 服務帳戶：選取服務帳戶 (
crowdstrike-filevantage-sa)。
- 服務帳戶：選取服務帳戶 (
前往「容器」分頁：
- 按一下「變數與密鑰」。
- 針對每個環境變數，按一下「+ 新增變數」：
變數名稱 範例值 GCS_BUCKETcrowdstrike-filevantage-logsGCS_PREFIXfilevantage/STATE_KEYfilevantage/state.jsonFALCON_CLIENT_IDyour-client-idFALCON_CLIENT_SECRETyour-client-secretFALCON_BASE_URLhttps://api.crowdstrike.com(美國 - 1) /https://api.us-2.crowdstrike.com(美國 - 2) /https://api.eu-1.crowdstrike.com(歐盟 - 1)在「變數與密鑰」部分，向下捲動至「要求」：
- 要求逾時：輸入
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 # 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 CrowdStrike FileVantage logs 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', 'filevantage/') state_key = os.environ.get('STATE_KEY', 'filevantage/state.json') client_id = os.environ.get('FALCON_CLIENT_ID') client_secret = os.environ.get('FALCON_CLIENT_SECRET') base_url = os.environ.get('FALCON_BASE_URL') if not all([bucket_name, client_id, client_secret, base_url]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(bucket_name) # Get OAuth token token_url = f"{base_url}/oauth2/token" token_headers = { 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json' } token_data = f"client_id={client_id}&client_secret={client_secret}&grant_type=client_credentials" token_response = http.request( 'POST', token_url, body=token_data.encode('utf-8'), headers=token_headers ) if token_response.status != 200: print(f"Failed to get OAuth token: {token_response.status}") print(f"Response: {token_response.data.decode('utf-8')}") return token_data_json = json.loads(token_response.data.decode('utf-8')) access_token = token_data_json['access_token'] # Get last checkpoint last_timestamp = get_last_checkpoint(bucket, state_key) # Fetch file changes using v3 endpoint (high volume query) changes_url = f"{base_url}/filevantage/queries/changes/v3" headers = { 'Authorization': f'Bearer {access_token}', 'Accept': 'application/json' } # Build query parameters params = [] params.append('limit=5000') params.append('sort=action_timestamp|asc') if last_timestamp: params.append(f"filter=action_timestamp:>'{last_timestamp}'") query_url = f"{changes_url}?{'&'.join(params)}" backoff = 1.0 max_retries = 3 for attempt in range(max_retries): response = http.request('GET', query_url, headers=headers) 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) continue if response.status != 200: print(f"Failed to query changes: {response.status}") print(f"Response: {response.data.decode('utf-8')}") return break else: print("Max retries exceeded") return response_data = json.loads(response.data.decode('utf-8')) change_ids = response_data.get('resources', []) if not change_ids: print("No new changes found") return # Get detailed change information using v2 endpoint details_url = f"{base_url}/filevantage/entities/changes/v2" batch_size = 500 all_changes = [] latest_timestamp = last_timestamp for i in range(0, len(change_ids), batch_size): batch_ids = change_ids[i:i + batch_size] # Build query string with multiple ids parameters ids_params = '&'.join([f'ids={id}' for id in batch_ids]) details_query_url = f"{details_url}?{ids_params}" backoff = 1.0 for attempt in range(max_retries): details_response = http.request('GET', details_query_url, headers=headers) if details_response.status == 429: retry_after = int(details_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) continue if details_response.status == 200: details_data = json.loads(details_response.data.decode('utf-8')) changes = details_data.get('resources', []) all_changes.extend(changes) # Track latest timestamp for change in changes: change_time = change.get('action_timestamp') if change_time and (not latest_timestamp or change_time > latest_timestamp): latest_timestamp = change_time break else: print(f"Failed to get change details (batch {i//batch_size + 1}): {details_response.status}") break if all_changes: # Store logs in GCS timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') blob_name = f"{prefix}filevantage_changes_{timestamp}.json" blob = bucket.blob(blob_name) log_lines = '\n'.join(json.dumps(change) for change in all_changes) blob.upload_from_string(log_lines, content_type='application/json') # Update checkpoint save_checkpoint(bucket, state_key, latest_timestamp) print(f"Stored {len(all_changes)} changes in GCS: {blob_name}") except Exception as e: print(f"Error: {str(e)}") raise def get_last_checkpoint(bucket, key): """Get the 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') except Exception as e: print(f"Error reading checkpoint: {e}") return None def save_checkpoint(bucket, key, timestamp): """Save the last processed timestamp to GCS state file""" try: state = { 'last_timestamp': timestamp, 'updated_at': datetime.now(timezone.utc).isoformat() } blob = bucket.blob(key) blob.upload_from_string( json.dumps(state), content_type='application/json' ) except Exception as e: print(f"Error saving checkpoint: {e}")- 第二個檔案：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-filevantage-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 */15 * * * *(每 15 分鐘)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取 Pub/Sub 主題 ( crowdstrike-filevantage-trigger)郵件內文 {}(空白 JSON 物件)點選「建立」。
測試排程器工作
- 在 Cloud Scheduler 控制台中，找出您的作業 (
crowdstrike-filevantage-hourly)。 - 按一下「強制執行」即可手動觸發。
- 等待幾秒鐘，然後依序前往「Cloud Run」>「Services」(服務) >「crowdstrike-filevantage-collector」>「Logs」(記錄)。
- 確認函式是否已順利執行。
- 檢查 GCS 值區 (
crowdstrike-filevantage-logs)，確認記錄是否已寫入。
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶，從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
CrowdStrike FileVantage logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「CrowdStrike Filevantage」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬的服務帳戶電子郵件地址，例如：
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址，以便在下一步中使用。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (例如
crowdstrike-filevantage-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料：
- 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色：選取「Storage 物件檢視者」。
按一下 [儲存]。
在 Google SecOps 中設定動態消息，擷取 CrowdStrike FileVantage 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
CrowdStrike FileVantage logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「CrowdStrike Filevantage」做為「記錄類型」。
- 點選 [下一步]。
指定下列輸入參數的值：
儲存空間 bucket URL：輸入 GCS bucket URI，並加上前置路徑：
gs://crowdstrike-filevantage-logs/filevantage/取代：
crowdstrike-filevantage-logs：您的 GCS bucket 名稱。filevantage/：儲存記錄的前置字元/資料夾路徑。
來源刪除選項：根據偏好設定選取刪除選項：
- 永不：移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案：成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄：成功轉移後刪除檔案和空白目錄。
檔案存在時間上限：包含在過去天數內修改的檔案。預設值為 180 天。
資產命名空間：資產命名空間。
擷取標籤：要套用至這個動態饋給事件的標籤。
點選 [下一步]。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定，然後按一下「Submit」(提交)。
需要其他協助嗎？向社群成員和 Google SecOps 專業人員尋求答案。