收集 PingIDM (原名 ForgeRock OpenIDM) 記錄

剖析器版本：4.0

本文說明如何使用 Google Cloud Storage V2，將 PingIDM (舊稱 ForgeRock OpenIDM) 記錄檔擷取至 Google Security Operations。

PingIDM 是身分管理平台，提供使用者佈建、同步處理、密碼管理和存取控管功能。這項服務會將身分生命週期事件、驗證嘗試、對帳作業和設定變更記錄到稽核記錄中，並透過 REST 提供存取權。

事前準備

請確認您已完成下列事前準備事項：

• Google SecOps 執行個體。
• 已啟用 Cloud Storage API 的 GCP 專案。
• 具備建立及管理 GCS 值區和 IAM 政策的權限。
• 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限。
• 具備管理員憑證的 ForgeRock OpenIDM 或 PingIDM 執行個體特殊存取權。

建立 Google Cloud Storage bucket

1. 前往 Google Cloud 控制台。
2. 選取專案或建立新專案。
3. 在導覽選單中，依序前往「Cloud Storage」>「bucket」。
4. 按一下「建立值區」。

5. 請提供下列設定詳細資料：

   設定	值
   為 bucket 命名	輸入全域不重複的名稱 (例如 forgerock-openidm-audit-logs)
   位置類型	根據需求選擇 (區域、雙區域、多區域)
   位置	選取位置 (例如 us-central1)
   儲存空間級別	標準 (建議用於經常存取的記錄)
   存取控管	統一 (建議)
   保護工具	選用：啟用物件版本管理或保留政策

6. 點選「建立」。

收集 ForgeRock OpenIDM 憑證

取得 ForgeRock OpenIDM 基準網址

1. 登入 ForgeRock OpenIDM 或 PingIDM 執行個體。
2. 記下瀏覽器網址列中的基本網址。
   • 格式：https://openidm.example.com
   • 請勿加入結尾斜線或路徑，例如 /admin。

取得管理員憑證

1. 取得 ForgeRock OpenIDM 執行個體的管理員憑證。
2. 你需要：
   • 使用者名稱：管理員使用者名稱 (例如 openidm-admin)
   • 密碼：管理員密碼

驗證權限

1. 登入 ForgeRock OpenIDM。
2. 依序前往「設定」>「系統偏好設定」>「稽核」。
3. 如果可以查看稽核設定和主題，表示您具備必要權限。
4. 如果沒有看到這個選項，請與管理員聯絡，要求對方授予稽核讀取權限。

測試 API 存取權

• 請先測試憑證，再繼續進行整合：

  # Replace with your actual credentials
  OPENIDM_BASE_URL="[https://openidm.example.com](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 叫用。

建立服務帳戶

1. 在 GCP 主控台中，依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)。
2. 按一下「Create Service Account」(建立服務帳戶)。
3. 請提供下列設定詳細資料：
   • 服務帳戶名稱：forgerock-openidm-collector-sa
   • 服務帳戶說明：Service account for Cloud Run function to collect ForgeRock OpenIDM logs
4. 按一下「建立並繼續」。
5. 在「將專案存取權授予這個服務帳戶」部分，新增下列角色：
   1. Storage 物件管理員：將記錄檔寫入 GCS bucket，並管理狀態檔案。
   2. Cloud Run 叫用者：允許 Pub/Sub 叫用函式。
   3. Cloud Functions 叫用者：允許叫用函式。
6. 然後依序點選「繼續」和「完成」。

授予 GCS 值區的 IAM 權限

1. 依序前往「Cloud Storage」>「Buckets」。
2. 點選 bucket 名稱 (例如 forgerock-openidm-audit-logs)。
3. 前往「權限」分頁標籤。
4. 按一下「授予存取權」。
5. 請提供下列設定詳細資料：
   • 新增主體：輸入服務帳戶電子郵件地址 (例如 forgerock-openidm-collector-sa@PROJECT_ID.iam.gserviceaccount.com)。
   • 指派角色：選取「Storage 物件管理員」。
6. 按一下 [儲存]。

建立 Pub/Sub 主題

1. 在 GCP 控制台中，前往「Pub/Sub」>「主題」。
2. 按一下「建立主題」。
3. 請提供下列設定詳細資料：
   • 主題 ID：forgerock-openidm-trigger
   • 其他設定保留預設值。
4. 點選「建立」。

建立 Cloud Run 函式來收集記錄

1. 前往 GCP Console 的「Cloud Run」。
2. 按一下「Create service」(建立服務)。
3. 選取「函式」 (使用內嵌編輯器建立函式)。

4. 在「設定」部分，提供下列設定詳細資料：

   設定	值
   服務名稱	forgerock-openidm-collector
   區域	選取與 GCS bucket 相符的區域 (例如 us-central1)
   執行階段	選取 Python 3.12 以上版本

5. 在「Trigger」(觸發條件) 專區：

   1. 按一下「+ 新增觸發條件」。
   2. 選取「Cloud Pub/Sub」。
   3. 在「Select a Cloud Pub/Sub topic」(選取 Cloud Pub/Sub 主題) 中，選擇 forgerock-openidm-trigger。
   4. 按一下 [儲存]。

6. 在「Authentication」(驗證) 部分：

   1. 選取「需要驗證」。
   2. 檢查 Identity and Access Management (IAM)。

7. 向下捲動並展開「容器」、「網路」和「安全性」。

8. 前往「安全性」分頁：

   • 服務帳戶：選取 forgerock-openidm-collector-sa。

9. 前往「容器」分頁：

   1. 按一下「變數與密鑰」。
   2. 針對每個環境變數，按一下「+ 新增變數」：

   變數名稱	範例值	說明
   GCS_BUCKET	forgerock-openidm-audit-logs	GCS bucket 名稱
   GCS_PREFIX	openidm	記錄檔的前置字串
   STATE_KEY	openidm/state.json	狀態檔案路徑
   OPENIDM_BASE_URL	https://openidm.example.com	OpenIDM 基準網址
   OPENIDM_USERNAME	openidm-admin	OpenIDM 管理員使用者名稱
   OPENIDM_PASSWORD	your-admin-password	OpenIDM 管理員密碼
   AUDIT_TOPICS	access,activity,authentication,config,sync	以半形逗號分隔的主題
   PAGE_SIZE	100	每頁記錄數
   MAX_PAGES	50	每個主題的頁數上限

10. 在「要求」部分中：

    • 要求逾時：600 秒。

11. 前往「設定」分頁：

    • 資源：512 MiB 記憶體、1 個 CPU。

12. 點按「Create」(建立)。服務建立完成後，系統會開啟內嵌程式碼編輯器。

新增函式程式碼

1. 在「進入點」欄位中輸入「main」。

2. 在編輯器中建立兩個檔案：

   • 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.
       """
       if not all([GCS_BUCKET, OPENIDM_BASE_URL, OPENIDM_USERNAME, OPENIDM_PASSWORD]):
           print('Error: Missing required environment variables')
           return
   
       try:
           bucket = storage_client.bucket(GCS_BUCKET)
           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):
       try:
           blob = bucket.blob(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, key, state):
       try:
           blob = bucket.blob(key)
           blob.upload_from_string(json.dumps(state, indent=2), content_type='application/json')
       except Exception as e:
           print(f"Warning: Could not save state: {e}")
   
   def fetch_audit_logs(topic, topic_state):
       base_url = OPENIDM_BASE_URL.rstrip('/')
       all_events = []
       last_timestamp = topic_state.get('last_timestamp')
       query_filter = 'true' if not last_timestamp else 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'
               }
   
               headers = {
                   'X-OpenIDM-Username': OPENIDM_USERNAME,
                   'X-OpenIDM-Password': OPENIDM_PASSWORD,
                   'Accept-API-Version': 'resource=1.0',
                   'Accept': 'application/json'
               }
   
               response = http.request('GET', url, fields=params, headers=headers)
               if response.status != 200:
                   print(f"API error for topic {topic}: {response.status}")
                   break
   
               data = json.loads(response.data.decode('utf-8'))
               events = data.get('result', [])
               if not events:
                   break
   
               all_events.extend(events)
               page_offset += PAGE_SIZE
               page_count += 1
               if len(events) < PAGE_SIZE:
                   break
           except Exception as e:
               print(f"Error for topic {topic}: {str(e)}")
               break
       return all_events
   
   def write_to_gcs(bucket, events):
       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])
       bucket.blob(filename).upload_from_string(ndjson_content, content_type='application/x-ndjson')
   

   • requirements.txt:

   functions-framework==3.*
   google-cloud-storage==2.*
   urllib3>=2.0.0
   

3. 按一下「Deploy」(部署)。部署作業需要 2 到 3 分鐘。

建立 Cloud Scheduler 工作

1. 前往 GCP 主控台的「Cloud Scheduler」。
2. 點選「建立工作」。

3. 請提供下列設定詳細資料：

   設定	值
   名稱	forgerock-openidm-collector-hourly
   區域	選取與 Cloud Run 函式相同的區域
   頻率	0 * * * * (每小時)
   時區	選取時區 (建議使用世界標準時間)
   目標類型	Pub/Sub
   主題	forgerock-openidm-trigger
   郵件內文	{}

4. 點選「建立」。

排程頻率選項

頻率	Cron 運算式	用途
每 5 分鐘	*/5 * * * *	大量、低延遲
每小時	0 * * * *	標準 (建議)
每日	0 0 * * *	批次收集

測試整合項目

1. 在 Cloud Scheduler 中，按一下工作的「強制執行」。
2. 在 Cloud Run 記錄中驗證執行作業。找出：Successfully processed X audit events。
3. 確認 Cloud Storage bucket 的 openidm/ 資料夾中，是否出現新的 json 檔案。

擷取 Google SecOps 服務帳戶

在 Google SecOps 中設定動態饋給

1. 依序前往「SIEM 設定」>「動態消息」。
2. 依序點選「新增動態消息」>「設定單一動態消息」。
3. 動態饋給名稱：ForgeRock OpenIDM Audit Logs。
4. 來源類型：Google Cloud Storage V2。
5. 記錄類型：FORGEROCK_OPENIDM。
6. 按一下「取得服務帳戶」，然後複製提供的電子郵件地址 (例如 chronicle-12345...)。
7. 點選「下一步」。
8. 儲存空間值區網址：gs://forgerock-openidm-audit-logs/openidm/。
9. 來源刪除選項：根據偏好設定選取 (例如「永不」)。
10. 點選「下一步」，檢查內容，然後點選「提交」。

授予 IAM 權限

1. 依序前往「Cloud Storage」>「Buckets」。
2. 選取 bucket，然後按一下「Permissions」(權限) 分頁標籤。
3. 按一下「授予存取權」。
4. 新增主體：貼上 Google SecOps 服務帳戶電子郵件地址。
5. 指派角色：選取「Storage 物件檢視者」。
6. 按一下 [儲存]。

UDM 對應表

記錄欄位	UDM 對應	邏輯
fluenttag_label、topic_label、context_roles_label	additional.fields	內容的鍵/值組合。
Via	intermediary.hostname	Proxy/中介主機名稱。
x_forwarded_ip、caller.callerIps	intermediary.ip	中介機構的 IP。
timestamp	metadata.event_timestamp	事件時間戳記。
transactionId	metadata.product_deployment_id	專屬交易 ID。
eventName	metadata.product_event_type	ForgeRock 的事件名稱。
_id	metadata.product_log_id	不重複的記錄 ID。
request_method	network.http.method	使用的 HTTP 方法。
response.statusCode	network.http.response_code	HTTP 狀態碼。
user_agent	network.http.user_agent	完整的使用者代理程式字串。
client.ip、context.ipAddress	principal.ip	執行者的來源 IP。
userId、principalData.0	principal.user.userid	執行動作的使用者 ID。
security_action	security_result.action	已採取安全性措施。
level	security_result.severity	對應的嚴重性等級。
server.ip	target.ip	目標系統的 IP。
http.request.path	target.url	目標網址/路徑。
不適用	metadata.product_name	設為 ForgeRock OpenIDM。
不適用	metadata.vendor_name	設為 ForgeRock。
elapsed_time" and "X-Content-Type-Options	additional.fields	從變更記錄對應
trackingIds.0	metadata.product_log_id	從變更記錄對應
context.userRolesProperty", "context.authenticationId", and "context.authenticationIdProperty	principal.user.attribute.roles	從變更記錄對應
context.component", "context.moduleId", and "context.queryId	principal.resource.attribute.labels	從變更記錄對應
context.roles", "_normalized.fieldNames", and "_eventId	additional.fields	從變更記錄對應

變更記錄

查看這個剖析器的變更記錄

還有其他問題嗎？向社群成員和 Google SecOps 專業人員尋求答案。