收集 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 值區

  1. 前往 Google Cloud Console
  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"
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. 按一下「選擇角色」
    2. 搜尋並選取「Storage 物件管理員」
    3. 點選「+ 新增其他角色」
    4. 搜尋並選取「Cloud Run Invoker」
    5. 點選「+ 新增其他角色」
    6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)
  6. 按一下「繼續」
  7. 按一下 [完成]

這些角色適用於:

  • Storage 物件管理員:將記錄檔寫入 GCS 值區,並管理狀態檔案
  • Cloud Run 叫用者:允許 Pub/Sub 叫用函式
  • Cloud Functions 叫用者:允許函式叫用

授予 GCS 值區的 IAM 權限

授予服務帳戶 GCS bucket 的寫入權限:

  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 主題

建立 Pub/Sub 主題,Cloud Scheduler 會將訊息發布至該主題,而 Cloud Run 函式會訂閱該主題。

  1. GCP Console 中,前往「Pub/Sub」>「Topics」(主題)
  2. 按一下「建立主題」
  3. 請提供下列設定詳細資料:
    • 主題 ID:輸入 forgerock-openidm-trigger
    • 其他設定保留預設值。
  4. 點選「建立」

建立 Cloud Run 函式來收集記錄

Cloud Run 函式會由 Cloud Scheduler 的 Pub/Sub 訊息觸發,從 ForgeRock OpenIDM API 擷取記錄,並寫入 GCS。

  1. 前往 GCP Console 的「Cloud Run」
  2. 按一下「Create service」(建立服務)
  3. 選取「函式」 (使用內嵌編輯器建立函式)。
  4. 在「設定」部分,提供下列設定詳細資料:

    設定
    服務名稱 forgerock-openidm-collector
    區域 選取與 GCS bucket 相符的區域 (例如 us-central1)
    執行階段 選取「Python 3.12」以上版本
  5. 在「Trigger (optional)」(觸發條件 (選用)) 專區:

    1. 按一下「+ 新增觸發條件」
    2. 選取「Cloud Pub/Sub」
    3. 在「選取 Cloud Pub/Sub 主題」中,選擇 Pub/Sub 主題 forgerock-openidm-trigger
    4. 按一下 [儲存]
  6. 在「Authentication」(驗證) 部分:

    1. 選取「需要驗證」
    2. 檢查 Identity and Access Management (IAM)
  7. 捲動至「Containers, Networking, Security」(容器、網路、安全性) 並展開

  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 秒 (10 分鐘)。
  11. 前往「設定」分頁:

    • 在「資源」部分:
      • 記憶體:選取 512 MiB 以上。
      • CPU:選取 1
    • 在「修訂版本資源調度」部分:
      • 執行個體數量下限:輸入 0
      • 「Maximum number of instances」(執行個體數量上限):輸入 100 (或根據預期負載調整)。
  12. 點選「建立」

  13. 等待服務建立完成 (1 到 2 分鐘)。

  14. 服務建立完成後,系統會自動開啟內嵌程式碼編輯器

新增函式程式碼

  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.
    
        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
    
  3. 點選「部署」來儲存並部署函式。

  4. 等待部署作業完成 (2 到 3 分鐘)。

建立 Cloud Scheduler 工作

Cloud Scheduler 會定期將訊息發布至 Pub/Sub 主題,觸發 Cloud Run 函式。

  1. 前往 GCP 主控台的「Cloud Scheduler」
  2. 點選「建立工作」
  3. 請提供下列設定詳細資料:

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

排程頻率選項

根據記錄檔量和延遲時間要求選擇頻率:

頻率 Cron 運算式 用途
每 5 分鐘 */5 * * * * 高容量、低延遲
每 15 分鐘 */15 * * * * 中等
每小時 0 * * * * 標準 (建議)
每 6 小時 0 */6 * * * 少量、批次處理
每日 0 0 * * * 歷來資料集合

測試整合項目

  1. Cloud Scheduler 控制台中,找出您的工作 forgerock-openidm-collector-hourly
  2. 按一下「強制執行」,手動觸發工作。
  3. 稍等幾秒鐘。
  4. 前往「Cloud Run」>「Services」
  5. 按一下函式名稱 forgerock-openidm-collector
  6. 按一下 [Logs] (記錄) 分頁標籤。
  7. 確認函式是否已順利執行。尋找:

    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
    
  8. 依序前往「Cloud Storage」>「Buckets」

  9. 按一下 bucket 名稱 forgerock-openidm-audit-logs

  10. 前往前置字元資料夾 openidm/

  11. 確認是否已建立含有目前時間戳記的新 .json 檔案。

如果在記錄中發現錯誤:

  • HTTP 401:檢查環境變數中的 OpenIDM 憑證
  • HTTP 403:確認帳戶是否具備稽核讀取權限
  • HTTP 404:確認 OPENIDM_BASE_URL 正確無誤,且不包含尾端路徑
  • 缺少環境變數:檢查是否已設定所有必要變數

擷取 Google SecOps 服務帳戶

Google SecOps 會使用專屬服務帳戶,從 GCS bucket 讀取資料。您必須授予這個服務帳戶值區存取權。

在 Google SecOps 中設定動態饋給,擷取 ForgeRock OpenIDM 記錄

  1. 依序前往「SIEM 設定」>「動態饋給」
  2. 按一下「新增動態消息」
  3. 按一下「設定單一動態饋給」
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 ForgeRock OpenIDM Audit Logs)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」
  6. 選取「FORGEROCK_OPENIDM」做為「記錄類型」

  7. 按一下「取得服務帳戶」

  8. 系統會顯示不重複的服務帳戶電子郵件地址,例如:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  9. 複製這個電子郵件地址,後續步驟會用到。

  10. 點選「下一步」

  11. 指定下列輸入參數的值:

    • 儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:
    gs://forgerock-openidm-audit-logs/openidm/
    
    • 取代:
      • forgerock-openidm-audit-logs:您的 GCS bucket 名稱。
      • openidm:儲存記錄的前置路徑。
    • 來源刪除選項:根據偏好選取刪除選項:
      • 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
      • 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
      • 刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。
    • 檔案存在時間上限:包含在過去天數內修改的檔案。預設值為 180 天。
    • 資產命名空間資產命名空間
    • 擷取標籤:要套用至這個動態饋給事件的標籤。
  12. 點選「下一步」

  13. 在「Finalize」(完成) 畫面中檢查新的動態饋給設定,然後按一下「Submit」(提交)

將 IAM 權限授予 Google SecOps 服務帳戶

Google SecOps 服務帳戶需要 GCS bucket 的「Storage 物件檢視者」角色。

  1. 依序前往「Cloud Storage」>「Buckets」
  2. 按一下 bucket 名稱 forgerock-openidm-audit-logs
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」
  5. 請提供下列設定詳細資料:
    • 新增主體:貼上 Google SecOps 服務帳戶電子郵件地址。
    • 指派角色:選取「Storage 物件檢視者」
  6. 按一下 [儲存]

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 專業人員尋求答案。