收集 CrowdStrike Falcon Event Streams 記錄

剖析器版本：12.0

支援的國家/地區：

本文說明如何使用 Google Cloud Storage V2，將 CrowdStrike Falcon Event Streams 記錄檔擷取至 Google Security Operations。CrowdStrike Falcon Event Streams 提供即時串流 API，可從 Falcon 平台傳送安全性事件資料，包括偵測事件、稽核事件、驗證活動和事件更新。Event Streams API 會使用與 /sensors/entities/datafeed/v2 端點的持續性 HTTP 連線，近乎即時地推送事件，並支援以偏移量為基礎的續傳功能，確保傳送作業可靠。

事前準備

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

  • Google SecOps 執行個體。
  • 已啟用 Cloud Storage API 的 GCP 專案。
  • 建立及管理 GCS bucket 的權限。
  • 管理 Google Cloud Storage 值區 IAM 政策的權限。
  • 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限。
  • 具備建立 API 用戶端權限的 CrowdStrike Falcon Console 存取權。

建立 Google Cloud Storage bucket

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

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

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

  6. 點選「建立」。

設定 CrowdStrike Falcon API 存取權

建立 API 用戶端

  1. 登入 CrowdStrike Falcon 控制台。
  2. 在導覽選單中，依序前往「支援與資源」>「資源和工具」>「API 用戶端和金鑰」。
  3. 按一下「建立 API 用戶端」。
  4. 請提供下列設定詳細資料：
    • 用戶端名稱：輸入描述性名稱 (例如 Google SecOps Event Streams Integration)。
    • 說明 (選填)：輸入說明 (例如 API client for streaming events to GCS)。
  5. 在「API 範圍」部分，選取下列權限：
    • 事件串流：選取「讀取」 (使用串流 API 時必須選取)。
  6. 點選「建立」。

記錄 API 憑證

建立 API 用戶端後，您會收到下列憑證：

  • 用戶端 ID：32 個字元的小寫十六進位字串。
  • 用戶端密鑰：40 個字元的英數字元字串。
  • 基準網址：CrowdStrike API 的完整網域名稱 (例如 api.crowdstrike.com 或 api.us-2.crowdstrike.com)。

重要事項：請立即複製並儲存「Client Secret」。這個值只會顯示一次，之後便無法擷取。

地區端點

CrowdStrike 在多個區域營運，並提供不同的 API 端點：

區域 基準網址
US-1 api.crowdstrike.com
US-2 api.us-2.crowdstrike.com
EU-1 api.eu-1.crowdstrike.com
US-GOV-1 api.laggar.gcw.crowdstrike.com

重要事項：請使用與 CrowdStrike Falcon Console 登入區域對應的基準網址。

驗證權限

如要確認 API 用戶端是否具備必要權限，請按照下列步驟操作：

  1. 登入 CrowdStrike Falcon 控制台。
  2. 依序前往「支援與資源」>「資源和工具」>「API 用戶端和金鑰」。
  3. 找出您建立的 API 用戶端，並確認「事件串流」範圍顯示「讀取」存取權。

測試 API 存取權

  • 請先測試憑證，再繼續進行整合：

    CLIENT_ID="<your-client-id>"
CLIENT_SECRET="<your-client-secret>"
BASE_URL="https://api.crowdstrike.com"

# Get OAuth2 token
TOKEN=$(curl -s -X POST "${BASE_URL}/oauth2/token" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}" | python3 -c "import sys,json; print(json.load(sys.stdin)['access_token'])")

# Test Event Streams API access (discover available data feeds)
curl -s -H "Authorization: Bearer ${TOKEN}" \
  "${BASE_URL}/sensors/entities/datafeed/v2?appId=secops-test&format=flatjson"

如果回應成功，系統會傳回 JSON 物件，其中包含每個可用串流分割區的 resources、dataFeedURL 和 sessionToken 欄位。

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

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

這些角色適用於：

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

授予 GCS bucket 的 IAM 權限

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

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

建立 Pub/Sub 主題

建立 Pub/Sub 主題，供 Cloud Scheduler 發布訊息，以及 Cloud Run 函式訂閱訊息。

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

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發，連線至 CrowdStrike Event Streams API、取用事件，並將事件寫入 GCS。

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

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

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

  5. 在「觸發條件 (選用)」部分：

    1. 按一下「+ 新增觸發條件」。
    2. 選取「Cloud Pub/Sub」。
    3. 在「選取 Cloud Pub/Sub 主題」中，選擇 cs-stream-trigger。
    4. 按一下 [儲存]。

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

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

  7. 向下捲動並展開「容器、網路、安全性」。

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

    • 服務帳戶：選取 cs-stream-collector-sa

  9. 前往「容器」分頁：

    1. 按一下「變數與密鑰」。
    2. 針對每個環境變數，按一下「+ 新增變數」：
    變數名稱 範例值 說明
    GCS_BUCKET cs-stream-logs GCS bucket 名稱
    GCS_PREFIX crowdstrike/stream 記錄檔的前置字串
    STATE_KEY crowdstrike/stream/state.json 狀態檔案路徑
    CS_BASE_URL https://api.crowdstrike.com CrowdStrike API 基準網址
    CS_CLIENT_ID your-client-id API 用戶端 ID
    CS_CLIENT_SECRET your-client-secret API 用戶端密鑰
    CS_APP_ID secops-stream-collector Event Streams 的專屬應用程式 ID (最多 32 個英數字元)
    STREAM_TIMEOUT 300 串流讀取逾時時間 (以秒為單位)
    MAX_RECORDS 10000 每次執行的記錄數上限

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

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

  11. 前往「設定」分頁：

    • 在「資源」部分：
      • 記憶體：選取 512 MiB 以上
      • CPU：選取 1

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

    • 執行個體數量下限：輸入 0
    • 執行個體數量上限：輸入 1 (一次只能執行一個串流消費者)

  13. 點選「建立」。

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

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

新增函式程式碼

  1. 在「進入點」欄位中輸入 main。

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

    • main.py:

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

# 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', 'crowdstrike/stream')
STATE_KEY = os.environ.get('STATE_KEY', 'crowdstrike/stream/state.json')
CS_BASE_URL = os.environ.get('CS_BASE_URL', 'https://api.crowdstrike.com')
CS_CLIENT_ID = os.environ.get('CS_CLIENT_ID')
CS_CLIENT_SECRET = os.environ.get('CS_CLIENT_SECRET')
CS_APP_ID = os.environ.get('CS_APP_ID', 'secops-stream-collector')
STREAM_TIMEOUT = int(os.environ.get('STREAM_TIMEOUT', '300'))
MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000'))

def get_oauth_token(base_url: str, client_id: str, client_secret: str) -> str:
    """Get CrowdStrike OAuth2 access token."""
    token_url = f"{base_url}/oauth2/token"
    body = f"client_id={client_id}&client_secret={client_secret}"
    response = http.request(
        'POST', token_url,
        body=body.encode('utf-8'),
        headers={'Content-Type': 'application/x-www-form-urlencoded'}
    )
    if response.status != 200:
        raise Exception(f"OAuth token request failed: {response.status}")
    data = json.loads(response.data.decode('utf-8'))
    return data['access_token']

def refresh_stream_session(base_url: str, token: str, app_id: str, partition: int = 0):
    """Refresh an active stream session to keep it alive."""
    refresh_url = (
        f"{base_url}/sensors/entities/datafeed-actions/v1/{partition}"
        f"?appId={app_id}&action_name=refresh_active_stream_session"
    )
    response = http.request(
        'POST', refresh_url,
        headers={
            'Authorization': f'Bearer {token}',
            'Content-Type': 'application/json',
        }
    )
    if response.status == 200:
        print("Stream session refreshed successfully")
    else:
        print(f"Warning: Stream session refresh returned {response.status}")

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch CrowdStrike
    Event Streams data and write to GCS.
    """

    if not all([GCS_BUCKET, CS_BASE_URL, CS_CLIENT_ID, CS_CLIENT_SECRET]):
        print('Error: Missing required environment variables')
        return

    try:
        bucket = storage_client.bucket(GCS_BUCKET)

        # Load state
        state = load_state(bucket, STATE_KEY)

        # Get OAuth token
        token = get_oauth_token(CS_BASE_URL, CS_CLIENT_ID, CS_CLIENT_SECRET)
        headers = {
            'Authorization': f'Bearer {token}',
            'Accept': 'application/json',
        }

        # Discover available data feeds
        discover_url = (
            f"{CS_BASE_URL}/sensors/entities/datafeed/v2"
            f"?appId={CS_APP_ID}&format=flatjson"
        )
        response = http.request('GET', discover_url, headers=headers)

        if response.status != 200:
            print(f"Failed to discover data feed: {response.status}")
            print(f"Response: {response.data.decode('utf-8')}")
            return

        feed_data = json.loads(response.data.decode('utf-8'))
        resources = feed_data.get('resources', [])

        if not resources:
            print("No data feed resources available")
            return

        feed_url = resources[0].get('dataFeedURL', '')
        session_token = resources[0].get(
            'sessionToken', {}
        ).get('token', '')

        if not feed_url or not session_token:
            print("Missing feed URL or session token")
            return

        # Build stream URL with offset if available
        stream_url = feed_url
        offset = state.get('offset')
        if offset:
            stream_url = f"{feed_url}&offset={offset}"

        # Connect to Event Stream
        stream_headers = {
            'Authorization': f'Token {session_token}',
            'Accept': 'application/json',
            'Connection': 'keep-alive',
        }

        stream_http = urllib3.PoolManager(
            timeout=urllib3.Timeout(
                connect=10.0, read=float(STREAM_TIMEOUT)
            ),
            retries=False,
        )

        print(f"Connecting to Event Stream...")
        resp = stream_http.request(
            'GET', stream_url,
            headers=stream_headers,
            preload_content=False
        )

        if resp.status != 200:
            print(f"Stream connection failed: {resp.status}")
            return

        records = []
        latest_offset = offset
        now = datetime.now(timezone.utc)

        for line in resp.stream(4096):
            decoded = line.decode('utf-8').strip()
            if not decoded:
                continue

            try:
                event = json.loads(decoded)
                records.append(event)

                event_offset = event.get(
                    'metadata', {}
                ).get('offset')
                if event_offset is not None:
                    latest_offset = event_offset

                if len(records) >= MAX_RECORDS:
                    print(
                        f"Reached max records limit ({MAX_RECORDS})"
                    )
                    break
            except json.JSONDecodeError:
                continue

        resp.release_conn()

        if not records:
            print("No new events found in stream.")
            # Refresh stream session to keep it alive
            refresh_stream_session(
                CS_BASE_URL, token, CS_APP_ID
            )
            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(r, ensure_ascii=False) for r in records]
        ) + '\n'
        blob.upload_from_string(
            ndjson, content_type='application/x-ndjson'
        )

        print(
            f"Wrote {len(records)} records to "
            f"gs://{GCS_BUCKET}/{object_key}"
        )

        # Save state with latest offset
        new_state = {
            'offset': latest_offset,
            'last_run': now.isoformat()
        }
        save_state(bucket, STATE_KEY, new_state)

        # Refresh stream session to keep it alive
        refresh_stream_session(
            CS_BASE_URL, token, CS_APP_ID
        )

        print(f"Successfully processed {len(records)} events")

    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, state: dict):
    """Save state to GCS."""
    try:
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, indent=2),
            content_type='application/json'
        )
        print(f"Saved state: {state}")
    except Exception as e:
        print(f"Warning: Could not save state: {e}")

    requirements.txt:

    ```none
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. 提供下列設定詳細資料：

    設定 值
    名稱 cs-stream-collector-scheduled
    區域 選取與 Cloud Run 函式相同的區域
    頻率 */15 * * * * (每 15 分鐘)
    時區 選取時區 (建議使用世界標準時間)
    目標類型 Pub/Sub
    主題 選取「cs-stream-trigger」
    郵件內文 {} (空白 JSON 物件)

  4. 點選「建立」。

測試整合項目

  1. 在 Cloud Scheduler 控制台中，找出您的工作。
  2. 按一下「強制執行」，手動觸發工作。
  3. 稍等幾秒鐘。
  4. 前往「Cloud Run」>「服務」。
  5. 按一下 cs-stream-collector。
  6. 點選「記錄」分頁標籤。

  7. 確認函式是否執行成功。尋找：

    Connecting to Event Stream...
Wrote X records to gs://cs-stream-logs/crowdstrike/stream/logs_YYYYMMDD_HHMMSS.ndjson
Stream session refreshed successfully
Successfully processed X events

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

  9. 按一下 cs-stream-logs。

  10. 前往 crowdstrike/stream/ 資料夾。

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

如果記錄中顯示錯誤：

  • HTTP 401：檢查環境變數中的 API 憑證
  • HTTP 403：確認 API 用戶端具有「Event streams: Read」範圍
  • HTTP 429：速率限制 - 調整排程器頻率或增加 STREAM_TIMEOUT
  • 缺少環境變數：檢查是否已設定所有必要變數
  • 沒有可用的資料動態饋給資源：確認 CS_APP_ID 是唯一的，且不超過 32 個英數字元

擷取 Google SecOps 服務帳戶

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

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

  1. 依序前往「SIEM 設定」>「動態饋給」。
  2. 按一下「新增動態饋給」。
  3. 按一下「設定單一動態饋給」。
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 CrowdStrike Event Streams logs)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」。
  6. 選取「CrowdStrike Falcon Streaming」做為「記錄類型」。
  7. 按一下「取得服務帳戶」。

  8. 系統會顯示專屬服務帳戶電子郵件地址，例如：

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

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

  10. 點選「下一步」。

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

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

      gs://cs-stream-logs/crowdstrike/stream/

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

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

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

    • 檔案存在時間上限：納入在過去天數內修改的檔案 (預設為 180 天)

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

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

  12. 點選「下一步」。

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

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

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

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

  6. 按一下 [儲存]。

需要進一步協助嗎？

如要進一步瞭解 Google Security Operations 資訊提供，請參閱 Google Security Operations 資訊提供說明文件。如要瞭解各動態饋給類型的規定，請參閱「依類型設定動態饋給」。

如需參考資料，請參閱 CrowdStrike Falcon Event Streams 剖析器說明文件。

UDM 對應差異參考資料

2026 年 9 月 15 日，Google SecOps 發布新版 CrowdStrike Falcon Event Streams 剖析器，其中包含 CrowdStrike Falcon Event Streams 記錄欄位對應至 UDM 欄位的重大變更。

記錄欄位對應差異

下表列出 2026 年 9 月 15 日前和之後 (分別列於「舊版對應」和「新版對應」欄位) 顯示的 CrowdStrike Falcon Event Streams 記錄到 UDM 欄位對應差異：

記錄欄位 舊對應 新增對應 變更原因
deviceId additional.fields principal.asset.asset_id 對應至更明確的 UDM 欄位，而非標籤。
event.AuditKeyValues (report_name) security_result.detection_fields target.file.names 對應至更明確的 UDM 欄位，而非標籤。
event.IdpPolicyRuleName security_result.detection_fields security_result.rule_name 對應至更明確的 UDM 欄位，而非標籤。
event.MitreAttack.TacticID security_result.detection_fields security_result.tactic.id 對應至更明確的 UDM 欄位，而非標籤。
event.MitreAttack.TechniqueID security_result.detection_fields security_result.technique.id 對應至更明確的 UDM 欄位，而非標籤。
event.NetworkAccesse.RemoteAddress principal.ip target.ip 已更新，因為 event.NetworkAccesse.RemoteAddress 代表目標裝置地址。
event.TacticId security_result.detection_fields security_result.tactic.id 對應至更明確的 UDM 欄位，而非標籤。
event.TechniqueId security_result.detection_fields security_result.technique.id 對應至更明確的 UDM 欄位，而非標籤。
event.UserId principal.user.userid principal.user.email_addresses、
principal.user.userid		 如果 event.UserId 是電子郵件地址，則會對應至 principal.user.email_addresses；否則會對應至 principal.user.userid。
remoteAddress principal.ip target.ip 已更新，因為 remoteAddress 代表目標裝置地址。
ruleName security_result.detection_fields security_result.rule_name 對應至更明確的 UDM 欄位，而非標籤。
serviceName metadata.description、
target.application		 target.application 移除多餘的對應關係。

變更記錄

查看這個剖析器的變更記錄

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求解答。