Group-IB 脅威インテリジェンス のログを収集する

以下でサポートされています。

Google SecOps SIEM

このドキュメントでは、Google Cloud Storage を使用して Group-IB 脅威インテリジェンス ログを Google Security Operations に取り込む方法について説明します。

Group-IB Threat Intelligence & Attribution（TI&A）は、脅威アクター、セキュリティ侵害インジケーター（IOC）、マルウェア、コマンド＆コントロール（C2）インフラストラクチャ、漏洩した認証情報、フィッシング キャンペーン、脆弱性に関するリアルタイム データを提供するサイバー脅威インテリジェンス プラットフォームです。オープン Web、ディープ Web、ダーク Web のソースからインテリジェンスを集約し、セキュリティ チームが脅威を事前に検出して対応できるようにします。

始める前に

次の前提条件を満たしていることを確認してください。

• Google SecOps インスタンス
• API アクセスが有効になっている Group-IB TI&A アカウント
• Group-IB TI&A ポータルへのアクセス（tap.group-ib.com）
• 次の API が有効になっている Google Cloud プロジェクト。
  • Cloud Storage API
  • Cloud Functions API
  • Cloud Scheduler API
  • Cloud Build API

Group-IB API キーを生成する

1. https://tap.group-ib.com で Group-IB TI&A ポータルにログインします。
2. 右上の名前をクリックし、[プロフィール] を選択します。
3. [設定に移動] をクリックします。
4. [Security and Access] タブに移動します。
5. [Personal token] セクションで、[Generate new token] をクリックします。

6. API キーをコピーして安全に保存します。

   注: API キーは、HTTP 基本認証のメールアドレスとともにパスワードとして使用されます。安全に保管してください。鍵を紛失した場合は、新しい鍵を生成する必要があります。

重要: API リクエストがブロックされないように、Cloud Functions（または NAT ゲートウェイ）の IP アドレスを許可リストに登録するには、Group-IB サポートにお問い合わせください。

Google Cloud Storage バケットを作成する

1. Google Cloud コンソールに移動します。
2. プロジェクトを選択するか、新しいプロジェクトを作成します。
3. ナビゲーション メニューで、[Cloud Storage > バケット] に移動します。
4. [バケットを作成] をクリックします。

5. 次の構成情報を提供してください。

   設定	値
   バケットに名前を付ける	グローバルに一意の名前（groupib-ti-logs など）を入力します。
   ロケーション タイプ	ニーズに応じて選択します（リージョン、デュアルリージョン、マルチリージョン）。
   ロケーション	場所を選択します（例: us-central1）。
   ストレージ クラス	Standard（アクセス頻度の高いログにおすすめ）
   アクセス制御	均一（推奨）

6. [作成] をクリックします。

Group-IB データを取得する Cloud Functions をデプロイする

Group-IB TI&A API から脅威インテリジェンス データを取得し、Google SecOps が取り込むための NDJSON ファイルとして GCS バケットに書き込む Cloud Functions の関数を作成します。

Cloud Functions の関数を作成する

1. Google Cloud コンソールに移動します。
2. Cloud Functions に移動します。
3. [関数を作成] をクリックします。

4. 次の構成情報を提供してください。

   設定	値
   環境	第 2 世代
   関数名	groupib-to-gcs
   リージョン	GCS バケットに最も近いリージョンを選択します。
   トリガーのタイプ	HTTPS
   認証	認証が必要です
   割り当てられたメモリ	512 MB（大規模なコレクションを取得する場合は増やす）
   タイムアウト	540 秒

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

6. [ランタイム] を Python 3.11（以降）に設定します。

7. [エントリ ポイント] を main に設定します。

8. main.py の内容を次のコードに置き換えます。

   import json
   import os
   import requests
   import functions_framework
   from datetime import datetime, timedelta, timezone
   from urllib.parse import urljoin
   from requests.auth import HTTPBasicAuth
   from google.cloud import storage
   
   GIB_API_URL = os.environ.get('GIB_API_URL', 'https://tap.group-ib.com/api/v2/')
   GIB_USERNAME = os.environ.get('GIB_USERNAME')
   GIB_API_KEY = os.environ.get('GIB_API_KEY')
   GCS_BUCKET = os.environ.get('GCS_BUCKET')
   GCS_PREFIX = os.environ.get('GCS_PREFIX', 'groupib-ti')
   COLLECTIONS = os.environ.get('GIB_COLLECTIONS', 'compromised/account,malware/cnc,apt/threat,hi/threat').split(',')
   DEFAULT_DAYS_BACK = int(os.environ.get('DEFAULT_DAYS_BACK', '3'))
   
   # Max items per request: 100 for most collections, 20 for apt/threat and hi/threat
   BIG_DATA_COLLECTIONS = ['apt/threat', 'hi/threat']
   
   # File to persist seqUpdate values between runs (use GCS for durability)
   STATE_BLOB = '_state/seq_updates.json'
   
   def load_state():
       """Load seqUpdate state from GCS."""
       client = storage.Client()
       bucket = client.bucket(GCS_BUCKET)
       blob = bucket.blob(f"{GCS_PREFIX}/{STATE_BLOB}")
       if blob.exists():
           return json.loads(blob.download_as_text())
       return {}
   
   def save_state(state):
       """Save seqUpdate state to GCS."""
       client = storage.Client()
       bucket = client.bucket(GCS_BUCKET)
       blob = bucket.blob(f"{GCS_PREFIX}/{STATE_BLOB}")
       blob.upload_from_string(json.dumps(state), content_type='application/json')
   
   def gib_request(session, url, params=None):
       """Send authenticated GET request to Group-IB API."""
       resp = session.get(url, params=params)
       if resp.status_code == 301:
           raise Exception('IP not whitelisted by Group-IB. Contact Group-IB support.')
       resp.raise_for_status()
       return resp.json()
   
   def get_seq_update_by_date(session, collection, date_str):
       """Get seqUpdate value for a collection starting from a given date."""
       url = urljoin(GIB_API_URL, 'sequence_list')
       data = gib_request(session, url, {'date': date_str, 'collection': collection})
       return data.get('list', {}).get(collection)
   
   def fetch_collection(session, collection, seq_update):
       """Fetch all new items from a collection starting after the given seqUpdate."""
       limit = 20 if collection in BIG_DATA_COLLECTIONS else 100
       url = urljoin(GIB_API_URL, f"{collection}/updated")
       all_items = []
       last_seq = seq_update
   
       while True:
           data = gib_request(session, url, {'seqUpdate': str(last_seq), 'limit': limit})
           items = data.get('items', [])
           if not items:
               break
           all_items.extend(items)
           last_seq = items[-1].get('seqUpdate')
   
       return all_items, last_seq
   
   def write_to_gcs(items, collection_name):
       """Write items to GCS as NDJSON."""
       if not items:
           return 0
       client = storage.Client()
       bucket = client.bucket(GCS_BUCKET)
       timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S')
       safe_name = collection_name.replace('/', '_')
       blob_path = f"{GCS_PREFIX}/{safe_name}_{timestamp}.ndjson"
       blob = bucket.blob(blob_path)
       ndjson = '\n'.join(json.dumps(item, ensure_ascii=False) for item in items) + '\n'
       blob.upload_from_string(ndjson, content_type='application/x-ndjson')
       return len(items)
   
   @functions_framework.http
   def main(request):
       """Cloud Function entry point."""
       session = requests.Session()
       session.auth = HTTPBasicAuth(GIB_USERNAME, GIB_API_KEY)
       session.headers.update({'Accept': '*/*'})
   
       state = load_state()
       total = 0
   
       for collection in COLLECTIONS:
           collection = collection.strip()
           seq_update = state.get(collection)
   
           if seq_update is None:
               default_date = (datetime.now(timezone.utc) - timedelta(days=DEFAULT_DAYS_BACK)).strftime('%Y-%m-%d')
               seq_update = get_seq_update_by_date(session, collection, default_date)
               if seq_update is None:
                   continue
   
           items, last_seq = fetch_collection(session, collection, seq_update)
           if items:
               write_to_gcs(items, collection)
               total += len(items)
           state[collection] = last_seq
   
       save_state(state)
       return json.dumps({'status': 'success', 'total_items': total}), 200
   

9. requirements.txt の内容を次の依存関係に置き換えます。

   functions-framework==3.*
   requests>=2.28.0
   google-cloud-storage>=2.0.0
   

10. [デプロイ] をクリックします。

環境変数を構成する

1. デプロイ後、関数の詳細ページに移動します。
2. [編集] をクリックします。
3. [ランタイム、ビルド、接続、セキュリティの設定] セクションを開きます。

4. [ランタイム環境変数] で、次の変数を追加します。

   変数	値
   GIB_API_URL	https://tap.group-ib.com/api/v2/（または地域に応じて https://bt.group-ib.com/api/v2/）
   GIB_USERNAME	Group-IB アカウントのメールアドレス
   GIB_API_KEY	Group-IB API キー（個人用トークン）
   GCS_BUCKET	GCS バケット名（例: groupib-ti-logs）
   GCS_PREFIX	ログファイルの接頭辞（例: groupib-ti）
   GIB_COLLECTIONS	取得するコレクションのカンマ区切りのリスト（下記参照）
   DEFAULT_DAYS_BACK	最初の実行で遡る日数（デフォルト: 3）

5. [デプロイ] をクリックします。

   注: 本番環境では、機密性の高い値（GIB_API_KEY）を平文の環境変数ではなく、Secret Manager に保存します。これらを参照するには、関数構成の [シークレットの参照] オプションを使用します。

利用可能な Group-IB コレクション

ユースケースに関連するコレクションを使用して GIB_COLLECTIONS 変数を構成します。

コレクション	説明	最大値
compromised/account	不正使用されたアカウント認証情報（ログイン、パスワード、ドメイン）	100
compromised/card	不正使用された銀行カード	100
compromised/mule	マネー ミュールのアカウント	100
compromised/imei	不正使用されたモバイル デバイスの IMEI	100
compromised/file	マルウェアの属性を持つ侵害されたファイル	100
attacks/ddos	DDoS 攻撃データ（ターゲット IP、ドメイン）	100
attacks/deface	ウェブサイトの改ざんインシデント	100
attacks/phishing	フィッシング URL とドメイン	100
attacks/phishing_kit	フィッシング キットのハッシュとターゲット ブランド	100
bp/phishing	ブランド保護 - フィッシング インシデント	100
bp/phishing_kit	ブランド保護 - フィッシング キット	100
hi/threat	IOC と MITRE ATT&CK マッピングを含むサイバー犯罪者（HI）の脅威レポート	20
hi/threat_actor	サイバー犯罪者の脅威アクター プロファイル	100
apt/threat	IOC と MITRE ATT&CK マッピングを含む APT（国家）脅威レポート	20
apt/threat_actor	APT 脅威アクターのプロファイル	100
malware/cnc	コマンド＆コントロール（C＆C）サーバーのインジケーター（IP、ドメイン）	100
malware/malware	マルウェアの説明と脅威レベル	100
malware/targeted_malware	標的型マルウェアのサンプル（ハッシュ、ファイル名）	100
osi/git_leak	Git リポジトリのデータ漏洩	100
osi/public_leak	一般公開データの漏洩（貼り付け、ダンプ）	100
osi/vulnerability	CVSS スコアを含む脆弱性データ	100
suspicious_ip/tor_node	Tor 出口ノードの IP アドレス	100
suspicious_ip/open_proxy	オープン プロキシの IP アドレス	100
suspicious_ip/socks_proxy	SOCKS プロキシの IP アドレス	100

• 例: 侵害された認証情報、C2 インフラストラクチャ、APT データを収集するには:

  compromised/account,malware/cnc,apt/threat,hi/threat
  

  注: apt/threat コレクションと hi/threat コレクションは、レスポンス サイズが大きいため、API リクエストごとに最大 20 個のアイテムに制限されています。Cloud Functions の関数はこれを自動的に処理します。POC ライセンスまたはパートナー ライセンスを使用している場合、データへのアクセスは 30 日間に制限されることがあります。attacks/phishing の IP アドレスを検出に使用しないでください。誤検出が多数発生する可能性があります。URL のみに焦点を当てます。

Cloud Functions のスケジュールを設定する

Cloud Scheduler を使用して、関数を定期的にトリガーする。

1. Google Cloud コンソールで [Cloud Scheduler] に移動します。
2. [ジョブを作成] をクリックします。

3. 次の構成情報を提供してください。

   設定	値
   名前	groupib-to-gcs-schedule
   リージョン	Cloud Functions の関数と同じリージョン
   周波数	0 */1 * * *（1 時間ごと）または 0 0 * * *（毎日）
   タイムゾーン	タイムゾーンを選択する

4. [実行内容を構成する] で、次の操作を行います。

   • ターゲット タイプ: [HTTP] を選択します。
   • URL: Cloud Functions のトリガー URL を入力します。
   • HTTP メソッド: [POST] を選択します。
   • Auth ヘッダー: [OIDC トークンを追加] を選択します。
   • サービス アカウント: roles/cloudfunctions.invoker 権限を持つサービス アカウントを選択します。

5. [作成] をクリックします。

   注: Cloud Functions の関数は、GCS に保存されている seqUpdate 値を使用して位置を追跡します。初回実行時には、DEFAULT_DAYS_BACK 日（デフォルト: 3）遡って検索します。後続の実行は、スケジューラの頻度に関係なく、前の実行が終了したところから再開されます。

データ エクスポートをテストする

1. Cloud Scheduler コンソールで、ジョブの横にある [強制実行] をクリックして、関数を手動でトリガーします。
2. Cloud Logging で Cloud Functions のログを確認し、Group-IB からデータが取得されたことを確認します。
3. Google Cloud コンソールで、[Cloud Storage] > [バケット] に移動します。
4. バケット名（groupib-ti-logs など）をクリックします。
5. 接頭辞フォルダ（groupib-ti/ など）に移動します。

6. 新しい .ndjson ファイルがバケットに表示されていることを確認します。

   注: 各ログファイルには、{collection}_{YYYYMMDD_HHMMSS}.ndjson 形式のタイムスタンプ（ioc_common_20260318_120000.ndjson など）で名前が付けられます。

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

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

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

1. [SIEM 設定] > [フィード] に移動します。
2. [Add New Feed] をクリックします。
3. [単一フィードを設定] をクリックします。
4. [フィード名] フィールドに、フィードの名前を入力します（例: Group-IB Threat Intelligence）。
5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
6. [ログタイプ] として [Group-IB 脅威インテリジェンス] を選択します。

7. [サービス アカウントを取得する] をクリックします。一意のサービス アカウントのメールアドレスが表示されます。例:

   [e-mail]
   

8. このメールアドレスをコピーして、次のステップで使用します。

   注: 各 Google SecOps インスタンスには一意のサービス アカウントがあります。他のドキュメントや例のサービス アカウントは使用しないでください。

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

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

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

6. [保存] をクリックします。

   注: 削除オプションの [転送されたファイルを削除] または [転送されたファイルと空のディレクトリを削除] を使用する場合は、Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者ロールを付与します。

Group-IB 脅威インテリジェンスのログを取り込むように Google SecOps でフィードを構成する

1. [SIEM 設定] > [フィード] に移動します。
2. [Add New Feed] をクリックします。
3. [単一フィードを設定] をクリックします。
4. [フィード名] フィールドに、フィードの名前を入力します（例: Group-IB Threat Intelligence）。
5. [ソースタイプ] として [Google Cloud Storage V2] を選択します。
6. [ログタイプ] として [Group-IB 脅威インテリジェンス] を選択します。
7. [次へ] をクリックします。

8. 次の入力パラメータの値を指定します。

   • ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。

     gs://groupib-ti-logs/groupib-ti/
     

     次のように置き換えます。

     • groupib-ti-logs: GCS バケット名。
     • groupib-ti: ログが保存される接頭辞/フォルダパス。
     注: URI の末尾には常にスラッシュ（/）を含めてください。

   • Source deletion option: 必要に応じて削除オプションを選択します。

     • なし: 転送後にファイルを削除しません（テストに推奨）。
     • 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
     • 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
     注: 削除オプションを選択する場合、サービス アカウントには、ストレージ オブジェクト閲覧者ではなく、ストレージ オブジェクト管理者のロールが必要です。必要に応じて IAM 権限を更新します。

   • ファイルの最長存続期間: 過去数日間に変更されたファイルを含めます。デフォルトは 180 日です。

   • Asset namespace: アセットの名前空間。

   • Ingestion labels: このフィードのイベントに適用されるラベル。

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

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

さらにサポートが必要な場合 コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。