Code42 Incydr のコア データセットを収集する
このドキュメントでは、Google Cloud Storage を使用して Code42 Incydr のコア データセット(ユーザー、監査、ケース、必要に応じてファイル イベント)を Google Security Operations に取り込む方法について説明します。
Code42 Incydr は、エンドポイント、クラウド サービス、メール全体でファイル アクティビティをリアルタイムでモニタリングし、デバイス全体でのデータ流出を検出、調査、対応する内部リスク管理ソリューションです。
始める前に
次の前提条件を満たしていることを確認します。
- Google SecOps インスタンス
- Cloud Storage API が有効になっている GCP プロジェクト
- GCS バケットを作成および管理する権限
- GCS バケットの IAM ポリシーを管理する権限
- Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
- 内部リスク管理者ロールを持つ Code42 Incydr API または管理コンソールへの特権アクセス
Google Cloud Storage バケットを作成する
- Google Cloud Console に移動します。
- プロジェクトを選択するか、新しいプロジェクトを作成します。
- ナビゲーション メニューで、[Cloud Storage > バケット] に移動します。
- [バケットを作成] をクリックします。
次の構成情報を提供してください。
設定 値 バケットに名前を付ける グローバルに一意の名前( code42-incydr-logsなど)を入力します。ロケーション タイプ ニーズに基づいて選択します(リージョン、デュアルリージョン、マルチリージョン)。 ロケーション ロケーションを選択します(例: us-central1)。ストレージ クラス Standard(頻繁にアクセスされるログにおすすめ) アクセス制御 均一(推奨) 保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする [作成] をクリックします。
Code42 Incydr API 認証情報を収集する
API クライアントを作成する
- Code42 Incydr ウェブ コンソールにログインします。
- [管理> 統合> API クライアント] に移動します。
- [Create new API client] をクリックします。
- [Create new API client] ダイアログで、クライアントの名前(
Google Security Operations Integrationなど)を入力します。 - 次の詳細をコピーして安全な場所に保存します。
- クライアント ID: API クライアントの識別子。
- Secret: API クライアント シークレット キー。
- [作成] をクリックします。
API ベース URL を特定する
API ベース URL は、Code42 Incydr コンソール URL によって異なります。Incydr デベロッパー ポータルまたはテナントの環境ドキュメントで API ゲートウェイの URL を確認します。
一般的なデフォルト:
コンソールの URL API ベース URL https://console.us.code42.comhttps://api.us.code42.comhttps://console.us2.code42.comhttps://api.us2.code42.comhttps://console.ie.code42.comhttps://api.ie.code42.comhttps://console.gov.code42.comhttps://api.gov.code42.com
API クライアントの権限を確認する
API クライアントには、必要なエンドポイントにアクセスするための適切な権限が必要です。
- Code42 Incydr コンソールで、[Administration] > [Integrations] > [API Clients] に移動します。
- 作成した API クライアント名をクリックします。
API クライアントが次のスコープにアクセスできることを確認します。
- ユーザー: ユーザーデータへの読み取りアクセス権
- 監査ログ: 監査ログへの読み取りアクセス
- Cases: ケースデータへの読み取りアクセス
- ファイル イベント(省略可): ファイル イベント データへの読み取りアクセス
テスト API アクセス
統合に進む前に、認証情報をテストします。
# Replace with your actual credentials CLIENT_ID="your-client-id" CLIENT_SECRET="your-client-secret" API_BASE="https://api.us.code42.com" # Get OAuth token TOKEN=$(curl -s -X POST "${API_BASE}/v1/oauth/token" \ -u "${CLIENT_ID}:${CLIENT_SECRET}" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "grant_type=client_credentials" | jq -r '.access_token') # Test API access curl -v -H "Authorization: Bearer ${TOKEN}" "${API_BASE}/v1/users?pageSize=1"
Cloud Run functions のサービス アカウントを作成する
Cloud Run 関数には、GCS バケットに書き込み、Pub/Sub によって呼び出される権限を持つサービス アカウントが必要です。
サービス アカウントの作成
- GCP Console で、[IAM と管理>サービス アカウント] に移動します。
- [サービス アカウントを作成] をクリックします。
- 次の構成の詳細を指定します。
- サービス アカウント名: 「
code42-incydr-collector-sa」と入力します。 - サービス アカウントの説明: 「
Service account for Cloud Run function to collect Code42 Incydr logs」と入力します。
- サービス アカウント名: 「
- [作成して続行] をクリックします。
- [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
- [ロールを選択] をクリックします。
- [ストレージ オブジェクト管理者] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Run 起動元] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Functions 起動元] を検索して選択します。
- [続行] をクリックします。
- [完了] をクリックします。
これらのロールは、次の目的で必要です。
- Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
- Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可する
- Cloud Functions 起動元: 関数の呼び出しを許可する
GCS バケットに対する IAM 権限を付与する
GCS バケットに対する書き込み権限をサービス アカウントに付与します。
- [Cloud Storage] > [バケット] に移動します。
- バケット名(
code42-incydr-logsなど)をクリックします。 - [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成の詳細を指定します。
- プリンシパルを追加: サービス アカウントのメールアドレス(例:
code42-incydr-collector-sa@your-project.iam.gserviceaccount.com)を入力します。 - ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
- プリンシパルを追加: サービス アカウントのメールアドレス(例:
- [保存] をクリックします。
Pub/Sub トピックの作成
Cloud Scheduler がパブリッシュし、Cloud Run functions がサブスクライブする Pub/Sub トピックを作成します。
- GCP Console で、[Pub/Sub> トピック] に移動します。
- [トピックを作成] をクリックします。
- 次の構成の詳細を指定します。
- トピック ID: 「
code42-incydr-trigger」と入力します。 - その他の設定はデフォルトのままにします。
- トピック ID: 「
- [作成] をクリックします。
ログを収集する Cloud Run 関数を作成する
Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Code42 Incydr API からログを取得して GCS に書き込みます。
- GCP Console で、[Cloud Run] に移動します。
- [サービスを作成] をクリックします。
- [関数] を選択します(インライン エディタを使用して関数を作成します)。
[構成] セクションで、次の構成の詳細を指定します。
設定 値 サービス名 code42-incydr-collectorリージョン GCS バケットと一致するリージョンを選択します(例: us-central1)。ランタイム [Python 3.12] 以降を選択します。 [トリガー(省略可)] セクションで、次の操作を行います。
- [+ トリガーを追加] をクリックします。
- [Cloud Pub/Sub] を選択します。
- [Cloud Pub/Sub トピックを選択してください] で、トピック
code42-incydr-triggerを選択します。 - [保存] をクリックします。
[認証] セクションで、次の操作を行います。
- [認証が必要] を選択します。
- Identity and Access Management(IAM)を確認します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービス アカウント: サービス アカウント
code42-incydr-collector-saを選択します。
- サービス アカウント: サービス アカウント
[コンテナ] タブに移動します。
- [変数とシークレット] をクリックします。
- 環境変数ごとに [+ 変数を追加] をクリックします。
変数名 値の例 説明 INCYDR_BASE_URLhttps://api.us.code42.comテナントの API ベース URL INCYDR_CLIENT_IDyour-client-idAPI クライアント ID INCYDR_CLIENT_SECRETyour-client-secretAPI クライアント シークレット GCS_BUCKETcode42-incydr-logsGCS バケット名 GCS_PREFIXcode42/ログファイルの接頭辞 PAGE_SIZE5001 ページあたりのレコード数 LOOKBACK_MINUTES60最初のルックバック期間 STREAMSusers,audit,casesカンマ区切りのデータ ストリーム FE_QUERY_JSON`` 省略可: ファイル イベントのクエリ JSON FE_PAGE_SIZE1000省略可: ファイル イベント ページのサイズ [変数とシークレット] タブで [リクエスト] まで下にスクロールします。
- リクエストのタイムアウト:
600秒(10 分)を入力します。
- リクエストのタイムアウト:
[コンテナ] の [設定] タブに移動します。
- [リソース] セクションで次の操作を行います。
- メモリ: 1,024 MiB 以上を選択します。
- CPU: [1] を選択します。
- [完了] をクリックします。
- [リソース] セクションで次の操作を行います。
[実行環境] までスクロールします。
- [デフォルト](推奨)を選択します。
[リビジョン スケーリング] セクションで、次の操作を行います。
- [インスタンスの最小数] に「
0」と入力します。 - インスタンスの最大数:
100と入力します(または、予想される負荷に基づいて調整します)。
- [インスタンスの最小数] に「
[作成] をクリックします。
サービスが作成されるまで待ちます(1 ~ 2 分)。
サービスを作成すると、インライン コードエディタが自動的に開きます。
関数コードを追加する
- [関数のエントリ ポイント] に「main」と入力します。
インライン コードエディタで、次の 2 つのファイルを作成します。
- 最初のファイル: main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timedelta, timezone import time # Initialize HTTP client http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=5.0, read=30.0), retries=False, ) # Initialize Storage client storage_client = storage.Client() # Environment variables BASE = os.environ.get("INCYDR_BASE_URL", "").rstrip("/") CID = os.environ.get("INCYDR_CLIENT_ID", "") CSECRET = os.environ.get("INCYDR_CLIENT_SECRET", "") BUCKET = os.environ.get("GCS_BUCKET", "") PREFIX_BASE = os.environ.get("GCS_PREFIX", "code42/") PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "500")) LOOKBACK_MINUTES = int(os.environ.get("LOOKBACK_MINUTES", "60")) STREAMS = [s.strip() for s in os.environ.get("STREAMS", "users").split(",") if s.strip()] FE_QUERY_JSON = os.environ.get("FE_QUERY_JSON", "").strip() FE_PAGE_SIZE = int(os.environ.get("FE_PAGE_SIZE", "1000")) def now_utc(): return datetime.now(timezone.utc) def iso_minus(minutes: int): return (now_utc() - timedelta(minutes=minutes)).strftime("%Y-%m-%dT%H:%M:%SZ") def put_json(bucket, prefix: str, page_label: str, data): ts = now_utc().strftime("%Y/%m/%d/%H%M%S") key = f"{PREFIX_BASE}{prefix}{ts}-{page_label}.json" blob = bucket.blob(key) blob.upload_from_string(json.dumps(data), content_type='application/json') return key def get_token(): """Get OAuth 2.0 access token using client credentials flow.""" token_url = f"{BASE}/v1/oauth/token" # Encode credentials import base64 credentials = f"{CID}:{CSECRET}" encoded_credentials = base64.b64encode(credentials.encode('utf-8')).decode('utf-8') headers = { 'Authorization': f'Basic {encoded_credentials}', 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json' } body = 'grant_type=client_credentials' backoff = 1.0 max_retries = 3 for attempt in range(max_retries): response = http.request('POST', token_url, body=body, headers=headers) if response.status == 429: retry_after = int(response.headers.get('Retry-After', str(int(backoff)))) print(f"Rate limited (429) on token request. Retrying after {retry_after}s...") time.sleep(retry_after) backoff = min(backoff * 2, 30.0) continue if response.status != 200: raise RuntimeError(f"Failed to get access token: {response.status} - {response.data.decode('utf-8')}") data = json.loads(response.data.decode('utf-8')) return data['access_token'] raise RuntimeError(f"Failed to get token after {max_retries} retries due to rate limiting") def auth_header(): token = get_token() return { "Authorization": f"Bearer {token}", "Accept": "application/json" } def http_get(path: str, params: dict = None, headers: dict = None): url = f"{BASE}{path}" if params: from urllib.parse import urlencode url += "?" + urlencode(params) backoff = 1.0 max_retries = 3 for attempt in range(max_retries): response = http.request('GET', 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 return response.data raise RuntimeError(f"Failed after {max_retries} retries due to rate limiting") def http_post_json(path: str, body: dict, headers: dict = None): url = f"{BASE}{path}" backoff = 1.0 max_retries = 3 for attempt in range(max_retries): response = http.request( 'POST', url, body=json.dumps(body), headers={**headers, 'Content-Type': 'application/json'} ) 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 return response.data raise RuntimeError(f"Failed after {max_retries} retries due to rate limiting") # USERS (/v1/users) def pull_users(bucket, hdrs): next_token = None pages = 0 while True: params = {"active": "true", "blocked": "false", "pageSize": PAGE_SIZE} if next_token: params["pageToken"] = next_token raw = http_get("/v1/users", params, hdrs) data = json.loads(raw.decode('utf-8')) put_json(bucket, "users/", f"users-page-{pages}", data) pages += 1 next_token = data.get("nextPageToken") or data.get("next_page_token") if not next_token: break return pages # AUDIT LOG (/v1/audit/log) def pull_audit(bucket, hdrs): start_iso = iso_minus(LOOKBACK_MINUTES) next_token = None pages = 0 while True: params = {"startTime": start_iso, "pageSize": PAGE_SIZE} if next_token: params["pageToken"] = next_token raw = http_get("/v1/audit/log", params, hdrs) try: data = json.loads(raw.decode('utf-8')) put_json(bucket, "audit/", f"audit-page-{pages}", data) next_token = data.get("nextPageToken") or data.get("next_page_token") pages += 1 if not next_token: break except Exception as e: print(f"Error parsing audit log response: {e}") # Save raw response ts = now_utc().strftime("%Y/%m/%d/%H%M%S") key = f"{PREFIX_BASE}audit/{ts}-audit-export.bin" blob = bucket.blob(key) blob.upload_from_string(raw, content_type='application/octet-stream') pages += 1 break return pages # CASES (/v1/cases) def pull_cases(bucket, hdrs): next_token = None pages = 0 while True: params = {"pageSize": PAGE_SIZE} if next_token: params["pageToken"] = next_token raw = http_get("/v1/cases", params, hdrs) data = json.loads(raw.decode('utf-8')) put_json(bucket, "cases/", f"cases-page-{pages}", data) pages += 1 next_token = data.get("nextPageToken") or data.get("next_page_token") if not next_token: break return pages # FILE EVENTS (/v2/file-events/search) def pull_file_events(bucket, hdrs): if not FE_QUERY_JSON: return 0 try: base_query = json.loads(FE_QUERY_JSON) except Exception as e: print(f"Error: FE_QUERY_JSON is not valid JSON: {e}") return 0 pages = 0 next_token = None while True: body = dict(base_query) body["pageSize"] = FE_PAGE_SIZE if next_token: body["pageToken"] = next_token raw = http_post_json("/v2/file-events/search", body, hdrs) data = json.loads(raw.decode('utf-8')) put_json(bucket, "file_events/", f"fileevents-page-{pages}", data) pages += 1 next_token = ( data.get("nextPageToken") or data.get("next_page_token") or (data.get("file_events") or {}).get("nextPageToken") ) if not next_token: break return pages @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch logs from Code42 Incydr API and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([BASE, CID, CSECRET, BUCKET]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(BUCKET) hdrs = auth_header() report = {} if "users" in STREAMS: print("Fetching users...") report["users_pages"] = pull_users(bucket, hdrs) if "audit" in STREAMS: print("Fetching audit logs...") report["audit_pages"] = pull_audit(bucket, hdrs) if "cases" in STREAMS: print("Fetching cases...") report["cases_pages"] = pull_cases(bucket, hdrs) if "file_events" in STREAMS: print("Fetching file events...") report["file_events_pages"] = pull_file_events(bucket, hdrs) print(f'Successfully processed logs: {json.dumps(report)}') except Exception as e: print(f'Error processing logs: {str(e)}') raise- 2 つ目のファイル: requirements.txt:
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] に移動します。
- [ジョブを作成] をクリックします。
次の構成情報を提供してください。
設定 値 名前 code42-incydr-hourlyリージョン Cloud Run functions と同じリージョンを選択する 周波数 0 * * * *(1 時間ごとに正時)タイムゾーン タイムゾーンを選択します(UTC を推奨)。 ターゲット タイプ Pub/Sub トピック トピック code42-incydr-triggerを選択します。メッセージ本文 {}(空の JSON オブジェクト)[作成] をクリックします。
スケジュールの頻度のオプション
ログの量とレイテンシの要件に基づいて頻度を選択します。
頻度 CRON 式 ユースケース 5 分毎 */5 * * * *大容量、低レイテンシ 15 分ごと */15 * * * *検索量が普通 1 時間ごと 0 * * * *標準(推奨) 6 時間ごと 0 */6 * * *少量、バッチ処理 毎日 0 0 * * *履歴データの収集
スケジューラ ジョブをテストする
- Cloud Scheduler コンソールで、ジョブ(
code42-incydr-hourly)を見つけます。 - [強制実行] をクリックして手動でトリガーします。
- 数秒待ってから、[Cloud Run> サービス> code42-incydr-collector > ログ] に移動します。
関数が正常に実行されたことを確認します。求めていること:
Fetching users... Fetching audit logs... Fetching cases... Successfully processed logs: {"users_pages": X, "audit_pages": Y, "cases_pages": Z}GCS バケット(
code42-incydr-logs)をチェックして、ログが書き込まれたことを確認します。
ログにエラーが表示された場合:
- HTTP 401: 環境変数で API 認証情報を確認する
- HTTP 403: Code42 Incydr コンソールで API クライアントに必要な権限があることを確認する
- HTTP 429: レート制限 - 関数はバックオフで自動的に再試行されます
- アクセス トークンの取得に失敗しました:
INCYDR_BASE_URL、INCYDR_CLIENT_ID、INCYDR_CLIENT_SECRETが正しいことを確認します
Google SecOps サービス アカウントを取得する
Google SecOps は、一意のサービス アカウントを使用して GCS バケットからデータを読み取ります。このサービス アカウントにバケットへのアクセス権を付与する必要があります。
サービス アカウントのメールアドレスを取得する
- [SIEM 設定] > [フィード] に移動します。
- [Add New Feed] をクリックします。
- [単一フィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します(例:
Code42 Incydr Datasets)。 - [ソースタイプ] として [Google Cloud Storage V2] を選択します。
- [ログタイプ] として [Code42 Incydr] を選択します。
[サービス アカウントを取得する] をクリックします。一意のサービス アカウント メールアドレスが表示されます(例:)。
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.comこのメールアドレスをコピーして、次のステップで使用します。
Google SecOps サービス アカウントに IAM 権限を付与する
Google SecOps サービス アカウントには、GCS バケットに対する Storage オブジェクト閲覧者ロールが必要です。
- [Cloud Storage] > [バケット] に移動します。
- バケット名(
code42-incydr-logs)をクリックします。 - [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成の詳細を指定します。
- プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [ストレージ オブジェクト閲覧者] を選択します。
[保存] をクリックします。
Code42 Incydr のログを取り込むように Google SecOps でフィードを構成する
- [SIEM 設定] > [フィード] に移動します。
- [Add New Feed] をクリックします。
- [単一フィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します(例:
Code42 Incydr Datasets)。 - [ソースタイプ] として [Google Cloud Storage V2] を選択します。
- [ログタイプ] として [Code42 Incydr] を選択します。
- [次へ] をクリックします。
次の入力パラメータの値を指定します。
ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
gs://code42-incydr-logs/code42/次のように置き換えます。
code42-incydr-logs: GCS バケット名。code42/: ログが保存されるオプションの接頭辞/フォルダパス(ルートの場合は空のままにします)。
例:
- ルートバケット:
gs://code42-incydr-logs/ - 接頭辞あり:
gs://code42-incydr-logs/code42/
- ルートバケット:
Source deletion option: 必要に応じて削除オプションを選択します。
- なし: 転送後にファイルを削除しません(テストにおすすめ)。
- 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
- 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
アセットの名前空間: アセットの名前空間。
Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。
ご不明な点がございましたら、コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。