收集 ForgeRock OpenIDM 記錄
本文說明如何使用 Google Cloud Storage V2,將 ForgeRock OpenIDM 記錄檔擷取至 Google Security Operations。
ForgeRock OpenIDM (現已更名為 PingIDM) 是一個身分管理平台,提供使用者佈建、同步、密碼管理和存取控管功能。這項服務會將身分生命週期事件、驗證嘗試、對帳作業和設定變更記錄到稽核記錄中,並透過 REST 提供存取權。
事前準備
請確認您已完成下列事前準備事項:
- Google SecOps 執行個體
- 已啟用 Cloud Storage API 的 GCP 專案
- 建立及管理 GCS 值區的權限
- 管理 Google Cloud Storage 值區 IAM 政策的權限
- 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
- 具備管理員憑證,可存取 ForgeRock OpenIDM 或 PingIDM 執行個體的特殊權限
建立 Google Cloud Storage 值區
- 前往 Google Cloud Console。
- 選取專案或建立新專案。
- 在導覽選單中,依序前往「Cloud Storage」>「Bucket」。
- 按一下「建立值區」。
請提供下列設定詳細資料:
設定 值 為 bucket 命名 輸入全域不重複的名稱 (例如 forgerock-openidm-audit-logs)位置類型 根據需求選擇 (區域、雙區域、多區域) 位置 選取位置 (例如 us-central1)儲存空間級別 標準 (建議用於經常存取的記錄) 存取控管 統一 (建議) 保護工具 選用:啟用物件版本管理或保留政策 點選「建立」。
收集 ForgeRock OpenIDM 憑證
取得 ForgeRock OpenIDM 基準網址
- 登入 ForgeRock OpenIDM 或 PingIDM 執行個體。
- 記下瀏覽器網址列中的基本網址。
- 格式:
https://openidm.example.com - 請勿加入結尾斜線或路徑,例如
/admin
- 格式:
取得管理員憑證
- 取得 ForgeRock OpenIDM 執行個體的管理員憑證。
你需要:
- 使用者名稱:管理員使用者名稱 (例如
openidm-admin) - 密碼:管理員密碼
- 使用者名稱:管理員使用者名稱 (例如
驗證權限
如要確認帳戶是否具備必要權限,請按照下列步驟操作:
- 登入 ForgeRock OpenIDM。
- 依序前往「設定」>「系統偏好設定」>「稽核」。
- 如果可以查看稽核設定和主題,表示您具備必要權限。
- 如果沒有看到這個選項,請與管理員聯絡,要求對方授予稽核讀取權限。
測試 API 存取權
請先測試憑證,再繼續進行整合:
# Replace with your actual credentials
OPENIDM_BASE_URL="https://openidm.example.com"
OPENIDM_USERNAME="openidm-admin"
OPENIDM_PASSWORD="your-admin-password"
# Test API access to authentication audit topic
curl -v \
-H "X-OpenIDM-Username: ${OPENIDM_USERNAME}" \
-H "X-OpenIDM-Password: ${OPENIDM_PASSWORD}" \
-H "Accept-API-Version: resource=1.0" \
-H "Accept: application/json" \
"${OPENIDM_BASE_URL}/openidm/audit/authentication?_queryFilter=true&_pageSize=1"
預期回應:HTTP 200,並包含稽核事件的 JSON。
為 Cloud Run 函式建立服務帳戶
Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶,並由 Pub/Sub 叫用。
建立服務帳戶
- 在 GCP 主控台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
- 按一下「Create Service Account」(建立服務帳戶)。
- 請提供下列設定詳細資料:
- 服務帳戶名稱:輸入
forgerock-openidm-collector-sa。 - 服務帳戶說明:輸入
Service account for Cloud Run function to collect ForgeRock OpenIDM logs。
- 服務帳戶名稱:輸入
- 按一下「建立並繼續」。
- 在「將專案存取權授予這個服務帳戶」部分,新增下列角色:
- 按一下「選擇角色」。
- 搜尋並選取「Storage 物件管理員」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Run Invoker」。
- 點選「+ 新增其他角色」。
- 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)。
- 按一下「繼續」。
- 按一下 [完成]。
這些角色適用於:
- Storage 物件管理員:將記錄檔寫入 GCS 值區,並管理狀態檔案
- Cloud Run 叫用者:允許 Pub/Sub 叫用函式
- Cloud Functions 叫用者:允許函式叫用
授予 GCS 值區的 IAM 權限
授予服務帳戶 GCS bucket 的寫入權限:
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱 (例如
forgerock-openidm-audit-logs)。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:輸入服務帳戶電子郵件地址 (例如
forgerock-openidm-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。 - 指派角色:選取「Storage 物件管理員」。
- 新增主體:輸入服務帳戶電子郵件地址 (例如
- 按一下 [儲存]。
建立 Pub/Sub 主題
建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。
- 在 GCP Console 中,前往「Pub/Sub」>「Topics」(主題)。
- 按一下「建立主題」。
- 請提供下列設定詳細資料:
- 主題 ID:輸入
forgerock-openidm-trigger。 - 其他設定保留預設值。
- 主題 ID:輸入
- 點選「建立」。
建立 Cloud Run 函式來收集記錄
Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 ForgeRock OpenIDM API 擷取記錄,並寫入 GCS。
- 前往 GCP Console 的「Cloud Run」。
- 按一下「Create service」(建立服務)。
- 選取「函式」 (使用內嵌編輯器建立函式)。
在「設定」部分,提供下列設定詳細資料:
設定 值 服務名稱 forgerock-openidm-collector區域 選取與 GCS bucket 相符的區域 (例如 us-central1)執行階段 選取「Python 3.12」以上版本 在「Trigger (optional)」(觸發條件 (選用)) 專區:
- 按一下「+ 新增觸發條件」。
- 選取「Cloud Pub/Sub」。
- 在「選取 Cloud Pub/Sub 主題」中,選擇 Pub/Sub 主題
forgerock-openidm-trigger。 - 按一下 [儲存]。
在「Authentication」(驗證) 部分:
- 選取「需要驗證」。
- 檢查 Identity and Access Management (IAM)。
捲動至「Containers, Networking, Security」(容器、網路、安全性) 並展開。
前往「安全性」分頁:
- 服務帳戶:選取服務帳戶
forgerock-openidm-collector-sa。
- 服務帳戶:選取服務帳戶
前往「容器」分頁:
- 按一下「變數與密鑰」。
針對每個環境變數,按一下「+ 新增變數」:
變數名稱 範例值 說明 GCS_BUCKETforgerock-openidm-audit-logsGCS bucket 名稱 GCS_PREFIXopenidm記錄檔的前置字串 STATE_KEYopenidm/state.json狀態檔案路徑 OPENIDM_BASE_URLhttps://openidm.example.comOpenIDM 基準網址 OPENIDM_USERNAMEopenidm-adminOpenIDM 管理員使用者名稱 OPENIDM_PASSWORDyour-admin-passwordOpenIDM 管理員密碼 AUDIT_TOPICSaccess,activity,authentication,config,sync以半形逗號分隔的稽核主題 PAGE_SIZE100每頁記錄數 MAX_PAGES50每個主題的頁數上限
在「變數與密鑰」部分,向下捲動至「要求」:
- 要求逾時:輸入
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 # 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', 'openidm') STATE_KEY = os.environ.get('STATE_KEY', 'openidm/state.json') OPENIDM_BASE_URL = os.environ.get('OPENIDM_BASE_URL') OPENIDM_USERNAME = os.environ.get('OPENIDM_USERNAME') OPENIDM_PASSWORD = os.environ.get('OPENIDM_PASSWORD') AUDIT_TOPICS = os.environ.get('AUDIT_TOPICS', 'access,activity,authentication,config,sync').split(',') PAGE_SIZE = int(os.environ.get('PAGE_SIZE', '100')) MAX_PAGES = int(os.environ.get('MAX_PAGES', '50')) @functions_framework.cloud_event def main(cloud_event): """ Cloud Run function triggered by Pub/Sub to fetch ForgeRock OpenIDM logs and write to GCS. Args: cloud_event: CloudEvent object containing Pub/Sub message """ if not all([GCS_BUCKET, OPENIDM_BASE_URL, OPENIDM_USERNAME, OPENIDM_PASSWORD]): print('Error: Missing required environment variables') return try: # Get GCS bucket bucket = storage_client.bucket(GCS_BUCKET) # Load state state = load_state(bucket, STATE_KEY) all_events = [] for topic in AUDIT_TOPICS: topic = topic.strip() print(f"Fetching audit logs for topic: {topic}") events = fetch_audit_logs(topic, state.get(topic, {})) all_events.extend(events) if events: latest_timestamp = max(e.get('timestamp', '') for e in events) state[topic] = { 'last_timestamp': latest_timestamp, 'last_run': datetime.now(timezone.utc).isoformat(), 'events_count': len(events) } if all_events: write_to_gcs(bucket, all_events) save_state(bucket, STATE_KEY, state) print(f"Successfully processed {len(all_events)} audit events") else: print("No new audit events to process") 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, state): """Save state to GCS.""" try: blob = bucket.blob(key) blob.upload_from_string( json.dumps(state, indent=2), content_type='application/json' ) print(f"Saved state: {json.dumps(state)}") except Exception as e: print(f"Warning: Could not save state: {e}") def fetch_audit_logs(topic, topic_state): """ Fetch audit logs from ForgeRock OpenIDM API with pagination. Args: topic: Audit topic name topic_state: State dictionary for this topic Returns: List of audit events """ base_url = OPENIDM_BASE_URL.rstrip('/') all_events = [] last_timestamp = topic_state.get('last_timestamp') query_filter = 'true' if last_timestamp: query_filter = f'timestamp gt "{last_timestamp}"' page_offset = 0 page_count = 0 while page_count < MAX_PAGES: try: url = f"{base_url}/openidm/audit/{topic}" params = { '_queryFilter': query_filter, '_pageSize': str(PAGE_SIZE), '_pagedResultsOffset': str(page_offset), '_sortKeys': 'timestamp' } query_string = '&'.join([f"{k}={urllib3.request.urlencode({k: v})[len(k)+1:]}" for k, v in params.items()]) full_url = f"{url}?{query_string}" headers = { 'X-OpenIDM-Username': OPENIDM_USERNAME, 'X-OpenIDM-Password': OPENIDM_PASSWORD, 'Accept-API-Version': 'resource=1.0', 'Accept': 'application/json' } response = http.request('GET', full_url, headers=headers) if response.status != 200: print(f"API error for topic {topic}: {response.status} - {response.data.decode('utf-8')}") break data = json.loads(response.data.decode('utf-8')) events = data.get('result', []) if not events: print(f"No more events for topic {topic}") break all_events.extend(events) page_offset += PAGE_SIZE page_count += 1 print(f"Fetched page {page_count} for topic {topic}, total events: {len(all_events)}") if len(events) < PAGE_SIZE: break except urllib3.exceptions.HTTPError as e: print(f"HTTP error for topic {topic}: {str(e)}") break except json.JSONDecodeError as e: print(f"JSON decode error for topic {topic}: {str(e)}") break except Exception as e: print(f"Unexpected error for topic {topic}: {str(e)}") break return all_events def write_to_gcs(bucket, events): """Write events to GCS as NDJSON.""" timestamp = datetime.now(timezone.utc).strftime('%Y%m%d_%H%M%S') filename = f"{GCS_PREFIX}/openidm_audit_{timestamp}.json" ndjson_content = '\n'.join([json.dumps(event) for event in events]) blob = bucket.blob(filename) blob.upload_from_string( ndjson_content.encode('utf-8'), content_type='application/x-ndjson' ) print(f"Wrote {len(events)} events to gs://{GCS_BUCKET}/{filename}")- 第二個檔案: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」。
- 點選「建立工作」。
請提供下列設定詳細資料:
設定 值 名稱 forgerock-openidm-collector-hourly區域 選取與 Cloud Run 函式相同的區域 頻率 0 * * * *(每小時整點)時區 選取時區 (建議使用世界標準時間) 目標類型 Pub/Sub 主題 選取 Pub/Sub 主題 forgerock-openidm-trigger郵件內文 {}(空白 JSON 物件)點選「建立」。
排程頻率選項
根據記錄檔量和延遲時間要求選擇頻率:
| 頻率 | Cron 運算式 | 用途 |
|---|---|---|
| 每 5 分鐘 | */5 * * * * |
高容量、低延遲 |
| 每 15 分鐘 | */15 * * * * |
中等 |
| 每小時 | 0 * * * * |
標準 (建議) |
| 每 6 小時 | 0 */6 * * * |
少量、批次處理 |
| 每日 | 0 0 * * * |
歷來資料集合 |
測試整合項目
- 在 Cloud Scheduler 控制台中,找出您的工作
forgerock-openidm-collector-hourly。 - 按一下「強制執行」,手動觸發工作。
- 稍等幾秒鐘。
- 前往「Cloud Run」>「Services」。
- 按一下函式名稱
forgerock-openidm-collector。 - 按一下 [Logs] (記錄) 分頁標籤。
確認函式是否已順利執行。尋找:
Fetching audit logs for topic: access Fetched page 1 for topic access, total events: X Wrote X events to gs://forgerock-openidm-audit-logs/openidm/openidm_audit_YYYYMMDD_HHMMSS.json Successfully processed X audit events依序前往「Cloud Storage」>「Buckets」。
按一下 bucket 名稱
forgerock-openidm-audit-logs。前往前置字元資料夾
openidm/。確認是否已建立含有目前時間戳記的新
.json檔案。
如果在記錄中發現錯誤:
- HTTP 401:檢查環境變數中的 OpenIDM 憑證
- HTTP 403:確認帳戶是否具備稽核讀取權限
- HTTP 404:確認 OPENIDM_BASE_URL 正確無誤,且不包含尾端路徑
- 缺少環境變數:檢查是否已設定所有必要變數
擷取 Google SecOps 服務帳戶
Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。
在 Google SecOps 中設定動態饋給,擷取 ForgeRock OpenIDM 記錄
- 依序前往「SIEM 設定」>「動態饋給」。
- 按一下「新增動態消息」。
- 按一下「設定單一動態饋給」。
- 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如
ForgeRock OpenIDM Audit Logs)。 - 選取「Google Cloud Storage V2」做為「來源類型」。
選取「FORGEROCK_OPENIDM」做為「記錄類型」。
按一下「取得服務帳戶」。
系統會顯示不重複的服務帳戶電子郵件地址,例如:
chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com複製這個電子郵件地址,後續步驟會用到。
點選「下一步」。
指定下列輸入參數的值:
- 儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:
gs://forgerock-openidm-audit-logs/openidm/- 取代:
forgerock-openidm-audit-logs:您的 GCS bucket 名稱。openidm:儲存記錄的前置路徑。
- 來源刪除選項:根據偏好選取刪除選項:
- 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
- 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
- 刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。
- 檔案存在時間上限:包含在過去天數內修改的檔案。預設值為 180 天。
- 資產命名空間:資產命名空間。
- 擷取標籤:要套用至這個動態饋給事件的標籤。
點選「下一步」。
在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)。
將 IAM 權限授予 Google SecOps 服務帳戶
Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。
- 依序前往「Cloud Storage」>「Buckets」。
- 按一下 bucket 名稱
forgerock-openidm-audit-logs。 - 前往「權限」分頁標籤。
- 按一下「授予存取權」。
- 請提供下列設定詳細資料:
- 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址。
- 指派角色:選取「Storage 物件檢視者」。
按一下 [儲存]。
UDM 對應表
| 記錄欄位 | UDM 對應 | 邏輯 |
|---|---|---|
| additional_label、additional_elapsed_time、additional_ContentLength、additional_accept_encoding、additional_Accept、additional_accept_language、additional_origin_hop、additional_cache_control、additional_Connection、additional_Cookie、additional_Pragma、additional_exchange_id、additional_contentType、additional_X-content_type-Options、fluenttag_label、source_label、topic_label、request_protocol_label、taskName_label、linkQualifier_label、situation_label、mapping_label、eventid_label、context_roles_label、field_names_label | additional.fields | 其他鍵/值組合 |
| Via | intermediary.hostname | 中介主機名稱 |
| x_forwarded_ip、ip、caller.callerIps | intermediary.ip | 中介裝置的 IP 位址 |
| 時間戳記 | metadata.event_timestamp | 事件時間戳記 |
| metadata.event_type | 活動類型 | |
| transactionId | metadata.product_deployment_id | 產品的部署 ID |
| eventName | metadata.product_event_type | 產品的事件類型 |
| _id、trackingIds | metadata.product_log_id | 產品的記錄 ID |
| http.request.secure | network.application_protocol | 應用程式通訊協定 |
| http_version | network.application_protocol_version | 應用程式通訊協定版本 |
| request_method、http.request.method | network.http.method | HTTP 方法 |
| user_agent、http.request.headers.user_agent.0 | network.http.parsed_user_agent | 已剖析的使用者代理程式 |
| refferal_url | network.http.referral_url | 參照連結網址 |
| response.statusCode、status_code | network.http.response_code | HTTP 回應碼 |
| user_agent、http.request.headers.user_agent | network.http.user_agent | 使用者代理程式字串 |
| transaction_id、transactionId | network.session_id | 工作階段 ID |
| 主機 | principal.asset.hostname | 主體資產的主機名稱 |
| true_client_ip、client.ip、context.ipAddress、entry.info.ipAddress、src_ip | principal.asset.ip | 主體資產的 IP 位址 |
| 主機 | principal.hostname | 主體的主機名稱 |
| true_client_ip、client.ip、context.ipAddress、entry.info.ipAddress、src_ip | principal.ip | 主體的 IP 位址 |
| client.port、src_port | principal.port | 主體通訊埠 |
| component_label、moduleId_label、query_id_label | principal.resource.attribute.labels | 主體資源的屬性標籤 |
| entry.info.treeName | principal.resource.name | 主體資源名稱 |
| sourceObjectId、objectId、entry.info.nodeId | principal.resource.product_object_id | 主體資源的產品物件 ID |
| entry.info.authLevel | principal.resource.resource_subtype | 主體資源的子類型 |
| user_roles_property_label、authentication_id_label、authentication_id_property_label | principal.user.attribute.labels | 主體使用者的屬性標籤 |
| userId、principalData.0 | principal.user.userid | 主體的使用者 ID |
| security_action | security_result.action | 安全性結果中採取的動作 |
| 結果、動作 | security_result.action_details | 動作的詳細資料 |
| result_label、moduleId_label、nodeType_label、displayName_label、nodeOutcome_label、elapsedTimeUnits_label、elapsedTime_label、operation_label、taskName_label、linkQualifier_label、situation_label、mapping_label | security_result.detection_fields | 偵測欄位 |
| level | security_result.severity | 安全性結果的嚴重程度 |
| level | security_result.severity_details | 嚴重程度詳細資料 |
| response_detail_reason | security_result.summary | 安全性結果摘要 |
| http.request.headers.host.0 | target.asset.hostname | 目標資產的主機名稱 |
| server.ip、x_forwarded_ip | target.asset.ip | 目標資產的 IP 位址 |
| http.request.headers.host.0 | target.hostname | 目標主機名稱 |
| server.ip、x_forwarded_ip | target.ip | 目標的 IP 位址 |
| server.port | target.port | 目標通訊埠 |
| targetObjectId | target.resource.product_object_id | 目標資源的產品物件 ID |
| http.request.path | target.url | 目標網址 |
| metadata.product_name | 產品名稱 | |
| metadata.vendor_name | 供應商名稱 |
還有其他問題嗎?向社群成員和 Google SecOps 專業人員尋求答案。