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 バケットを作成する

  1. Google Cloud Console に移動します。
  2. プロジェクトを選択するか、新しいプロジェクトを作成します。
  3. ナビゲーション メニューで、[Cloud Storage > バケット] に移動します。
  4. [バケットを作成] をクリックします。
  5. 次の構成情報を提供してください。

    設定
    バケットに名前を付ける グローバルに一意の名前(code42-incydr-logs など)を入力します。
    ロケーション タイプ ニーズに基づいて選択します(リージョン、デュアルリージョン、マルチリージョン)。
    ロケーション ロケーションを選択します(例: us-central1)。
    ストレージ クラス Standard(頻繁にアクセスされるログにおすすめ)
    アクセス制御 均一(推奨)
    保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする
  6. [作成] をクリックします。

Code42 Incydr API 認証情報を収集する

API クライアントを作成する

  1. Code42 Incydr ウェブ コンソールにログインします。
  2. [管理> 統合> API クライアント] に移動します。
  3. [Create new API client] をクリックします。
  4. [Create new API client] ダイアログで、クライアントの名前(Google Security Operations Integration など)を入力します。
  5. 次の詳細をコピーして安全な場所に保存します。
    • クライアント ID: API クライアントの識別子。
    • Secret: API クライアント シークレット キー。
  6. [作成] をクリックします。

API ベース URL を特定する

API ベース URL は、Code42 Incydr コンソール URL によって異なります。Incydr デベロッパー ポータルまたはテナントの環境ドキュメントで API ゲートウェイの URL を確認します。

  • 一般的なデフォルト:

    コンソールの URL API ベース URL
    https://console.us.code42.com https://api.us.code42.com
    https://console.us2.code42.com https://api.us2.code42.com
    https://console.ie.code42.com https://api.ie.code42.com
    https://console.gov.code42.com https://api.gov.code42.com

API クライアントの権限を確認する

API クライアントには、必要なエンドポイントにアクセスするための適切な権限が必要です。

  1. Code42 Incydr コンソールで、[Administration] > [Integrations] > [API Clients] に移動します。
  2. 作成した API クライアント名をクリックします。
  3. 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 によって呼び出される権限を持つサービス アカウントが必要です。

サービス アカウントの作成

  1. GCP Console で、[IAM と管理>サービス アカウント] に移動します。
  2. [サービス アカウントを作成] をクリックします。
  3. 次の構成の詳細を指定します。
    • サービス アカウント名: 「code42-incydr-collector-sa」と入力します。
    • サービス アカウントの説明: 「Service account for Cloud Run function to collect Code42 Incydr logs」と入力します。
  4. [作成して続行] をクリックします。
  5. [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
    1. [ロールを選択] をクリックします。
    2. [ストレージ オブジェクト管理者] を検索して選択します。
    3. [+ 別のロールを追加] をクリックします。
    4. [Cloud Run 起動元] を検索して選択します。
    5. [+ 別のロールを追加] をクリックします。
    6. [Cloud Functions 起動元] を検索して選択します。
  6. [続行] をクリックします。
  7. [完了] をクリックします。

これらのロールは、次の目的で必要です。

  • Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
  • Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可する
  • Cloud Functions 起動元: 関数の呼び出しを許可する

GCS バケットに対する IAM 権限を付与する

GCS バケットに対する書き込み権限をサービス アカウントに付与します。

  1. [Cloud Storage] > [バケット] に移動します。
  2. バケット名(code42-incydr-logs など)をクリックします。
  3. [権限] タブに移動します。
  4. [アクセス権を付与] をクリックします。
  5. 次の構成の詳細を指定します。
    • プリンシパルを追加: サービス アカウントのメールアドレス(例: code42-incydr-collector-sa@your-project.iam.gserviceaccount.com)を入力します。
    • ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
  6. [保存] をクリックします。

Pub/Sub トピックの作成

Cloud Scheduler がパブリッシュし、Cloud Run functions がサブスクライブする Pub/Sub トピックを作成します。

  1. GCP Console で、[Pub/Sub> トピック] に移動します。
  2. [トピックを作成] をクリックします。
  3. 次の構成の詳細を指定します。
    • トピック ID: 「code42-incydr-trigger」と入力します。
    • その他の設定はデフォルトのままにします。
  4. [作成] をクリックします。

ログを収集する Cloud Run 関数を作成する

Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Code42 Incydr API からログを取得して GCS に書き込みます。

  1. GCP Console で、[Cloud Run] に移動します。
  2. [サービスを作成] をクリックします。
  3. [関数] を選択します(インライン エディタを使用して関数を作成します)。
  4. [構成] セクションで、次の構成の詳細を指定します。

    設定
    サービス名 code42-incydr-collector
    リージョン GCS バケットと一致するリージョンを選択します(例: us-central1)。
    ランタイム [Python 3.12] 以降を選択します。
  5. [トリガー(省略可)] セクションで、次の操作を行います。

    1. [+ トリガーを追加] をクリックします。
    2. [Cloud Pub/Sub] を選択します。
    3. [Cloud Pub/Sub トピックを選択してください] で、トピック code42-incydr-trigger を選択します。
    4. [保存] をクリックします。
  6. [認証] セクションで、次の操作を行います。

    1. [認証が必要] を選択します。
    2. Identity and Access Management(IAM)を確認します。
  7. 下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。

  8. [セキュリティ] タブに移動します。

    • サービス アカウント: サービス アカウント code42-incydr-collector-sa を選択します。
  9. [コンテナ] タブに移動します。

    1. [変数とシークレット] をクリックします。
    2. 環境変数ごとに [+ 変数を追加] をクリックします。
    変数名 値の例 説明
    INCYDR_BASE_URL https://api.us.code42.com テナントの API ベース URL
    INCYDR_CLIENT_ID your-client-id API クライアント ID
    INCYDR_CLIENT_SECRET your-client-secret API クライアント シークレット
    GCS_BUCKET code42-incydr-logs GCS バケット名
    GCS_PREFIX code42/ ログファイルの接頭辞
    PAGE_SIZE 500 1 ページあたりのレコード数
    LOOKBACK_MINUTES 60 最初のルックバック期間
    STREAMS users,audit,cases カンマ区切りのデータ ストリーム
    FE_QUERY_JSON `` 省略可: ファイル イベントのクエリ JSON
    FE_PAGE_SIZE 1000 省略可: ファイル イベント ページのサイズ
  10. [変数とシークレット] タブで [リクエスト] まで下にスクロールします。

    • リクエストのタイムアウト: 600 秒(10 分)を入力します。
  11. [コンテナ] の [設定] タブに移動します。

    • [リソース] セクションで次の操作を行います。
      • メモリ: 1,024 MiB 以上を選択します。
      • CPU: [1] を選択します。
    • [完了] をクリックします。
  12. [実行環境] までスクロールします。

    • [デフォルト](推奨)を選択します。
  13. [リビジョン スケーリング] セクションで、次の操作を行います。

    • [インスタンスの最小数] に「0」と入力します。
    • インスタンスの最大数: 100 と入力します(または、予想される負荷に基づいて調整します)。
  14. [作成] をクリックします。

  15. サービスが作成されるまで待ちます(1 ~ 2 分)。

  16. サービスを作成すると、インライン コードエディタが自動的に開きます。

関数コードを追加する

  1. [関数のエントリ ポイント] に「main」と入力します。
  2. インライン コードエディタで、次の 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
    
  3. [デプロイ] をクリックして、関数を保存してデプロイします。

  4. デプロイが完了するまで待ちます(2 ~ 3 分)。

Cloud Scheduler ジョブの作成

Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run functions の関数をトリガーします。

  1. GCP Console で、[Cloud Scheduler] に移動します。
  2. [ジョブを作成] をクリックします。
  3. 次の構成情報を提供してください。

    設定
    名前 code42-incydr-hourly
    リージョン Cloud Run functions と同じリージョンを選択する
    周波数 0 * * * *(1 時間ごとに正時)
    タイムゾーン タイムゾーンを選択します(UTC を推奨)。
    ターゲット タイプ Pub/Sub
    トピック トピック code42-incydr-trigger を選択します。
    メッセージ本文 {}(空の JSON オブジェクト)
  4. [作成] をクリックします。

スケジュールの頻度のオプション

  • ログの量とレイテンシの要件に基づいて頻度を選択します。

    頻度 CRON 式 ユースケース
    5 分毎 */5 * * * * 大容量、低レイテンシ
    15 分ごと */15 * * * * 検索量が普通
    1 時間ごと 0 * * * * 標準(推奨)
    6 時間ごと 0 */6 * * * 少量、バッチ処理
    毎日 0 0 * * * 履歴データの収集

スケジューラ ジョブをテストする

  1. Cloud Scheduler コンソールで、ジョブ(code42-incydr-hourly)を見つけます。
  2. [強制実行] をクリックして手動でトリガーします。
  3. 数秒待ってから、[Cloud Run> サービス> code42-incydr-collector > ログ] に移動します。
  4. 関数が正常に実行されたことを確認します。求めていること:

    Fetching users...
    Fetching audit logs...
    Fetching cases...
    Successfully processed logs: {"users_pages": X, "audit_pages": Y, "cases_pages": Z}
    
  5. GCS バケット(code42-incydr-logs)をチェックして、ログが書き込まれたことを確認します。

ログにエラーが表示された場合:

  • HTTP 401: 環境変数で API 認証情報を確認する
  • HTTP 403: Code42 Incydr コンソールで API クライアントに必要な権限があることを確認する
  • HTTP 429: レート制限 - 関数はバックオフで自動的に再試行されます
  • アクセス トークンの取得に失敗しました: INCYDR_BASE_URLINCYDR_CLIENT_IDINCYDR_CLIENT_SECRET が正しいことを確認します

Google SecOps サービス アカウントを取得する

Google SecOps は、一意のサービス アカウントを使用して GCS バケットからデータを読み取ります。このサービス アカウントにバケットへのアクセス権を付与する必要があります。

サービス アカウントのメールアドレスを取得する

  1. [SIEM 設定] > [フィード] に移動します。
  2. [Add New Feed] をクリックします。
  3. [単一フィードを設定] をクリックします。
  4. [フィード名] フィールドに、フィードの名前を入力します(例: Code42 Incydr Datasets)。
  5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
  6. [ログタイプ] として [Code42 Incydr] を選択します。
  7. [サービス アカウントを取得する] をクリックします。一意のサービス アカウント メールアドレスが表示されます(例:)。

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. このメールアドレスをコピーして、次のステップで使用します。

Google SecOps サービス アカウントに IAM 権限を付与する

Google SecOps サービス アカウントには、GCS バケットに対する Storage オブジェクト閲覧者ロールが必要です。

  1. [Cloud Storage] > [バケット] に移動します。
  2. バケット名(code42-incydr-logs)をクリックします。
  3. [権限] タブに移動します。
  4. [アクセス権を付与] をクリックします。
  5. 次の構成の詳細を指定します。
    • プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
    • ロールを割り当てる: [ストレージ オブジェクト閲覧者] を選択します。
  6. [保存] をクリックします。

Code42 Incydr のログを取り込むように Google SecOps でフィードを構成する

  1. [SIEM 設定] > [フィード] に移動します。
  2. [Add New Feed] をクリックします。
  3. [単一フィードを設定] をクリックします。
  4. [フィード名] フィールドに、フィードの名前を入力します(例: Code42 Incydr Datasets)。
  5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
  6. [ログタイプ] として [Code42 Incydr] を選択します。
  7. [次へ] をクリックします。
  8. 次の入力パラメータの値を指定します。

    • ストレージ バケットの 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: このフィードのイベントに適用されるラベル。

  9. [次へ] をクリックします。

  10. [Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

ご不明な点がございましたら、コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。