IBM Security Identity Manager のログを収集する
このドキュメントでは、Google Cloud Storage V2 を使用して IBM Security Identity Manager ログを Google Security Operations に取り込む方法について説明します。
IBM Security Identity Manager(ISIM)は、IT 環境全体でユーザー アクセスを管理する自動化されたポリシーベースのソリューションです。企業全体で効果的な ID 管理とガバナンスを推進するのに役立ちます。ロール、アカウント、アクセス権限を使用して、ユーザー ライフサイクル全体にわたるユーザー ID の作成、変更、再認証、終了を自動化します。このプロダクトは、新しいバージョンでは IBM Security Verify Governance - Identity Manager とも呼ばれます。
始める前に
次の前提条件を満たしていることを確認します。
- Google SecOps インスタンス
- Cloud Storage API が有効になっている GCP プロジェクト
- GCS バケットを作成および管理する権限
- GCS バケットの IAM ポリシーを管理する権限
- Cloud Run サービス、Pub/Sub トピック、Cloud Scheduler ジョブを作成する権限
- IBM Security Identity Manager REST API への特権アクセス
- IBM Security Identity Manager バージョン 7.0.2 以降、または IBM Security Verify Governance 10.0.1 以降
- REST API エンドポイント(
/itim/rest/activitiesと/itim/rest/requests)にアクセスするための管理者権限を持つ ISIM ユーザー アカウント - GCP プロジェクトと ISIM 仮想アプライアンス間のネットワーク接続(ファイアウォール ルールで ISIM ポートの HTTPS トラフィックを許可する必要があります)
IBM Security Identity Manager API アクセスを構成する
Google SecOps がアクティビティとリクエストの監査データを取得できるようにするには、REST API アクセスに必要な権限を持つ専用のサービス アカウントを ISIM で構成する必要があります。
API アクセス用の専用の ISIM アカウントを作成する
- IBM Security Identity Manager 管理コンソールにログインします。
- [Manage Users> Create User] に移動します。
- 統合用の新しいユーザー アカウントを作成します(例:
secops-api-user)。 - REST API を介してアクティビティとリクエストを表示する権限を含むシステム管理者ロールまたはカスタムロールを割り当てます。
- アカウントの認証情報を安全に保存します。
- ユーザー名: ISIM ログイン ユーザー名(例:
secops-api-user) - Password: ISIM のログイン パスワード
- ユーザー名: ISIM ログイン ユーザー名(例:
ISIM API ベース URL を特定する
ISIM API のベース URL は次の形式です。
https://<isim-hostname>:<port>たとえば、ISIM 仮想アプライアンスのホスト名が
isim.example.comで、デフォルトの HTTPS ポートを使用している場合:https://isim.example.com:9082
テスト API へのアクセス
統合に進む前に、認証情報をテストします。
# Replace with your actual credentials and ISIM URL ISIM_BASE_URL="https://isim.example.com:9082" ISIM_USER="chronicle-api-user" ISIM_PASS="your-password" # Authenticate and obtain session cookie curl -v -k -c cookies.txt \ -d "j_username=${ISIM_USER}&j_password=${ISIM_PASS}" \ "${ISIM_BASE_URL}/itim/j_security_check" # Retrieve CSRF token from systemusers/me endpoint curl -v -k -b cookies.txt \ -H "Accept: application/json" \ "${ISIM_BASE_URL}/itim/rest/systemusers/me" # Test activities endpoint curl -v -k -b cookies.txt \ -H "Accept: application/json" \ "${ISIM_BASE_URL}/itim/rest/activities"
成功すると、最近の ISIM アクティビティの JSON 配列が返されます。
Google Cloud Storage バケットを作成する
- Google Cloud コンソールに移動します。
- プロジェクトを選択するか、新しいプロジェクトを作成します。
- ナビゲーション メニューで、[Cloud Storage > バケット] に移動します。
- [バケットを作成] をクリックします。
次の構成情報を提供してください。
設定 値 バケットに名前を付ける グローバルに一意の名前( ibm-sim-logs-gcsなど)を入力します。ロケーション タイプ ニーズに応じて選択します(リージョン、デュアルリージョン、マルチリージョン)。 ロケーション 場所を選択します(例: us-central1)。ストレージ クラス Standard(アクセス頻度の高いログにおすすめ) アクセス制御 均一(推奨) 保護ツール 省略可: オブジェクトのバージョニングまたは保持ポリシーを有効にする [作成] をクリックします。
Cloud Run functions のサービス アカウントを作成する
Cloud Run functions には、GCS バケットへの書き込み権限があり、Pub/Sub によって呼び出されるサービス アカウントが必要です。
サービス アカウントの作成
- GCP コンソールで、[IAM と管理>サービス アカウント] に移動します。
- [サービス アカウントを作成] をクリックします。
- 次の構成情報を指定します。
- サービス アカウント名: 「
ibm-sim-collector-sa」と入力します。 - サービス アカウントの説明:
Service account for Cloud Run function to collect IBM Security Identity Manager logsと入力します。
- サービス アカウント名: 「
- [作成して続行] をクリックします。
- [このサービス アカウントにプロジェクトへのアクセスを許可する] セクションで、次のロールを追加します。
- [ロールを選択] をクリックします。
- [ストレージ オブジェクト管理者] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Run 起動元] を検索して選択します。
- [+ 別のロールを追加] をクリックします。
- [Cloud Functions 起動元] を検索して選択します。
- [続行] をクリックします。
- [完了] をクリックします。
これらのロールは、次の操作に必要です。
- Storage オブジェクト管理者: ログを GCS バケットに書き込み、状態ファイルを管理する
- Cloud Run 起動元: Pub/Sub が関数を呼び出すことを許可します。
- Cloud Functions 起動元: 関数の呼び出しを許可する
GCS バケットに対する IAM 権限を付与する
GCS バケットに対する書き込み権限をサービス アカウントに付与します。
- [Cloud Storage] > [バケット] に移動します。
- バケット名(
ibm-sim-logs-gcs)をクリックします。 - [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルの追加: サービス アカウントのメールアドレス(
ibm-sim-collector-sa@PROJECT_ID.iam.gserviceaccount.com)を入力します。 - ロールを割り当てる: [ストレージ オブジェクト管理者] を選択します。
- プリンシパルの追加: サービス アカウントのメールアドレス(
- [保存] をクリックします。
Pub/Sub トピックの作成
Cloud Scheduler がパブリッシュし、Cloud Run 関数がサブスクライブする Pub/Sub トピックを作成します。
- GCP Console で、[Pub/Sub> トピック] に移動します。
- [トピックを作成] をクリックします。
- 次の構成情報を指定します。
- トピック ID: 「
ibm-sim-logs-trigger」と入力します。 - その他の設定はデフォルトのままにします。
- トピック ID: 「
- [作成] をクリックします。
ログを収集する Cloud Run 関数を作成する
Cloud Run 関数は、Cloud Scheduler からの Pub/Sub メッセージによってトリガーされ、IBM Security Identity Manager REST API からログを取得して GCS に書き込みます。
ログを収集する Cloud Run functions を作成するには、次の操作を行います。
- GCP Console で、[Cloud Run] に移動します。
- [サービスを作成] をクリックします。
- [関数] を選択します(インライン エディタを使用して関数を作成します)。
[構成] セクションで、次の構成の詳細を指定します。
設定 値 サービス名 ibm-sim-collectorリージョン GCS バケットと一致するリージョンを選択します(例: us-central1)。ランタイム [Python 3.12] 以降を選択します。 [トリガー(省略可)] セクションで、次の操作を行います。
- [+ トリガーを追加] をクリックします。
- [Cloud Pub/Sub] を選択します。
- [Cloud Pub/Sub トピックを選択してください] で、
ibm-sim-logs-triggerを選択します。 - [保存] をクリックします。
[認証] セクションで、次の操作を行います。
- [認証が必要] を選択します。
- Identity and Access Management(IAM)を確認します。
下にスクロールして、[コンテナ、ネットワーキング、セキュリティ] を開きます。
[セキュリティ] タブに移動します。
- サービス アカウント:
ibm-sim-collector-saを選択します。
- サービス アカウント:
[コンテナ] タブに移動します。
- [変数とシークレット] をクリックします。
環境変数ごとに [+ 変数を追加] をクリックします。
変数名 値の例 説明 GCS_BUCKETibm-sim-logs-gcsGCS バケット名 GCS_PREFIXibm-simログファイルの接頭辞 STATE_KEYibm-sim/state.json状態ファイルのパス ISIM_BASE_URLhttps://isim.example.com:9082ISIM サーバーのベース URL ISIM_USERNAMEsecops-api-userISIM API ユーザー名 ISIM_PASSWORDyour-passwordISIM API パスワード LOOKBACK_HOURS24初期ルックバック期間 PAGE_SIZE100API ページあたりのレコード数 MAX_RECORDS5000実行あたりの最大レコード数
[変数とシークレット] セクションで、[リクエスト] まで下にスクロールします。
- リクエストのタイムアウト:
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 import urllib.parse from datetime import datetime, timezone, timedelta # Disable SSL warnings for self-signed certificates urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) http = urllib3.PoolManager( timeout=urllib3.Timeout(connect=10.0, read=60.0), retries=False, cert_reqs='CERT_NONE', ) storage_client = storage.Client() GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'ibm-sim') STATE_KEY = os.environ.get('STATE_KEY', 'ibm-sim/state.json') ISIM_BASE_URL = os.environ.get('ISIM_BASE_URL', '').rstrip('/') ISIM_USERNAME = os.environ.get('ISIM_USERNAME') ISIM_PASSWORD = os.environ.get('ISIM_PASSWORD') LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100')) MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000')) @functions_framework.cloud_event def main(cloud_event): if not all([GCS_BUCKET, ISIM_BASE_URL, ISIM_USERNAME, ISIM_PASSWORD]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) state = load_state(bucket) now = datetime.now(timezone.utc) if isinstance(state, dict) and state.get('last_event_time'): try: last_val = state['last_event_time'] if last_val.endswith('Z'): last_val = last_val[:-1] + '+00:00' last_time = datetime.fromisoformat(last_val) last_time = last_time - timedelta(minutes=2) except Exception as e: print(f"Warning: Could not parse last_event_time: {e}") last_time = now - timedelta(hours=LOOKBACK_HOURS) else: last_time = now - timedelta(hours=LOOKBACK_HOURS) print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}") session_cookies = authenticate() csrf_token = get_csrf_token(session_cookies) activities = fetch_activities(session_cookies, csrf_token, last_time, now) requests_data = fetch_requests(session_cookies, csrf_token, last_time, now) all_records = [] for a in activities: a['_isim_record_type'] = 'activity' all_records.append(a) for r in requests_data: r['_isim_record_type'] = 'request' all_records.append(r) if not all_records: print("No new records found.") save_state(bucket, now.isoformat()) return timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/ibm_sim_audit_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join( [json.dumps(r, ensure_ascii=False, default=str) for r in all_records] ) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(all_records)} records to gs://{GCS_BUCKET}/{object_key}") newest = find_newest_time(all_records) save_state(bucket, newest if newest else now.isoformat()) print(f"Successfully processed {len(all_records)} records " f"(activities: {len(activities)}, requests: {len(requests_data)})") except Exception as e: print(f'Error processing logs: {str(e)}') raise def authenticate(): url = f"{ISIM_BASE_URL}/itim/j_security_check" encoded_body = urllib.parse.urlencode({ 'j_username': ISIM_USERNAME, 'j_password': ISIM_PASSWORD }).encode('utf-8') response = http.request( 'POST', url, body=encoded_body, headers={'Content-Type': 'application/x-www-form-urlencoded'}, redirect=False ) cookies = {} set_cookie_headers = response.headers.getlist('Set-Cookie') if hasattr(response.headers, 'getlist') else [] if not set_cookie_headers: raw = response.headers.get('Set-Cookie', '') if raw: set_cookie_headers = [raw] for cookie_header in set_cookie_headers: parts = cookie_header.split(';')[0] if '=' in parts: name, value = parts.split('=', 1) cookies[name.strip()] = value.strip() if not cookies: raise Exception( f"Authentication failed: no session cookies returned " f"(HTTP {response.status})" ) print("Successfully authenticated to ISIM") return cookies def get_csrf_token(cookies): url = f"{ISIM_BASE_URL}/itim/rest/systemusers/me" cookie_str = '; '.join([f"{k}={v}" for k, v in cookies.items()]) response = http.request( 'GET', url, headers={ 'Accept': 'application/json', 'Cookie': cookie_str } ) if response.status != 200: raise Exception( f"Failed to get CSRF token: {response.status} - " f"{response.data.decode('utf-8')}" ) data = json.loads(response.data.decode('utf-8')) csrf_token = None if isinstance(data, dict): csrf_token = data.get('_csrf', data.get('CSRFToken')) if not csrf_token: csrf_header = response.headers.get('CSRFToken', '') if csrf_header: csrf_token = csrf_header print("Successfully retrieved CSRF token") return csrf_token def fetch_activities(cookies, csrf_token, start_time, end_time): url = f"{ISIM_BASE_URL}/itim/rest/activities" cookie_str = '; '.join([f"{k}={v}" for k, v in cookies.items()]) headers = { 'Accept': 'application/json', 'Cookie': cookie_str } if csrf_token: headers['CSRFToken'] = csrf_token all_records = [] start_index = 0 while len(all_records) < MAX_RECORDS: params = { '_start': str(start_index), '_count': str(min(PAGE_SIZE, MAX_RECORDS - len(all_records))) } query_string = urllib.parse.urlencode(params) request_url = f"{url}?{query_string}" response = http.request('GET', request_url, headers=headers) if response.status != 200: print(f"Activities query failed: {response.status} - " f"{response.data.decode('utf-8')}") break page_results = json.loads(response.data.decode('utf-8')) if isinstance(page_results, dict): page_results = page_results.get('activities', page_results.get('items', [])) if not page_results: break all_records.extend(page_results) print(f"Activities: Retrieved {len(page_results)} records " f"(total: {len(all_records)})") if len(page_results) < PAGE_SIZE: break start_index += len(page_results) print(f"Total activities fetched: {len(all_records)}") return all_records def fetch_requests(cookies, csrf_token, start_time, end_time): url = f"{ISIM_BASE_URL}/itim/rest/requests" cookie_str = '; '.join([f"{k}={v}" for k, v in cookies.items()]) headers = { 'Accept': 'application/json', 'Cookie': cookie_str } if csrf_token: headers['CSRFToken'] = csrf_token all_records = [] start_index = 0 while len(all_records) < MAX_RECORDS: params = { '_start': str(start_index), '_count': str(min(PAGE_SIZE, MAX_RECORDS - len(all_records))) } query_string = urllib.parse.urlencode(params) request_url = f"{url}?{query_string}" response = http.request('GET', request_url, headers=headers) if response.status != 200: print(f"Requests query failed: {response.status} - " f"{response.data.decode('utf-8')}") break page_results = json.loads(response.data.decode('utf-8')) if isinstance(page_results, dict): page_results = page_results.get('requests', page_results.get('items', [])) if not page_results: break all_records.extend(page_results) print(f"Requests: Retrieved {len(page_results)} records " f"(total: {len(all_records)})") if len(page_results) < PAGE_SIZE: break start_index += len(page_results) print(f"Total requests fetched: {len(all_records)}") return all_records def find_newest_time(records): newest = None for r in records: for field in ['completedTime', 'scheduledTime', 'startedTime', 'lastModified', 'requestDate', 'timestamp']: t = r.get(field) if t: try: if isinstance(t, (int, float)): dt = datetime.fromtimestamp(t / 1000, tz=timezone.utc) t_iso = dt.isoformat() else: t_iso = str(t) if newest is None or t_iso > newest: newest = t_iso except Exception: continue return newest def load_state(bucket): try: blob = bucket.blob(STATE_KEY) if blob.exists(): return json.loads(blob.download_as_text()) except Exception as e: print(f"Warning: Could not load state: {e}") return {} def save_state(bucket, last_event_time_iso): try: state = { 'last_event_time': last_event_time_iso, 'last_run': datetime.now(timezone.utc).isoformat() } blob = bucket.blob(STATE_KEY) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: last_event_time={last_event_time_iso}") except Exception as e: print(f"Warning: Could not save state: {e}")requirements.txt:
functions-framework==3.* google-cloud-storage==2.* urllib3>=2.0.0
[デプロイ] をクリックして、関数を保存してデプロイします。
デプロイが完了するまで待ちます(2 ~ 3 分)。
Cloud Scheduler ジョブの作成
Cloud Scheduler は、定期的に Pub/Sub トピックにメッセージをパブリッシュし、Cloud Run functions の関数をトリガーします。
- GCP Console で、[Cloud Scheduler] に移動します。
- [ジョブを作成] をクリックします。
次の構成情報を提供してください。
設定 値 名前 ibm-sim-collector-hourlyリージョン Cloud Run functions と同じリージョンを選択する 周波数 0 * * * *(1 時間ごと、毎正時)タイムゾーン タイムゾーンを選択(UTC を推奨) ターゲット タイプ Pub/Sub トピック ibm-sim-logs-triggerを選択メッセージ本文 {}(空の JSON オブジェクト)[作成] をクリックします。
スケジュールの頻度オプション
ログの量とレイテンシの要件に基づいて頻度を選択します。
| 頻度 | CRON 式 | ユースケース |
|---|---|---|
| 5 分おき | */5 * * * * |
大容量、低レイテンシ |
| 15 分ごと | */15 * * * * |
検索量が普通 |
| 1 時間ごと | 0 * * * * |
標準(推奨) |
| 6 時間ごと | 0 */6 * * * |
少量、バッチ処理 |
| 毎日 | 0 0 * * * |
過去のデータの収集 |
統合をテストする
- Cloud Scheduler コンソールで、ジョブ(
ibm-sim-collector-hourly)を見つけます。 - [強制実行] をクリックして、ジョブを手動でトリガーします。
- 数秒待ちます。
- Cloud Run > [サービス] に移動します。
ibm-sim-collectorをクリックします。- [ログ] タブをクリックします。
関数が正常に実行されたことを確認します。以下のものを探します。
Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Successfully authenticated to ISIM Successfully retrieved CSRF token Activities: Retrieved X records (total: X) Requests: Retrieved X records (total: X) Wrote X records to gs://ibm-sim-logs-gcs/ibm-sim/ibm_sim_audit_YYYYMMDD_HHMMSS.ndjson Successfully processed X records (activities: X, requests: X)[Cloud Storage] > [バケット] に移動します。
ibm-sim-logs-gcsをクリックします。ibm-sim/フォルダに移動します。現在のタイムスタンプで新しい
.ndjsonファイルが作成されたことを確認します。
ログにエラーが表示された場合:
- HTTP 401/302:
ISIM_USERNAME環境変数とISIM_PASSWORD環境変数が正しく、ユーザー アカウントがロックされていないことを確認します。 - HTTP 403: ISIM ユーザー アカウントに REST API アクセス用の管理者権限があることを確認します。
- HTTP 404:
ISIM_BASE_URLが正しく、適切なポート番号が含まれていることを確認します。 - 接続エラー: Cloud Run functions と ISIM 仮想アプライアンス間のネットワーク接続を確認します
- 環境変数がありません: 必要な変数がすべて Cloud Run functions の構成で設定されていることを確認します。
Google SecOps サービス アカウントを取得する
Google SecOps は、一意のサービス アカウントを使用して GCS バケットからデータを読み取ります。このサービス アカウントにバケットへのアクセス権を付与する必要があります。
サービス アカウントのメールアドレスを取得する
- [SIEM 設定] > [フィード] に移動します。
- [Add New Feed] をクリックします。
- [単一フィードを設定] をクリックします。
- [フィード名] フィールドに、フィードの名前を入力します(例:
IBM SIM Logs GCS)。 - [ソースタイプ] として [Google Cloud Storage V2] を選択します。
- [ログタイプ] として [IBM Security Identity Manager] を選択します。
[サービス アカウントを取得する] をクリックします。一意のサービス アカウントのメールアドレスが表示されます(例:)。
secops-12345678@secops-gcp-prod.iam.gserviceaccount.comこのメールアドレスをコピーして、次のステップで使用します。
[次へ] をクリックします。
次の入力パラメータの値を指定します。
ストレージ バケットの URL: 接頭辞パスを含む GCS バケット URI を入力します。
gs://ibm-sim-logs-gcs/ibm-sim/
Source deletion option: 必要に応じて削除オプションを選択します。
- なし: 転送後にファイルを削除しません(テストにおすすめ)。
- 転送されたファイルを削除する: 転送が完了した後にファイルを削除します。
転送されたファイルと空のディレクトリを削除する: 転送が完了した後にファイルと空のディレクトリを削除します。
ファイルの最大経過日数: 過去の日数以内に変更されたファイルを含めます(デフォルトは 180 日)。
アセットの名前空間: アセットの名前空間
Ingestion labels: このフィードのイベントに適用されるラベル
[次へ] をクリックします。
[Finalize] 画面で新しいフィードの設定を確認し、[送信] をクリックします。
Google SecOps サービス アカウントに IAM 権限を付与する
Google SecOps サービス アカウントには、GCS バケットに対する Storage オブジェクト閲覧者のロールが必要です。
Google SecOps サービス アカウントに IAM 権限を付与する手順は次のとおりです。
- [Cloud Storage] > [バケット] に移動します。
ibm-sim-logs-gcsをクリックします。- [権限] タブに移動します。
- [アクセス権を付与] をクリックします。
- 次の構成情報を指定します。
- プリンシパルを追加: Google SecOps サービス アカウントのメールアドレスを貼り付けます。
- ロールを割り当てる: [Storage オブジェクト閲覧者] を選択します。
- [保存] をクリックします。
UDM マッピング テーブル
| ログフィールド | UDM マッピング | ロジック |
|---|---|---|
| 説明 | metadata.description | イベントの説明 |
| receivedTime | metadata.event_timestamp | イベントが発生したときのタイムスタンプ |
| metadata.event_type | metadata.event_type | イベントのタイプ |
| version | metadata.product_version | プロダクト バージョン |
| connectionId | network.session_id | セッション ID |
| tls | network.tls.version | TLS バージョン |
| ip | principal.asset.ip | アセットの IP アドレス |
| hostname | principal.hostname | ホスト名 |
| ip | principal.ip | IP アドレス |
| ポート | principal.port | ポート番号 |
| kv_data | principal.user.attribute.labels | ユーザー属性ラベル |
| アクション | security_result.action_details | 実施した対応の詳細 |
| metadata.product_name | metadata.product_name | 製品名 |
| metadata.vendor_name | metadata.vendor_name | ベンダー名 |
変更ログ
さらにサポートが必要な場合 コミュニティ メンバーや Google SecOps のプロフェッショナルから回答を得ることができます。