CrowdStrike Falcon Event Streams のログを収集する
パーサーのバージョン: 12.0
このドキュメントでは、Google Cloud Storage V2 を使用して CrowdStrike Falcon Event Streams ログを Google Security Operations に取り込む方法について説明します。CrowdStrike Falcon Event Streams は、検出イベント、監査イベント、認証アクティビティ、インシデントの更新など、Falcon プラットフォームからセキュリティ イベント データを配信するリアルタイム ストリーミング API を提供します。Event Streams API は、/sensors/entities/datafeed/v2 エンドポイントとの永続的な HTTP 接続を使用して、ほぼリアルタイムでイベントをプッシュします。また、信頼性の高い配信のためにオフセットベースの再開をサポートしています。
始める前に
次の前提条件を満たしていることを確認してください。
- Google SecOps インスタンス。
- Cloud Storage API が有効になっている GCP プロジェクト。
- GCS バケットを作成および管理する権限。
- GCS バケットの IAM ポリシーを管理する権限。
- Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限。
- API クライアントを作成する権限を持つ CrowdStrike Falcon Console への特権アクセス。
Google Cloud Storage バケットを作成する
- Google Cloud コンソールに移動します。
- プロジェクトを選択するか、新しいプロジェクトを作成します。
- ナビゲーション メニューで、[Cloud Storage > バケット] に移動します。
- [バケットを作成] をクリックします。
次の構成情報を提供してください。
設定 値 バケットに名前を付ける グローバルに一意の名前( cs-stream-logsなど)を入力します。ロケーション タイプ ニーズに応じて選択します(リージョン、デュアルリージョン、マルチリージョン)。 ロケーション 場所を選択します(例: us-central1)。ストレージ クラス Standard(アクセス頻度の高いログにおすすめ) アクセス制御 均一(推奨) 保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする [作成] をクリックします。
CrowdStrike Falcon API アクセスを構成する
API クライアントを作成する
- CrowdStrike Falcon コンソールにログインします。
- ナビゲーション メニューで、[サポートとリソース> リソースとツール > API クライアントとキー] に移動します。
- [API クライアントを作成] をクリックします。
- 次の構成情報を指定します。
- クライアント名: わかりやすい名前を入力します(例:
Google SecOps Event Streams Integration)。 - 説明(省略可): 説明を入力します(例:
API client for streaming events to GCS)。
- クライアント名: わかりやすい名前を入力します(例:
- [API スコープ] セクションで、次の権限を選択します。
- イベント ストリーム: [読み取り] を選択します(ストリーミング API を使用するために必要)。
- [作成] をクリックします。
API 認証情報を記録する
API クライアントを作成すると、次の認証情報が返されます。
- クライアント ID: 32 文字の小文字の 16 進文字列。
- クライアント シークレット: 40 文字の英数字文字列。
- Base URL: CrowdStrike API の完全修飾ドメイン名(
api.crowdstrike.comやapi.us-2.crowdstrike.comなど)。
重要: クライアント シークレットをコピーしてすぐに保存します。この値は一度しか表示されず、後で取得することはできません。
リージョン エンドポイント
CrowdStrike は、異なる API エンドポイントを持つ複数のリージョンで動作します。
| リージョン | ベース URL |
|---|---|
| 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 のログイン リージョンに対応するベース URL を使用します。
権限を確認する
API クライアントに必要な権限があることを確認するには:
- CrowdStrike Falcon コンソールにログインします。
- [サポートとリソース] > [リソースとツール] > [API クライアントとキー] に移動します。
- 作成した 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"
レスポンスが成功すると、使用可能な各ストリーム パーティションの dataFeedURL フィールドと sessionToken フィールドを含む resources を含む JSON オブジェクトが返されます。
Cloud Run functions のサービス アカウントを作成する
Cloud Run functions には、GCS バケットへの書き込み権限があり、Pub/Sub によって呼び出されるサービス アカウントが必要です。
サービス アカウントの作成
- GCP コンソールで、[IAM と管理>サービス アカウント] に移動します。
- [サービス アカウントを作成] をクリックします。
- 次の構成情報を指定します。
- サービス アカウント名: 「
cs-stream-collector-sa」と入力します。 - サービス アカウントの説明:
Service account for Cloud Run function to collect CrowdStrike Event Streams logsと入力します。
- サービス アカウント名: 「
- [作成して続行] をクリックします。
- [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
- [ロールを選択] をクリックします。
- [ストレージ オブジェクト管理者] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Run 起動元] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Functions 起動元] を検索して選択します。
- [続行] をクリックします。
- [完了] をクリックします。
これらのロールは、次の操作に必要です。
- Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
- Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可します。
- Cloud Functions 起動元: 関数の呼び出しを許可する
GCS バケットに対する IAM 権限を付与する
GCS バケットに対する書き込み権限をサービス アカウントに付与します。
- [Cloud Storage] > [バケット] に移動します。
- バケット名(
cs-stream-logsなど)をクリックします。 - [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルを追加: サービス アカウントのメールアドレス(例:
cs-stream-collector-sa@PROJECT_ID.iam.gserviceaccount.com)を入力します。 - ロールを割り当てる: [ストレージ オブジェクト管理者] を選択します。
- プリンシパルを追加: サービス アカウントのメールアドレス(例:
- [保存] をクリックします。
Pub/Sub トピックの作成
Cloud Scheduler がパブリッシュし、Cloud Run 関数がサブスクライブする Pub/Sub トピックを作成します。
- GCP Console で、[Pub/Sub> トピック] に移動します。
- [トピックを作成] をクリックします。
- 次の構成情報を指定します。
- トピック ID: 「
cs-stream-trigger」と入力します。 - その他の設定はデフォルトのままにします。
- トピック ID: 「
- [作成] をクリックします。
ログを収集する Cloud Run 関数を作成する
Cloud Run functions は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、CrowdStrike Event Streams API に接続してイベントを使用し、GCS に書き込みます。
- GCP Console で、[Cloud Run] に移動します。
- [サービスを作成] をクリックします。
- [関数] を選択します(インライン エディタを使用して関数を作成します)。
[構成] セクションで、次の構成の詳細を指定します。
設定 値 サービス名 cs-stream-collectorリージョン GCS バケットと一致するリージョンを選択します(例: us-central1)。ランタイム [Python 3.12] 以降を選択します。 [トリガー(省略可)] セクションで、次の操作を行います。
- [+ トリガーを追加] をクリックします。
- [Cloud Pub/Sub] を選択します。
- [Cloud Pub/Sub トピックを選択してください] で、
cs-stream-triggerを選択します。 - [保存] をクリックします。
[認証] セクションで、次の操作を行います。
- [認証が必要] を選択します。
- Identity and Access Management(IAM)を確認します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービス アカウント:
cs-stream-collector-saを選択します。
- サービス アカウント:
[コンテナ] タブに移動します。
- [変数とシークレット] をクリックします。
- 環境変数ごとに [+ 変数を追加] をクリックします。
変数名 値の例 説明 GCS_BUCKETcs-stream-logsGCS バケット名 GCS_PREFIXcrowdstrike/streamログファイルの接頭辞 STATE_KEYcrowdstrike/stream/state.json状態ファイルのパス CS_BASE_URLhttps://api.crowdstrike.comCrowdStrike API のベース URL CS_CLIENT_IDyour-client-idAPI クライアント ID CS_CLIENT_SECRETyour-client-secretAPI クライアント シークレット CS_APP_IDsecops-stream-collectorイベント ストリームの一意のアプリ ID(英数字 32 文字以内) STREAM_TIMEOUT300ストリーム読み取りタイムアウト(秒) MAX_RECORDS10000実行あたりの最大レコード数 [変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。
- リクエストのタイムアウト:
600秒(10 分)を入力します。
- リクエストのタイムアウト:
[設定] タブに移動します。
- [リソース] セクションで次の操作を行います。
- メモリ: 512 MiB 以上を選択します。
- CPU: 1 を選択します。
- [リソース] セクションで次の操作を行います。
[リビジョン スケーリング] セクションで、次の操作を行います。
- インスタンスの最小数:
0と入力します。 - インスタンスの最大数:
1と入力します(一度に実行できるストリーム コンシューマーは 1 つのみです)。
- インスタンスの最小数:
[作成] をクリックします。
サービスが作成されるまで待ちます(1 ~ 2 分)。
サービスを作成すると、インライン コードエディタが自動的に開きます。
関数コードを追加する
- [エントリ ポイント] フィールドに「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 ```[デプロイ] をクリックして、関数を保存してデプロイします。
デプロイが完了するまで待ちます(2 ~ 3 分)。
Cloud Scheduler ジョブの作成
Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run functions の関数をトリガーします。
- GCP Console で、[Cloud Scheduler] に移動します。
- [ジョブを作成] をクリックします。
次の構成情報を提供してください。
設定 値 名前 cs-stream-collector-scheduledリージョン Cloud Run functions と同じリージョンを選択する 周波数 */15 * * * *(15 分ごと)タイムゾーン タイムゾーンを選択(UTC を推奨) ターゲット タイプ Pub/Sub トピック cs-stream-triggerを選択メッセージ本文 {}(空の JSON オブジェクト)[作成] をクリックします。
統合をテストする
- Cloud Scheduler コンソールで、ジョブを見つけます。
- [強制実行] をクリックして、ジョブを手動でトリガーします。
- 数秒待ちます。
- Cloud Run > [サービス] に移動します。
cs-stream-collectorをクリックします。- [ログ] タブをクリックします。
関数が正常に実行されたことを確認します。以下のものを探します。
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[Cloud Storage] > [バケット] に移動します。
cs-stream-logsをクリックします。crowdstrike/stream/フォルダに移動します。現在のタイムスタンプで新しい
.ndjsonファイルが作成されたことを確認します。
ログにエラーが表示された場合:
- HTTP 401: 環境変数で API 認証情報を確認する
- HTTP 403: API クライアントにイベント ストリーム: 読み取りスコープがあることを確認する
- HTTP 429: レート制限 - スケジューラ頻度を調整するか、
STREAM_TIMEOUTを増やす - 環境変数が不足している: 必要な変数がすべて設定されていることを確認します。
- No data feed resources available:
CS_APP_IDが一意であり、32 文字の英数字を超えていないことを確認します
Google SecOps サービス アカウントを取得する
Google SecOps は、一意のサービス アカウントを使用して GCS バケットからデータを読み取ります。このサービス アカウントにバケットへのアクセス権を付与する必要があります。
サービス アカウントのメールアドレスを取得する
- [SIEM 設定] > [フィード] に移動します。
- [Add New Feed] をクリックします。
- [単一フィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します(例:
CrowdStrike Event Streams logs)。 - [ソースタイプ] として [Google Cloud Storage V2] を選択します。
- [ログタイプ] として [CrowdStrike Falcon Streaming] を選択します。
- [サービス アカウントを取得する] をクリックします。
一意のサービス アカウントのメールアドレスが表示されます(例:)。
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comこのメールアドレスをコピーして、次のステップで使用します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
gs://cs-stream-logs/crowdstrike/stream/
Source deletion option: 必要に応じて削除オプションを選択します。
- なし: 転送後にファイルを削除しません(テストにおすすめ)。
- 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
ファイルの最大経過日数: 過去の日数以内に変更されたファイルを含めます(デフォルトは 180 日)。
アセットの名前空間: アセットの名前空間
Ingestion labels: このフィードのイベントに適用されるラベル
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。
Google SecOps サービス アカウントに IAM 権限を付与する
Google SecOps サービス アカウントには、GCS バケットに対する Storage オブジェクト閲覧者のロールが必要です。
- [Cloud Storage] > [バケット] に移動します。
cs-stream-logsをクリックします。- [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [Storage オブジェクト閲覧者] を選択します。
[保存] をクリックします。
さらにサポートが必要な場合
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 のプロフェッショナルから回答を得ることができます。