Team Cymru Scout Threat Intelligence のログを収集する

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

Google SecOps SIEM

このドキュメントでは、Google Cloud Storage を使用して Team Cymru Scout Threat Intelligence データを Google Security Operations に取り込む方法について説明します。Team Cymru Scout は、アカウントの使用状況指標、クエリの上限、基盤クエリの統計情報などの脅威インテリジェンスデータを提供し、組織がセキュリティ体制と脅威インテリジェンスの消費量をモニタリングできるようにします。

始める前に

次の前提条件を満たしていることを確認します。

Google SecOps インスタンス
Cloud Storage API が有効になっている GCP プロジェクト
GCS バケットを作成および管理する権限
GCS バケットの IAM ポリシーを管理する権限
Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
Team Cymru Scout テナントへの特権アクセス

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

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

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

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

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

Team Cymru Scout API 認証情報を収集する

Team Cymru Scout Platform にログインします。
[API キー] ページに移動します。
[作成] ボタンをクリックします。
必要に応じて、鍵の説明を入力します。
[キーを作成] ボタンをクリックして、API キーを生成します。
次の詳細をコピーして安全な場所に保存します。
- SCOUT_API_TOKEN: API アクセストークン
- SCOUT_BASE_URL: Scout API のベース URL（通常は https://scout.cymru.com）
注: API トークンは、Cloud Run 関数が Authorization: Token <api_token> ヘッダー形式を使用して Team Cymru Scout API に対する認証を行うために使用されます。Team Cymru Scout は、トークン認証の代替としてユーザー名/パスワードによる基本認証もサポートしていますが、この統合ではトークン認証を使用します。

テスト API アクセス

統合に進む前に、認証情報をテストします。

# Replace with your actual credentials
SCOUT_API_TOKEN="your-api-token"
SCOUT_BASE_URL="https://scout.cymru.com"

# Test API access to usage endpoint
curl -v --request GET \
  --url "${SCOUT_BASE_URL}/api/scout/usage" \
  --header "Authorization: Token ${SCOUT_API_TOKEN}"

Cloud Run functions のサービスアカウントを作成する

Cloud Run 関数には、GCS バケットに書き込む権限を持つサービスアカウントが必要です。

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

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

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

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

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

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

[Cloud Storage] > [バケット] に移動します。
バケット名をクリックします。
[権限] タブに移動します。
[アクセス権を付与] をクリックします。
次の構成の詳細を指定します。
- プリンシパルを追加: サービスアカウントのメールアドレス（例: team-cymru-scout-ti-sa@PROJECT_ID.iam.gserviceaccount.com）を入力します。
- ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
[保存] をクリックします。

Pub/Sub トピックの作成

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

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

脅威インテリジェンスデータを収集する Cloud Run functions の関数を作成する

Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、Team Cymru Scout API から脅威インテリジェンスデータを取得して GCS に書き込みます。

GCP Console で、[Cloud Run] に移動します。
[サービスを作成] をクリックします。
[関数] を選択します（インラインエディタを使用して関数を作成します）。

[構成] セクションで、次の構成の詳細を指定します。

設定	値
サービス名	`team-cymru-scout-ti-collector`
リージョン	GCS バケットと一致するリージョンを選択します（例: `us-central1`）。
ランタイム	[Python 3.12] 以降を選択します。

[トリガー（省略可）] セクションで、次の操作を行います。
1. [+ トリガーを追加] をクリックします。
2. [Cloud Pub/Sub] を選択します。
3. [Cloud Pub/Sub トピックを選択してください] で、トピック team-cymru-scout-ti-trigger を選択します。
4. [保存] をクリックします。
[認証] セクションで、次の操作を行います。
1. [認証が必要] を選択します。
2. Identity and Access Management（IAM）を確認します。
注: Pub/Sub は、関数を呼び出すときに認証を自動的に処理します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービスアカウント: サービスアカウント team-cymru-scout-ti-sa を選択します。

[コンテナ] タブに移動します。

[変数とシークレット] をクリックします。
環境変数ごとに [+ 変数を追加] をクリックします。

変数名	値の例
`GCS_BUCKET`	`team-cymru-scout-ti`
`GCS_PREFIX`	`team-cymru/scout-ti/`
`STATE_KEY`	`team-cymru/scout-ti/state.json`
`SCOUT_BASE_URL`	`https://scout.cymru.com`
`SCOUT_API_TOKEN`	`your-scout-api-token`
`COLLECTION_INTERVAL_HOURS`	`1`
`HTTP_TIMEOUT`	`60`
`HTTP_RETRIES`	`3`

[変数とシークレット] タブで [リクエスト] まで下にスクロールします。
- リクエストのタイムアウト: 600 秒（10 分）を入力します。
[コンテナ] の [設定] タブに移動します。
- [リソース] セクションで次の操作を行います。
  - メモリ: 512 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, timezone
import time

# Initialize HTTP client
http = urllib3.PoolManager()

# Initialize Storage client
storage_client = storage.Client()

@functions_framework.cloud_event
def main(cloud_event):
    """
    Cloud Run function triggered by Pub/Sub to fetch usage data from Team Cymru Scout API and write to GCS.

    Args:
        cloud_event: CloudEvent object containing Pub/Sub message
    """

    # Get environment variables
    bucket_name = os.environ.get('GCS_BUCKET')
    prefix = os.environ.get('GCS_PREFIX', 'team-cymru/scout-ti/')
    state_key = os.environ.get('STATE_KEY', 'team-cymru/scout-ti/state.json')
    collection_interval_hours = int(os.environ.get('COLLECTION_INTERVAL_HOURS', '1'))
    http_timeout = int(os.environ.get('HTTP_TIMEOUT', '60'))
    http_retries = int(os.environ.get('HTTP_RETRIES', '3'))

    # Team Cymru Scout API credentials
    scout_base_url = os.environ.get('SCOUT_BASE_URL', 'https://scout.cymru.com')
    scout_api_token = os.environ.get('SCOUT_API_TOKEN')

    if not all([bucket_name, scout_api_token]):
        print('Error: Missing required environment variables')
        return

    try:
        # Get GCS bucket
        bucket = storage_client.bucket(bucket_name)

        # Load state (last collection timestamp)
        state = load_state(bucket, state_key)
        now = time.time()
        last_collection = state.get('last_collection_ts', now - (collection_interval_hours * 3600))

        print(f'Collecting usage data at {iso_format(now)} (last collection: {iso_format(last_collection)})')

        # Fetch usage data from Team Cymru Scout API
        usage_data = fetch_usage_data(
            scout_base_url, 
            scout_api_token, 
            http_timeout,
            http_retries
        )

        if usage_data:
            # Add timestamp and event type
            usage_data['event_type'] = 'account_usage'
            usage_data['collection_timestamp'] = iso_format(now)

            # Write to GCS
            write_to_gcs(bucket, prefix, usage_data, now)

            # Update state
            save_state(bucket, state_key, {'last_collection_ts': now})

            print(f'Successfully collected and stored usage data')
        else:
            print('No usage data retrieved')

    except Exception as e:
        print(f'Error processing usage data: {str(e)}')
        raise

def iso_format(ts):
    """Convert Unix timestamp to ISO 8601 format."""
    return time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(ts))

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: {str(e)}')
    return {}

def save_state(bucket, key, state):
    """Save state to GCS."""
    try:
        blob = bucket.blob(key)
        blob.upload_from_string(
            json.dumps(state, separators=(',', ':')),
            content_type='application/json'
        )
    except Exception as e:
        print(f'Warning: Could not save state: {str(e)}')

def http_request(url, method='GET', body=None, headers=None, timeout=60, retries=3):
    """Make HTTP request with retry logic."""
    attempt = 0
    while True:
        try:
            req_headers = headers or {}
            if body is not None:
                req_headers['Content-Type'] = 'application/json'
                body_bytes = body.encode('utf-8') if isinstance(body, str) else body
            else:
                body_bytes = None

            response = http.request(
                method,
                url,
                body=body_bytes,
                headers=req_headers,
                timeout=timeout
            )

            if response.status == 200:
                return response.data, response.headers.get('Content-Type', 'application/json')
            elif response.status in (429, 500, 502, 503, 504) and attempt < retries:
                delay = 1 + attempt
                retry_after = response.headers.get('Retry-After')
                if retry_after:
                    try:
                        delay = int(retry_after)
                    except:
                        pass
                time.sleep(max(1, delay))
                attempt += 1
                continue
            else:
                raise Exception(f'HTTP {response.status}: {response.data.decode("utf-8")}')
        except urllib3.exceptions.HTTPError as e:
            if attempt < retries:
                time.sleep(1 + attempt)
                attempt += 1
                continue
            raise

def fetch_usage_data(base_url, api_token, timeout, retries):
    """
    Fetch usage data from Team Cymru Scout API.

    Implementation mirrors the official Scout API example:
    curl --request GET --url 'https://scout.cymru.com/api/scout/usage' --header 'Authorization: Token valid_api_token'
    """
    # Use the documented /api/scout/usage endpoint
    url = f'{base_url}/api/scout/usage'

    # Use Token authentication as documented
    headers = {
        'Authorization': f'Token {api_token}',
        'Accept': 'application/json'
    }

    print(f'Fetching usage data from {url}')

    try:
        # Fetch data
        blob_data, content_type = http_request(url, method='GET', headers=headers, timeout=timeout, retries=retries)

        # Parse response
        usage_data = json.loads(blob_data.decode('utf-8'))

        print(f'Retrieved usage data: used_queries={usage_data.get("used_queries")}, query_limit={usage_data.get("query_limit")}')

        return usage_data

    except Exception as e:
        print(f'Error fetching usage data: {e}')
        return None

def write_to_gcs(bucket, prefix, data, timestamp):
    """Write data to GCS."""
    # Create date-based path
    date_path = time.strftime('%Y/%m/%d', time.gmtime(timestamp))
    key = f'{prefix}{date_path}/usage_{int(timestamp)}.json'

    # Write as JSON
    blob = bucket.blob(key)
    blob.upload_from_string(
        json.dumps(data, separators=(',', ':')),
        content_type='application/json'
    )
    print(f'Wrote data to gs://{bucket.name}/{key}')

2 つ目のファイル: requirements.txt:

functions-framework==3.*
google-cloud-storage==2.*
urllib3>=2.0.0

[デプロイ] をクリックして、関数を保存してデプロイします。
デプロイが完了するまで待ちます（2 ～ 3 分）。

注: Pub/Sub トリガー構成では、必要なサブスクリプションと権限が自動的に作成されます。この実装では、公式の Scout API ドキュメントで指定されているように、ドキュメント化された Team Cymru Scout API エンドポイント /api/scout/usage と Authorization: Token <api_token> 認証を使用します。

Cloud Scheduler ジョブの作成

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

GCP Console で、[Cloud Scheduler] に移動します。
[ジョブを作成] をクリックします。

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

設定	値
名前	`team-cymru-scout-ti-hourly`
リージョン	Cloud Run functions と同じリージョンを選択する
周波数	`0 * * * *`（1 時間ごとに正時）
タイムゾーン	タイムゾーンを選択します（UTC を推奨）。
ターゲットタイプ	Pub/Sub
トピック	トピック `team-cymru-scout-ti-trigger` を選択します。
メッセージ本文	`{}`（空の JSON オブジェクト）

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

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

データ量とレイテンシ要件に基づいて頻度を選択します。

頻度	CRON 式	ユースケース
5 分毎	`/5 * * *`	高頻度モニタリング
15 分ごと	`/15 * * *`	中頻度
1 時間ごと	`0 * * * *`	標準（推奨）
6 時間ごと	`0 /6 * *`	低頻度
毎日	`0 0 * * *`	1 日の使用状況のトラッキング

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

Cloud Scheduler コンソールで、ジョブを見つけます。
[強制実行] をクリックして手動でトリガーします。
数秒待ってから、[Cloud Run> サービス> team-cymru-scout-ti-collector > ログ] に移動します。
関数が正常に実行されたことを確認します。
GCS バケットをチェックして、使用状況データが書き込まれたことを確認します。

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

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

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

[SIEM 設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Team Cymru Scout Threat Intelligence）。
[ソースタイプ] として [Google Cloud Storage V2] を選択します。
[Log type] として [Team Cymru Scout Threat Intelligence] を選択します。
[サービスアカウントを取得する] をクリックします。一意のサービスアカウントメールアドレスが表示されます（例:）。
```
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
```
このメールアドレスをコピーして、次のステップで使用します。

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

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

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

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

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

Team Cymru Scout Threat Intelligence のデータを取り込むように Google SecOps でフィードを構成する

[SIEM 設定] > [フィード] に移動します。
[Add New Feed] をクリックします。
[単一フィードを設定] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: Team Cymru Scout Threat Intelligence）。
[ソースタイプ] として [Google Cloud Storage V2] を選択します。
[Log type] として [Team Cymru Scout Threat Intelligence] を選択します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- ストレージバケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
```
gs://team-cymru-scout-ti/team-cymru/scout-ti/
```
  - 次のように置き換えます。
    - team-cymru-scout-ti: GCS バケット名。
    - team-cymru/scout-ti/: ログが保存される接頭辞/フォルダパス。
  注: URI の末尾には常にスラッシュ（/）を付けます。
- Source deletion option: 必要に応じて削除オプションを選択します。
  - なし: 転送後にファイルを削除しません（テストにおすすめ）。
  - 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
  - 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
    
    注: 削除オプションを選択する場合は、サービスアカウントに Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者のロールが必要です。必要に応じて IAM 権限を更新します。
- ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
- アセットの名前空間: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

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