收集 IBM Security Identity Manager 記錄
本文說明如何使用 Google Cloud Storage V2,將 IBM Security Identity Manager 記錄檔擷取至 Google Security Operations。
IBM Security Identity Manager (ISIM) 是一項以政策為依據的自動化解決方案,可管理 IT 環境中的使用者存取權,協助企業有效管理及控管身分。這項解決方案會使用角色、帳戶和存取權限,在使用者生命週期中自動建立、修改、重新認證及終止使用者身分。在較新版本中,這項產品也稱為 IBM Security Verify Governance - Identity Manager。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 Google Cloud Storage 值區 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 以上版本
- 具備管理員權限的 ISIM 使用者帳戶,可存取 REST API 端點 (
/itim/rest/activities和/itim/rest/requests) - GCP 專案與 ISIM 虛擬設備之間的網路連線 (防火牆規則必須允許 ISIM 通訊埠上的 HTTPS 流量)
設定 IBM Security Identity Manager API 存取權
如要讓 Google SecOps 擷取活動和要求稽核資料,您必須在 ISIM 中設定專屬服務帳戶,並授予 REST API 存取權。
建立專用的 ISIM 帳戶,以存取 API
- 登入 IBM Security Identity Manager 管理控制台。
- 依序前往「管理使用者」>「建立使用者」。
- 為整合服務建立新的使用者帳戶 (例如
secops-api-user)。 - 指派「系統管理員」角色,或是包含透過 REST API 查看活動和要求權限的自訂角色。
- 以安全方式儲存帳戶憑證:
- 使用者名稱:ISIM 登入使用者名稱 (例如
secops-api-user) - 密碼:ISIM 登入密碼
- 使用者名稱:ISIM 登入使用者名稱 (例如
找出 ISIM API 基礎網址
ISIM API 基本網址的格式如下:
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 bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 ibm-sim-logs-gcs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用:啟用物件版本管理或資料保留政策 點選「建立」。
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限,且可由 Pub/Sub 叫用的服務帳戶。
建立服務帳戶
- 在 GCP 控制台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
ibm-sim-collector-sa - 服務帳戶說明:輸入
Service account for Cloud Run function to collect IBM Security Identity Manager logs
- 服務帳戶名稱:輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分,新增下列角色:
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下 [完成]。
這些角色適用於:
- Storage 物件管理員:將記錄檔寫入 GCS bucket,並管理狀態檔案
- Cloud Run 叫用者:允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者:允許函式叫用
授予 GCS 值區的 IAM 權限
將 GCS bucket 的寫入權限授予服務帳戶:
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (
ibm-sim-logs-gcs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (
ibm-sim-collector-sa@PROJECT_ID.iam.gserviceaccount.com) - 指派角色:選取「Storage 物件管理員」
- 新增主體:輸入服務帳戶電子郵件地址 (
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中,前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料:
- 主題 ID:輸入
ibm-sim-logs-trigger - 其他設定保留預設值
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 傳送的 Pub/Sub 訊息觸發,從 IBM Security Identity Manager REST API 擷取記錄,並寫入 GCS。
如要建立 Cloud Run 函式來收集記錄,請按照下列步驟操作:
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 ibm-sim-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中,選擇
ibm-sim-logs-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「容器」、「網路」和「安全性」。
前往「安全性」分頁:
- 服務帳戶:選取
ibm-sim-collector-sa
- 服務帳戶:選取
前往「容器」分頁:
- 按一下「變數與密鑰」。
針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 說明 GCS_BUCKETibm-sim-logs-gcsGCS bucket 名稱 GCS_PREFIXibm-sim記錄檔的前置字串 STATE_KEYibm-sim/state.json狀態檔案路徑 ISIM_BASE_URLhttps://isim.example.com:9082ISIM 伺服器基本網址 ISIM_USERNAMEsecops-api-userISIM API 使用者名稱 ISIM_PASSWORDyour-passwordISIM API 密碼 LOOKBACK_HOURS24初始回溯期 PAGE_SIZE100每個 API 頁面的記錄 MAX_RECORDS5000每次執行的記錄數上限
在「變數與密鑰」部分,向下捲動至「要求」:
- 要求逾時:輸入
600秒 (10 分鐘)
- 要求逾時:輸入
前往「設定」分頁:
- 在「資源」部分:
- 記憶體:選取 512 MiB 以上
- CPU:選取 1
- 在「資源」部分:
在「修訂版本資源調度」部分:
- 執行個體數量下限:輸入
0 - 執行個體數量上限:輸入
100
- 執行個體數量下限:輸入
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立完成後,系統會自動開啟內嵌程式碼編輯器。
新增函式程式碼
- 在「進入點」欄位中輸入「main」。
在內嵌程式碼編輯器中建立兩個檔案:
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 函式。
- 前往 GCP 主控台的「Cloud Scheduler」。
- 點選「建立工作」。
請提供下列設定詳細資料:
設定 值 名稱 ibm-sim-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取「 ibm-sim-logs-trigger」郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間要求選擇頻率:
| 頻率 | Cron 運算式 | 用途 |
|---|---|---|
| 每 5 分鐘 | */5 * * * * |
大量、低延遲 |
| 每 15 分鐘 | */15 * * * * |
中等 |
| 每小時 | 0 * * * * |
標準 (建議) |
| 每 6 小時 | 0 */6 * * * |
少量、批次處理 |
| 每日 | 0 0 * * * |
歷來資料集合 |
測試整合項目
- 在 Cloud Scheduler 控制台中,找出您的工作 (
ibm-sim-collector-hourly)。 - 按一下「強制執行」即可手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「Services」。
- 按一下
ibm-sim-collector。 - 按一下 [Logs] (記錄) 分頁標籤。
確認函式是否已順利執行。請注意以下事項:
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」>「Buckets」。
按一下
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 函式與 ISIM 虛擬設備之間的網路連線
- 缺少環境變數:確認 Cloud Run 函式設定中已設定所有必要變數
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
取得服務帳戶電子郵件地址
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
IBM SIM Logs GCS)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「IBM Security Identity Manager」做為「記錄類型」。
按一下「取得服務帳戶」,系統會顯示專屬服務帳戶電子郵件地址,例如:
secops-12345678@secops-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,以便在下一步中使用。
點選「下一步」。
指定下列輸入參數的值:
儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:
gs://ibm-sim-logs-gcs/ibm-sim/
來源刪除選項:根據偏好設定選取刪除選項:
- 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。
檔案存在時間上限:包含在過去天數內修改的檔案 (預設為 180 天)
資產命名空間:資產命名空間
擷取標籤:要套用至這個動態饋給事件的標籤
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
如要將 IAM 權限授予 Google SecOps 服務帳戶,請按照下列步驟操作:
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下
ibm-sim-logs-gcs。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址
- 指派角色:選取「Storage 物件檢視者」
- 按一下 [儲存]。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| 說明 | metadata.description | 事件說明 |
| receivedTime | metadata.event_timestamp | 事件發生時的時間戳記 |
| metadata.event_type | metadata.event_type | 活動類型 |
| 版本 | metadata.product_version | 產品版本 |
| connectionId | network.session_id | 工作階段 ID |
| tls | network.tls.version | 傳輸層安全標準 (TLS) 版本 |
| ip | principal.asset.ip | 資產的 IP 位址 |
| 主機名稱 | 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 專業人員尋求答案。