收集 Delinea 單一登入 (SSO) 記錄

支援的國家/地區:

本文說明如何使用 Google Cloud Storage,將 Delinea (前身為 Centrify) 單一登入 (SSO) 記錄檔擷取至 Google Security Operations。剖析器會擷取記錄,並處理 JSON 和系統記錄格式。這項服務會剖析鍵值組、時間戳記和其他相關欄位,並將這些欄位對應至 UDM 模型,同時採用特定邏輯處理登入失敗、使用者代理程式、嚴重程度、驗證機制和各種事件類型。在失敗事件中,系統會優先使用 FailUserName,而非 NormalizedUser 做為目標電子郵件地址。

事前準備

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

  • Google SecOps 執行個體
  • 已啟用 Cloud Storage API 的 GCP 專案
  • 建立及管理 GCS 值區的權限
  • 管理 Google Cloud Storage 值區 IAM 政策的權限
  • 建立 Cloud Run 函式、Pub/Sub 主題和 Cloud Scheduler 工作的權限
  • Delinea (Centrify) SSO 租戶的特殊存取權

收集 Delinea (Centrify) 單一登入 (SSO) 憑證

建立 OAuth2 用戶端應用程式

  1. 登入 Delinea 管理入口網站
  2. 依序前往「應用程式」>「新增網頁應用程式」
  3. 按一下「自訂」分頁標籤。
  4. 搜尋「OAuth2 Client」,然後按一下「Add」
  5. 在「新增網頁應用程式」對話方塊中,按一下「是」
  6. 按一下「新增網頁應用程式」對話方塊中的「關閉」
  7. 在「Application Configuration」頁面中,設定下列項目:
    • 「設定」分頁:
      • 「Application ID」:輸入專屬 ID (例如 secops-oauth-client)。
      • 應用程式名稱:輸入描述性名稱 (例如 SecOps Data Export)。
      • 應用程式說明:輸入說明 (例如 OAuth client for exporting audit events to SecOps)。
    • 「一般使用方式」部分:
      • 用戶端 ID 類型:選取「機密」
      • 已核發的用戶端 ID:複製並儲存這個值。
      • 核發的用戶端密鑰:複製並儲存這個值。
    • 「符記」分頁:
      • 驗證方法:選取「用戶端憑證」
      • 權杖類型:選取「JwtRS256」
    • 「範圍」分頁:
      • 新增範圍 redrock/query 和說明 Query API Access
  8. 按一下「儲存」即可建立 OAuth 用戶端。
  9. 複製並儲存以下詳細資料,存放在安全的位置:
    • 租戶網址:您的 Centrify 租戶網址 (例如 https://yourtenant.my.centrify.com)。
    • 用戶端 ID:來自步驟 7。
    • 用戶端密鑰:來自步驟 7。
    • OAuth 應用程式 ID:來自應用程式設定。

驗證權限

如要確認 OAuth 用戶端是否具備必要權限,請按照下列步驟操作:

  1. 登入 Delinea 管理入口網站
  2. 依序前往「設定」圖示 (⚙️) >「資源」>「角色」
  3. 確認指派給 OAuth 用戶端的角色包含「報表:稽核事件:查看」權限。
  4. 如果缺少權限,請與 Delinea 管理員聯絡,要求對方授予必要權限。

測試 API 存取權

  • 請先測試憑證,再繼續進行整合:

    # Replace with your actual credentials
    TENANT_URL="https://yourtenant.my.centrify.com"
    CLIENT_ID="your-client-id"
    CLIENT_SECRET="your-client-secret"
    OAUTH_APP_ID="your-oauth-application-id"
    
    # Get OAuth token
    TOKEN=$(curl -s -X POST "${TENANT_URL}/oauth2/token/${OAUTH_APP_ID}" \
      -H "Authorization: Basic $(echo -n "${CLIENT_ID}:${CLIENT_SECRET}" | base64)" \
      -H "X-CENTRIFY-NATIVE-CLIENT: True" \
      -H "Content-Type: application/x-www-form-urlencoded" \
      -d "grant_type=client_credentials&scope=redrock/query" | jq -r '.access_token')
    
    # Test query API access
    curl -v -X POST "${TENANT_URL}/Redrock/query" \
      -H "Authorization: Bearer ${TOKEN}" \
      -H "X-CENTRIFY-NATIVE-CLIENT: True" \
      -H "Content-Type: application/json" \
      -d '{"Script":"Select * from Event where WhenOccurred > datefunc('"'"'now'"'"', '"'"'-1'"'"') ORDER BY WhenOccurred ASC","args":{"PageNumber":1,"PageSize":10,"Limit":10,"Caching":-1}}'
    

如果成功,您應該會看到包含稽核事件的 JSON 回應。如果收到 401 或 403 錯誤,請確認您的憑證和權限。

建立 Google Cloud Storage 值區

  1. 前往 Google Cloud 控制台
  2. 選取專案或建立新專案。
  3. 在導覽選單中,依序前往「Cloud Storage」>「Bucket」
  4. 按一下「建立值區」
  5. 請提供下列設定詳細資料:

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

為 Cloud Run 函式建立服務帳戶

Cloud Run 函式需要具備 GCS bucket 寫入權限的服務帳戶,並由 Pub/Sub 叫用。

建立服務帳戶

  1. GCP 主控台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)
  2. 按一下 [Create Service Account] (建立服務帳戶)
  3. 請提供下列設定詳細資料:
    • 服務帳戶名稱:輸入 delinea-sso-collector-sa
    • 服務帳戶說明:輸入 Service account for Cloud Run function to collect Delinea SSO 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 名稱 (例如 delinea-centrify-logs-bucket)。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」
  5. 請提供下列設定詳細資料:
    • 新增主體:輸入服務帳戶電子郵件地址 (例如 delinea-sso-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:輸入 delinea-sso-logs-trigger
    • 其他設定保留預設值。
  4. 點選「建立」

建立 Cloud Run 函式來收集記錄

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

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

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

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

    1. 選取「需要驗證」
    2. 檢查 Identity and Access Management (IAM)
  7. 向下捲動並展開「Containers, Networking, Security」

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

    • 服務帳戶:選取服務帳戶 delinea-sso-collector-sa
  9. 前往「容器」分頁:

    1. 按一下「變數與密鑰」
    2. 針對每個環境變數,按一下「+ 新增變數」
    變數名稱 範例值 說明
    GCS_BUCKET delinea-centrify-logs-bucket GCS bucket 名稱
    GCS_PREFIX centrify-sso-logs 記錄檔的前置字串
    STATE_KEY centrify-sso-logs/state.json 狀態檔案路徑
    TENANT_URL https://yourtenant.my.centrify.com Delinea 租戶網址
    CLIENT_ID your-client-id OAuth 用戶端 ID
    CLIENT_SECRET your-client-secret OAuth 用戶端密鑰
    OAUTH_APP_ID your-oauth-application-id OAuth 應用程式 ID
    PAGE_SIZE 1000 每頁記錄數
    MAX_PAGES 10 要擷取的頁面數量上限
  10. 在「變數與密鑰」部分,向下捲動至「要求」

    • 要求逾時:輸入 600 秒 (10 分鐘)。
  11. 前往「設定」分頁:

    • 在「資源」部分:
      • 記憶體:選取 512 MiB 以上。
      • CPU:選取 1
  12. 在「修訂版本資源調度」部分:

    • 執行個體數量下限:輸入 0
    • 「Maximum number of instances」(執行個體數量上限):輸入 100 (或根據預期負載調整)。
  13. 點選「建立」

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

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

新增函式程式碼

  1. 在「Function entry point」(函式進入點) 中輸入 main
  2. 在內嵌程式碼編輯器中建立兩個檔案:

    • 第一個檔案: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()
    
    @functions_framework.cloud_event
    def main(cloud_event):
        """
        Cloud Run function triggered by Pub/Sub to fetch Delinea Centrify SSO audit events and write to GCS.
    
        Args:
            cloud_event: CloudEvent object containing Pub/Sub message
        """
    
        # Get environment variables
        bucket_name = os.environ.get('GCS_BUCKET')
        prefix = os.environ.get('GCS_PREFIX', 'centrify-sso-logs')
        state_key = os.environ.get('STATE_KEY', 'centrify-sso-logs/state.json')
    
        # Centrify API credentials
        tenant_url = os.environ.get('TENANT_URL')
        client_id = os.environ.get('CLIENT_ID')
        client_secret = os.environ.get('CLIENT_SECRET')
        oauth_app_id = os.environ.get('OAUTH_APP_ID')
    
        # Optional parameters
        page_size = int(os.environ.get('PAGE_SIZE', '1000'))
        max_pages = int(os.environ.get('MAX_PAGES', '10'))
    
        if not all([bucket_name, tenant_url, client_id, client_secret, oauth_app_id]):
            print('Error: Missing required environment variables')
            return
    
        try:
            # Get GCS bucket
            bucket = storage_client.bucket(bucket_name)
    
            # Load state (last processed timestamp)
            state = load_state(bucket, state_key)
            last_timestamp = state.get('last_timestamp')
    
            print(f'Processing logs since {last_timestamp if last_timestamp else "24 hours ago"}')
    
            # Get OAuth access token
            access_token = get_oauth_token(tenant_url, client_id, client_secret, oauth_app_id)
    
            # Fetch audit events
            events = fetch_audit_events(tenant_url, access_token, last_timestamp, page_size, max_pages)
    
            if events:
                # Write events to GCS
                current_timestamp = datetime.now(timezone.utc)
                blob_name = f"{prefix}/centrify-sso-events-{current_timestamp.strftime('%Y%m%d_%H%M%S')}.json"
                blob = bucket.blob(blob_name)
    
                # Convert to JSONL format (one JSON object per line)
                jsonl_content = '\n'.join([json.dumps(event, default=str) for event in events])
                blob.upload_from_string(jsonl_content, content_type='application/x-ndjson')
    
                print(f'Wrote {len(events)} events to {blob_name}')
    
                # Update state with latest timestamp
                latest_timestamp = get_latest_event_timestamp(events)
                save_state(bucket, state_key, {'last_timestamp': latest_timestamp, 'updated_at': current_timestamp.isoformat() + 'Z'})
    
                print(f'Successfully processed {len(events)} events')
            else:
                print('No new events found')
    
        except Exception as e:
            print(f'Error processing Centrify SSO logs: {str(e)}')
            raise
    
    def get_oauth_token(tenant_url, client_id, client_secret, oauth_app_id):
        """Get OAuth access token using client credentials flow."""
        credentials = f"{client_id}:{client_secret}"
        basic_auth = base64.b64encode(credentials.encode('utf-8')).decode('utf-8')
    
        token_url = f"{tenant_url}/oauth2/token/{oauth_app_id}"
        headers = {
            'Authorization': f'Basic {basic_auth}',
            'X-CENTRIFY-NATIVE-CLIENT': 'True',
            'Content-Type': 'application/x-www-form-urlencoded'
        }
    
        data = {
            'grant_type': 'client_credentials',
            'scope': 'redrock/query'
        }
    
        response = http.request('POST', token_url, headers=headers, fields=data)
    
        if response.status != 200:
            raise Exception(f"OAuth token request failed: {response.status} {response.data.decode('utf-8')}")
    
        token_data = json.loads(response.data.decode('utf-8'))
        return token_data['access_token']
    
    def fetch_audit_events(tenant_url, access_token, last_timestamp, page_size, max_pages):
        """Fetch audit events from Centrify using the Redrock/query API with proper pagination."""
        query_url = f"{tenant_url}/Redrock/query"
        headers = {
            'Authorization': f'Bearer {access_token}',
            'X-CENTRIFY-NATIVE-CLIENT': 'True',
            'Content-Type': 'application/json'
        }
    
        # Build SQL query with timestamp filter
        if last_timestamp:
            sql_query = f"Select * from Event where WhenOccurred > '{last_timestamp}' ORDER BY WhenOccurred ASC"
        else:
            # First run - get events from last 24 hours
            sql_query = "Select * from Event where WhenOccurred > datefunc('now', '-1') ORDER BY WhenOccurred ASC"
    
        all_events = []
        page_num = 1
        backoff = 1.0
    
        while page_num <= max_pages:
            payload = {
                "Script": sql_query,
                "args": {
                    "PageNumber": page_num,
                    "PageSize": page_size,
                    "Limit": page_size * max_pages,
                    "Caching": -1
                }
            }
    
            try:
                response = http.request('POST', query_url, headers=headers, body=json.dumps(payload))
    
                # Handle rate limiting with exponential backoff
                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:
                    raise Exception(f"API query failed: {response.status} {response.data.decode('utf-8')}")
    
                response_data = json.loads(response.data.decode('utf-8'))
    
                if not response_data.get('success', False):
                    raise Exception(f"API query failed: {response_data.get('Message', 'Unknown error')}")
    
                # Parse the response
                result = response_data.get('Result', {})
                columns = {col['Name']: i for i, col in enumerate(result.get('Columns', []))}
                raw_results = result.get('Results', [])
    
                if not raw_results:
                    print(f"No more results on page {page_num}")
                    break
    
                print(f"Page {page_num}: Retrieved {len(raw_results)} events")
    
                for raw_event in raw_results:
                    event = {}
                    row_data = raw_event.get('Row', {})
    
                    # Map column names to values
                    for col_name, col_index in columns.items():
                        if col_name in row_data and row_data[col_name] is not None:
                            event[col_name] = row_data[col_name]
    
                    # Add metadata
                    event['_source'] = 'centrify_sso'
                    event['_collected_at'] = datetime.now(timezone.utc).isoformat() + 'Z'
    
                    all_events.append(event)
    
                # Check if we've reached the end
                if len(raw_results) < page_size:
                    print(f"Reached last page (page {page_num} returned {len(raw_results)} < {page_size})")
                    break
    
                page_num += 1
    
            except Exception as e:
                print(f"Error fetching page {page_num}: {e}")
                raise
    
        print(f"Retrieved {len(all_events)} total events from {page_num} pages")
        return all_events
    
    def get_latest_event_timestamp(events):
        """Get the latest timestamp from the events for state tracking."""
        if not events:
            return datetime.now(timezone.utc).isoformat() + 'Z'
    
        latest = None
        for event in events:
            when_occurred = event.get('WhenOccurred')
            if when_occurred:
                if latest is None or when_occurred > latest:
                    latest = when_occurred
    
        return latest or datetime.now(timezone.utc).isoformat() + 'Z'
    
    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: {str(e)}')
        return {}
    
    def save_state(bucket, key, state):
        """Save state to GCS."""
        try:
            blob = bucket.blob(key)
            blob.upload_from_string(
                json.dumps(state),
                content_type='application/json'
            )
        except Exception as e:
            print(f'Warning: Could not save state: {str(e)}')
    
    • 第二個檔案: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. 請提供下列設定詳細資料:

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

排程頻率選項

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

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

測試整合項目

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

    Processing logs since YYYY-MM-DDTHH:MM:SS+00:00
    Page 1: Retrieved X events
    Wrote X events to centrify-sso-logs/centrify-sso-events_YYYYMMDD_HHMMSS.json
    Successfully processed X events
    
  8. 依序前往「Cloud Storage」>「Buckets」

  9. 按一下 bucket 名稱 (例如 delinea-centrify-logs-bucket)。

  10. 前往前置字元資料夾 centrify-sso-logs/

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

如果在記錄中發現錯誤:

  • HTTP 401:檢查環境變數中的 API 憑證 (CLIENT_ID、CLIENT_SECRET、OAUTH_APP_ID)
  • HTTP 403:確認 OAuth 用戶端具有「報表:稽核事件:查看」權限
  • HTTP 429:頻率限制 - 函式會自動重試並延遲
  • 缺少環境變數:檢查 Cloud Run 函式設定中是否已設定所有必要變數

擷取 Google SecOps 服務帳戶

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

取得服務帳戶電子郵件地址

  1. 依序前往「SIEM 設定」>「動態饋給」
  2. 按一下「新增動態消息」
  3. 按一下「設定單一動態饋給」
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Delinea Centrify SSO logs)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」
  6. 選取「Centrify」做為「記錄類型」
  7. 按一下「取得服務帳戶」。系統會顯示不重複的服務帳戶電子郵件地址,例如:

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. 複製這個電子郵件地址,以便在下一步中使用。

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

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

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

在 Google SecOps 中設定動態饋給,擷取 Delinea (Centrify) 單一登入記錄

  1. 依序前往「SIEM 設定」>「動態饋給」
  2. 按一下「新增動態消息」
  3. 按一下「設定單一動態饋給」
  4. 在「動態饋給名稱」欄位中輸入動態饋給名稱 (例如 Delinea Centrify SSO logs)。
  5. 選取「Google Cloud Storage V2」做為「來源類型」
  6. 選取「Centrify」做為「記錄類型」
  7. 點選 [下一步]。
  8. 指定下列輸入參數的值:

    • 儲存空間 bucket URL:輸入 GCS bucket URI,並加上前置路徑:

      gs://delinea-centrify-logs-bucket/centrify-sso-logs/
      
      • 取代:

        • delinea-centrify-logs-bucket:您的 GCS bucket 名稱。
        • centrify-sso-logs:儲存記錄的選用前置字元/資料夾路徑 (如為根目錄,請留空)。
      • 範例:

        • 根層級 bucket:gs://company-logs/
        • 前置字串:gs://company-logs/centrify-sso-logs/
        • 有子資料夾:gs://company-logs/delinea/sso/
    • 來源刪除選項:根據偏好設定選取刪除選項:

      • 永不:移轉後一律不刪除任何檔案 (建議用於測試)。
      • 刪除已轉移的檔案:成功轉移檔案後刪除檔案。
      • 刪除已轉移的檔案和空白目錄:成功轉移後刪除檔案和空白目錄。

    • 檔案存在時間上限:包含在過去天數內修改的檔案。預設值為 180 天。

    • 資產命名空間資產命名空間

    • 擷取標籤:要套用至這個動態饋給事件的標籤。

  9. 點選 [下一步]。

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

UDM 對應表

記錄欄位 UDM 對應 邏輯
AccountID security_result.detection_fields.value 原始記錄中的 AccountID 值會指派給 security_result.detection_fields 物件,並以「帳戶 ID」做為鍵。
ApplicationName target.application 原始記錄中的 ApplicationName 值會指派給 target.application 欄位。
AuthorityFQDN target.asset.network_domain 原始記錄中的 AuthorityFQDN 值會指派給 target.asset.network_domain 欄位。
AuthorityID target.asset.asset_id 原始記錄中的 AuthorityID 值會指派給 target.asset.asset_id 欄位,並加上「AuthorityID:」前置字串。
AzDeploymentId security_result.detection_fields.value 原始記錄中的 AzDeploymentId 值會指派給 security_result.detection_fields 物件,並以 AzDeploymentId 做為鍵。
AzRoleId additional.fields.value.string_value 原始記錄中的 AzRoleId 值會指派給 additional.fields 物件,並以 AzRoleId 做為鍵。
AzRoleName target.user.attribute.roles.name 原始記錄中的 AzRoleName 值會指派給 target.user.attribute.roles.name 欄位。
ComputerFQDN principal.asset.network_domain 原始記錄中的 ComputerFQDN 值會指派給 principal.asset.network_domain 欄位。
ComputerID principal.asset.asset_id 原始記錄中的 ComputerID 值會指派給 principal.asset.asset_id 欄位,並加上「ComputerId:」前置字串。
ComputerName about.hostname 原始記錄中的 ComputerName 值會指派給 about.hostname 欄位。
CredentialId security_result.detection_fields.value 原始記錄中的 CredentialId 值會指派給 security_result.detection_fields 物件,並以「Credential Id」做為鍵。
DirectoryServiceName security_result.detection_fields.value 原始記錄中的 DirectoryServiceName 值會指派給 security_result.detection_fields 物件,並以「Directory Service Name」做為鍵。
DirectoryServiceNameLocalized security_result.detection_fields.value 原始記錄中的 DirectoryServiceNameLocalized 值會指派給 security_result.detection_fields 物件,並以「Directory Service Name Localized」做為鍵。
DirectoryServiceUuid security_result.detection_fields.value 原始記錄中的 DirectoryServiceUuid 值會指派給 security_result.detection_fields 物件,並以「Directory Service Uuid」做為鍵。
EventMessage security_result.summary 原始記錄中的 EventMessage 值會指派給 security_result.summary 欄位。
EventType metadata.product_event_type 原始記錄中的 EventType 值會指派給 metadata.product_event_type 欄位。這項資訊也會用於判斷 metadata.event_type。
FailReason security_result.summary 如果原始記錄中存在 FailReason 值,系統會將該值指派給 security_result.summary 欄位。
FailUserName target.user.email_addresses 如果原始記錄中存在 FailUserName 值,系統會將該值指派給 target.user.email_addresses 欄位。
FromIPAddress principal.ip 原始記錄中的 FromIPAddress 值會指派給 principal.ip 欄位。
ID security_result.detection_fields.value 原始記錄中的 ID 值會指派給 security_result.detection_fields 物件,並以 ID 做為鍵。
InternalTrackingID metadata.product_log_id 原始記錄中的 InternalTrackingID 值會指派給 metadata.product_log_id 欄位。
JumpType additional.fields.value.string_value 原始記錄中的 JumpType 值會指派給 additional.fields 物件,並以「Jump Type」做為鍵。
NormalizedUser target.user.email_addresses 原始記錄中的 NormalizedUser 值會指派給 target.user.email_addresses 欄位。
OperationMode additional.fields.value.string_value 原始記錄中的 OperationMode 值會指派給 additional.fields 物件,並以「Operation Mode」做為鍵。
ProxyId security_result.detection_fields.value 原始記錄中的 ProxyId 值會指派給 security_result.detection_fields 物件,並以「Proxy ID」做為鍵。
RequestUserAgent network.http.user_agent 原始記錄中的 RequestUserAgent 值會指派給 network.http.user_agent 欄位。
SessionGuid network.session_id 原始記錄中的 SessionGuid 值會指派給 network.session_id 欄位。
用戶群 additional.fields.value.string_value 原始記錄中的「Tenant」值會指派給 additional.fields 物件,並以「Tenant」做為鍵。
ThreadType additional.fields.value.string_value 原始記錄中的 ThreadType 值會指派給 additional.fields 物件,並以「Thread Type」做為鍵。
UserType principal.user.attribute.roles.name 原始記錄中的 UserType 值會指派給 principal.user.attribute.roles.name 欄位。
WhenOccurred metadata.event_timestamp 系統會剖析原始記錄中的 WhenOccurred 值,並指派給 metadata.event_timestamp 欄位。這個欄位也會填入頂層時間戳記欄位。
硬式編碼值 metadata.product_name 「SSO」。
硬式編碼值 metadata.event_type 由 EventType 欄位決定。如果沒有 EventType 或不符合任何特定條件,則預設為 STATUS_UPDATE。可以是 USER_LOGIN、USER_CREATION、USER_RESOURCE_ACCESS、USER_LOGOUT 或 USER_CHANGE_PASSWORD。
硬式編碼值 metadata.vendor_name 「CENTRIFY_SSO」。
硬式編碼值 metadata.product_version 「SSO」。
硬式編碼值 metadata.log_type 「Centrify」。
從訊息欄位擷取 network.session_id 如果訊息欄位含有工作階段 ID,系統會擷取並使用該 ID。否則預設為「1」。
從主機欄位擷取 principal.hostname 如果主機欄位有值 (來自系統記錄標頭),則會從該欄位擷取。
從 PID 欄位擷取 principal.process.pid 如果有的話,會從 pid 欄位擷取,該欄位來自 syslog 標頭。
UserGuid 或從訊息中擷取 target.user.userid 如果存在 UserGuid,系統會使用該值。否則,如果訊息欄位含有使用者 ID,系統會擷取並使用該 ID。
由 Level 和 FailReason 決定 security_result.action 如果層級為「資訊」,請設為「ALLOW」;如果 FailReason 存在,請設為「BLOCK」。
由 FailReason 決定 security_result.category 如果 FailReason 存在,則設為「AUTH_VIOLATION」。
由「等級」欄位決定 security_result.severity 由「層級」欄位決定。如果「層級」為「資訊」,請設為「INFORMATIONAL」;如果「層級」為「警告」,請設為「MEDIUM」;如果「層級」為「錯誤」,請設為「ERROR」。

需要其他協助嗎?向社群成員和 Google SecOps 專業人員尋求答案。