このページは Cloud Translation API によって翻訳されました。

MuleSoft Anypoint のログを収集する

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

Google SecOps SIEM

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

始める前に

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

Google SecOps インスタンス
MuleSoft への特権アクセス
AWS への特権アクセス

MuleSoft 組織 ID を取得する

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

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

Google SecOps 用に AWS S3 バケットと IAM を構成する

バケットの作成のユーザーガイドに沿って、Amazon S3 バケットを作成します。
後で参照できるように、バケットの名前とリージョンを保存します（例: mulesoft-audit-logs）。
IAM ユーザーの作成のユーザーガイドに沿って、ユーザーを作成します。
作成したユーザーを選択します。
[セキュリティ認証情報] タブを選択します。
[アクセスキー] セクションで [アクセスキーを作成] をクリックします。
[ユースケース] として [サードパーティサービス] を選択します。
[次へ] をクリックします。
省略可: 説明タグを追加します。
[アクセスキーを作成] をクリックします。
[CSV ファイルをダウンロード] をクリックし、[アクセスキー] と [シークレットアクセスキー] を保存して、今後の参照に備えます。
[完了] をクリックします。
[権限] タブを選択します。
[権限ポリシー] セクションの [権限を追加] をクリックします。
[権限を追加] を選択します。
[ポリシーを直接アタッチする] を選択します。
AmazonS3FullAccess ポリシーを検索して選択します。
[次へ] をクリックします。
[権限を追加] をクリックします。

MuleSoft 接続アプリを作成する

Anypoint Platform にログインします。
[アクセス管理> 接続済みアプリ> アプリを作成] に移動します。
次の構成の詳細を入力します。
- アプリ名: 一意の名前を入力します（例: Google SecOps export）。
- [アプリが独自の権限で動作する（クライアント認証情報）] を選択します。
- [スコープを追加] → [監査ログビューア] → [次へ] をクリックします。
- ログが必要なビジネスグループをすべて選択します。
- [次へ> スコープを追加] をクリックします。
[保存] をクリックし、クライアント ID とクライアントシークレットをコピーします。

S3 アップロードの IAM ポリシーとロールを構成する

ポリシー JSON（mulesoft-audit-logs はバケット名に置き換えます）。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowPutAuditObjects",
      "Effect": "Allow",
      "Action": ["s3:PutObject"],
      "Resource": "arn:aws:s3:::mulesoft-audit-logs/*"
    }
  ]
}

AWS コンソール > IAM > ポリシー > ポリシーの作成 > [JSON] タブ に移動します。
ポリシーをコピーして貼り付けます。
[次へ] > [ポリシーを作成] をクリックします。
[IAM] > [ロール] > [ロールの作成] > [AWS サービス] > [Lambda] に移動します。
新しく作成したポリシーを関連付けます。
ロールに「WriteMulesoftToS3Role」という名前を付けて、[ロールを作成] をクリックします。

Lambda 関数を作成する

設定	値
名前	`mulesoft_audit_to_s3`
ランタイム	Python 3.13
アーキテクチャ	x86_64
実行ロール	既存の > `WriteMulesoftToS3Role` を使用する

関数を作成したら、[コード] タブを開き、スタブを削除して次のコード（mulesoft_audit_to_s3.py）を入力します。

#!/usr/bin/env python3

import os, json, gzip, io, uuid, datetime as dt, urllib.request, urllib.error, urllib.parse
import boto3

ORG_ID        = os.environ["MULE_ORG_ID"]
CLIENT_ID     = os.environ["CLIENT_ID"]
CLIENT_SECRET = os.environ["CLIENT_SECRET"]
S3_BUCKET     = os.environ["S3_BUCKET_NAME"]

TOKEN_URL = "https://anypoint.mulesoft.com/accounts/api/v2/oauth2/token"
QUERY_URL = f"https://anypoint.mulesoft.com/audit/v2/organizations/{ORG_ID}/query"

def http_post(url, data, headers=None):
    raw = json.dumps(data).encode() if headers else urllib.parse.urlencode(data).encode()
    req = urllib.request.Request(url, raw, headers or {})
    try:
        with urllib.request.urlopen(req, timeout=30) as r:
            return json.loads(r.read())
    except urllib.error.HTTPError as e:
        print("MuleSoft error body →", e.read().decode())
        raise

def get_token():
    return http_post(TOKEN_URL, {
        "grant_type": "client_credentials",
        "client_id":  CLIENT_ID,
        "client_secret": CLIENT_SECRET
    })["access_token"]

def fetch_audit(token, start, end):
    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
    }
    while True:
        data = http_post(QUERY_URL, body, headers)
        if not data.get("data"):
            break
        yield from data["data"]
        body["offset"] += body["limit"]

def upload(events, ts):
    key = f"{ts:%Y/%m/%d}/mulesoft-audit-{uuid.uuid4()}.json.gz"
    buf = io.BytesIO()
    with gzip.GzipFile(fileobj=buf, mode="w") as gz:
        for ev in events:
            gz.write((json.dumps(ev) + "\n").encode())
    buf.seek(0)
    boto3.client("s3").upload_fileobj(buf, S3_BUCKET, key)

def lambda_handler(event=None, context=None):
    now   = dt.datetime.utcnow().replace(microsecond=0)
    start = now - dt.timedelta(days=1)

    token  = get_token()
    events = list(fetch_audit(token, start, now))

    if events:
        upload(events, start)
        print(f"Uploaded {len(events)} events")
    else:
        print("No events in the last 24 h")

# For local testing
if __name__ == "__main__":
    lambda_handler()

[構成> 環境変数 > 編集 > 新しい環境変数を追加] に移動します。
指定された次の環境変数を入力し、値を置き換えます。

キー値の例

MULE_ORG_ID your_org_id

CLIENT_ID your_client_id

CLIENT_SECRET your_client_secret

S3_BUCKET_NAME mulesoft-audit-logs

キー	値の例
`MULE_ORG_ID`	`your_org_id`
`CLIENT_ID`	`your_client_id`
`CLIENT_SECRET`	`your_client_secret`
`S3_BUCKET_NAME`	`mulesoft-audit-logs`

Lambda 関数をスケジュールする（EventBridge Scheduler）

[構成> トリガー> トリガーを追加> EventBridge Scheduler > ルールを作成] に移動します。
次の構成の詳細を入力します。
- 名前: daily-mulesoft-audit export
- スケジュールパターン: Cron 式。
- 式: 0 2 * * *（毎日 02:00 UTC に実行）。
残りの設定はデフォルトのままにして、[作成] をクリックします。

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

[SIEM 設定] > [フィード] に移動します。
[新しく追加] をクリックします。
[フィード名] フィールドに、フィードの名前を入力します（例: MuleSoft Logs）。
[ソースタイプ] として [Amazon S3 V2] を選択します。
[ログタイプ] として [Mulesoft] を選択します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
- S3 URI: バケット URI
  - s3://mulesoft-audit-logs/
    - mulesoft-audit-logs は、バケットの実際の名前に置き換えます。
- Source deletion options: 必要に応じて削除オプションを選択します。
  
  注: Delete transferred files オプションまたは Delete transferred files and empty directories オプションを選択する場合は、サービスアカウントに適切な権限が付与されていることを確認してください。
- ファイルの最大経過日数: 指定した日数以内に変更されたファイルを含めます。デフォルトは 180 日です。
- アクセスキー ID: s3 バケットにアクセスできるユーザーアクセスキー。
- シークレットアクセスキー: s3 バケットにアクセスできるユーザーのシークレットキー。
- アセットの名前空間: アセットの名前空間。
- Ingestion labels: このフィードのイベントに適用されるラベル。
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。

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