收集 IBM Security Verify SaaS 記錄
本文說明如何使用 Google Cloud Storage V2,將 IBM Security Verify SaaS 記錄檔擷取至 Google Security Operations。
IBM Security Verify SaaS 是雲端身分與存取權管理平台,提供單一登入、多重驗證和適應性存取權控管功能。並產生驗證事件、政策決策和使用者生命週期管理的稽核記錄。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS bucket 的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- IBM Security Verify SaaS 的特殊存取權 (管理員角色)
建立 Google Cloud Storage bucket
- 前往 Google Cloud 控制台。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 ibm-verify-saas-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用:啟用物件版本管理或資料保留政策 點選「建立」。
收集 IBM Security Verify SaaS API 憑證
建立 API 用戶端
- 登入 IBM Security Verify 管理控制台。
- 依序前往「安全性」>「API 存取權」。
- 按一下「新增 API 用戶端」。
- 輸入 API 用戶端的名稱 (例如
Google Security Operations Integration)。 - 指派下列權限:
- 讀取事件記錄
- 閱讀報告
- 按一下 [儲存]。
- 複製下列詳細資料並儲存到安全位置:
- 用戶端 ID:API 用戶端 ID。
- 用戶端密鑰:API 用戶端密鑰。
判斷租戶網址
API 基礎網址衍生自 IBM Security Verify 租戶。格式為:
| 控制台網址 | API 基礎網址 |
|---|---|
https://YOUR_TENANT.verify.ibm.com |
https://YOUR_TENANT.verify.ibm.com |
- 請將
YOUR_TENANT替換成實際的 IBM Security Verify 租戶名稱。
驗證權限
如要確認 API 用戶端是否具備必要權限,請按照下列步驟操作:
- 登入 IBM Security Verify 管理控制台。
- 依序前往「安全性」>「API 存取權」。
- 按一下 API 用戶端名稱。
- 確認已啟用「讀取事件記錄」和「讀取報表」權限。
- 如果缺少權限,請與 IBM Security Verify 管理員聯絡。
測試 API 存取權
請先測試憑證,再繼續進行整合:
# Replace with your actual credentials CLIENT_ID="your-client-id" CLIENT_SECRET="your-client-secret" TENANT_URL="https://YOUR_TENANT.verify.ibm.com" # Get OAuth token TOKEN=$(curl -s -X POST "${TENANT_URL}/v1.0/endpoint/default/token" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "grant_type=client_credentials&client_id=${CLIENT_ID}&client_secret=${CLIENT_SECRET}" \ | jq -r '.access_token') # Test API access curl -v -H "Authorization: Bearer ${TOKEN}" "${TENANT_URL}/v1.0/events?size=1"
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限,且可由 Pub/Sub 叫用的服務帳戶。
建立服務帳戶
- 在 GCP 控制台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
ibm-verify-collector-sa。 - 服務帳戶說明:輸入
Service account for Cloud Run function to collect IBM Security Verify SaaS 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-verify-saas-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (例如
ibm-verify-collector-sa@your-project.iam.gserviceaccount.com)。 - 指派角色:選取「Storage 物件管理員」。
- 新增主體:輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中,前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料:
- 主題 ID:輸入
ibm-verify-trigger。 - 其他設定保留預設值。
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 IBM Security Verify SaaS API 擷取記錄,並寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 ibm-verify-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取 Python 3.12 以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中,選擇主題
ibm-verify-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
向下捲動並展開「容器」、「網路」和「安全性」。
前往「安全性」分頁:
- 服務帳戶:選取服務帳戶
ibm-verify-collector-sa。
- 服務帳戶:選取服務帳戶
前往「容器」分頁:
- 按一下「變數與密鑰」。
- 針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 說明 GCS_BUCKETibm-verify-saas-logsGCS bucket 名稱 GCS_PREFIXibm-verify記錄檔的前置字串 STATE_KEYibm-verify/state.json狀態檔案路徑 TENANT_URLhttps://YOUR_TENANT.verify.ibm.comIBM Security Verify 租戶網址 CLIENT_IDyour-client-idAPI 用戶端 ID CLIENT_SECRETyour-client-secretAPI 用戶端密鑰 MAX_RECORDS10000每次執行的記錄數上限 PAGE_SIZE1000每頁記錄數 LOOKBACK_HOURS24初始回溯期 在「變數與密鑰」分頁中向下捲動至「要求」:
- 要求逾時:輸入
600秒 (10 分鐘)。
- 要求逾時:輸入
前往「容器」中的「設定」分頁:
- 在「資源」部分:
- 記憶體:選取 512 MiB 以上。
- CPU:選取 1。
- 在「資源」部分:
在「修訂版本資源調度」部分:
- 執行個體數量下限:輸入
0。 - 「Maximum number of instances」(執行個體數量上限):輸入
100(或根據預期負載調整)。
- 執行個體數量下限:輸入
點選「建立」。
等待服務建立完成 (1 到 2 分鐘)。
服務建立完成後,系統會自動開啟內嵌程式碼編輯器。
新增函式程式碼
- 在「函式進入點」中輸入 main。
在內嵌程式碼編輯器中,建立兩個檔案:
第一個檔案 - main.py:
import functions_framework from google.cloud import storage import json import os import urllib3 from datetime import datetime, timezone, timedelta import time import base64 # 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() # Environment variables GCS_BUCKET = os.environ.get('GCS_BUCKET') GCS_PREFIX = os.environ.get('GCS_PREFIX', 'ibm-verify') STATE_KEY = os.environ.get('STATE_KEY', 'ibm-verify/state.json') TENANT_URL = os.environ.get('TENANT_URL', '').rstrip('/') CLIENT_ID = os.environ.get('CLIENT_ID', '') CLIENT_SECRET = os.environ.get('CLIENT_SECRET', '') MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '10000')) PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '1000')) LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24')) def parse_datetime(value: str) -> datetime: """Parse ISO datetime string to datetime object.""" if value.endswith("Z"): value = value[:-1] + "+00:00" return datetime.fromisoformat(value) def get_token(): """Get OAuth 2.0 access token using client credentials flow.""" token_url = f"{TENANT_URL}/v1.0/endpoint/default/token" body = f"grant_type=client_credentials&client_id={CLIENT_ID}&client_secret={CLIENT_SECRET}" headers = { 'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json' } backoff = 1.0 max_retries = 3 for attempt in range(max_retries): response = http.request('POST', token_url, body=body.encode('utf-8'), headers=headers) 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 RuntimeError(f"Failed to get access token: {response.status} - {response.data.decode('utf-8')}") data = json.loads(response.data.decode('utf-8')) return data['access_token'] raise RuntimeError(f"Failed to get token after {max_retries} retries due to rate limiting") @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch IBM Security Verify SaaS logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, TENANT_URL, CLIENT_ID, CLIENT_SECRET]): print('Error: Missing required environment variables') return try: bucket = storage_client.bucket(GCS_BUCKET) # Load state state = load_state(bucket, STATE_KEY) # Determine time window now = datetime.now(timezone.utc) last_time = None if isinstance(state, dict) and state.get("last_event_time"): try: last_time = parse_datetime(state["last_event_time"]) last_time = last_time - timedelta(minutes=2) except Exception as e: print(f"Warning: Could not parse last_event_time: {e}") if last_time is None: last_time = now - timedelta(hours=LOOKBACK_HOURS) print(f"Fetching logs from {last_time.isoformat()} to {now.isoformat()}") # Get access token token = get_token() # Fetch logs records, newest_event_time = fetch_logs( token=token, start_time=last_time, end_time=now, page_size=PAGE_SIZE, max_records=MAX_RECORDS, ) if not records: print("No new log records found.") save_state(bucket, STATE_KEY, now.isoformat()) return # Write to GCS as NDJSON timestamp = now.strftime('%Y%m%d_%H%M%S') object_key = f"{GCS_PREFIX}/logs_{timestamp}.ndjson" blob = bucket.blob(object_key) ndjson = '\n'.join([json.dumps(record, ensure_ascii=False) for record in records]) + '\n' blob.upload_from_string(ndjson, content_type='application/x-ndjson') print(f"Wrote {len(records)} records to gs://{GCS_BUCKET}/{object_key}") if newest_event_time: save_state(bucket, STATE_KEY, newest_event_time) else: save_state(bucket, STATE_KEY, now.isoformat()) print(f"Successfully processed {len(records)} records") except Exception as e: print(f'Error processing logs: {str(e)}') raise 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: {e}") return {} def save_state(bucket, key, last_event_time_iso: str): """Save the last event timestamp to GCS state file.""" try: state = {'last_event_time': last_event_time_iso} blob = bucket.blob(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}") def fetch_logs(token: str, start_time: datetime, end_time: datetime, page_size: int, max_records: int): """ Fetch logs from IBM Security Verify SaaS Events API with pagination and rate limiting. Args: token: OAuth 2.0 access token start_time: Start time for log query end_time: End time for log query page_size: Number of records per page max_records: Maximum total records to fetch Returns: Tuple of (records list, newest_event_time ISO string) """ endpoint = f"{TENANT_URL}/v1.0/events" headers = { 'Authorization': f'Bearer {token}', 'Accept': 'application/json', 'User-Agent': 'GoogleSecOps-IBMVerifyCollector/1.0' } records = [] newest_time = None page_num = 0 backoff = 1.0 start_iso = start_time.strftime('%Y-%m-%dT%H:%M:%S.000Z') end_iso = end_time.strftime('%Y-%m-%dT%H:%M:%S.000Z') # IBM Verify Events API uses filter and sort_order with pagination search_after = None while True: page_num += 1 if len(records) >= max_records: print(f"Reached max_records limit ({max_records})") break params = [] params.append(f"size={min(page_size, max_records - len(records))}") params.append(f"filter=time+ge+\"{start_iso}\"+and+time+le+\"{end_iso}\"") params.append("sort_order=asc") if search_after: params.append(f"search_after={search_after}") url = f"{endpoint}?{'&'.join(params)}" try: response = http.request('GET', url, headers=headers) 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}") return [], None data = json.loads(response.data.decode('utf-8')) page_results = data.get('events', []) if not page_results: print(f"No more results (empty page)") break print(f"Page {page_num}: Retrieved {len(page_results)} events") records.extend(page_results) # Track newest event time for event in page_results: try: event_time = event.get('time') if event_time: if newest_time is None or parse_datetime(event_time) > parse_datetime(newest_time): newest_time = event_time except Exception as e: print(f"Warning: Could not parse event time: {e}") # Check for more results using search_after pagination if len(page_results) < page_size: print(f"Reached last page (size={len(page_results)} < limit={page_size})") break # Use the last event's sort value for pagination last_event = page_results[-1] search_after = last_event.get('time', '') except Exception as e: print(f"Error fetching logs: {e}") return [], None print(f"Retrieved {len(records)} total records from {page_num} pages") return records, newest_time第二個檔案 - 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-verify-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取主題 ibm-verify-trigger郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間要求選擇頻率:
| 頻率 | Cron 運算式 | 用途 |
|---|---|---|
| 每 5 分鐘 | */5 * * * * |
大量低延遲 |
| 每 15 分鐘 | */15 * * * * |
中等 |
| 每小時 | 0 * * * * |
標準 (建議) |
| 每 6 小時 | 0 */6 * * * |
少量、批次處理 |
| 每日 | 0 0 * * * |
歷來資料集合 |
測試整合項目
- 在 Cloud Scheduler 控制台中,找出您的工作 (
ibm-verify-collector-hourly)。 - 按一下「強制執行」即可手動觸發。
- 等待幾秒鐘,然後依序前往「Cloud Run」>「Services」(服務) >「ibm-verify-collector」>「Logs」(記錄)。
確認函式是否已順利執行。尋找:
Fetching logs from YYYY-MM-DDTHH:MM:SS+00:00 to YYYY-MM-DDTHH:MM:SS+00:00 Page 1: Retrieved X events Wrote X records to gs://ibm-verify-saas-logs/ibm-verify/logs_YYYYMMDD_HHMMSS.ndjson Successfully processed X records檢查 GCS 值區 (
ibm-verify-saas-logs),確認記錄是否已寫入。
如果記錄中顯示錯誤:
- HTTP 401:檢查環境變數中的 API 憑證
- HTTP 403:確認 API 用戶端在 IBM Security Verify 管理控制台中具備必要權限
- HTTP 429:頻率限制 - 函式會自動重試,並延遲一段時間
- 無法取得存取權杖:確認
TENANT_URL、CLIENT_ID和CLIENT_SECRET正確無誤
在 Google SecOps 中設定動態饋給,擷取 IBM Security Verify SaaS 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
IBM Security Verify SaaS Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
- 選取「IBM Security Verify SaaS」做為「記錄類型」。
按一下「取得服務帳戶」。系統會顯示專屬服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,後續步驟會用到。
點選「下一步」。
指定下列輸入參數的值:
儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:
gs://ibm-verify-saas-logs/ibm-verify/- 取代:
ibm-verify-saas-logs:您的 GCS bucket 名稱。ibm-verify:儲存記錄的選用前置字元/資料夾路徑 (如為根目錄,請留空)。
- 取代:
來源刪除選項:根據偏好設定選取刪除選項:
- 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。
檔案存在時間上限:包含在過去天數內修改的檔案 (預設為 180 天)。
資產命名空間:資產命名空間。
擷取標籤:要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (
ibm-verify-saas-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色:選取「Storage 物件檢視者」。
按一下 [儲存]。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| data.realm、geoip.continent_name、data.applicationtype、data.id_token、data.grant_id、data.grant_type、data.at_hash、data.rt_hash、id、data.scope、data.uasessionid、data.performedby_type、data.subject_type、data.target_type、data.templateid、data.subjectid、data.performedby、data.action、data.performedby_clientname、geoip.as_org、geoip.country_iso_code、operation.op、operation.path、operation.value、operation.value.name、operation.value.values | additional.fields | 標準 UDM 欄位未涵蓋的其他中繼資料欄位 |
| auth_type | extensions.auth.type | 使用的驗證類型 |
| meta_event_type | metadata.event_type | 事件類型 (例如 USER_LOGIN、NETWORK_CONNECTION) |
| tenantid | metadata.product_deployment_id | 產品部署的 ID |
| event_type | metadata.product_event_type | 產品專屬事件類型 |
| data.ib_request_id | metadata.product_log_id | 產品專屬記錄 ID |
| data.devicetype | network.http.parsed_user_agent | 已剖析的使用者代理程式字串 |
| data.devicetype | network.http.user_agent | 使用者代理程式字串 |
| data.sessionid | network.session_id | 工作階段 ID |
| tenantname | observer.hostname | 觀察者的主機名稱 |
| data.applicationname | principal.application | 與主體相關聯的應用程式 |
| data.origin、geoip.ip | principal.ip | 主體的 IP 位址 |
| geoip.city_name | principal.location.city | 負責人所在地城市 |
| geoip.country_name | principal.location.country_or_region | 主要負責人所在國家/地區 |
| geoip.location.lat | principal.location.region_latitude | 主體所在位置的緯度 |
| geoip.location.lon | principal.location.region_longitude | 負責人所在位置的經度 |
| geoip.region_name | principal.location.state | 主體所在地點的州/省 |
| data.applicationid | principal.resource.product_object_id | 資源的產品物件 ID |
| data.subtype | principal.resource.resource_subtype | 資源的子類型 |
| data.redirecturl | principal.url | 與負責人相關聯的網址 |
| data.username | principal.user.user_display_name | 使用者的顯示名稱 |
| data.userid | principal.user.userid | 使用者 ID |
| security_result | security_result | 安全性結果資訊 |
| data.client_id、data.client_name、data.client_type | security_result.about.resource.attribute.labels | 資源屬性的標籤 |
| action1 | security_result.action | 保全系統採取的動作 |
| data.result | security_result.action_details | 安全性動作的詳細資料 |
| data.status_code、correlationid | security_result.detection_fields | 與偵測相關的欄位 |
| servicename | target.application | 指定應用程式 |
| data.targetid | target.resource.id | 目標資源的 ID |
| data.account_name | target.user.user_display_name | 目標使用者的顯示名稱 |
| data.target | target.user.userid | 目標的使用者 ID |
| metadata.product_name | 產品名稱 | |
| metadata.vendor_name | 供應商/公司名稱 | |
data.status_code", "data.performedby_type", "data.subject_type", "data.target_type", "data.templateid", "data.subjectid", "data.performedby", "data.action", and "data.performedby_clientname |
additional.fields |
從變更記錄對應 |
data.cause |
security_result.summary |
從變更記錄對應 |
data.json.Operations.path", "data.json.Operations.op", and "data.json.Operations.value |
additional.fields |
從變更記錄對應 |
geoip.as_org", "geoip.country_iso_code |
additional.fields |
從變更記錄對應 |
變更記錄
還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求答案。