收集 Vectra XDR 記錄

支援的國家/地區:

本文說明如何使用 Google Cloud Storage V2,將 Vectra XDR 記錄檔擷取至 Google Security Operations。

Vectra XDR 是一種擴展式偵測及回應平台,可關聯網路、身分、雲端和 SaaS 環境中的威脅。Vectra AI Platform REST API v3 提供程式化存取權,可存取偵測、評分、鎖定、稽核和健康狀態記錄。

事前準備

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

  • Google SecOps 執行個體
  • 已啟用 Cloud Storage API 的 GCP 專案
  • 建立及管理 GCS bucket 的權限
  • 管理 Google Cloud Storage 值區 IAM 政策的權限
  • 建立 Cloud Run 服務、Pub/Sub 主題和 Cloud Scheduler 工作的權限
  • 具備管理員權限的 Vectra AI Platform 特殊存取權
  • API 用戶端,且已在 Vectra AI Platform 中設定用戶端 ID 和用戶端密鑰

建立 Google Cloud Storage bucket

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

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

收集 Vectra XDR API 憑證

建立 API 用戶端

  1. 登入 Vectra AI Platform 執行個體 (例如 https://your-tenant.vectra.ai)。
  2. 依序前往「管理」>「API 用戶端」
  3. 按一下「新增 API 用戶端」
  4. 輸入 API 用戶端的名稱 (例如 Google SecOps Integration)。
  5. 選取所需角色 (例如「唯讀」,或具備偵測和稽核存取權的自訂角色)。
  6. 按一下「產生憑證」
  7. 複製下列詳細資料並存放在安全的位置:

    • 用戶端 ID:API 用戶端 ID
    • 用戶端密鑰:API 用戶端密鑰

判斷 API 基準網址

Vectra API 基準網址是您的租戶網址:

格式 範例
租戶網址 https://your-tenant.vectra.ai

測試 API 存取權

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

    # Replace with your actual credentials
    VECTRA_CLIENT_ID="your-client-id"
    VECTRA_CLIENT_SECRET="your-client-secret"
    VECTRA_BASE_URL="https://your-tenant.vectra.ai"
    
    # Get access token using HTTP Basic Auth
    TOKEN=$(curl -s -X POST "${VECTRA_BASE_URL}/oauth2/token" \
      -u "${VECTRA_CLIENT_ID}:${VECTRA_CLIENT_SECRET}" \
      -H "Content-Type: application/x-www-form-urlencoded" \
      -d "grant_type=client_credentials" \
      | jq -r '.access_token')
    
    # Test API access - list detection events
    curl -s -X GET "${VECTRA_BASE_URL}/api/v3.4/events/detections/?limit=1" \
      -H "Authorization: Bearer ${TOKEN}"
    

為 Cloud Run 函式建立服務帳戶

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

建立服務帳戶

  1. GCP 控制台中,依序前往「IAM & Admin」(IAM 與管理) >「Service Accounts」(服務帳戶)
  2. 按一下「Create Service Account」(建立服務帳戶)
  3. 請提供下列設定詳細資料:
    • 服務帳戶名稱:輸入 vectra-xdr-logs-collector-sa
    • 服務帳戶說明:輸入 Service account for Cloud Run function to collect Vectra XDR logs
  4. 按一下「建立並繼續」
  5. 在「將專案存取權授予這個服務帳戶」部分,新增下列角色:
    1. 按一下「選擇角色」
    2. 搜尋並選取「Storage 物件管理員」
    3. 點選「+ 新增其他角色」
    4. 搜尋並選取「Cloud Run Invoker」
    5. 點選「+ 新增其他角色」
    6. 搜尋並選取「Cloud Functions Invoker」(Cloud Functions 叫用者)
  6. 按一下「繼續」
  7. 按一下 [完成]

這些角色適用於:

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

授予 GCS 值區的 IAM 權限

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

  1. 依序前往「Cloud Storage」>「Buckets」
  2. 按一下 bucket 名稱 (例如 vectra-xdr-logs)。
  3. 前往「權限」分頁標籤。
  4. 按一下「授予存取權」
  5. 請提供下列設定詳細資料:
    • 新增主體:輸入服務帳戶電子郵件地址 (例如 vectra-xdr-logs-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:輸入 vectra-xdr-logs-trigger
    • 其他設定保留預設值
  4. 點選「建立」

建立 Cloud Run 函式來收集記錄

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

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

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

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

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

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

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

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

      變數名稱 範例值 說明
      GCS_BUCKET vectra-xdr-logs GCS bucket 名稱
      GCS_PREFIX vectra 記錄檔的前置字串
      STATE_KEY vectra/state.json 狀態檔案路徑
      VECTRA_CLIENT_ID your-client-id Vectra API 用戶端 ID
      VECTRA_CLIENT_SECRET your-client-secret Vectra API 用戶端密鑰
      VECTRA_BASE_URL https://your-tenant.vectra.ai Vectra 租戶網址
      MAX_RECORDS 5000 每次執行的記錄數上限
      BATCH_SIZE 500 每個 API 要求的事件數 (最多 1000 個)
      LOOKBACK_HOURS 24 初始回溯期
  10. 在「變數與密鑰」部分,捲動至「要求」

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

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

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

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

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

新增函式程式碼

  1. 在「進入點」欄位中輸入「main」
  2. 在內嵌程式碼編輯器中,建立兩個檔案:

    • 第一個檔案:main.py:

      import functions_framework
      from google.cloud import storage
      import base64
      import json
      import os
      import urllib3
      from datetime import datetime, timezone, timedelta
      import time
      
      # 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', 'vectra')
      STATE_KEY = os.environ.get('STATE_KEY', 'vectra/state.json')
      VECTRA_CLIENT_ID = os.environ.get('VECTRA_CLIENT_ID')
      VECTRA_CLIENT_SECRET = os.environ.get('VECTRA_CLIENT_SECRET')
      VECTRA_BASE_URL = os.environ.get('VECTRA_BASE_URL')
      MAX_RECORDS = int(os.environ.get('MAX_RECORDS', '5000'))
      BATCH_SIZE = int(os.environ.get('BATCH_SIZE', '500'))
      LOOKBACK_HOURS = int(os.environ.get('LOOKBACK_HOURS', '24'))
      
      def get_access_token():
        """
        Obtain OAuth2 access token using HTTP Basic Auth
        with client credentials grant per Vectra API docs.
        """
        base_url = VECTRA_BASE_URL.rstrip('/')
        token_url = f"{base_url}/oauth2/token"
      
        credentials = base64.b64encode(
          f"{VECTRA_CLIENT_ID}:{VECTRA_CLIENT_SECRET}".encode()
        ).decode()
      
        headers = {
          'Content-Type': 'application/x-www-form-urlencoded',
          'Authorization': f'Basic {credentials}',
          'Accept': 'application/json'
        }
      
        body = 'grant_type=client_credentials'
      
        backoff = 1.0
        for attempt in range(3):
          response = http.request('POST', token_url, body=body, 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("Failed to get access token after 3 retries")
      
      @functions_framework.cloud_event
      def main(cloud_event):
        """
        Cloud Run function triggered by Pub/Sub to fetch Vectra XDR
        detection and audit event logs and write to GCS.
        Uses the Vectra REST API v3.4 Events endpoints with
        checkpoint-based pagination.
        """
      
        if not all([GCS_BUCKET, VECTRA_CLIENT_ID, VECTRA_CLIENT_SECRET, VECTRA_BASE_URL]):
          print('Error: Missing required environment variables')
          return
      
        try:
          bucket = storage_client.bucket(GCS_BUCKET)
      
          # Load state (stores per-endpoint checkpoints)
          state = load_state(bucket, STATE_KEY)
      
          now = datetime.now(timezone.utc)
          token = get_access_token()
      
          # Fetch detection events and audit events
          all_records = []
          new_state = dict(state)
      
          for event_type in ['detections', 'audits']:
            checkpoint = state.get(f'{event_type}_checkpoint')
            last_time = state.get(f'{event_type}_last_time')
      
            # On first run, use lookback window
            start_time = None
            if not checkpoint and not last_time:
              start_time = now - timedelta(hours=LOOKBACK_HOURS)
      
            records, next_checkpoint = fetch_events(
              token=token,
              event_type=event_type,
              checkpoint=checkpoint,
              start_time=start_time,
              batch_size=BATCH_SIZE,
              max_records=MAX_RECORDS,
            )
            all_records.extend(records)
      
            # Save checkpoint for next run
            if next_checkpoint is not None:
              new_state[f'{event_type}_checkpoint'] = next_checkpoint
            new_state[f'{event_type}_last_time'] = now.isoformat()
      
          if not all_records:
            print("No new event records found.")
            save_state(bucket, STATE_KEY, new_state)
            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 all_records
          ) + '\n'
          blob.upload_from_string(ndjson, content_type='application/x-ndjson')
      
          print(f"Wrote {len(all_records)} records to gs://{GCS_BUCKET}/{object_key}")
      
          save_state(bucket, STATE_KEY, new_state)
          print(f"Successfully processed {len(all_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():
            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: dict):
        """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: {state}")
        except Exception as e:
          print(f"Warning: Could not save state: {e}")
      
      def fetch_events(token: str, event_type: str, checkpoint: int = None,
              start_time: datetime = None, batch_size: int = 500,
              max_records: int = 5000):
        """
        Fetch events from Vectra AI Platform REST API v3.4
        using checkpoint-based pagination.
      
        Endpoints:
          - /api/v3.4/events/detections/
          - /api/v3.4/events/audits/
      
        Args:
          token: OAuth2 access token
          event_type: 'detections' or 'audits'
          checkpoint: Resume from this event ID (from previous run)
          start_time: Only used on first run when no checkpoint exists
          batch_size: Number of events per request (max 1000)
          max_records: Maximum total events to fetch per run
      
        Returns:
          Tuple of (events list, next_checkpoint int or None)
        """
        base_url = VECTRA_BASE_URL.rstrip('/')
        endpoint = f"{base_url}/api/v3.4/events/{event_type}/"
      
        headers = {
          'Authorization': f'Bearer {token}',
          'Accept': 'application/json',
          'User-Agent': 'GoogleSecOps-VectraXDRCollector/1.0'
        }
      
        records = []
        batch_num = 0
        backoff = 1.0
        next_checkpoint = checkpoint
      
        while True:
          batch_num += 1
      
          if len(records) >= max_records:
            print(f"Reached max_records limit ({max_records}) for {event_type}")
            break
      
          # Build query parameters
          params = {
            'limit': min(batch_size, max_records - len(records)),
            'ordering': 'event_timestamp',
          }
      
          if next_checkpoint is not None:
            params['from'] = next_checkpoint
          elif start_time is not None:
            params['event_timestamp_gte'] = start_time.strftime('%Y-%m-%dT%H:%M:%SZ')
      
          # Include INFO detections (excluded by default in v3.4)
          if event_type == 'detections':
            params['include_info_category'] = 'true'
      
          query_string = '&'.join(f'{k}={v}' for k, v in params.items())
          url = f"{endpoint}?{query_string}"
      
          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 {response.status}: {response.data.decode('utf-8')}")
              break
      
            data = json.loads(response.data.decode('utf-8'))
      
            events = data.get('events', [])
            remaining = data.get('remaining_count', 0)
            batch_checkpoint = data.get('next_checkpoint')
      
            if not events:
              print(f"No more events for {event_type}")
              break
      
            print(f"{event_type} batch {batch_num}: {len(events)} events, {remaining} remaining")
      
            # Tag events with source type
            for event in events:
              event['_vectra_event_type'] = event_type
      
            records.extend(events)
      
            # Update checkpoint for next batch/run
            if batch_checkpoint is not None:
              next_checkpoint = batch_checkpoint
      
            # No more events remaining
            if remaining == 0:
              print(f"All events fetched for {event_type}")
              break
      
          except Exception as e:
            print(f"Error fetching {event_type} events: {e}")
            break
      
        print(f"Retrieved {len(records)} total {event_type} events from {batch_num} batches")
        return records, next_checkpoint
      
    • 第二個檔案: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. 請提供下列設定詳細資料:

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

排程頻率選項

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

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

測試整合項目

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

    detections batch 1: X events, Y remaining
    All events fetched for detections
    audits batch 1: X events, Y remaining
    All events fetched for audits
    Wrote X records to gs://vectra-xdr-logs/vectra/logs_YYYYMMDD_HHMMSS.ndjson
    Successfully processed X records
    
  8. 依序前往「Cloud Storage」>「Buckets」

  9. 按一下 bucket 名稱 (vectra-xdr-logs)。

  10. 前往 vectra/ 資料夾。

  11. 確認是否已建立新的 .ndjson 檔案,且檔案名稱包含目前的時間戳記。

如果記錄中顯示錯誤:

  • HTTP 401:檢查環境變數中的用戶端憑證
  • HTTP 403:確認 API 用戶端在 Vectra AI Platform 中是否具備必要權限
  • HTTP 429:頻率限制 - 函式會自動重試,並延遲一段時間
  • 缺少環境變數:檢查是否已設定所有必要變數

在 Google SecOps 中設定資訊提供,擷取 Vectra XDR 記錄

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

    chronicle-12345678@chronicle-gcp-prod.iam.gserviceaccount.com
    
  8. 複製這個電子郵件地址。

  9. 點選「下一步」

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

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

      gs://vectra-xdr-logs/vectra/
      
      • 取代:
        • vectra-xdr-logs:您的 GCS bucket 名稱。
        • vectra:儲存記錄的選用前置字元/資料夾路徑 (如為根目錄,請留空)。
    • 來源刪除選項:根據偏好設定選取刪除選項:

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

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

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

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

  11. 點選「下一步」

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

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

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

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

UDM 對應表

記錄欄位 UDM 對應 邏輯
log_type_lable、id_label、triaged_label、detail_label、properties_label、event_object_label additional.fields 根據記錄類型與各種標籤合併
msg、d_type_vname extensions.auth.auth_details 針對 USER_LOGIN 設為 msg,針對其他使用者登入設為 d_type_vname
d_type_vname、category、type、msg metadata.event_type 一開始設為 GENERIC_EVENT,之後會根據記錄類型和條件覆寫為特定類型
d_type_vname、category、type、event_action metadata.product_event_type 在「偵測」中設為 d_type_vname,在「評分」中設為 category,在「封鎖」中設為 type,在「稽核」中設為 event_action
detection_id、id metadata.product_log_id 在「偵測」、「鎖定」和「稽核」中,分別設為 detection_id、id
version, system.version.vectra_version metadata.product_version 稽核中的版本,健康狀態中的 system.version.vectra_version
_sensor.serial_number observer.asset_id 直接複製值
_sensor.name observer.hostname 直接複製值
_sensor.ip_address observer.ip 直接複製值
_sensor.location observer.location.name 直接複製值
entity_id principal.asset_id 直接複製的值,如果不是以半形冒號分隔,則會加上「Vectra:」前置字串
entity_uid principal.hostname 如果符合模式,則直接複製值
src_ip、source_ip principal.ip 如果 IP 有效,則從「偵測」中的 src_ip 設定;如果 IP 有效,則從「稽核」中的 source_ip 設定
var_entity_uid、var_username、var_api_client_id、var_user_type、var_entity_name principal.user.attribute.labels 根據記錄類型與各種 var_ 合併
user_role principal.user.attribute.roles 直接複製值
entity_uid、name、entity_name、username principal.user.email_addresses 如果電子郵件模式為「偵測」,請從 entity_uid 進行設定;如果電子郵件模式為「評分」,請從名稱進行設定;如果電子郵件模式為「封鎖」,請從 entity_name 進行設定;如果電子郵件模式為「稽核」,請從使用者名稱進行設定
locked_by principal.user.user_display_name 直接複製值
entity_id、user_id principal.user.userid 在「偵測/評分/封鎖」中從 entity_id 設定,在「稽核」中從 user_id 設定
result_status security_result.action 成功時設為 ALLOW,失敗時設為 FAIL
result_status security_result.action_details 直接複製值
斜接 security_result.attack_details.techniques 從斜接陣列合併
category security_result.category 根據 COMMAND & CONTROL、BOTNET ACTIVITY 等的規則運算式比對結果,對應至 category_temp。
category security_result.category_details 直接複製值
確定度 security_result.confidence_score 已轉換為浮點數
urgency_reason、msg security_result.description 在評分中設為 urgency_reason,在稽核中設為 msg
var_type、var_detection_type、var_triaged、var_id、var_active_detection_types、var_breadth_contrib、var_attack_rating、var_importance、var_last_detection_id、var_last_detection_type、var_last_detection_url、var_urgency_score、var_velocity_contrib、unlock_event_timestamp_label、var_detection_updated_at、var_sensor_connectivity_sensors_error、var_connectivity_sensors_affected_metadata_hours、var_connectivity_sensors_status、var_connectivity_updated_at、var_cpu_idle_percent、var_cpu_nice_percent、var_cpu_system_percent、var_cpu_updated_at、var_cpu_user_percent、var_disk_disk_utilization_free_bytes、var_disk_raid_disks_missing_output_label、var_disk_raid_disks_missing_error、var_disk_raid_disks_missing_status、var_disk_degraded_raid_volume_output、var_disk_degraded_raid_volume_error、var_disk_degraded_raid_volume_status、var_disk_disk_raid_error、var_disk_disk_raid_status、var_disk_disk_raid_output、var_disk_disk_utilization_total_bytes、var_disk_disk_utilization_usage_percent、var_disk_disk_utilization_used_bytes、var_disk_updated_at、var_hostid_artifact_counts_arsenic、var_hostid_artifact_counts_carbon_black、var_hostid_artifact_counts_cb_cloud、var_hostid_artifact_counts_clear_state、var_hostid_artifact_counts_cookie、var_hostid_artifact_counts_crowdstrike、var_hostid_artifact_counts_cybereason、var_hostid_artifact_counts_dhcp、var_hostid_artifact_counts_dns、var_hostid_artifact_counts_end_time、var_hostid_artifact_counts_fireeye、var_hostid_artifact_counts_generic_edr、var_hostid_artifact_counts_idle_end、var_hostid_artifact_counts_idle_start、var_hostid_artifact_counts_invalid、var_hostid_artifact_counts_kerberos_user、var_hostid_artifact_counts_kerberos、var_hostid_artifact_counts_mdns、var_hostid_artifact_counts_netbios、var_hostid_artifact_counts_proxy_ip、var_hostid_artifact_counts_rdns、var_hostid_artifact_counts_sentinelone、var_hostid_artifact_counts_split、var_hostid_artifact_counts_src_port、var_hostid_artifact_counts_static_ip、var_hostid_artifact_counts_TestEDR、var_hostid_artifact_counts_total、var_hostid_artifact_counts_uagent、var_hostid_artifact_counts_vmachine_info、var_hostid_artifact_counts_windows_defender、var_hostid_artifact_counts_zpa_user、var_hostid_ip_always_percent、var_hostid_ip_never_percent、var_hostid_ip_sometimes_percent、var_hostid_updated_at、var_dimm_stat_dimm、var_dimm_stat_status、var_memory_free_bytes、var_memory_total_bytes、var_memory_updated_at、var_memory_usage_percent、var_memory_used_bytes、network_interfaces_brain_label、var_network_interfaces_sensors_w4ftj0a8_eth0_link、var_network_traffic_brain_aggregated_peak_traffic_mbps、var_network_traffic_brain_interface_peak_traffic_label、var_network_traffic_sensors_edr_sensor_aggregated_peak_traffic_mbps、var_sensor_interface_peak_traffic_eth0_peak_traffic_mbps、var_network_updated_at、var_network_vlans_count、var_network_vlans_vlan_ids、var_power_error、var_power_status、var_power_updated_at、var_power_power_supply、var_sensors_headend_uri、var_sensors_id、var_sensors_last_seen、var_sensors_luid、var_sensors_mode、var_sensors_original_version、var_sensors_product_name、var_sensors_public_key、var_sensors_ssh_tunnel_port、var_sensors_status、var_sensors_update_count、var_sensors_version、var_trafficdrop_sensors_error、var_trafficdrop_sensors_ip_address、var_trafficdrop_sensors_luid、var_trafficdrop_sensors_name、var_trafficdrop_sensors_output_end、var_trafficdrop_sensors_output_interface_cutoff、var_trafficdrop_sensors_output_start、var_trafficdrop_sensors_output_interface_baseline、var_trafficdrop_sensors_output_interface_name、var_trafficdrop_sensors_output_interface_traffic、var_trafficdrop_sensors_serial_number、var_trafficdrop_sensors_status、var_trafficdrop_updated_at security_result.detection_fields 根據記錄類型與各種 var_ 和標籤合併
is_prioritized security_result.priority 如果為 true,請設為 HIGH;如果為 false,請設為 LOW
is_prioritized security_result.priority_details 直接複製值
威脅 security_result.risk_score 已轉換為浮點數
嚴重性 security_result.severity 如果符合 (?i)high,則設為 HIGH;如果符合 (?i)medium,則設為 MEDIUM;如果符合 (?i)low,則設為 LOW
嚴重性 security_result.severity_details 直接複製值
detection_href、url security_result.url_back_to_product 在「偵測」中設為 detection_href,在「評分」中設為網址
entity_id target.asset_id 直接複製的值,如果不是以半形冒號分隔,則會加上「Vectra:」前置字串
entity_name、name target.hostname 如果不是 IP,則從 Lockdown 中的 entity_name 設定,或從 Scoring 中的 name 設定
target_ip_temp target.ip 直接複製值
lock_event_timestamp target.user.account_lockout_time 從字串轉換為時間戳記
var_entity_name target.user.attribute.labels 已與 var_entity_name 合併
name、entity_name target.user.email_addresses 如果電子郵件模式為「評分」,則設定「寄件者名稱」;如果電子郵件模式為「封鎖」,則設定「實體名稱」
entity_id target.user.userid 直接複製值
metadata.product_name 設為「XDR」
metadata.vendor_name 設為「Vectra」
entity_uid principal.macprincipal.ip 從變更記錄對應

變更記錄

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

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