MuleSoft Anypoint プラットフォームのログを収集する

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

Google SecOps SIEM

注: この機能には、Google Security Operations サービス固有の規約の pre-GA サービス規約が適用されます。pre-GA の機能についてはサポートが制限されます。また、pre-GA の機能の変更は pre-GA の他のバージョンと互換性がない可能性があります。 詳細については、Google SecOps 技術サポート サービス ガイドラインと Google SecOps サービス固有の規約をご覧ください。

注: MuleSoft は事前構築済みのパーサーのないログタイプであるため、このガイドではログの取り込みのみを扱います。MuleSoft ログを UDM に変換するようにカスタム パーサーを構成する必要があります。

このドキュメントでは、Google Cloud Storage を使用して MuleSoft Anypoint プラットフォームのログから監査証跡イベントを Google Security Operations に取り込む方法について説明します。

始める前に

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

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

MuleSoft 組織 ID を取得する

1. Anypoint Platform にログインします。
2. [アクセス管理> 組織] に移動します。
3. [ビジネス グループ] テーブルで、組織の名前をクリックします。
4. 組織 ID（0a12b3c4-d5e6-789f-1021-1a2b34cd5e6f など）をコピーします。

または、MuleSoft Business Groups に移動して、URL から ID をコピーします。

MuleSoft 接続アプリを作成する

1. Anypoint Platform にログインします。
2. [Access Management] > [Connected Apps] > [Create App] に移動します。
3. 次の構成の詳細を指定します。
   • アプリ名: 一意の名前を入力します（例: Google SecOps export）。
   • [アプリが独自の権限で動作する（クライアント認証情報）] を選択します。
4. [スコープを追加] > [監査ログビューア] > [次へ] をクリックします。
5. ログが必要なビジネス グループをすべて選択します。
6. [次へ> スコープを追加] をクリックします。

7. [保存] をクリックし、クライアント ID とクライアント シークレットをコピーします。

   注: 組織の権限によっては、追加のスコープが必要になる場合があります。権限エラーが発生した場合は、接続アプリに必要なすべてのビジネス グループへのアクセス権が付与されていることを確認します。

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

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

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

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

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

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

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

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

1. GCP Console で、[IAM と管理>サービス アカウント] に移動します。
2. [サービス アカウントを作成] をクリックします。
3. 次の構成の詳細を指定します。
   • サービス アカウント名: 「mulesoft-logs-collector-sa」と入力します。
   • サービス アカウントの説明: 「Service account for Cloud Run function to collect MuleSoft Anypoint 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. バケット名をクリックします。
3. [権限] タブに移動します。
4. [アクセス権を付与] をクリックします。
5. 次の構成の詳細を指定します。
   • プリンシパルを追加: サービス アカウントのメールアドレス（例: mulesoft-logs-collector-sa@PROJECT_ID.iam.gserviceaccount.com）を入力します。
   • ロールを割り当てる: [Storage オブジェクト管理者] を選択します。
6. [保存] をクリックします。

Pub/Sub トピックの作成

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

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

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

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

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

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

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

5. [トリガー（省略可）] セクションで、次の操作を行います。

   1. [+ トリガーを追加] をクリックします。
   2. [Cloud Pub/Sub] を選択します。
   3. [Cloud Pub/Sub トピックを選択してください] で、トピック mulesoft-audit-trigger を選択します。
   4. [保存] をクリックします。

6. [認証] セクションで、次の操作を行います。

   1. [認証が必要] を選択します。
   2. Identity and Access Management（IAM）を確認します。
   注: Pub/Sub は、関数を呼び出すときに認証を自動的に処理します。

7. 下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。

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

   • サービス アカウント: サービス アカウント mulesoft-logs-collector-sa を選択します。

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

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

   変数名	値の例
   MULE_ORG_ID	your_org_id
   CLIENT_ID	your_client_id
   CLIENT_SECRET	your_client_secret
   GCS_BUCKET	mulesoft-audit-logs

10. [変数とシークレット] タブで [リクエスト] まで下にスクロールします。

    • リクエストのタイムアウト: 600 秒（10 分）を入力します。

11. [コンテナ] の [設定] タブに移動します。

    • [リソース] セクションで次の操作を行います。
      • メモリ: 512 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 uuid
   import time
   
   # 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()
   
   # MuleSoft API endpoints
   TOKEN_URL = "https://anypoint.mulesoft.com/accounts/api/v2/oauth2/token"
   
   @functions_framework.cloud_event
   def main(cloud_event):
       """
       Cloud Run function triggered by Pub/Sub to fetch MuleSoft audit logs and write to GCS.
   
       Args:
           cloud_event: CloudEvent object containing Pub/Sub message
       """
   
       # Get environment variables
       org_id = os.environ.get('MULE_ORG_ID')
       client_id = os.environ.get('CLIENT_ID')
       client_secret = os.environ.get('CLIENT_SECRET')
       bucket_name = os.environ.get('GCS_BUCKET')
   
       if not all([org_id, client_id, client_secret, bucket_name]):
           print('Error: Missing required environment variables')
           return
   
       query_url = f"https://anypoint.mulesoft.com/audit/v2/organizations/{org_id}/query"
   
       try:
           # Get GCS bucket
           bucket = storage_client.bucket(bucket_name)
   
           # Get OAuth token
           token = get_token(client_id, client_secret)
   
           # Calculate time range (last 24 hours)
           now = datetime.now(timezone.utc).replace(microsecond=0)
           start = now - timedelta(days=1)
   
           print(f'Fetching audit logs from {start.isoformat()} to {now.isoformat()}')
   
           # Fetch audit logs
           events = list(fetch_audit(query_url, token, start, now))
   
           # Upload to GCS
           if events:
               upload_to_gcs(bucket, events, start)
               print(f'Uploaded {len(events)} events')
           else:
               print('No events in the last 24 hours')
   
       except Exception as e:
           print(f'Error processing logs: {str(e)}')
           raise
   
   def get_token(client_id, client_secret):
       """Get OAuth 2.0 access token from MuleSoft."""
       data = {
           'grant_type': 'client_credentials',
           'client_id': client_id,
           'client_secret': client_secret
       }
   
       encoded_data = urllib3.request.urlencode(data).encode('utf-8')
   
       backoff = 1.0
       max_retries = 3
   
       for attempt in range(max_retries):
           try:
               response = http.request(
                   'POST',
                   TOKEN_URL,
                   body=encoded_data,
                   headers={'Content-Type': 'application/x-www-form-urlencoded'}
               )
   
               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 Exception(f'Failed to get token: {response.status} - {response.data.decode()}')
   
               token_data = json.loads(response.data.decode('utf-8'))
               return token_data['access_token']
   
           except Exception as e:
               if attempt == max_retries - 1:
                   raise
               print(f'Token request failed (attempt {attempt + 1}/{max_retries}): {e}')
               time.sleep(backoff)
               backoff = min(backoff * 2, 30.0)
   
       raise Exception('Failed to get token after maximum retries')
   
   def fetch_audit(query_url, token, start, end):
       """Fetch audit logs from MuleSoft API with pagination."""
       headers = {
           'Authorization': f'Bearer {token}',
           'Content-Type': 'application/json'
       }
   
       body = {
           'startDate': f"{start.isoformat(timespec='milliseconds')}Z",
           'endDate': f"{end.isoformat(timespec='milliseconds')}Z",
           'limit': 200,
           'offset': 0,
           'ascending': False
       }
   
       backoff = 1.0
   
       while True:
           try:
               response = http.request(
                   'POST',
                   query_url,
                   body=json.dumps(body).encode('utf-8'),
                   headers=headers
               )
   
               # Handle rate limiting with exponential backoff
               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
   
               backoff = 1.0
   
               if response.status != 200:
                   print(f'HTTP Error: {response.status}')
                   response_text = response.data.decode('utf-8')
                   print(f'Response body: {response_text}')
                   break
   
               data = json.loads(response.data.decode('utf-8'))
   
               if not data.get('data'):
                   break
   
               yield from data['data']
               body['offset'] += body['limit']
   
           except Exception as e:
               print(f'Error fetching audit logs: {e}')
               break
   
   def upload_to_gcs(bucket, events, timestamp):
       """Upload events to GCS as compressed JSON."""
       import gzip
       import io
   
       # Create blob name with timestamp and UUID
       blob_name = f"{timestamp.strftime('%Y/%m/%d')}/mulesoft-audit-{uuid.uuid4()}.json.gz"
   
       # Compress events
       buf = io.BytesIO()
       with gzip.GzipFile(fileobj=buf, mode='w') as gz:
           for event in events:
               gz.write((json.dumps(event) + '\n').encode('utf-8'))
   
       buf.seek(0)
   
       # Upload to GCS
       blob = bucket.blob(blob_name)
       blob.upload_from_file(buf, content_type='application/gzip')
   
       print(f'Uploaded to gs://{bucket.name}/{blob_name}')
   

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

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

3. [デプロイ] をクリックして、関数を保存してデプロイします。

4. デプロイが完了するまで待ちます（2 ～ 3 分）。

重要な考慮事項

レート制限: 監査ログクエリ エンドポイントは、3 つのコントロール プレーンの IP ごとにレート上限を適用します。米国のコントロール プレーンでは IP ごとに 1 分あたり 700 件のリクエストが許可されますが、EU と Gov のコントロール プレーンでは IP ごとに 1 分あたり 40 件のリクエストが許可されます。この関数は、レート制限を自動的に処理するために指数バックオフを実装します。

トークンの有効期限: 通常、アクセス トークンは発行後 30 ～ 60 分で有効期限が切れます。この関数は、実行ごとに新しいトークンをリクエストします。頻繁に実行される本番環境のデプロイでは、更新ロジックを使用してトークン キャッシュ保存を実装することを検討してください。

監査ログの保持: 監査ログのデフォルトの保持期間は 1 年です。組織が 2023 年 7 月 10 日より前に作成され、保持期間を手動で変更していない場合、保持期間は 6 年間です。構成された保持期間を超えてログを保持する必要がある場合は、ログを定期的にダウンロードします。

Cloud Scheduler ジョブの作成

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

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

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

   設定	値
   名前	daily-mulesoft-audit-export
   リージョン	Cloud Run functions と同じリージョンを選択する
   周波数	0 2 * * *（毎日 02:00 UTC に実行）
   タイムゾーン	タイムゾーンを選択します（UTC を推奨）。
   ターゲット タイプ	Pub/Sub
   トピック	トピック mulesoft-audit-trigger を選択します。
   メッセージ本文	{}（空の JSON オブジェクト）

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

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

1. Cloud Scheduler コンソールで、ジョブを見つけます。
2. [強制実行] をクリックして手動でトリガーします。
3. 数秒待ってから、[Cloud Run] > [サービス] > [mulesoft-audit-collector] > [ログ] に移動します。
4. 関数が正常に実行されたことを確認します。
5. GCS バケットをチェックして、ログが書き込まれたことを確認します。

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

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

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

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

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

   [e-mail]
   

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

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

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

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

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

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

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

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

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

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

   • ストレージ バケットの URL: GCS バケットの URI を入力します。

     gs://mulesoft-audit-logs/
     

     • mulesoft-audit-logs は、バケットの実際の名前に置き換えます。
     注: URI の末尾には常にスラッシュ（/）を付けます。

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

     • なし: 転送後にファイルを削除しません（テストにおすすめ）。
     • 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。

     • 転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。

       注: 削除オプションを選択する場合は、サービス アカウントに Storage オブジェクト閲覧者ではなく、Storage オブジェクト管理者のロールが必要です。必要に応じて IAM 権限を更新します。

   • ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。

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

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

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

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

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